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")) ||