Merge pull request 'A node says when it composes again, says what a long pass waits on, and composes on request (issue 774)' (#185) from fix/compose-retries-transient into main

This commit was merged in pull request #185.
This commit is contained in:
2026-10-31 00:21:33 +00:00
11 changed files with 371 additions and 25 deletions
+37
View File
@@ -18,6 +18,7 @@ import (
"github.com/novox/mesh-host/internal/accounts" "github.com/novox/mesh-host/internal/accounts"
"github.com/novox/mesh-host/internal/apply" "github.com/novox/mesh-host/internal/apply"
"github.com/novox/mesh-host/internal/link" "github.com/novox/mesh-host/internal/link"
"github.com/novox/mesh-host/internal/selfcompose"
"github.com/novox/mesh-host/internal/store" "github.com/novox/mesh-host/internal/store"
) )
@@ -292,3 +293,39 @@ func compositionFrom(current *agentNow) func(module, digest string, diff bool) (
return ag.Composition(ctx, module, digest, diff) 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")
}
out := map[string]any{}
if 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()
pass, err := c.ComposeNow(ctx)
if err != nil {
return nil, err
}
out["pass"] = pass
return out, nil
}
}
+1
View File
@@ -1142,6 +1142,7 @@ func runLink(ctx context.Context, opts options) error {
go reconcileActs(ctx, queue.Acts) go reconcileActs(ctx, queue.Acts)
current := &agentNow{} current := &agentNow{}
queue.Acts.Composition = compositionFrom(current) queue.Acts.Composition = compositionFrom(current)
queue.Acts.Compose = composeFrom(current, &composerNow)
setRunsHere(runsFromAgent(current)) setRunsHere(runsFromAgent(current))
factsOfNow := func(ctx context.Context, ag *agent.Agent) { factsOfNow := func(ctx context.Context, ag *agent.Agent) {
sayFacts(ctx, opts, mine.Membership.Signer, queue, say, ag) sayFacts(ctx, opts, mine.Membership.Signer, queue, say, ag)
+6
View File
@@ -668,6 +668,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
} }
if composer != nil { if composer != nil {
machinesNow.Store(composer) // the machines a failing network part points at, by the placements (issue 774) 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) composes.set(delivery.ComposesOwn)
env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy 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 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() stopRun()
cancel() cancel()
current.set(nil) current.set(nil)
composerNow.Store(nil)
if ctx.Err() != nil { if ctx.Err() != nil {
return 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. // (writeHealthReport); nil on a node that does not compose its own, whose parts then name no target.
var machinesNow atomic.Pointer[selfcompose.Composer] 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 // 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). // its own (placedAllHave), so what runs is the agent's alone (ADR 0550 §2, design 47's L3 s9).
var keptRetired atomic.Bool var keptRetired atomic.Bool
+1 -1
View File
@@ -3,7 +3,7 @@ module github.com/novox/mesh-host
go 1.26.0 go 1.26.0
require ( 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-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
golang.org/x/crypto v0.57.0 golang.org/x/crypto v0.57.0
+2 -2
View File
@@ -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.20261030235330-dfd5d8295e0a h1:tjbRfUfE6pIylHFXg53SI2Sfgi1+QzWUyaY00xz25z4=
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/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+40 -1
View File
@@ -210,6 +210,8 @@ type Agent struct {
retry chan []string retry chan []string
// reconcile carries the judge's edges: a resource found not active (ADR 0568 §2). // reconcile carries the judge's edges: a resource found not active (ADR 0568 §2).
reconcile chan []string 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 // 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. // transient asks for every transiently failed module at once. stopped is closed when Run returns.
tries map[string]int 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{}, 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{}, 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{})} transient: make(chan struct{}, 1), stopped: make(chan struct{})}
} }
@@ -302,6 +304,15 @@ func (a *Agent) Run(ctx context.Context) error {
dirty[m] = true dirty[m] = true
} }
a.mu.Unlock() 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: 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. // The wait of a transiently failed apply ran out: tried again from the keys it has, if nothing tried it since.
a.mu.Lock() a.mu.Lock()
@@ -1150,6 +1161,34 @@ func (a *Agent) Placed(except string) map[string]bool {
return out 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, // 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. // only their resources may go. Called without a.mu held.
func (a *Agent) RemovedByWant() map[string]bool { func (a *Agent) RemovedByWant() map[string]bool {
+28 -3
View File
@@ -457,13 +457,27 @@ func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) {
if n := len(b.appliedList()); n != 1 { if n := len(b.appliedList()); n != 1 {
t.Fatalf("a failed apply was retried by an unrelated key or the clock (%d applies)", n) 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.mu.Lock()
b.failWith = "" b.failWith = ""
b.mu.Unlock() b.mu.Unlock()
b.wants.put(wk, wraw) if err := b.ag.RetryFailed("web"); err != nil {
eventually(t, "the change of the want did not retry", func() bool { return len(b.appliedList()) == 2 }) 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) }) 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 // 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 // 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 { if got := b.appliedList(); len(got) != 0 {
t.Fatalf("something was applied behind a failed backup: %v", got) 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. // 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) wk, again := wantFor(t, "db", fp2, dig2, delivery.BackupAgain)
+10
View File
@@ -40,6 +40,8 @@ const (
ActSetuidSearch = "setuid-search" ActSetuidSearch = "setuid-search"
// ActComposition reads a module's composition beside what runs of it; read-only, no warrant (issue 774). // ActComposition reads a module's composition beside what runs of it; read-only, no warrant (issue 774).
ActComposition = "composition" ActComposition = "composition"
// ActCompose composes the node now, the recovery of a failed composition (issue 774).
ActCompose = "compose"
) )
// The states of an act. // 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): // 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. // 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) 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 says one of the seat's events, asked, done or refused; nil says nothing.
Say func(event string, body any) Say func(event string, body any)
Now func() time.Time Now func() time.Time
@@ -190,6 +195,11 @@ func (h *HandActs) Call(c EngineCall) (any, error) {
diff = b diff = b
} }
return h.Composition(arg("module"), arg("digest"), diff) 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": case "acts":
if id := arg("id"); id != "" { if id := arg("id"); id != "" {
return h.Act(id) return h.Act(id)
+5 -1
View File
@@ -239,7 +239,8 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
JudgeDirectories: func(by string) (string, error) { return "judged for " + by, nil }, JudgeDirectories: func(by string) (string, error) { return "judged for " + by, nil },
Composition: func(module, digest string, diff bool) (any, error) { Composition: func(module, digest string, diff bool) (any, error) {
return map[string]any{"module": module, "digest": digest, "diff": diff}, nil 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 }{ 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", "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"`}, {"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. // 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","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"`}, {"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\""`}, {"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, " ") subject, body, ok := strings.Cut(row.subject, " ")
+168 -13
View File
@@ -127,6 +127,72 @@ type Composer struct {
missed atomic.Uint64 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 is set when a pass refused for a bus that did not answer: the composer composes again after a bound.
owed atomic.Bool 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
// 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.
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"`
// 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
// 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)
}
// 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()
defer c.mu.Unlock()
return c.wait, c.tries
} }
// note counts err when it is the bus not answering, and returns it. // note counts err when it is the bus not answering, and returns it.
@@ -150,7 +216,39 @@ func New(env Env) *Composer {
} }
return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{}, 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{}, 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.
//
// 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:
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. // ownSealed is one own secret's sealed value at a generation and the key it is sealed to.
@@ -262,15 +360,17 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
}() }()
} }
out := make(chan agent.Update, 64) out := make(chan agent.Update, 64)
c.watching.Store(true)
go func() { go func() {
defer close(out) defer close(out)
defer c.watching.Store(false)
caught := make([]bool, len(ins)) caught := make([]bool, len(ins))
ready := false ready := false
dirty := false dirty := false
// A pass refused for a bus that did not answer is made again after a bound (laterAgain), not kept until another // 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). // input moves (issue 1126, 2026-10-28).
var again <-chan time.Time var again <-chan time.Time
tries := 0 var replies []chan Pass
for { for {
// Drain whatever came, then compose once for all of it. // Drain whatever came, then compose once for all of it.
select { select {
@@ -279,6 +379,13 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
case <-again: case <-again:
again = nil again = nil
dirty = true dirty = true
c.env.Say(c.againLine())
case reply := <-c.asked:
// 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)
again = nil
dirty = true
case t := <-merged: case t := <-merged:
if !t.ok { 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)) 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 +425,19 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
} }
} }
if c.owed.Load() { if c.owed.Load() {
wait := LaterWait(tries) wait, _ := c.waitNow()
tries++
c.env.Say(fmt.Sprintf("the node composes again in %s: a read on the bus did not answer", wait)) c.env.Say(fmt.Sprintf("the node composes again in %s: a read on the bus did not answer", wait))
again = time.After(wait) again = time.After(wait)
} else { } 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() { if !c.caughtSent() {
select { select {
case out <- agent.Update{Caught: true}: case out <- agent.Update{Caught: true}:
@@ -518,6 +631,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 // 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. // handed, and a deletion for each handed one no want names any more.
func (c *Composer) compose(ctx context.Context) []agent.Update { 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() c.mu.Lock()
fps := map[string]string{} fps := map[string]string{}
kept := map[string]string{} kept := map[string]string{}
@@ -542,7 +684,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 // 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). // 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.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 return nil
} }
// Composed from the build its want names (frombus.Scope.KeptAt, issue 1126). // Composed from the build its want names (frombus.Scope.KeptAt, issue 1126).
@@ -581,11 +723,20 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
return nil return nil
} }
if err != 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. // 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.sayOnce("the node cannot compose its modules, and keeps running what it runs: " + why)
c.refused(fmt.Sprintf("%v", err), nil) done := c.waiting("the agent marking each wanted module's have")
// A bus that did not answer past every try is no answer: composed again after a bound (issue 1126, 2026-10-28). c.refused(why, nil)
c.owed.Store(BusFailed(err)) done()
return nil 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. // 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 +747,17 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
if c.missed.Load() > missed { 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 // 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). // 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") pass.Refused = "the have of what it runs did not answer on the bus; " + c.owe()
c.owed.Store(true) c.sayOnce("the node composes nothing new for now, and keeps running what it runs: " + pass.Refused)
return nil 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 // 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). // 1126: one module's refusal froze the whole node).
done := c.waiting("the agent marking each held module's have")
c.refused("", held) c.refused("", held)
done()
if len(held) == 0 { if len(held) == 0 {
c.sayOnce("") c.sayOnce("")
} else { } else {
@@ -658,6 +812,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
out = append(out, agent.Update{Key: key, Deleted: true, Revision: c.rev}) out = append(out, agent.Update{Key: key, Deleted: true, Revision: c.rev})
} }
} }
pass.Handed = len(out)
return out return out
} }
+73 -4
View File
@@ -6,6 +6,7 @@ import (
"encoding/base64" "encoding/base64"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"os" "os"
"slices" "slices"
@@ -238,6 +239,29 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
} }
} }
b.mu.Unlock() 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)
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)
}
// 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) lemursKey, _ := delivery.ComposedKey("n1", "lemurs", fp)
nodeKey, _ := delivery.ComposedNodeKey("n1") nodeKey, _ := delivery.ComposedNodeKey("n1")
if !got[lemursKey] || !got[nodeKey] || len(got) != 2 { if !got[lemursKey] || !got[nodeKey] || len(got) != 2 {
@@ -277,15 +301,27 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
c.mu.Unlock() c.mu.Unlock()
return c2, &refusals return c2, &refusals
} }
setFor(t, &passSaysEvery, 5*time.Millisecond)
for name, row := range map[string]struct { for name, row := range map[string]struct {
failures int // reads failing before the bus answers failures int // reads failing before the bus answers
cancel bool cancel bool
composes 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 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 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 failed := 0
pctx, pcancel := context.WithCancel(ctx) pctx, pcancel := context.WithCancel(ctx)
c2, refusals := fresh(func(gctx context.Context, bucket, key string) ([]byte, error) { c2, refusals := fresh(func(gctx context.Context, bucket, key string) ([]byte, error) {
@@ -295,6 +331,10 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
pcancel() pcancel()
return nil, fmt.Errorf("the vault's own could not be asked: %w", context.Canceled) 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) return nil, fmt.Errorf("the revisions of %s in %s: %w", key, bucket, nats.ErrJetStreamNotEnabled)
} }
b.mu.Lock() b.mu.Lock()
@@ -303,14 +343,39 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
}) })
handed := c2.compose(pctx) handed := c2.compose(pctx)
pcancel() 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 { switch {
case row.composes && (len(handed) == 0 || len(*refusals) != 0): 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) 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)): 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, 1+len(composeAgain)) t.Errorf("%s: refusals %q after %d reads; want one refusal after %d tries", name, *refusals, failed, tries)
case row.cancel && len(*refusals) != 0: case row.cancel && len(*refusals) != 0:
t.Errorf("%s: a pass whose run ended refused %q", name, *refusals) t.Errorf("%s: a pass whose run ended refused %q", name, *refusals)
} }
if !row.composes && !row.cancel && len(*refusals) > 0 {
// 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)
}
}
// 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")) || 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")) || !Transient(fmt.Errorf("x: %w", nats.ErrNoResponders)) || Transient(fmt.Errorf("nats says ${dir:mesh-state}, and nats declares no directory")) ||
@@ -881,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 // 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. // indexer made it): the shadow tries again, and answers once it appears.
sb.missing(delivery.PlacementCandidatesBucket, true) 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) time.Sleep(30 * time.Millisecond)
sb.missing(delivery.PlacementCandidatesBucket, false) sb.missing(delivery.PlacementCandidatesBucket, false)
sb.made(7) sb.made(7)
@@ -1216,6 +1282,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
t.Errorf("a shadow not ready said %q", nsaid) t.Errorf("a shadow not ready said %q", nsaid)
} }
nmu.Unlock() 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 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}) nb.send(delivery.AssignerWantsBucket, agent.Update{Caught: true})
select { select {