package link import ( "context" "crypto/ed25519" "errors" "fmt" "os" "strings" "sync" "testing" "time" "github.com/nats-io/nats.go" ) // quietLink is a link with no broker behind it. type quietLink struct { lost chan error } func newQuietLink() *quietLink { return &quietLink{lost: make(chan error, 1)} } func (l *quietLink) Lost() <-chan error { return l.lost } func (l *quietLink) Close() {} func aMember(t *testing.T) (Membership, ed25519.PrivateKey) { t.Helper() public, private, err := ed25519.GenerateKey(nil) if err != nil { t.Fatal(err) } return Membership{Node: "n1", Signer: public}, private } // Facts made with no link open are not kept, and nothing is said: the next link says them as they are then. func TestFactsWithNoLinkOpenAreNotSaid(t *testing.T) { m, _ := aMember(t) l := newQuietLink() s := &Session{Membership: m, Timeout: time.Second} if s.SayFacts(context.Background(), Report{}) { t.Fatal("facts were said with no link") } s.Attach(context.Background(), l) s.Detach(l) if s.Linked() || s.SayFacts(context.Background(), Report{}) { t.Fatal("facts were said on a link that was let go") } } // A link opening says the node's facts, off the link's own loop. func TestALinkOpeningSaysTheFacts(t *testing.T) { m, _ := aMember(t) l := newQuietLink() opened := make(chan struct{}, 1) s := &Session{Membership: m, Timeout: time.Second, OnLink: func(context.Context) { opened <- struct{}{} }} s.Attach(context.Background(), l) select { case <-opened: case <-time.After(2 * time.Second): t.Fatal("a link opening did not ask for the facts") } } // aBus is a connection to a real server, because some claims here are about one: set MESH_TEST_NATS to a server with // JetStream, or the test is skipped. func aBus(t *testing.T) (*nats.Conn, nats.JetStreamContext) { t.Helper() url := os.Getenv("MESH_TEST_NATS") if url == "" { t.Skip("MESH_TEST_NATS unset") } conn, err := nats.Connect(url) if err != nil { t.Fatal(err) } t.Cleanup(conn.Close) js, err := conn.JetStream() if err != nil { t.Fatal(err) } return conn, js } // eventuallyHeld waits for a condition the bus reaches on its own time. func eventuallyHeld(t *testing.T, what string, ok func() bool) { t.Helper() deadline := time.Now().Add(5 * time.Second) for !ok() { if time.Now().After(deadline) { t.Fatal(what) } time.Sleep(5 * time.Millisecond) } } // refusingLink is a quiet link the bus refused subscriptions to. type refusingLink struct { *quietLink refused chan string } func (l *refusingLink) Refused() <-chan string { return l.refused } // The sweep of 2026-10-30: every engine was refused its node seat's subscriptions at 12:10, by a user list composed // without them, and nats asks a refused subscription again only on a new connection; the seat answered nobody all // afternoon. A refused subscription is read from the server's words, said, and the link opened again after a pause // that grows while it is refused again; a refused publish is not one. func TestASubscriptionTheBusRefusedIsAskedAgainOnANewLink(t *testing.T) { // The server's words, as the client hands them on (journal of nox-mesh-host on novox, 2026-10-30 12:10:44). for _, row := range []struct { err error want string }{ {fmt.Errorf("%w: %s", nats.ErrPermissionViolation, `Permissions Violation for Subscription to "mesh.seat.node-engine.tool.root-free.novox"`), "mesh.seat.node-engine.tool.root-free.novox"}, {errors.Join(fmt.Errorf("%w: %s", nats.ErrPermissionViolation, `Permissions Violation for Subscription to "$SRV.PING.node-engine"`), errors.New("stopping service: x")), "$SRV.PING.node-engine"}, {fmt.Errorf("%w: %s", nats.ErrPermissionViolation, `Permissions Violation for Publish to "$JS.API.STREAM.INFO.KV_mesh-host_health"`), ""}, {errors.New("nats: timeout"), ""}, } { got, ok := RefusedSubscription(row.err) if got != row.want || ok != (row.want != "") { t.Errorf("%v reads as %q, %v", row.err, got, ok) } l := &natsLink{refused: make(chan string, 1)} l.asyncError(nil, nil, row.err) if heard := len(l.refused); heard != map[bool]int{true: 1}[row.want != ""] { t.Errorf("%v handed %d refusals on", row.err, heard) } } m, _ := aMember(t) q := &Session{Membership: m, Timeout: time.Second, RefusedPause: 50 * time.Millisecond} var said []string var mu sync.Mutex say := func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() } for _, want := range []time.Duration{50 * time.Millisecond, 100 * time.Millisecond} { l := &refusingLink{quietLink: newQuietLink(), refused: make(chan string, 2)} l.refused <- "mesh.seat.node-engine.tool.root-free.n1" l.refused <- "$SRV.PING" began := time.Now() done := make(chan error, 1) go func() { done <- serve(context.Background(), l, m, q, say, time.Second) }() select { case err := <-done: if err == nil || !strings.Contains(err.Error(), "refused 2 of this node's subscriptions, "+ "mesh.seat.node-engine.tool.root-free.n1 first") { t.Fatalf("the link ended with %v", err) } if took := time.Since(began); took < want { t.Errorf("the link was dropped after %s, before its pause of %s", took, want) } case <-time.After(5 * time.Second): t.Fatal("a link the bus refused a subscription was kept, and the subscription never asked again") } } mu.Lock() defer mu.Unlock() if n := strings.Count(strings.Join(said, "\n"), "the bus refused this node a subscription to "+ "mesh.seat.node-engine.tool.root-free.n1"); n != 2 { t.Errorf("the refusal was said %d times in two links: %q", n, said) } if q.refusalsHeld(); q.refusedPause() != 50*time.Millisecond { t.Error("a link that held its subscriptions did not start from the first pause again") } }