Ask the bus again for a subscription it refused (novox/hq issue 222)

A push sends the machine that needs a grant and the bus's machine in the same breath, and the
runtime can subscribe the moment before the bus reloads its user list. Refused once, the
subscription stayed dead until some later membership re-served it, and a newly assigned module ran
unreachable. A refused subject the runtime answers is now asked for again for about five minutes.

node-tools/internal/bus/refused_test.go

@@ -0,0 +1,86 @@
+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")
+	}
+}