From ab22efd8b182d63ce82bef7d2c507f0b8dac12a6 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:39:12 +0100 Subject: [PATCH] Node agent: follow the placement key at start only and say why when it cannot; test the reached candidate and the switched rule; word the log by the seat's state (issue 1126) --- cmd/mesh-host/nodeagent.go | 78 ++++++++------------- cmd/mesh-host/nodeagent_integration_test.go | 2 +- internal/link/history_integration_test.go | 2 +- internal/selfcompose/selfcompose.go | 24 ++++++- internal/selfcompose/selfcompose_test.go | 38 +++++++++- 5 files changed, 89 insertions(+), 55 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index ba1e5ae..0a050bb 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -594,7 +594,9 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess session.NoteComposes(linked, delivery.ComposesController, why) // And once the catalogue places it, the agent begins again composing its own: the placement key appearing ends // this run, so a node does not wait for a restart to follow the seed of its placement (ADR 0549 §4). - go endWhenPlaced(run, session, node, stopRun, say) + go endWhenPlaced(run, session, node, stopRun, say, func(reason string) { + session.NoteComposes(linked, delivery.ComposesController, why+"; "+reason) + }) } ag = agent.New(env) own.set(ag) @@ -896,38 +898,29 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f return nil, err.Error() } env := composerEnv(session, node, say, hooks) - placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, say, composedItsOwn(ctx, env.Get, node)) + unreadable := "" + placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, func(m string) { + if strings.Contains(m, "cannot be read") { + unreadable = m + } + say(m) + }, selfcompose.ComposedItsOwn(ctx, env.Get, node)) if err != nil { return nil, "its placement could not be read before the link ended: " + err.Error() } if !placed { - why := "the catalogue does not place it" + why := "the seat's placement state holds no key for it" + if unreadable != "" { + why = "its placement cannot be read, and it never composed its own: " + unreadable + } say("this node applies the controller's compositions: " + why) return nil, why } - why := "the catalogue places it (" + key + ")" + why := "the seat's placement state holds its key (" + key + ")" say("this node composes its own modules: " + why) return selfcompose.New(env), why } -// composedItsOwn says whether the node's own have, kept on the bus, says it composed from its placement key at its last -// start. A have that cannot be read says yes: the safe side, a node waits and does not fall back. -func composedItsOwn(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string) bool { - key, err := delivery.HaveNodeKey(node) - if err != nil { - return true - } - raw, err := get(ctx, delivery.HaveBucket, key) - if err != nil { - return true - } - if len(raw) == 0 { - return false - } - n, err := delivery.DecodeHaveNode(raw) - return err != nil || n.Composes == delivery.ComposesOwn -} - // composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer // to placement candidates (selfcompose.Shadow) alike. func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env { @@ -966,7 +959,7 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st } updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) if err != nil { - say("whether the catalogue still places this node is not followed (" + err.Error() + "): it returns to the controller's compositions at its next start") + say("whether this node's key in the seat's placement state still exists is not followed (" + err.Error() + "): it returns to the controller's compositions at its next start") return } placed := false @@ -975,7 +968,7 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st case u.Caught && placed: continue case u.Caught || u.Deleted || len(u.Value) == 0: - say("the catalogue no longer places this node (" + key + "): the node agent begins again, applying the controller's compositions") + say("this node's key in the seat's placement state is gone (" + key + "): the node agent begins again, applying the controller's compositions") stop() return default: @@ -983,46 +976,37 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st } } if ctx.Err() == nil { - say("the watch of " + key + " ended while the link stays: whether the catalogue still places this node is followed again at the node agent's next start") + say("the watch of " + key + " ended while the link stays: whether this node's key in the seat's placement state still exists is followed again at the node agent's next start") } } // endWhenPlaced calls stop once this node's placement key appears in the seat's placement (a value after the watch // caught up, or one already there), so the agent begins again and composes its own. A watch that cannot be made is said; // the node then switches at its next start. -func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce) { +func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce, note func(string)) { key, err := delivery.PlacementKey(node) if err != nil { return } - // The seat's placement state appears when its holder first writes it (ADR 0567): until then the watch is refused, as a - // read of it finds no bucket (Placed), and the node stays on the controller's compositions. The watch is tried again, - // so the node switches when its key is written, not only at its next start. - var updates <-chan agent.Update - said := "" - for try := 1; ; try++ { - updates, err = bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) - if err == nil { - break - } - if why := err.Error(); why != said { - say("whether the catalogue places this node is not followed yet (" + why + "): the node stays on the controller's compositions and the watch is tried again") - said = why - } - select { - case <-ctx.Done(): - return - case <-time.After(link.Backoff(try)): - } + // Tried at the start only: the delivery brings the grants part and the holder first, and the holder makes its states + // at its own start, so the state exists and the read is granted by the time the node-engine starts. A watch still + // refused is said once in the journal and in the node's own have, where the watchman reads it, and the node stays + // on the controller's compositions until its next start. + updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) + if err != nil { + why := "this node's key in the seat's placement state cannot be followed (" + err.Error() + "): it stays on the controller's compositions until its next start" + say(why) + note(why) + return } for u := range updates { if !u.Caught && !u.Deleted && len(u.Value) > 0 { - say("the catalogue places this node now (" + key + "): the node agent begins again, composing its own modules") + say("this node's key in the seat's placement state is written now (" + key + "): the node agent begins again, composing its own modules") stop() return } } if ctx.Err() == nil { - say("the watch of " + key + " ended while the link stays: whether the catalogue places this node is followed again at the node agent's next start") + say("the watch of " + key + " ended while the link stays: whether this node's key in the seat's placement state is written is followed again at the node agent's next start") } } diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index ab3dde5..b214352 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -529,7 +529,7 @@ func h8back(t *testing.T, b *itBed, placement, resolved, candidates nats.KeyValu b.mu.Lock() said := strings.Join(b.said, "\n") b.mu.Unlock() - return strings.Contains(said, "the catalogue no longer places this node") && node().Composes == delivery.ComposesController + return strings.Contains(said, "key in the seat's placement state is gone") && node().Composes == delivery.ComposesController }) time.Sleep(3 * time.Second) // what a pass in flight would mark lands within it select { diff --git a/internal/link/history_integration_test.go b/internal/link/history_integration_test.go index 1c1f553..4f042b3 100644 --- a/internal/link/history_integration_test.go +++ b/internal/link/history_integration_test.go @@ -70,7 +70,7 @@ func TestNatsAKeysRevisionsAreReadBack(t *testing.T) { } // A bucket that is not there, and one the account may not read, are errors at once, never a wait: the node agent - // follows its placement key by trying the watch again (endWhenPlaced) and stays on the controller's compositions. + // follows its placement key by one watch at its start (endWhenPlaced) and, refused, stays on the controller's compositions and says why. if _, err := bus.KVWatch(ctx, "never-it", ">"); err == nil { t.Error("a watch of a bucket that is not there gave no error") } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 6781bee..f3167c7 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -483,7 +483,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { if !placed { // The catalogue no longer places it: the agent returns to the controller's compositions (endWhenUnplaced), and // this is no refusal of what it runs (issue 1126, the seed's revert). - c.sayOnce("the catalogue no longer places this node: it composes nothing of its own, and keeps running what it runs") + c.sayOnce("this node's key in the seat's placement state is gone: it composes nothing of its own, and keeps running what it runs") c.owed.Store(false) // nothing of its own to compose, so no pass is owed and no timer armed again return nil } @@ -1018,11 +1018,11 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin return false, ctx.Err() } if !switched { - say(fmt.Sprintf("whether the catalogue places this node cannot be read (%s): it has never composed its own, so it applies the controller's compositions and switches when its key is read: %v", key, err)) + say(fmt.Sprintf("this node's key in the seat's placement state cannot be read (%s): it has never composed its own, so it applies the controller's compositions and switches when its key is read: %v", key, err)) return false, nil } if why := err.Error(); why != said { - say(fmt.Sprintf("whether the catalogue places this node cannot be read (%s): the node agent waits, choosing neither its own compositions nor the controller's: %v", key, err)) + say(fmt.Sprintf("this node's key in the seat's placement state cannot be read (%s): the node agent waits, choosing neither its own compositions nor the controller's: %v", key, err)) said = why } select { @@ -1033,6 +1033,24 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin } } +// ComposedItsOwn says whether the node's own have, kept on the bus, says it composed from its placement key at its last +// start. A have that cannot be read says yes: the safe side, a node waits and does not fall back. +func ComposedItsOwn(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string) bool { + key, err := delivery.HaveNodeKey(node) + if err != nil { + return true + } + raw, err := get(ctx, delivery.HaveBucket, key) + if err != nil { + return true + } + if len(raw) == 0 { + return false + } + n, err := delivery.DecodeHaveNode(raw) + return err != nil || n.Composes == delivery.ComposesOwn +} + // composeAgain are the waits before a pass that met a bus not answering is tried again. var composeAgain = []time.Duration{time.Second, 2 * time.Second, 4 * time.Second} diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 98c1ec2..f588f0c 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -781,6 +781,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { k, _ := delivery.PlacementCandidateKey("deletes", "n1") sb.put(delivery.PlacementCandidatesBucket, k, raw, 7) } + // A proposal that changes another node's key and reaches n1's composition (ADR 0567 §3): n1's candidate holds no + // placement of its own change (Reached, nil), and n1 answers against its current placement, never as one whose key + // the proposal takes away. + if raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "reach", Reached: true}); err != nil { + t.Fatal(err) + } else { + k, _ := delivery.PlacementCandidateKey("reach", "n1") + sb.put(delivery.PlacementCandidatesBucket, k, raw, 8) + } sb.caught() var answers map[string]delivery.CandidateAnswer waitFor := func(why string, ok func(map[string]delivery.CandidateAnswer) bool) { @@ -794,7 +803,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } } } - waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 5 }) + waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 6 }) // Every watch and get the shadow makes is one the node's grant allows (issue 1126, 2026-10-27 09:21). sb.mu.Lock() for _, w := range sb.watches { @@ -824,6 +833,10 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { "ghost": func(a delivery.CandidateAnswer) bool { return a.Verdict == delivery.CandidateRefused && a.Refused != "" }, + "reach": func(a delivery.CandidateAnswer) bool { + return a.Verdict == delivery.CandidateSame && a.Returns == "" && a.Revision == 8 && len(a.Removed) == 0 && + fmt.Sprint(a.Same) == "[lemurs node]" + }, "deletes": func(a delivery.CandidateAnswer) bool { return a.Verdict == delivery.CandidateSame && a.Returns == delivery.ReturnsToController && a.Revision == 7 && len(a.Removed) == 0 && fmt.Sprint(a.Same) == "[lemurs node]" @@ -933,7 +946,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } k, _ := delivery.PlacementCandidateKey("same", "n1") sb.del(delivery.PlacementCandidatesBucket, k) - waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 4 }) + waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 5 }) // Review of L2b, round 2: a watch ends while the link stays; a group's key is deleted before the shadow watches // again. The new run reads afresh, so that group is no longer answered. @@ -945,7 +958,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { sb.missing(delivery.PlacementCandidatesBucket, false) waitFor("a group deleted while the shadow was down is still answered", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["drops"] - return !ok && len(a) == 3 + return !ok && len(a) == 4 }) // Whose compositions a node applies at a start (review of the shadow, round 1): only a read that succeeds chooses; a @@ -987,6 +1000,25 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } } + // Whether the node composed its own before is read from its own have: own says yes, the controller's or no have says + // no, and a have that cannot be read says yes (the safe side: a node waits and does not fall back). + for name, row := range map[string]struct { + raw []byte + err error + want bool + }{ + "own": {raw: []byte(`{"v":1,"composes":"own"}`), want: true}, + "the controller's": {raw: []byte(`{"v":1,"composes":"controller"}`), want: false}, + "no have yet": {want: false}, + "a have that cannot be read": {err: fmt.Errorf("nats: timeout"), want: true}, + "a have that is no have": {raw: []byte(`{`), want: true}, + } { + get := func(context.Context, string, string) ([]byte, error) { return row.raw, row.err } + if got := ComposedItsOwn(ctx, get, "n1"); got != row.want { + t.Errorf("composed its own, %s: %v, want %v", name, got, row.want) + } + } + // Issue 1126, the shadow shows its work: one input not read after the wait is said, naming it, in the journal and in // a refused answer to each group known; once read, the group is answered for real. setFor(t, ¬ReadyAfter, 30*time.Millisecond)