Keep the declarations retired once their stream is gone, so nothing expects, asserts or makes NODES again (issue 1945)
mesh/delivery merged: source.main moved to novox/mesh-controller@2814262ac0fa
mesh/check pass 0a536d7b6d79 on 2bcd85d2df4c | unit 966 run, 172 kept, 2 skipped | integration 1622 run, 236 kept, 7 skipped | over bound: integration…

A node-engine newer than the registered one, or a node asleep on an older
have, made the done test false after the start deleted NODES; the bus's
health check, probes D6 and D7 and healer H3 then expected the stream and
every node's consumer again, raised them urgent and made them anew. Release
6 sends no declaration, so a stream made again helps no node: a gone stream
is retired. The advisory watchdog also reads a consumer whose stream is gone
as gone instead of failing.
This commit is contained in:
2026-10-25 15:01:09 +01:00
parent 2bcd85d2df
commit 0a536d7b6d
5 changed files with 78 additions and 7 deletions
@@ -140,4 +140,35 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
if !declarationsRetiredNow(ctx, m.open.inventory) { if !declarationsRetiredNow(ctx, m.open.inventory) {
t.Errorf("a raise after the retirement would assert the stream again") 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)
}
} }
+16 -5
View File
@@ -135,9 +135,18 @@ func declarationsDone(nodes []string, runs map[string]string, current string) bo
return true return true
} }
// declarationsRetiredOn reads whether the declarations are done on the mesh now: every node's have of the node-engine // declarationsRetiredOn reads whether the declarations are done on the mesh now: their stream gone from the bus, or
// against the fingerprint it is registered at. // 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) { 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) fps, err := currentFingerprints(ctx, inv)
if err != nil { if err != nil {
return false, err 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 // 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) { func retireDeclarations(ctx context.Context, open *stores) (bool, error) {
if _, err := theLease.epoch(ctx); err != nil { if _, err := theLease.epoch(ctx); err != nil {
return false, err return false, err
@@ -324,8 +334,9 @@ func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Co
return c, nil 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 // declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects and for what
// store it cannot read is not done, so the stream is asserted as before. // 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 { func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool {
bus, err := aBus() bus, err := aBus()
if err != nil { if err != nil {
@@ -3,6 +3,7 @@ package main
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"strings" "strings"
"testing" "testing"
@@ -110,3 +111,28 @@ func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) {
t.Errorf("done with no fingerprint registered, or no node") 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)
}
}
+3 -1
View File
@@ -512,7 +512,9 @@ func (w *watchdogs) lostConsumers(ctx context.Context, heard []link.Advisory) (m
for _, a := range deleted { for _, a := range deleted {
_, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer) _, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer)
switch { 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] out[a.Stream+"."+a.Consumer] = want[a.Stream+"."+a.Consumer]
case err != nil: case err != nil:
return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err) return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err)
+2 -1
View File
@@ -165,7 +165,8 @@ func HandPathWrites(nodes []string) ([]string, error) {
// DeclarationsStream is the stream of each node's last declaration, `mesh.node.<node>.declare` (issue 107). Release 6 // DeclarationsStream is the stream of each node's last declaration, `mesh.node.<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 // 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 // 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" const DeclarationsStream = "NODES"
// RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the // RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the