From 4e8fdfdf04d7f091ce8612b1f823df111bc5342a Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 27 Oct 2026 23:25:59 +0100 Subject: [PATCH] The way back: a node whose placement key is deleted returns to the controller's compositions without a restart, says so in its journal and its have, and answers such a candidate as returning; pin mesh-sdk 472d5f3 (issue 1126, the seed's revert) The seed's revert (mesh-catalog #526) was refused on 'would stop running node', while a self-composing node would have refused its placement's absence, kept what runs and marked every have failed until a restart. endWhenUnplaced ends the agent's run when the key goes, so it begins again on the controller's compositions, as at a start without a key; the composer refuses nothing while unplaced; the node's have says whose compositions it applies (Composes); the shadow answers a deleted placement Returns, verdict same. --- cmd/mesh-host/nodeagent.go | 49 ++++++++++++++++++--- cmd/mesh-host/nodeagent_integration_test.go | 48 +++++++++++++++++++- go.mod | 2 +- go.sum | 2 + internal/link/have.go | 24 ++++++++++ internal/selfcompose/selfcompose.go | 9 +++- internal/selfcompose/selfcompose_test.go | 19 ++++++-- internal/selfcompose/shadow.go | 17 +++---- 8 files changed, 145 insertions(+), 25 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index e57b079..d71c8c5 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -544,9 +544,14 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess session.NoteCandidates(ctx, answers) }) go shadow.Keep(linked, link.Backoff) - if composer := selfComposer(linked, session, node, say, hooks); composer != nil { + if composer, why := selfComposer(linked, session, node, say, hooks); composer != nil { env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy + session.NoteComposes(linked, delivery.ComposesOwn, why) + // And once the catalogue no longer places it, the agent begins again on the controller's compositions: the + // placement key deleted ends this run, the way back (issue 1126, the seed's revert). + go endWhenUnplaced(run, session, node, stopRun, say) } else { + 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) @@ -843,23 +848,24 @@ type composerHooks struct { held func(module string) []string } -func selfComposer(ctx context.Context, session *link.Session, node string, say func(string), hooks composerHooks) *selfcompose.Composer { +func selfComposer(ctx context.Context, session *link.Session, node string, say func(string), hooks composerHooks) (*selfcompose.Composer, string) { key, err := delivery.PlacementKey(node) if err != nil { say("this node applies the controller's compositions: " + err.Error()) - return nil + return nil, err.Error() } got, err := session.KVHistory(ctx, delivery.PlacementBucket, key) if err != nil || len(got) == 0 || got[len(got)-1].Deleted { - why := "the catalogue does not place it yet" + why := "the catalogue does not place it" if err != nil { why = "its placement cannot be read: " + err.Error() } say("this node applies the controller's compositions: " + why) - return nil + return nil, why } - say("this node composes its own modules: the catalogue places it (" + key + ")") - return selfcompose.New(composerEnv(session, node, say, hooks)) + why := "the catalogue places it (" + key + ")" + say("this node composes its own modules: " + why) + return selfcompose.New(composerEnv(session, node, say, hooks)), why } // composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer @@ -889,6 +895,35 @@ func composerEnv(session *link.Session, node string, say func(string), hooks com Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held} } +// endWhenUnplaced calls stop once this node's placement key is deleted from the indexer's placement (after the watch +// caught up, or gone already), so the agent begins again and applies the controller's compositions, as a node the +// catalogue does not place does from its start: the way back (issue 1126, the seed's revert). A watch that cannot be +// made is said; the node then returns at its next start. +func endWhenUnplaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce) { + key, err := delivery.PlacementKey(node) + if err != nil { + return + } + 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") + return + } + placed := false + for u := range updates { + switch { + 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") + stop() + return + default: + placed = true + } + } +} + // endWhenPlaced calls stop once this node's placement key appears in the indexer'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. diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 2d5a7d9..d3a352f 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -402,8 +402,54 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { placement, resolved, candidates := st.placement, st.resolved, st.candidates hkey, place, fp := st.hkey, st.place, st.fp within(t, 5*time.Second, "the node did not compose and apply its own module once placed", func() bool { return b.file("web") == "greeting=bonjour\n" }) - _ = placement h8rest(t, b, resolved, candidates, hkey, place, fp) + h8back(t, b, placement, candidates, hkey) +} + +// h8back is the way back (issue 1126, the seed's revert): a candidate deleting the node's placement is answered as +// returning to the controller's compositions, and the placement key deleted returns the agent to them without a +// restart, said in the journal and the node's have, keeping what runs. +func h8back(t *testing.T, b *itBed, placement, candidates nats.KeyValue, hkey string) { + t.Helper() + node := func() delivery.HaveNode { + e, err := b.haves.Get(hkey) + if err != nil { + return delivery.HaveNode{} + } + n, _ := delivery.DecodeHaveNode(e.Value()) + return n + } + if n := node(); n.Composes != delivery.ComposesOwn { + t.Errorf("a node composing its own says it composes %q", n.Composes) + } + craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "revert/x", + Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 2, Head: "h2"}}) + if err != nil { + t.Fatal(err) + } + ckey, _ := delivery.PlacementCandidateKey("revert/x", itNode) + if _, err := candidates.Put(ckey, craw); err != nil { + t.Fatal(err) + } + within(t, 10*time.Second, "the node did not answer a candidate deleting its placement as returning", func() bool { + a := node().Candidates["revert/x"] + return a.Head == "h2" && a.Returns == delivery.ReturnsToController && a.Verdict == delivery.CandidateSame && + len(a.Removed) == 0 + }) + ran := b.have("web").Composed + pkey, _ := delivery.PlacementKey(itNode) + if err := placement.Delete(pkey); err != nil { + t.Fatal(err) + } + within(t, 15*time.Second, "the node did not return to the controller's compositions when its placement went", func() bool { + 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 + }) + if h := b.have("web"); h.Composed != ran || h.Phase == delivery.PhaseFailed || b.file("web") != "greeting=salut\n" { + t.Errorf("returning to the controller's compositions moved what runs: %+v, %q", h, b.file("web")) + } } // h8state is what h8 set up that the rest of H8 writes to. diff --git a/go.mod b/go.mod index b397ee6..678e26f 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027221705-2fd5ffdd6263 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 diff --git a/go.sum b/go.sum index 09f31d8..fd262e6 100644 --- a/go.sum +++ b/go.sum @@ -20,6 +20,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027201213-7a8e9f970648 h1:XoMsgg9k git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027201213-7a8e9f970648/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027221705-2fd5ffdd6263 h1:JZV8C+WR6NPiopj9ShzhbpjN0gqVQV60vOFxsxFwH00= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027221705-2fd5ffdd6263/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466 h1:xzbkI9dOkN/Iv78AS/fOwhjkgBEQOYtpbHzNYwb4eu0= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/internal/link/have.go b/internal/link/have.go index d923ee8..bdfe893 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -52,6 +52,9 @@ type haveState struct { nodeSealed []string // nodeCandidates is the node's answer to each placement candidate group (novox/hq ADR 0559 §4), written into its key. nodeCandidates map[string]delivery.CandidateAnswer + // nodeComposes and nodeComposesWhy are whose compositions the node applies and why (HaveNode.Composes), written into + // its key. + nodeComposes, nodeComposesWhy string // written is the build of each module this node-engine wrote a have key for and that the declaration it applied // still holds; removed is each such module the latest declaration no longer holds, with the build its key last // said, until the key is written once with the phase removed (issue 1369, novox/hq ADR 0408 §4: never deleted). @@ -423,6 +426,7 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { facts.Composed = s.nodeComposed facts.Sealed = s.nodeSealed facts.Candidates = s.nodeCandidates + facts.Composes, facts.ComposesWhy = s.nodeComposes, s.nodeComposesWhy if raw, err = delivery.EncodeHaveNode(facts); err == nil { want := facts if !q.putHave(write, bucket, key, raw, func(current []byte) bool { @@ -752,6 +756,26 @@ func (q *Session) NoteCandidates(ctx context.Context, answers map[string]deliver q.writeHaves(ctx, bus) } +// NoteComposes says, in the node's own key, whose compositions the node applies (delivery.ComposesOwn or +// delivery.ComposesController) and why: written at each start of its agent, so a node returning to the controller's +// compositions says so where every have is read (issue 1126). +func (q *Session) NoteComposes(ctx context.Context, composes, why string) { + q.init() + s := &q.haves + s.mu.Lock() + changed := s.nodeComposes != composes || s.nodeComposesWhy != why + s.nodeComposes, s.nodeComposesWhy = composes, why + known := s.node != nil + s.mu.Unlock() + if !known || !changed { + return // said with the node's first facts + } + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + q.writeHaves(ctx, bus) +} + // ReadNodeHave is the node's own key as the bucket holds it, and whether it holds one: what the node's own composition // it last ran is and carries sealed, read after a restart (ADR 0547 §3). func (q *Session) ReadNodeHave(ctx context.Context) (delivery.HaveNode, bool) { diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index f92405f..37afe58 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -429,9 +429,16 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { _, nodeWanted := c.wants[agent.NodeModule] // What it runs inside its own composition and no want names at a build of the new form stays there, as the // controller's composed writer folds it (issue 1126); compose folds none a want names. - inOwn := c.inOwnOf(c.placements[c.env.Node]) + own, placed := c.placements[c.env.Node] + inOwn := c.inOwnOf(own) r := reader{c: c} c.mu.Unlock() + 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") + return nil + } // Composed from the build its want names (frombus.Scope.KeptAt, issue 1126). for _, module := range behind { kept[module] = fps[module] diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index afe26b4..b225bcc 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -477,6 +477,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { candidate("adds", 4, "lemurs", "keeper") candidate("drops", 5) candidate("ghost", 6, "lemurs", "ghost") + // A head deleting n1's placement file (the seed's revert, mesh-catalog #526): n1 returns to the controller's + // compositions, answered as such, not as every module removed. + if raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "deletes", + Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-deletes"}}); err != nil { + t.Fatal(err) + } else { + k, _ := delivery.PlacementCandidateKey("deletes", "n1") + sb.put(delivery.PlacementCandidatesBucket, k, raw, 7) + } sb.caught() var answers map[string]delivery.CandidateAnswer waitFor := func(why string, ok func(map[string]delivery.CandidateAnswer) bool) { @@ -490,7 +499,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } } } - waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 4 }) + waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 5 }) // 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 { @@ -520,6 +529,10 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { "ghost": func(a delivery.CandidateAnswer) bool { return a.Verdict == delivery.CandidateRefused && a.Refused != "" }, + "deletes": func(a delivery.CandidateAnswer) bool { + return a.Verdict == delivery.CandidateSame && a.Returns == delivery.ReturnsToController && a.Revision == 7 && + len(a.Removed) == 0 + }, } { if !want(answers[group]) { t.Errorf("%s: %+v", group, answers[group]) @@ -534,7 +547,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) == 3 }) + waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 4 }) // 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. @@ -546,7 +559,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) == 2 + return !ok && len(a) == 3 }) // Issue 1126, the seed's review: a module n1 runs inside its own composition (nats, at a build of the old form) is diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index 6ebc4f1..1a776c8 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -292,18 +292,11 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid running = c.env.Running() } if p, ok := given[node]; ok && p == nil { - // The head removes this node's placement file: everything it runs is removed. - a := delivery.CandidateAnswer{Head: head, Revision: rev} - for _, r := range running { - if r.Module == "" { - a.Removed = append(a.Removed, delivery.HaveNodeWord) - } else { - a.Removed = append(a.Removed, r.Module) - } - } - sort.Strings(a.Removed) - a.Verdict = delivery.VerdictOf(a) - return a, true + // The head deletes this node's placement file: the node then returns to the controller's compositions, as one the + // catalogue does not place (endWhenUnplaced), and keeps what runs; nothing is removed (issue 1126, the seed's + // revert, whose check read "would stop running node"). + return delivery.CandidateAnswer{Head: head, Revision: rev, Returns: delivery.ReturnsToController, + Verdict: delivery.CandidateSame}, true } placement := given[node]