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…
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:
@@ -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)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user