117 lines
4.1 KiB
Go
117 lines
4.1 KiB
Go
package bus
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// novox/hq issue 222: a subscription the bus refused is asked for again. A push sends the machine
|
|
// that needs a grant and the bus's machine together; the runtime may subscribe the moment before the
|
|
// bus reloads, and a refusal must not leave the module unreachable until some later membership.
|
|
func TestARefusedSubscriptionIsAskedForAgain(t *testing.T) {
|
|
url := os.Getenv("MESH_TEST_NATS")
|
|
if url == "" {
|
|
t.Skip("MESH_TEST_NATS unset")
|
|
}
|
|
c, err := Connect(Credential{URL: url, Module: "alpha", Node: "anchor"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer c.Close()
|
|
var mu sync.Mutex
|
|
var said []string
|
|
c.Logf = func(f string, a ...any) { mu.Lock(); said = append(said, fmt.Sprintf(f, a...)); mu.Unlock() }
|
|
c.RetryAfter = []time.Duration{50 * time.Millisecond, 50 * time.Millisecond}
|
|
|
|
subject := "mesh.mod.alpha.tool.ping.anchor"
|
|
stop, err := c.answerOn(subject, "", func(json.RawMessage) (any, error) { return "pong", nil })
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
first := c.answering[answeringKey(subject, "")].sub
|
|
|
|
// What the bus says when the grant is not there yet.
|
|
c.refused(fmt.Errorf("%w: Permissions Violation for Subscription to %q", nats.ErrPermissionViolation, subject))
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for {
|
|
c.mu.Lock()
|
|
again := c.answering[answeringKey(subject, "")].sub
|
|
c.mu.Unlock()
|
|
if again != first {
|
|
break
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("the refused subscription was not asked for again")
|
|
}
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
asker, err := nats.Connect(url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer asker.Close()
|
|
reply, err := asker.Request(subject, []byte("{}"), 2*time.Second)
|
|
if err != nil {
|
|
t.Fatalf("the subject asked for again does not answer: %v", err)
|
|
}
|
|
if !strings.Contains(string(reply.Data), "pong") {
|
|
t.Fatalf("the subject asked for again answered %s", reply.Data)
|
|
}
|
|
|
|
// Past its attempts it is given up, and said.
|
|
for i := 0; i < 3; i++ {
|
|
c.refused(fmt.Errorf("%w: Permissions Violation for Subscription to %q", nats.ErrPermissionViolation, subject))
|
|
time.Sleep(80 * time.Millisecond)
|
|
}
|
|
mu.Lock()
|
|
gaveUp := strings.Contains(strings.Join(said, "\n"), "still refuses "+subject)
|
|
mu.Unlock()
|
|
if !gaveUp {
|
|
t.Errorf("never gave up on a subject that stays refused:\n%s", strings.Join(said, "\n"))
|
|
}
|
|
|
|
// A subject the runtime stopped answering is not asked for again.
|
|
stop()
|
|
c.refused(fmt.Errorf("%w: Permissions Violation for Subscription to %q", nats.ErrPermissionViolation, subject))
|
|
if _, still := c.answering[answeringKey(subject, "")]; still {
|
|
t.Error("a stopped subject is still tracked")
|
|
}
|
|
}
|
|
|
|
// Issue 1966: a refused publish of an answer (to a caller's inbox) is said, once per subject per minute, never dropped
|
|
// in silence while the caller waits out its time.
|
|
func TestARefusedAnswerIsSaidOncePerMinute(t *testing.T) {
|
|
c := &Conn{}
|
|
var said []string
|
|
c.Logf = func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }
|
|
refuse := func(subject string) bool {
|
|
return c.answerRefused(fmt.Errorf("%w: Permissions Violation for Publish to %q", nats.ErrPermissionViolation, subject))
|
|
}
|
|
// A reply subject ends in a token new to every request: a burst under one prefix is one line.
|
|
if !refuse("_INBOX.user.abc.t1") || !refuse("_INBOX.user.abc.t2") || !refuse("_INBOX.user.def.t1") {
|
|
t.Fatal("a refused answer was not taken for one")
|
|
}
|
|
if len(said) != 2 {
|
|
t.Fatalf("want one line per caller, got %v", said)
|
|
}
|
|
c.inboxSaid["_INBOX.user.abc"] = time.Now().Add(-2 * time.Minute)
|
|
c.inboxSaid["_INBOX.user.old"] = time.Now().Add(-2 * time.Minute)
|
|
refuse("_INBOX.user.abc.t3")
|
|
if len(said) != 3 || !strings.Contains(said[2], "_INBOX.user.abc") {
|
|
t.Fatalf("a minute later it is said again, got %v", said)
|
|
}
|
|
if _, kept := c.inboxSaid["_INBOX.user.old"]; kept || len(c.inboxSaid) != 2 {
|
|
t.Fatalf("entries older than a minute are dropped on write, kept %v", c.inboxSaid)
|
|
}
|
|
if refuse("mesh.mod.x.tool.y") {
|
|
t.Fatal("a refused publish that is no answer was taken for one")
|
|
}
|
|
}
|