From 4993f1b1d7536e770ac88b5ee8bc330c47b9046b Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 00:34:10 +0100 Subject: [PATCH 1/2] Make the switch between composers safe: a composer tells only its own run's agent, a pass whose run ended refuses nothing, a bus that pauses is tried again, a mark keeps what runs and clears at the next start; pin mesh-sdk 7359d18 (issue 1126) Live 2026-10-28: the own run's last pass, failing on its run's end, marked every module of the controller-mode agent that replaced it (ace 60, shanks 63), and the marks dropped the build and digest because the have read failed in the bus's reload pause. Now: the refusal hook is bound to the run's agent and ctx; a pass whose context ended returns nothing; JetStream not enabled or no responders are tried again after 1, 2 and 4 s before any mark; NotComposed reads the have again while the bus does not answer and marks nothing it cannot read; baseline takes a have a composer marked as what runs, so a mark it did not make clears; each run's shadow lives with the run. --- cmd/mesh-host/nodeagent.go | 26 ++++++-- cmd/mesh-host/nodeagent_integration_test.go | 40 ++++++++++++- go.mod | 2 +- go.sum | 2 + internal/agent/agent.go | 58 +++++++++++++++++- internal/agent/agent_test.go | 55 +++++++++++++++++ internal/link/have.go | 25 ++++++-- internal/selfcompose/selfcompose.go | 46 +++++++++++++- internal/selfcompose/selfcompose_test.go | 66 +++++++++++++++++++++ 9 files changed, 302 insertions(+), 18 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 1f0b801..8368dd3 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -72,6 +72,20 @@ func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool Sealed: got.Sealed}, true } +// ReadErr is Read telling a have absent from one the bus could not read (agent.HavesReadErr). +func (h sessionHaves) ReadErr(ctx context.Context, module string) (agent.Have, bool, error) { + if module == agent.NodeModule { + got, ok := h.Read(ctx, module) + return got, ok, nil + } + got, ok, err := h.session.ReadHaveErr(ctx, module) + if !ok || err != nil { + return agent.Have{}, false, err + } + return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail, + Sealed: got.Sealed}, true, nil +} + func (h sessionHaves) Write(ctx context.Context, x agent.Have) { if x.Module == agent.NodeModule { // The node's own key holds its facts and the digest of the composition it runs. @@ -509,9 +523,12 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess run, stopRun := context.WithCancel(linked) // What the node runs and how it opens a sealed value, for the composer to keep the boxes it runs; and the agent // told when the node cannot compose, so each wanted module's have says it. + // The agent of this run alone: its composer tells it, never the agent of a run begun since (issue 1126: a pass in + // flight at the switch marked the next agent's modules failed). + own := &agentNow{} hooks := composerHooks{ running: func() []delivery.Composed { - if now := current.get(); now != nil { + if now := own.get(); now != nil { return now.Everything() } return nil @@ -530,8 +547,8 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess return got.Sealed }, refused: func(why string) { - if now := current.get(); now != nil { - now.NotComposed(linked, why) + if now := own.get(); now != nil && run.Err() == nil { + now.NotComposed(run, why) } }, } @@ -543,7 +560,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess func(ctx context.Context, answers map[string]delivery.CandidateAnswer) { session.NoteCandidates(ctx, answers) }) - go shadow.Keep(linked, link.Backoff) + go shadow.Keep(run, link.Backoff) // this run's alone: a switch begins a run, and its shadow, again composer, why := selfComposer(linked, session, node, say, hooks) if linked.Err() != nil { // The link ended before the node knew whose compositions to apply: it chose neither, and waits for the next. @@ -564,6 +581,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess go endWhenPlaced(run, session, node, stopRun, say) } ag = agent.New(env) + own.set(ag) current.set(ag) err = ag.Run(run) stopRun() diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 7d152c1..025cd54 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -403,13 +403,13 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { 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" }) h8rest(t, b, resolved, candidates, hkey, place, fp) - h8back(t, b, placement, candidates, hkey) + h8back(t, b, placement, resolved, candidates, hkey) } // h8back is the way back (issue 1126, the seed's revert): a candidate deleting the node's placement is compared with // the controller's compositions at its wants (here none: refused, naming web), 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) { +func h8back(t *testing.T, b *itBed, placement, resolved, candidates nats.KeyValue, hkey string) { t.Helper() node := func() delivery.HaveNode { e, err := b.haves.Get(hkey) @@ -440,20 +440,56 @@ func h8back(t *testing.T, b *itBed, placement, candidates nats.KeyValue, hkey st }) ran := b.have("web").Composed pkey, _ := delivery.PlacementKey(itNode) + // A pass in flight at the switch (issue 1126, 2026-10-28: the last pass of the own run, failing on its run's end, + // marked the next agent's modules failed): a config value changed starts one, and the key goes at once. + if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut"}`)); err != nil { + t.Fatal(err) + } if err := placement.Delete(pkey); err != nil { t.Fatal(err) } + stopWatch := make(chan struct{}) + marked := make(chan Have1126, 1) + go func() { + for { + select { + case <-stopWatch: + return + case <-time.After(20 * time.Millisecond): + } + if h := b.have("web"); h.Phase == delivery.PhaseFailed { + select { + case marked <- Have1126{h.Phase, h.Detail}: + default: + } + return + } + } + }() + defer close(stopWatch) 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 }) + time.Sleep(3 * time.Second) // what a pass in flight would mark lands within it + select { + case h := <-marked: + t.Errorf("a pass in flight at the switch marked web failed: %+v", h) + default: + } 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")) } } +// Have1126 is what H8's way back watches of web's have. +type Have1126 struct { + Phase delivery.Phase + Detail string +} + // h8state is what h8 set up that the rest of H8 writes to. type h8state struct { placement, resolved, candidates nats.KeyValue diff --git a/go.mod b/go.mod index 412919b..927562c 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.20261027223231-e32d77753024 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027232145-7359d187db8d 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 17fbe1e..76c4f22 100644 --- a/go.sum +++ b/go.sum @@ -24,6 +24,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466 h1:xzbkI9dO git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024 h1:MnmyspVoaCldYmm4Ut1O39RDR9WpiL4TfFgDEnY+QRg= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027232145-7359d187db8d h1:DhR7DRKaF0OtA5l/ikIk7gZGeIM1lRz65PUE1DGEkvU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027232145-7359d187db8d/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/agent/agent.go b/internal/agent/agent.go index dd101be..0eb72e2 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -314,6 +314,44 @@ type notComposedMark struct { prior Have } +// NotComposedDetail opens the detail of a have NotComposed marks: the module keeps running what its build and digest +// name, and only the node's composition of it failed. +const NotComposedDetail = "the node cannot compose it, and keeps running what it ran: " + +// HavesReadErr is what a Haves may answer besides Read: a have absent (false, nil) told from one that could not be read +// (false, the error). +type HavesReadErr interface { + ReadErr(ctx context.Context, module string) (Have, bool, error) +} + +// readHave reads a module's have: found, absent, or unread. A read that fails is tried again after a short wait while +// the bus does not answer (a reload pauses it for a few seconds, issue 1126), and is unread when it still fails. A +// Haves that cannot tell an absent key from a failed read says absent. +func (a *Agent) readHave(ctx context.Context, module string) (h Have, found, unread bool) { + re, tells := a.env.Haves.(HavesReadErr) + if !tells { + h, found = a.env.Haves.Read(ctx, module) + return h, found, false + } + for try := 0; ; try++ { + h, found, err := re.ReadErr(ctx, module) + if err == nil { + return h, found, false + } + if try == 2 || ctx.Err() != nil { + return Have{}, false, true + } + select { + case <-ctx.Done(): + return Have{}, false, true + case <-time.After(time.Duration(1< Date: Wed, 28 Oct 2026 00:41:25 +0100 Subject: [PATCH 2/2] A composer's mark never covers another cause, and a run that ended marks and clears nothing (issue 1126, review of the switch, round 1) NotComposed marked from a read have whatever its phase: an apply that failed, naming the want's build and digest, became a composer's mark, which baseline then took for what runs, and never applied again. It now marks only a have that runs, says nothing, or is a composer's mark already. And its loop, whose reads can take seconds, checks the run's context at each module, so a run that ended writes no have. --- internal/agent/agent.go | 26 +++++++++++++++++++++++++- internal/agent/agent_test.go | 31 +++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 0eb72e2..7c70a47 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -318,6 +318,11 @@ type notComposedMark struct { // name, and only the node's composition of it failed. const NotComposedDetail = "the node cannot compose it, and keeps running what it ran: " +// composerMark says h is a mark NotComposed wrote: failed, its detail opening with NotComposedDetail. +func composerMark(h Have) bool { + return h.Phase == string(delivery.PhaseFailed) && strings.HasPrefix(h.Detail, NotComposedDetail) +} + // HavesReadErr is what a Haves may answer besides Read: a have absent (false, nil) told from one that could not be read // (false, the error). type HavesReadErr interface { @@ -375,6 +380,9 @@ func (a *Agent) NotComposed(ctx context.Context, why string) { a.mu.Unlock() sort.Strings(modules) for _, module := range modules { + if ctx.Err() != nil { + return // the run ended: what it would mark or clear is the next run's to say (issue 1126) + } a.mu.Lock() mark, marked := a.notComposed[module] r, ran := a.running[module] @@ -405,6 +413,9 @@ func (a *Agent) NotComposed(ctx context.Context, why string) { } else if h, found, unread := a.readHave(ctx, module); found { mark.prior = h // what its last have said runs: this agent has run nothing of it since its start } else if unread { + if ctx.Err() != nil { + return + } // What its have names cannot be read (the bus paused, issue 1126): a mark written now would drop the build // and digest it names. Nothing is written; the next refusal marks it once the have reads. a.mu.Lock() @@ -413,6 +424,19 @@ func (a *Agent) NotComposed(ctx context.Context, why string) { continue } } + // A have that already says another cause (an apply that failed, a backup held, an apply under way) keeps it: a + // composer's mark over it would hide that cause, and baseline would take the mark for what runs (review of the + // switch, round 1). + if p := mark.prior.Phase; !marked && p != "" && p != string(delivery.PhaseRunning) && !composerMark(mark.prior) { + a.mu.Lock() + a.sayOnce("unmarked-cause:"+module, fmt.Sprintf("%s is not marked: its have says %s (%s), and the node cannot compose it: %s", + module, p, mark.prior.Detail, why)) + a.mu.Unlock() + continue + } + if ctx.Err() != nil { + return + } mark.why = why a.mu.Lock() a.notComposed[module] = mark @@ -948,7 +972,7 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com // A have a composer marked failed (NotComposed) names what still runs: taken as running, so a mark this // agent did not make clears once what it names is what the want names (issue 1126, the stale marks after // the seed's revert). - marked := h.Phase == string(delivery.PhaseFailed) && strings.HasPrefix(h.Detail, NotComposedDetail) + marked := composerMark(h) if marked { h.Detail = "" } diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index d9d1e67..c3d4a56 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -342,6 +342,21 @@ func TestStartingOnWhatTheKeptDeclarationHoldsAppliesNothing(t *testing.T) { // What the node-engine itself last applied is what runs after a restart, not the kept declaration. func TestStartingOnItsOwnHaveAppliesNothing(t *testing.T) { t.Run("says the have again", testStartingOnItsOwnHaveSaysTheHaveAgainSoItsHealthIsThisEngines) + // An apply that failed is not what runs, though its have names the want's build and digest: the agent starting on + // it applies again (review of the switch, round 1). + t.Run("applies again after an apply failure", func(t *testing.T) { + b := newBed(t) + fp := fpOf("a") + ck, craw, dig := composition(t, "avahi", fp, false, file("avahi", "x", "1")) + wk, wraw := wantFor(t, "avahi", fp, dig, "") + b.haves.stored = map[string]Have{"avahi": {Module: "avahi", FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed), + Detail: "apply: unit failed to start"}} + b.comps.before(ck, craw) + b.wants.before(wk, wraw) + stop := b.run() + defer stop() + eventually(t, "a failed apply was not applied again", func() bool { return len(b.appliedList()) == 1 }) + }) // Issue 1126, the stale marks after the seed's revert: a have a composer marked failed names what still runs; the // agent starting on it applies nothing and says it running, so a mark it did not make clears. t.Run("clears a composer's mark", func(t *testing.T) { @@ -1084,6 +1099,11 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( b.haves.mu.Unlock() want("unread", fpOf("d"), "") want("fresh", fpOf("e"), "") // never ran, its have absent: marked, naming nothing it runs + // A have that already says another cause keeps it (review of the switch, round 1): an apply that failed. + brokenFP, brokenDig := fpOf("f"), "sha256:"+strings.Repeat("8", 64) + b.haves.Write(context.Background(), Have{Module: "broken", FP: brokenFP, Composed: brokenDig, Phase: string(delivery.PhaseFailed), + Detail: "apply: unit failed to start"}) + want("broken", brokenFP, "") eventually(t, "unread's want was not taken", func() bool { b.ag.mu.Lock() defer b.ag.mu.Unlock() @@ -1096,6 +1116,17 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( if h := b.haves.last("fresh"); h.Phase != string(delivery.PhaseFailed) || h.FP != "" { t.Errorf("a module that never ran, its have absent, is not marked: %+v", h) } + if h := b.haves.last("broken"); h.Detail != "apply: unit failed to start" || h.Composed != brokenDig { + t.Errorf("an apply failure was overwritten by a composer's mark: %+v", h) + } + // A run that ended marks nothing: what it would say is the next run's (review of the switch, round 1). + ended, end := context.WithCancel(context.Background()) + end() + before := len(b.haves.log) + b.ag.NotComposed(ended, "a refusal from a run that ended") + if len(b.haves.log) != before { + t.Errorf("a run that ended wrote %d haves", len(b.haves.log)-before) + } b.ag.NotComposed(context.Background(), "") for _, row := range []struct{ module, composed string }{{"web", dig}, {"idle", idleDig}} { if h := b.haves.last(row.module); h.Phase != string(delivery.PhaseRunning) || h.Composed != row.composed || len(h.Sealed) != 1 {