diff --git a/cmd/mesh-controller/composed_integration_test.go b/cmd/mesh-controller/composed_integration_test.go index b9d88388..5c87f775 100644 --- a/cmd/mesh-controller/composed_integration_test.go +++ b/cmd/mesh-controller/composed_integration_test.go @@ -140,4 +140,35 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) { if !declarationsRetiredNow(ctx, m.open.inventory) { t.Errorf("a raise after the retirement would assert the stream again") } + + // Issue 1945: a node-engine other than the registered one landing on laptop (a newer build on its way) leaves the + // declarations retired, so what the health check, the probes and healer H3 expect of the bus names neither the + // stream nor a node's consumer on it, and asserting the bus's objects again does not make the stream anew. + have("laptop", fpOf('7')) + if !declarationsRetiredNow(ctx, m.open.inventory) { + t.Fatalf("with laptop on another node-engine after the stream went, the declarations are no longer retired") + } + streams, consumers, err := expectedBusObjects(ctx, m.open.inventory) + if err != nil { + t.Fatal(err) + } + for _, s := range streams { + if s.Name == broker.DeclarationsStream { + t.Errorf("the bus's health check and probe D7 still expect the stream %s", s.Name) + } + } + for _, c := range consumers { + if c.Stream == broker.DeclarationsStream { + t.Errorf("the bus's health check and probe D6 still expect %s on %s", c.Name, c.Stream) + } + } + bus, err := broker.Dial(m.conn.ConnectedUrl()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(bus.Close) + _ = assertOnSend(ctx, m.open.inventory, bus, " ") + if _, err := m.js.Stream(ctx, broker.DeclarationsStream); !errors.Is(err, jetstream.ErrStreamNotFound) { + t.Errorf("healer H3's assertion of the bus's objects made the stream anew: %v", err) + } } diff --git a/cmd/mesh-controller/composed_start.go b/cmd/mesh-controller/composed_start.go index 6b5b0f61..4db346e9 100644 --- a/cmd/mesh-controller/composed_start.go +++ b/cmd/mesh-controller/composed_start.go @@ -135,9 +135,18 @@ func declarationsDone(nodes []string, runs map[string]string, current string) bo return true } -// declarationsRetiredOn reads whether the declarations are done on the mesh now: every node's have of the node-engine -// against the fingerprint it is registered at. +// declarationsRetiredOn reads whether the declarations are done on the mesh now: their stream gone from the bus, or +// every node's have of the node-engine against the fingerprint it is registered at. +// +// **A stream that is gone stays gone** (issue 1945). Release 6 sends no declaration (ADR 0539 clauses 5 and 11), so a +// stream made again would hold no node's last declaration and help no node; and the done test alone is not lasting: +// a node-engine newer than the one registered landing, or a node asleep on an older have, makes it false again after +// the start deleted the stream, and then the raise, the bus's health check (H-bus), the probes (D6, D7) and healer H3 +// expected the stream and every node's consumer again, raised them missing and made them anew (2026-10-25 14:52). func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jetstream.JetStream) (bool, error) { + if _, err := js.Stream(ctx, broker.DeclarationsStream); errors.Is(err, jetstream.ErrStreamNotFound) { + return true, nil + } fps, err := currentFingerprints(ctx, inv) if err != nil { return false, err @@ -162,7 +171,8 @@ func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jet } // retireDeclarations deletes the declarations' stream, with every node's consumer on it, once every node runs the new -// node-engine, and says it once; a stream already gone is said by nobody. It answers whether the declarations are done. +// node-engine, and says it once; a stream already gone is said by nobody, and is done. It answers whether the +// declarations are done. func retireDeclarations(ctx context.Context, open *stores) (bool, error) { if _, err := theLease.epoch(ctx); err != nil { return false, err @@ -324,8 +334,9 @@ func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Co return c, nil } -// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects; a bus or a -// store it cannot read is not done, so the stream is asserted as before. +// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects and for what +// the health check, the probes and the healers expect of the bus; a bus or a store it cannot read is not done, so a +// stream still there is asserted as before. func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool { bus, err := aBus() if err != nil { diff --git a/cmd/mesh-controller/composed_start_test.go b/cmd/mesh-controller/composed_start_test.go index 35831c35..83de4ab6 100644 --- a/cmd/mesh-controller/composed_start_test.go +++ b/cmd/mesh-controller/composed_start_test.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "fmt" "strings" "testing" @@ -110,3 +111,28 @@ func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) { t.Errorf("done with no fingerprint registered, or no node") } } + +// streamsGone is a bus that holds none of the streams asked for; every other part of it is unread. +type streamsGone struct{ jetstream.JetStream } + +func (streamsGone) Stream(context.Context, string) (jetstream.Stream, error) { + return nil, jetstream.ErrStreamNotFound +} + +// Issue 1945 (the alarm of a stream the start deleted): once the declarations' stream is gone from the bus, the +// declarations are retired whatever the nodes run now, so no raise, health check, probe or healer expects the stream +// or a node's consumer on it again. A node-engine newer than the registered one, or a node asleep on an older have, +// made the done test false after the stream went, and the bus's objects named it missing again (2026-10-25 14:52). +func TestTheDeclarationsStayRetiredOnceTheirStreamIsGone(t *testing.T) { + retired, err := func() (retired bool, err error) { + defer func() { + if p := recover(); p != nil { + err = fmt.Errorf("it read what the nodes run although the stream is gone: %v", p) + } + }() + return declarationsRetiredOn(t.Context(), nil, streamsGone{}) + }() + if err != nil || !retired { + t.Fatalf("with no declarations' stream on the bus: retired %v (%v), want retired", retired, err) + } +} diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index cdce1875..af6bc530 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -512,7 +512,9 @@ func (w *watchdogs) lostConsumers(ctx context.Context, heard []link.Advisory) (m for _, a := range deleted { _, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer) switch { - case errors.Is(err, nats.ErrConsumerNotFound): + // A consumer whose stream is gone is gone with it: the retired declarations' stream took every node's + // consumer, and its deletion said each as deleted (issue 1945). + case errors.Is(err, nats.ErrConsumerNotFound) || errors.Is(err, nats.ErrStreamNotFound): out[a.Stream+"."+a.Consumer] = want[a.Stream+"."+a.Consumer] case err != nil: return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err) diff --git a/internal/broker/composed.go b/internal/broker/composed.go index da5c495a..8856186e 100644 --- a/internal/broker/composed.go +++ b/internal/broker/composed.go @@ -165,7 +165,8 @@ func HandPathWrites(nodes []string) ([]string, error) { // DeclarationsStream is the stream of each node's last declaration, `mesh.node..declare` (issue 107). Release 6 // sends none (ADR 0539 clauses 5 and 11): the stream is kept, with each node's consumer, only so a node away during the // switch finds its last declaration and with it the new node-engine; the controller deletes it once every node's have -// names the new node-engine, and from then on neither asserts it nor any node's consumer on it. +// names the new node-engine, and from then on neither asserts it nor any node's consumer on it; nor makes it again once +// it is gone from the bus for any reason, since a stream made anew would hold no node's last declaration (issue 1945). const DeclarationsStream = "NODES" // RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the