From 9c15610f70a535b3818d9e171bd2c146906f59a0 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 31 Oct 2026 00:56:47 +0100 Subject: [PATCH 1/3] Say when the node composes again after a seat that did not answer, say what a long pass waits on, and serve the verb compose (issue 774) On 2026-10-31 ace's route-proxy said only that the vault could not be asked, and ace said nothing for five minutes while a pass waited after the vault restarted; the retry that already existed was invisible. A refusal for a seat or a read that did not answer now says when the node composes again ("the node composes again at HH:MM:SS (try N)"), a retry pass says it starts, a pass that runs on says every minute what it waits on, and the node-engine's verb compose makes a pass now and says how it went, so a person recovers without a restart. --- cmd/mesh-host/engineseat.go | 34 +++++ cmd/mesh-host/main.go | 1 + cmd/mesh-host/nodeagent.go | 6 + internal/link/handacts.go | 10 ++ internal/link/handover_test.go | 6 +- internal/selfcompose/selfcompose.go | 155 +++++++++++++++++++++-- internal/selfcompose/selfcompose_test.go | 60 ++++++++- 7 files changed, 255 insertions(+), 17 deletions(-) diff --git a/cmd/mesh-host/engineseat.go b/cmd/mesh-host/engineseat.go index 76727ef..a7fcbbe 100644 --- a/cmd/mesh-host/engineseat.go +++ b/cmd/mesh-host/engineseat.go @@ -18,6 +18,7 @@ import ( "github.com/novox/mesh-host/internal/accounts" "github.com/novox/mesh-host/internal/apply" "github.com/novox/mesh-host/internal/link" + "github.com/novox/mesh-host/internal/selfcompose" "github.com/novox/mesh-host/internal/store" ) @@ -292,3 +293,36 @@ func compositionFrom(current *agentNow) func(module, digest string, diff bool) ( return ag.Composition(ctx, module, digest, diff) } } + +// composeReadFor bounds the verb compose's wait for its pass: a pass that waits on a seat longer says what it waits on. +const composeReadFor = 2 * time.Minute + +// composeFrom answers the verb compose (issue 774): a pass of the composer that runs now, and with module that module's +// apply tried again by the node agent; refused in words while the node composes nothing of its own. +func composeFrom(current *agentNow, composer *atomic.Pointer[selfcompose.Composer]) func(module string) (any, error) { + return func(module string) (any, error) { + c := composer.Load() + if c == nil { + return nil, errors.New("this node composes none of its own modules now (it applies the controller's " + + "compositions, or its node agent is starting), so there is no pass to make") + } + ag := current.get() + if module != "" { + if ag == nil || !ag.Placed("")[module] { + return nil, fmt.Errorf("%s has no want on this machine, so there is no apply of it to try again", module) + } + } + ctx, cancel := context.WithTimeout(context.Background(), composeReadFor) + defer cancel() + pass, err := c.ComposeNow(ctx) + if err != nil { + return nil, err + } + out := map[string]any{"pass": pass} + if module != "" { + ag.Retry(module) + out["retried"] = module + } + return out, nil + } +} diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 278fa82..40325bf 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1142,6 +1142,7 @@ func runLink(ctx context.Context, opts options) error { go reconcileActs(ctx, queue.Acts) current := &agentNow{} queue.Acts.Composition = compositionFrom(current) + queue.Acts.Compose = composeFrom(current, &composerNow) setRunsHere(runsFromAgent(current)) factsOfNow := func(ctx context.Context, ag *agent.Agent) { sayFacts(ctx, opts, mine.Membership.Signer, queue, say, ag) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 9e55e69..145dc17 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -668,6 +668,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus } if composer != nil { machinesNow.Store(composer) // the machines a failing network part points at, by the placements (issue 774) + composerNow.Store(composer) // what the verb compose asks a pass of, this run's alone composes.set(delivery.ComposesOwn) env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy env.HoldsData = composer.HoldsData // what runs from the kept declaration says Data as its manifest does @@ -692,6 +693,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus stopRun() cancel() current.set(nil) + composerNow.Store(nil) if ctx.Err() != nil { return } @@ -814,6 +816,10 @@ func busyModules() []string { // (writeHealthReport); nil on a node that does not compose its own, whose parts then name no target. var machinesNow atomic.Pointer[selfcompose.Composer] +// composerNow is the composer of the run that composes the node's own now, nil while the node applies the controller's +// compositions or no run is going: what the verb compose asks a pass of (issue 774). +var composerNow atomic.Pointer[selfcompose.Composer] + // keptRetired says the kept declaration is read no more: every module the seat's placement puts on this node has a have of // its own (placedAllHave), so what runs is the agent's alone (ADR 0550 §2, design 47's L3 s9). var keptRetired atomic.Bool diff --git a/internal/link/handacts.go b/internal/link/handacts.go index f77f0f7..6682982 100644 --- a/internal/link/handacts.go +++ b/internal/link/handacts.go @@ -40,6 +40,8 @@ const ( ActSetuidSearch = "setuid-search" // ActComposition reads a module's composition beside what runs of it; read-only, no warrant (issue 774). ActComposition = "composition" + // ActCompose composes the node now, the recovery of a failed composition (issue 774). + ActCompose = "compose" ) // The states of an act. @@ -132,6 +134,9 @@ type HandActs struct { // Composition reads one module's composition beside what runs of it, read-only (the verb composition, issue 774): // the latest of its want's build or the one whose digest is given, and with diff the resources that differ. Composition func(module, digest string, diff bool) (any, error) + // Compose makes a pass of the node's composition now and says how it went; with module, its apply is also tried + // again (the verb compose, issue 774). + Compose func(module string) (any, error) // Say says one of the seat's events, asked, done or refused; nil says nothing. Say func(event string, body any) Now func() time.Time @@ -190,6 +195,11 @@ func (h *HandActs) Call(c EngineCall) (any, error) { diff = b } return h.Composition(arg("module"), arg("digest"), diff) + case ActCompose: + if h.Compose == nil { + return nil, errors.New("this engine composes nothing on a call yet: its node agent has not started") + } + return h.Compose(arg("module")) case "acts": if id := arg("id"); id != "" { return h.Act(id) diff --git a/internal/link/handover_test.go b/internal/link/handover_test.go index a4b8688..63cdd81 100644 --- a/internal/link/handover_test.go +++ b/internal/link/handover_test.go @@ -239,7 +239,8 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) { JudgeDirectories: func(by string) (string, error) { return "judged for " + by, nil }, Composition: func(module, digest string, diff bool) (any, error) { return map[string]any{"module": module, "digest": digest, "diff": diff}, nil - }} + }, + Compose: func(module string) (any, error) { return map[string]any{"composed": true, "retried": module}, nil }} for _, row := range []struct{ verb, subject, want string }{ {"judge-directories", "mesh.seat.node-engine.call.judge-directories.n1.shanks~node-tools", `"said":"judged for shanks.node-tools"`}, {"judge-directories", EngineToolSubject("judge-directories", "n1"), `"said":"judged for unproven"`}, @@ -248,6 +249,9 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) { // The read verb composition (issue 774) takes its module, digest and diff, as a word or a boolean. {"composition", EngineToolSubject("composition", "n1") + " " + `{"module":"postgres","digest":"sha256:a","diff":true}`, `"diff":true,"digest":"sha256:a","module":"postgres"`}, {"composition", EngineToolSubject("composition", "n1") + " " + `{"module":"postgres","diff":"true"}`, `"diff":true,"digest":"","module":"postgres"`}, + // The recovery verb compose (issue 774) makes a pass now, with a module's apply tried again when named. + {"compose", EngineToolSubject("compose", "n1"), `"composed":true,"retried":""`}, + {"compose", EngineToolSubject("compose", "n1") + " " + `{"module":"route-proxy"}`, `"retried":"route-proxy"`}, {"unserved", "mesh.seat.node-engine.call.unserved.n1.x", `"error":"the node seat node-engine serves no verb \"unserved\""`}, } { subject, body, ok := strings.Cut(row.subject, " ") diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 280b777..1040c67 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -127,6 +127,58 @@ type Composer struct { missed atomic.Uint64 // owed is set when a pass refused for a bus that did not answer: the composer composes again after a bound. owed atomic.Bool + // tries counts the passes owed in a row, and wait is the bound before the next (LaterWait): set by the pass that owes + // one, so the refusal it marks says when the node composes again (issue 774, 2026-10-31). Under mu. + tries int + wait time.Duration + // asked are the compose verb's asks (ComposeNow), each answered once the pass it asks is made. + asked chan chan Pass + // last is how the last pass went, under mu. + last Pass +} + +// Pass is how one pass went, as the node-engine's verb compose answers it. +type Pass struct { + // At is when the pass ended. + At time.Time `json:"at"` + // Refused is why the node composed none of its modules; empty when it composed. + Refused string `json:"refused,omitempty"` + // Held is each module refused alone, with why. + Held map[string]string `json:"held,omitempty"` + // Handed is how many compositions the pass handed the agent that differ from what it handed before. + Handed int `json:"handed"` + // Again is when the node composes again on its own, after a seat or a read that did not answer; zero when it does + // not need to. + Again time.Time `json:"again,omitempty"` +} + +// passSaysEvery is how often a pass that runs on says, in the journal, how long it has run and what it waits on. +var passSaysEvery = time.Minute + +// owe marks a pass owed after a seat or a read that did not answer: the next wait of the growing bound (LaterWait), and +// the words that say when the node composes again, for the refusal it marks. +func (c *Composer) owe() string { + c.mu.Lock() + defer c.mu.Unlock() + c.wait = LaterWait(c.tries) + c.tries++ + c.owed.Store(true) + return fmt.Sprintf("the node composes again at %s (try %d)", time.Now().Add(c.wait).Format("15:04:05"), c.tries) +} + +// paid clears what owe marked: the pass composed, or owes nothing. +func (c *Composer) paid() { + c.mu.Lock() + defer c.mu.Unlock() + c.tries, c.wait = 0, 0 + c.owed.Store(false) +} + +// waitNow is the bound before the pass owed, and its try. +func (c *Composer) waitNow() (time.Duration, int) { + c.mu.Lock() + defer c.mu.Unlock() + return c.wait, c.tries } // note counts err when it is the bus not answering, and returns it. @@ -150,7 +202,26 @@ func New(env Env) *Composer { } return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{}, records: map[string]delivery.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{}, - manifests: map[string][]byte{}, archives: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}} + manifests: map[string][]byte{}, archives: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}, + asked: make(chan chan Pass, 8)} +} + +// ComposeNow asks the composer for a pass now (the node-engine's verb compose, issue 774): the growing bound after a seat +// that did not answer starts again, and the answer is how the pass went. Refused in words when no pass is made before +// ctx ends: the composer not running, or its inputs not all read yet. +func (c *Composer) ComposeNow(ctx context.Context) (Pass, error) { + reply := make(chan Pass, 1) + select { + case c.asked <- reply: + case <-ctx.Done(): + return Pass{}, errors.New("the node's composer took no ask: it composes nothing now") + } + select { + case p := <-reply: + return p, nil + case <-ctx.Done(): + return Pass{}, fmt.Errorf("the node made no pass in time; it waits on %s", c.waitingOn()) + } } // ownSealed is one own secret's sealed value at a generation and the key it is sealed to. @@ -270,7 +341,7 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er // A pass refused for a bus that did not answer is made again after a bound (laterAgain), not kept until another // input moves (issue 1126, 2026-10-28). var again <-chan time.Time - tries := 0 + var replies []chan Pass for { // Drain whatever came, then compose once for all of it. select { @@ -279,6 +350,16 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er case <-again: again = nil dirty = true + _, try := c.waitNow() + c.env.Say(fmt.Sprintf("the node composes again now (try %d), after a seat or a read that did not answer", try+1)) + case reply := <-c.asked: + // Asked by hand (the verb compose): composed now, the bound starting again. + replies = append(replies, reply) + c.mu.Lock() + c.tries = 0 + c.mu.Unlock() + again = nil + dirty = true case t := <-merged: if !t.ok { c.env.Say(fmt.Sprintf("the watch of %s ended; the node composes no more until its agent starts again", ins[t.in].bucket)) @@ -318,13 +399,19 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er } } if c.owed.Load() { - wait := LaterWait(tries) - tries++ + wait, _ := c.waitNow() c.env.Say(fmt.Sprintf("the node composes again in %s: a read on the bus did not answer", wait)) again = time.After(wait) } else { - tries, again = 0, nil + again = nil } + c.mu.Lock() + last := c.last + c.mu.Unlock() + for _, r := range replies { + r <- last + } + replies = nil if !c.caughtSent() { select { case out <- agent.Update{Caught: true}: @@ -518,6 +605,35 @@ func (c *Composer) takeValue(u agent.Update) { // compose composes the node once and answers the updates to hand: each composition whose digest differs from the one // handed, and a deletion for each handed one no want names any more. func (c *Composer) compose(ctx context.Context) []agent.Update { + pass := Pass{} + defer func() { + pass.At = time.Now() + if c.owed.Load() { + wait, _ := c.waitNow() + pass.Again = pass.At.Add(wait) + } + c.mu.Lock() + c.last = pass + c.mu.Unlock() + }() + // A pass that runs on says so, and what it waits on, every passSaysEvery (issue 774, 2026-10-31: ace said nothing for + // five minutes while a pass waited after the vault restarted). + began := time.Now() + long, stopLong := context.WithCancel(ctx) + defer stopLong() + go func() { + t := time.NewTicker(passSaysEvery) + defer t.Stop() + for { + select { + case <-long.Done(): + return + case <-t.C: + c.env.Say(fmt.Sprintf("the node's composition has run for %s and waits on %s", + time.Since(began).Round(time.Second), c.waitingOn())) + } + } + }() c.mu.Lock() fps := map[string]string{} kept := map[string]string{} @@ -542,7 +658,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { // 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("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 + c.paid() // nothing of its own to compose, so no pass is owed and no timer armed again return nil } // Composed from the build its want names (frombus.Scope.KeptAt, issue 1126). @@ -581,11 +697,20 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { return nil } if err != nil { + why := fmt.Sprintf("%v", err) + // A bus that did not answer past every try is no answer: composed again after a bound (issue 1126, 2026-10-28), + // and the refusal says when (issue 774, 2026-10-31). A refusal is final until an input moves. + if BusFailed(err) { + why += "; " + c.owe() + } else { + c.paid() + } + pass.Refused = why // Said before the agent marks the modules: the marks read each have first, which a paused bus delays. - c.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err)) - c.refused(fmt.Sprintf("%v", err), nil) - // A bus that did not answer past every try is no answer: composed again after a bound (issue 1126, 2026-10-28). - c.owed.Store(BusFailed(err)) + c.sayOnce("the node cannot compose its modules, and keeps running what it runs: " + why) + done := c.waiting("the agent marking each wanted module's have") + c.refused(why, nil) + done() return nil } // The boxes it runs, slot by slot: read before c.mu is taken, since the agent's lock is taken to read what runs. @@ -596,14 +721,17 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { if c.missed.Load() > missed { // A have of what runs the bus did not answer for: its boxes would be the vault's, sealed again, and the // composition another; nothing is handed, and the pass is made again after a bound (issue 1126, 2026-10-28). - c.sayOnce("the node composes nothing new for now, and keeps running what it runs: the have of what it runs did not answer on the bus") - c.owed.Store(true) + pass.Refused = "the have of what it runs did not answer on the bus; " + c.owe() + c.sayOnce("the node composes nothing new for now, and keeps running what it runs: " + pass.Refused) return nil } - c.owed.Store(false) + c.paid() + pass.Held = held // A module refused alone is held: it keeps what it runs, its have says why, and every other module is applied (issue // 1126: one module's refusal froze the whole node). + done := c.waiting("the agent marking each held module's have") c.refused("", held) + done() if len(held) == 0 { c.sayOnce("") } else { @@ -658,6 +786,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { out = append(out, agent.Update{Key: key, Deleted: true, Revision: c.rev}) } } + pass.Handed = len(out) return out } diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 7139041..c4fe830 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -6,6 +6,7 @@ import ( "encoding/base64" "encoding/hex" "encoding/json" + "errors" "fmt" "os" "slices" @@ -238,6 +239,20 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } } b.mu.Unlock() + // Issue 774, 2026-10-31: the verb compose makes a pass now and says how it went; a composer that runs no watch + // makes none, and says so rather than waiting. + madeBefore := composed.Load() + cctx, ccancel := context.WithTimeout(ctx, 5*time.Second) + nowPass, err := c.ComposeNow(cctx) + ccancel() + if err != nil || nowPass.At.IsZero() || nowPass.Refused != "" || !nowPass.Again.IsZero() || composed.Load() == madeBefore { + t.Errorf("compose now: %+v (%v), %d pass(es) made; want one pass, composed", nowPass, err, composed.Load()-madeBefore) + } + idle, icancel := context.WithTimeout(ctx, 20*time.Millisecond) + if _, err := New(env).ComposeNow(idle); err == nil { + t.Error("a composer that watches nothing answered compose now") + } + icancel() lemursKey, _ := delivery.ComposedKey("n1", "lemurs", fp) nodeKey, _ := delivery.ComposedNodeKey("n1") if !got[lemursKey] || !got[nodeKey] || len(got) != 2 { @@ -277,15 +292,27 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { c.mu.Unlock() return c2, &refusals } + setFor(t, &passSaysEvery, 5*time.Millisecond) for name, row := range map[string]struct { failures int // reads failing before the bus answers cancel bool composes bool + fail error // what each failing read says; JetStream not enabled when nil + slow time.Duration // each failing read takes this long + // again: the refusal says when the node composes again, and a pass is owed (issue 774, 2026-10-31: ace's + // route-proxy said only "the vault's own could not be asked ... context deadline exceeded"). + again bool }{ "the bus paused for one read": {failures: 1, composes: true}, - "the bus paused past every try": {failures: 100}, + "the bus paused past every try": {failures: 100, again: true}, "the run ended mid-pass": {failures: 100, cancel: true}, + "the vault did not answer in time": {failures: 100, again: true, slow: 30 * time.Millisecond, + fail: fmt.Errorf("the vault's own could not be asked: no answer from mesh.seat.mesh-vault.call.own.node~n1: %w", context.DeadlineExceeded)}, + "a refusal": {failures: 100, fail: errors.New("nats says ${dir:mesh-state}, and nats declares no directory")}, } { + b.mu.Lock() + saidBefore := len(b.said) + b.mu.Unlock() failed := 0 pctx, pcancel := context.WithCancel(ctx) c2, refusals := fresh(func(gctx context.Context, bucket, key string) ([]byte, error) { @@ -295,6 +322,10 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { pcancel() return nil, fmt.Errorf("the vault's own could not be asked: %w", context.Canceled) } + time.Sleep(row.slow) + if row.fail != nil { + return nil, fmt.Errorf("the revisions of %s in %s: %w", key, bucket, row.fail) + } return nil, fmt.Errorf("the revisions of %s in %s: %w", key, bucket, nats.ErrJetStreamNotEnabled) } b.mu.Lock() @@ -303,14 +334,37 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { }) handed := c2.compose(pctx) pcancel() + tries := 1 + len(composeAgain) // a bus paused for a moment is tried again at once; a timeout or a refusal is not + if row.fail != nil { + tries = 1 + } switch { case row.composes && (len(handed) == 0 || len(*refusals) != 0): t.Errorf("%s: handed %d, refusals %q; want composed and nothing refused", name, len(handed), *refusals) - case !row.composes && !row.cancel && (len(*refusals) != 1 || failed != 1+len(composeAgain)): - t.Errorf("%s: refusals %q after %d reads; want one refusal after %d tries", name, *refusals, failed, 1+len(composeAgain)) + case !row.composes && !row.cancel && (len(*refusals) != 1 || failed != tries): + t.Errorf("%s: refusals %q after %d reads; want one refusal after %d tries", name, *refusals, failed, tries) case row.cancel && len(*refusals) != 0: t.Errorf("%s: a pass whose run ended refused %q", name, *refusals) } + if !row.composes && !row.cancel && len(*refusals) > 0 { + says := strings.Contains((*refusals)[0], "; the node composes again at ") && strings.HasSuffix((*refusals)[0], "(try 1)") + if says != row.again || c2.owed.Load() != row.again || (c2.last.Again.IsZero() == row.again) || c2.last.Refused != (*refusals)[0] { + t.Errorf("%s: refused %q, owes a pass %v, last pass %+v; want the time it composes again said and owed: %v", + name, (*refusals)[0], c2.owed.Load(), c2.last, row.again) + } + } + // A pass that runs on says so, and what it waits on (issue 774, 2026-10-31: ace was silent for five minutes). + if row.slow > 0 { + b.mu.Lock() + long := false + for _, line := range b.said[saidBefore:] { + long = long || strings.Contains(line, "the node's composition has run for ") && strings.Contains(line, " and waits on ") + } + b.mu.Unlock() + if !long { + t.Errorf("%s: a pass that ran on said nothing of what it waits on", name) + } + } } if !Transient(fmt.Errorf("the manifest of avahi: the revisions of x in mesh-delivery-indexer_manifests: nats: jetstream not enabled")) || !Transient(fmt.Errorf("x: %w", nats.ErrNoResponders)) || Transient(fmt.Errorf("nats says ${dir:mesh-state}, and nats declares no directory")) || From 56eb0b792c1cf1a18e4d9a2551779b86221e9bd2 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 31 Oct 2026 00:57:53 +0100 Subject: [PATCH 2/3] Pin mesh-sdk dfd5d82, so the node-engine announces the verb compose (issue 774) --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index f89f63a..b03b4e6 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.20261030233444-8ed3b872e0ba + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030235330-dfd5d8295e0a 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 a698341..c355c53 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030233444-8ed3b872e0ba h1:vRjCeYiVsX8Q8FrT8IHa3z+zpzh2WVaDCnu7l/XYATI= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030233444-8ed3b872e0ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030235330-dfd5d8295e0a h1:tjbRfUfE6pIylHFXg53SI2Sfgi1+QzWUyaY00xz25z4= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030235330-dfd5d8295e0a/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= From b0d6684d373192a7b55a9407e368d7ada14f82d0 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 31 Oct 2026 01:08:57 +0100 Subject: [PATCH 3/3] Retry only a failed apply that is not held, answer a repeated compose from the last pass, and say the try the refusal named (issue 774, review round 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The verb compose has no warrant (ADR 0563 §2): Agent.Retry released a module held for a failed backup and re-applied one already running its want. RetryFailed tries again only a failed apply not held, and refuses in words otherwise. A hand ask no longer restarts the growing bound, and a pass less than askedWithin old is the answer, so a caller in a loop drives no passes against the vault. A composer whose loop ended refuses at once. The retry line said try N+1. The test's first shadow now ends before slowAfter is set, a race the race detector found under the new timing. --- cmd/mesh-host/engineseat.go | 19 ++++++----- internal/agent/agent.go | 41 +++++++++++++++++++++++- internal/agent/agent_test.go | 31 ++++++++++++++++-- internal/selfcompose/selfcompose.go | 38 ++++++++++++++++++---- internal/selfcompose/selfcompose_test.go | 25 ++++++++++++--- 5 files changed, 131 insertions(+), 23 deletions(-) diff --git a/cmd/mesh-host/engineseat.go b/cmd/mesh-host/engineseat.go index a7fcbbe..13afd82 100644 --- a/cmd/mesh-host/engineseat.go +++ b/cmd/mesh-host/engineseat.go @@ -306,11 +306,18 @@ func composeFrom(current *agentNow, composer *atomic.Pointer[selfcompose.Compose return nil, errors.New("this node composes none of its own modules now (it applies the controller's " + "compositions, or its node agent is starting), so there is no pass to make") } - ag := current.get() + out := map[string]any{} if module != "" { - if ag == nil || !ag.Placed("")[module] { - return nil, fmt.Errorf("%s has no want on this machine, so there is no apply of it to try again", module) + // Only a failed apply that is not held is tried again; what runs is never taken away (review round 1: the + // verb has no warrant, ADR 0563 §2, and a hold for a failed backup is the operator's to release, ADR 0568 §4). + ag := current.get() + if ag == nil { + return nil, errors.New("this engine's node agent has not started, so no apply is tried again") } + if err := ag.RetryFailed(module); err != nil { + return nil, err + } + out["retried"] = module } ctx, cancel := context.WithTimeout(context.Background(), composeReadFor) defer cancel() @@ -318,11 +325,7 @@ func composeFrom(current *agentNow, composer *atomic.Pointer[selfcompose.Compose if err != nil { return nil, err } - out := map[string]any{"pass": pass} - if module != "" { - ag.Retry(module) - out["retried"] = module - } + out["pass"] = pass return out, nil } } diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 45bf7d8..7b59348 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -210,6 +210,8 @@ type Agent struct { retry chan []string // reconcile carries the judge's edges: a resource found not active (ADR 0568 §2). reconcile chan []string + // failedAgain carries the modules whose failed apply a person asked to try again (RetryFailed). + failedAgain chan string // tries counts the transient failures in a row of each module; again carries a retry whose wait ran out, and // transient asks for every transiently failed module at once. stopped is closed when Run returns. tries map[string]int @@ -256,7 +258,7 @@ func New(env Env) *Agent { } return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, applied: map[string]compNow{}, had: map[string]compNow{}, keptData: map[string]bool{}, attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}, notComposed: map[string]notComposedMark{}, - retry: make(chan []string, 8), reconcile: make(chan []string, 64), tries: map[string]int{}, again: make(chan againAsk, 64), + retry: make(chan []string, 8), reconcile: make(chan []string, 64), failedAgain: make(chan string, 8), tries: map[string]int{}, again: make(chan againAsk, 64), transient: make(chan struct{}, 1), stopped: make(chan struct{})} } @@ -302,6 +304,15 @@ func (a *Agent) Run(ctx context.Context) error { dirty[m] = true } a.mu.Unlock() + case m := <-a.failedAgain: + // A failed apply a person asked to try again: only one still failed and not held (RetryFailed); what runs is + // left as it is. + a.mu.Lock() + if att, ok := a.attempt[m]; ok && !att.Held { + delete(a.attempt, m) + dirty[m] = true + } + a.mu.Unlock() case ask := <-a.again: // The wait of a transiently failed apply ran out: tried again from the keys it has, if nothing tried it since. a.mu.Lock() @@ -1150,6 +1161,34 @@ func (a *Agent) Placed(except string) map[string]bool { return out } +// RetryFailed asks a module's failed apply to be tried again (the node-engine's verb compose, issue 774): only an apply +// that failed and is not held for a failed backup, whose release is the operator's answer (ADR 0568 §4); what runs is +// never taken away, so nothing is applied that the engine would not apply on its own at the next change. Refused in +// words otherwise. +func (a *Agent) RetryFailed(module string) error { + a.mu.Lock() + w := a.wants[module] + att, tried := a.attempt[module] + r, ran := a.running[module] + a.mu.Unlock() + switch { + case !w.Present || w.Want.Removed: + return fmt.Errorf("%s has no want on this machine, so there is no apply of it to try again", module) + case tried && att.Held: + return fmt.Errorf("%s is held for a failed backup: the operator's answer releases it", module) + case !tried && ran && r.FP == w.Want.FP && r.Phase != delivery.PhaseRemoved: + return fmt.Errorf("%s runs what its want names, nothing to retry", module) + case !tried: + return fmt.Errorf("%s has no failed apply to try again: it waits for its composition, or settles on its own", module) + } + select { + case a.failedAgain <- module: + return nil + default: + return fmt.Errorf("%s's retry is not taken now: the engine has too many asked; ask again in a moment", module) + } +} + // RemovedByWant is the modules whose want says removed (apply.Scope.RemovedByWant, issue 774): in the node's own apply, // only their resources may go. Called without a.mu held. func (a *Agent) RemovedByWant() map[string]bool { diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 4423a9f..e2bb085 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -457,13 +457,27 @@ func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) { if n := len(b.appliedList()); n != 1 { t.Fatalf("a failed apply was retried by an unrelated key or the clock (%d applies)", n) } - // The want is written again: the retry. + // A person asks it again (the node-engine's verb compose, issue 774): the failed apply is tried again with no key + // changed; once it runs, nothing is left to retry, and a module with no want has none. b.mu.Lock() b.failWith = "" b.mu.Unlock() - b.wants.put(wk, wraw) - eventually(t, "the change of the want did not retry", func() bool { return len(b.appliedList()) == 2 }) + if err := b.ag.RetryFailed("web"); err != nil { + t.Fatalf("a failed apply was refused a retry: %v", err) + } + eventually(t, "the asked retry did not apply", func() bool { return len(b.appliedList()) == 2 }) eventually(t, "not running", func() bool { return b.haves.last("web").Phase == string(delivery.PhaseRunning) }) + for module, words := range map[string]string{"web": "runs what its want names", "ghost": "has no want"} { + if err := b.ag.RetryFailed(module); err == nil || !strings.Contains(err.Error(), words) { + t.Errorf("a retry of %s: %v, want refused saying %q", module, err, words) + } + } + // The want is written again: no apply, it runs what the want names. + b.wants.put(wk, wraw) + time.Sleep(50 * time.Millisecond) + if n := len(b.appliedList()); n != 2 { + t.Errorf("a module running its want was applied again (%d applies)", n) + } }) // The bus did not answer for a moment (mesh-vault on the bus node, 2026-10-25 23:04:24): tried again with no new // revision of either key, by the backoff or by RetryTransient once the engine knows the bus is back; the have says @@ -540,6 +554,17 @@ func TestADataModuleIsBackedUpFirstAndHeldWhenItFails(t *testing.T) { if got := b.appliedList(); len(got) != 0 { t.Fatalf("something was applied behind a failed backup: %v", got) } + // A person's retry (the verb compose, issue 774) never releases it: the operator's answer does (ADR 0568 §4). + if err := b.ag.RetryFailed("db"); err == nil || !strings.Contains(err.Error(), "the operator's answer releases it") { + t.Errorf("a retry of db held for a failed backup: %v", err) + } + time.Sleep(50 * time.Millisecond) + b.mu.Lock() + asked := len(b.backups) + b.mu.Unlock() + if got := b.appliedList(); len(got) != 0 || asked != 1 { + t.Fatalf("a retry released db held for a failed backup: applied %v, %d backups asked", got, asked) + } // The answer arrives in the module's own want: back up again (it fails again), then skip. wk, again := wantFor(t, "db", fp2, dig2, delivery.BackupAgain) diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 1040c67..aeb2730 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -135,6 +135,8 @@ type Composer struct { asked chan chan Pass // last is how the last pass went, under mu. last Pass + // watching says the composer's loop runs (Watch), so an ask of a pass can be answered. + watching atomic.Bool } // Pass is how one pass went, as the node-engine's verb compose answers it. @@ -150,8 +152,14 @@ type Pass struct { // Again is when the node composes again on its own, after a seat or a read that did not answer; zero when it does // not need to. Again time.Time `json:"again,omitempty"` + // Earlier says the answer is the last pass, made less than askedWithin ago, and no new one was made for the ask. + Earlier bool `json:"earlier,omitempty"` } +// askedWithin is how recent a pass answers an ask of the verb compose without a new one: the first wait after a seat +// that did not answer (laterAgain's first). +var askedWithin = 5 * time.Second + // passSaysEvery is how often a pass that runs on says, in the journal, how long it has run and what it waits on. var passSaysEvery = time.Minute @@ -174,6 +182,12 @@ func (c *Composer) paid() { c.owed.Store(false) } +// againLine is what a retry pass says as it starts: the try its refusal named. +func (c *Composer) againLine() string { + _, try := c.waitNow() + return fmt.Sprintf("the node composes again now (try %d), after a seat or a read that did not answer", try) +} + // waitNow is the bound before the pass owed, and its try. func (c *Composer) waitNow() (time.Duration, int) { c.mu.Lock() @@ -209,7 +223,20 @@ func New(env Env) *Composer { // ComposeNow asks the composer for a pass now (the node-engine's verb compose, issue 774): the growing bound after a seat // that did not answer starts again, and the answer is how the pass went. Refused in words when no pass is made before // ctx ends: the composer not running, or its inputs not all read yet. +// +// A pass ended less than askedWithin ago is the answer, and no new one is made, so a caller asking in a loop never +// drives passes back to back against the vault; an ask never restarts the growing bound either (review round 1). func (c *Composer) ComposeNow(ctx context.Context) (Pass, error) { + if !c.watching.Load() { + return Pass{}, errors.New("the node's composer runs no loop now: it composes nothing until its agent starts again") + } + c.mu.Lock() + last := c.last + c.mu.Unlock() + if !last.At.IsZero() && time.Since(last.At) < askedWithin { + last.Earlier = true + return last, nil + } reply := make(chan Pass, 1) select { case c.asked <- reply: @@ -333,8 +360,10 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er }() } out := make(chan agent.Update, 64) + c.watching.Store(true) go func() { defer close(out) + defer c.watching.Store(false) caught := make([]bool, len(ins)) ready := false dirty := false @@ -350,14 +379,11 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er case <-again: again = nil dirty = true - _, try := c.waitNow() - c.env.Say(fmt.Sprintf("the node composes again now (try %d), after a seat or a read that did not answer", try+1)) + c.env.Say(c.againLine()) case reply := <-c.asked: - // Asked by hand (the verb compose): composed now, the bound starting again. + // Asked by hand (the verb compose): composed now; the growing bound is not restarted, so a pass that + // fails again waits longer, not less. replies = append(replies, reply) - c.mu.Lock() - c.tries = 0 - c.mu.Unlock() again = nil dirty = true case t := <-merged: diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index c4fe830..01647d9 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -241,6 +241,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { b.mu.Unlock() // Issue 774, 2026-10-31: the verb compose makes a pass now and says how it went; a composer that runs no watch // makes none, and says so rather than waiting. + // The pass made at the start is less than askedWithin old for a moment: the ask waits that out first. + setFor(t, &askedWithin, 200*time.Millisecond) + time.Sleep(250 * time.Millisecond) madeBefore := composed.Load() cctx, ccancel := context.WithTimeout(ctx, 5*time.Second) nowPass, err := c.ComposeNow(cctx) @@ -248,9 +251,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { if err != nil || nowPass.At.IsZero() || nowPass.Refused != "" || !nowPass.Again.IsZero() || composed.Load() == madeBefore { t.Errorf("compose now: %+v (%v), %d pass(es) made; want one pass, composed", nowPass, err, composed.Load()-madeBefore) } - idle, icancel := context.WithTimeout(ctx, 20*time.Millisecond) - if _, err := New(env).ComposeNow(idle); err == nil { - t.Error("a composer that watches nothing answered compose now") + // Asked again at once, the last pass is the answer: a caller in a loop drives no passes back to back (review round 1). + madeBefore = composed.Load() + if again, err := c.ComposeNow(ctx); err != nil || !again.Earlier || !again.At.Equal(nowPass.At) || composed.Load() != madeBefore { + t.Errorf("compose now asked again at once: %+v (%v), %d pass(es) made; want the last pass, none made", again, err, composed.Load()-madeBefore) + } + // A composer whose loop runs not refuses at once, not after the verb's whole bound. + idle, icancel := context.WithTimeout(ctx, 5*time.Second) + if _, err := New(env).ComposeNow(idle); err == nil || !strings.Contains(err.Error(), "runs no loop") || idle.Err() != nil { + t.Errorf("a composer that watches nothing answered compose now: %v", err) } icancel() lemursKey, _ := delivery.ComposedKey("n1", "lemurs", fp) @@ -347,7 +356,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { t.Errorf("%s: a pass whose run ended refused %q", name, *refusals) } if !row.composes && !row.cancel && len(*refusals) > 0 { - says := strings.Contains((*refusals)[0], "; the node composes again at ") && strings.HasSuffix((*refusals)[0], "(try 1)") + // The retry pass says the same try the refusal named (review round 1: it said try 2 for try 1). + says := strings.Contains((*refusals)[0], "; the node composes again at ") && strings.HasSuffix((*refusals)[0], "(try 1)") && + strings.Contains(c2.againLine(), "(try 1)") if says != row.again || c2.owed.Load() != row.again || (c2.last.Again.IsZero() == row.again) || c2.last.Refused != (*refusals)[0] { t.Errorf("%s: refused %q, owes a pass %v, last pass %+v; want the time it composes again said and owed: %v", name, (*refusals)[0], c2.owed.Load(), c2.last, row.again) @@ -935,7 +946,8 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { // Review of L2b, round 1, finding 2: the candidates' state is not there when the node starts (a delivery before the // indexer made it): the shadow tries again, and answers once it appears. sb.missing(delivery.PlacementCandidatesBucket, true) - go sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond }) + kept := make(chan struct{}) + go func() { sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond }); close(kept) }() time.Sleep(30 * time.Millisecond) sb.missing(delivery.PlacementCandidatesBucket, false) sb.made(7) @@ -1270,6 +1282,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { t.Errorf("a shadow not ready said %q", nsaid) } nmu.Unlock() + // The first shadow ends before slowAfter is set: it reads it while it answers (a race the race detector names). + scancel() + <-kept setFor(t, &slowAfter, time.Nanosecond) // every answer is slow: said, and noted in words that do not move with the time nb.send(delivery.AssignerWantsBucket, agent.Update{Caught: true}) select {