diff --git a/cmd/mesh-host/engineseat.go b/cmd/mesh-host/engineseat.go index 6cba6a6..772ff55 100644 --- a/cmd/mesh-host/engineseat.go +++ b/cmd/mesh-host/engineseat.go @@ -3,6 +3,7 @@ package main import ( "bufio" "context" + "encoding/json" "errors" "fmt" "os" @@ -15,6 +16,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" meshasker "git.novox.be/novox/mesh-sdk/go/asker" "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/novox/mesh-host/internal/accounts" "github.com/novox/mesh-host/internal/apply" "github.com/novox/mesh-host/internal/link" @@ -359,3 +361,27 @@ func packagesFrom(node string, session *link.Session, composer *atomic.Pointer[s return v, nil } } + +// composeCandidateFrom answers the verb compose-candidate (novox/hq ADR 0593 §7): the check's candidate composed on this +// node in shadow by the run's shadow, applying and writing nothing, within delivery.ComposeAnswerBound. The answer is +// the seat's shape (delivery.ComposeCandidateAnswer), checked as it is encoded; a node whose shadow does not run yet +// refuses in words, which the checker takes as no answer. +func composeCandidateFrom(shadow *atomic.Pointer[selfcompose.Shadow]) func(candidate string, input bool) (any, error) { + return func(candidate string, input bool) (any, error) { + ask := delivery.ComposeCandidateAsk{Candidate: candidate, Input: input} + if _, err := delivery.EncodeComposeCandidateAsk(ask); err != nil { + return nil, err + } + s := shadow.Load() + if s == nil { + return nil, errors.New("this node's shadow does not run now (its node agent is starting), so it composes no candidate") + } + ctx, cancel := context.WithTimeout(context.Background(), delivery.ComposeAnswerBound) + defer cancel() + raw, err := delivery.EncodeComposeCandidateAnswer(s.ComposeCandidate(ctx, ask)) + if err != nil { + return nil, fmt.Errorf("this node's answer to %s does not fit the seat's shape: %w", candidate, err) + } + return json.RawMessage(raw), nil + } +} diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 3045eb9..1bd7840 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1001,6 +1001,7 @@ func runLink(ctx context.Context, opts options) error { queue.Acts.Composition = compositionFrom(current) queue.Acts.Compose = composeFrom(current, &composerNow) queue.Acts.Packages = packagesFrom(mine.Node, queue, &composerNow) + queue.Acts.ComposeCandidate = composeCandidateFrom(&shadowNow) 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 d6b2c40..8001ee3 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -623,6 +623,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus session.NoteCandidates(ctx, delivery.WithValuesDiffer(answers)) }) go shadow.Keep(run, link.Backoff) // this run's alone: a switch begins a run, and its shadow, again + shadowNow.Store(shadow) // what the verb compose-candidate composes a check's candidate with // Its own composer, always (issue 1126, A7): a composition made elsewhere (the controller's, in the node's // composition bucket) is never applied. While the seat's placement holds no key for this node it composes // nothing and keeps running what it runs; a placement it cannot watch stops the agent, said, and it begins again. @@ -640,6 +641,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus cancel() current.set(nil) composerNow.Store(nil) + shadowNow.Store(nil) if ctx.Err() != nil { return } @@ -766,6 +768,10 @@ var machinesNow atomic.Pointer[selfcompose.Composer] // compositions or no run is going: what the verb compose asks a pass of (issue 774). var composerNow atomic.Pointer[selfcompose.Composer] +// shadowNow is the shadow of the run now, which answers placement candidates and the verb compose-candidate; nil +// between runs. +var shadowNow atomic.Pointer[selfcompose.Shadow] + // 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 2a0f538..3412de9 100644 --- a/internal/link/handacts.go +++ b/internal/link/handacts.go @@ -45,6 +45,8 @@ const ( // ActPackages reads the packages this node's compositions declare, and with write writes its key again (novox/hq // ADR 0565 §6, §15). ActPackages = "packages" + // ActComposeCandidate composes a check's candidate on this node in shadow, read-only (novox/hq ADR 0593 §7). + ActComposeCandidate = "compose-candidate" ) // The states of an act. @@ -143,6 +145,10 @@ type HandActs struct { // Packages answers the node's key of the engine's state packages beside the controller's mirror's key and their // difference, and with write writes the key again from what the engine composed (the verb packages, ADR 0565 §6, §15). Packages func(write bool) (any, error) + // ComposeCandidate composes a check's candidate (the indexer's candidate key) on this node in shadow and answers per + // module, applying and writing nothing; with input, also the composition input it composed from (the verb + // compose-candidate, ADR 0593 §7). + ComposeCandidate func(candidate string, input bool) (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 @@ -215,6 +221,15 @@ func (h *HandActs) Call(c EngineCall) (any, error) { write = b } return h.Packages(write) + case ActComposeCandidate: + if h.ComposeCandidate == nil { + return nil, errors.New("this engine composes no candidate yet: its node agent has not started") + } + input := arg("input") == "true" + if b, ok := c.Args["input"].(bool); ok { + input = b + } + return h.ComposeCandidate(arg("candidate"), input) case "acts": if id := arg("id"); id != "" { return h.Act(id) diff --git a/internal/selfcompose/candidate.go b/internal/selfcompose/candidate.go new file mode 100644 index 0000000..c215cc0 --- /dev/null +++ b/internal/selfcompose/candidate.go @@ -0,0 +1,289 @@ +package selfcompose + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "sort" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/compose/frombus" +) + +// **A check's candidate, composed in shadow on a call** (novox/hq ADR 0593 §7): the node seat's verb compose-candidate. +// Given the key of the indexer's candidate (delivery.CandidatesBucket, `.`), the node composes itself +// twice from what its shadow holds (placement, records, the indexer's modules, its config values), with its own sealed +// values: once as it is, and once with the candidate's modules laid over the indexer's (each at the candidate's +// fingerprint, its manifest by digest and the indexer's verdict on it, ADR 0578 §3; a module the candidate removes left +// out). It answers per module placed here composed (its digest, and whether the candidate changes it), refused by the +// candidate, or held without it too, and the verdict of its own composition. It applies nothing and writes nothing: it +// composes on a copy of the shadow's inputs, so neither the shadow's answers nor the node's own composer move. + +// errNotReady is a shadow that has not read every input yet. +var errNotReady = errors.New("this node's shadow has not read every input yet") + +// ComposeCandidate is the node's answer to ask, within delivery.ComposeAnswerBound (or ctx's deadline, if sooner). +func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandidateAsk) delivery.ComposeCandidateAnswer { + began := time.Now() + s.mu.Lock() + c := s.c + s.mu.Unlock() + node := c.env.Node + a := delivery.ComposeCandidateAnswer{Node: node, Candidate: ask.Candidate} + refuse := func(why string) delivery.ComposeCandidateAnswer { + a.Verdict, a.Refused, a.Modules, a.Own = delivery.ComposeCandidateRefused, why, nil, delivery.ModuleComposed{} + a.Took = time.Since(began).Milliseconds() + return a + } + if _, _, err := delivery.ParseCompositionKey(ask.Candidate); err != nil { + return refuse(err.Error()) + } + ctx, cancel := context.WithTimeout(ctx, delivery.ComposeAnswerBound) + defer cancel() + if !s.ready.Load() { + return refuse(errNotReady.Error()) + } + base := c.scratch() + raw, err := base.get(ctx, delivery.CandidatesBucket, ask.Candidate) + if err != nil { + return refuse(fmt.Sprintf("the indexer's candidate %s cannot be read: %v", ask.Candidate, err)) + } + if raw == nil { + return refuse("the indexer holds no candidate " + ask.Candidate) + } + cand, err := delivery.DecodeCandidate(raw) + if err != nil { + return refuse(fmt.Sprintf("the indexer's candidate %s cannot be read: %v", ask.Candidate, err)) + } + laid := c.scratch() + touched, err := laid.layCandidate(ctx, cand) + if err != nil { + return refuse(err.Error()) + } + + var running []delivery.Composed + if c.env.Running != nil { + running = c.env.Running() + } + baseComps, baseHeld, baseErr := base.composeShadow(ctx, running, nil, nil) + laidComps, laidHeld, laidErr := laid.composeShadow(ctx, running, touched, func(scope frombus.Scope) { + if !ask.Input { + return + } + in, err := frombus.Input(ctx, reader{c: laid}, laid.env.Mesh, scope) + if err != nil { + return + } + // The request's input carries no catalogue (composer.Request.Input): the cases give the manifests. + in.Catalogue = nil + for i := range in.Nodes { + in.Nodes[i].Kept = nil + } + if body, err := json.Marshal(in); err == nil { + a.Input = body + } + }) + if ctx.Err() != nil { + return refuse(fmt.Sprintf("the bound of %s passed before the node composed the candidate", delivery.ComposeAnswerBound)) + } + switch { + case laidErr != nil && BusFailed(laidErr): + return refuse("a read on the bus did not answer: " + laidErr.Error()) + case laidErr != nil && baseErr != nil: + // The node composes none of its own without the candidate either: the candidate changes nothing of that. + a.Verdict, a.Modules = delivery.ComposeCandidateComposed, []delivery.ModuleComposed{} + a.Own = delivery.ModuleComposed{Module: delivery.HaveNodeWord, State: delivery.ModuleHeldState, Words: laidErr.Error()} + a.Took = time.Since(began).Milliseconds() + return a + case laidErr != nil: + return refuse(laidErr.Error()) + } + byModule := func(comps map[string]delivery.Composed) map[string]delivery.Composed { + out := map[string]delivery.Composed{} + for _, comp := range comps { + out[comp.Module] = comp + } + return out + } + was, now := byModule(baseComps), byModule(laidComps) + one := func(module string) (delivery.ModuleComposed, bool) { + m := delivery.ModuleComposed{Module: module} + if module == "" { + m.Module = delivery.HaveNodeWord + } + if why, held := laidHeld[module]; held { + m.State, m.Words = delivery.ModuleRefusedState, why + if _, before := baseHeld[module]; before || baseErr != nil { + m.State = delivery.ModuleHeldState + } + return m, true + } + comp, ok := now[module] + if !ok { + return m, false + } + dig, err := delivery.ComposedDigestOf(comp) + if err != nil { + m.State, m.Words = delivery.ModuleRefusedState, err.Error() + return m, true + } + m.State, m.FP, m.Digest = delivery.ModuleComposedState, comp.FP, dig + if before, ok := was[module]; !ok { + m.Changed = true + } else if d, err := delivery.ComposedDigestOf(before); err != nil || d != dig { + m.Changed = true + } + return m, true + } + modules := map[string]bool{} + for m := range now { + modules[m] = m != "" + } + for m := range laidHeld { + modules[m] = m != "" && m != agent.NodeModule + } + names := make([]string, 0, len(modules)) + for m, isModule := range modules { + if isModule { + names = append(names, m) + } + } + sort.Strings(names) + a.Modules = []delivery.ModuleComposed{} + refused := false + for _, m := range names { + if mc, ok := one(m); ok { + a.Modules = append(a.Modules, mc) + refused = refused || mc.State == delivery.ModuleRefusedState + } + } + if own, ok := one(""); ok { + a.Own = own + } else { + a.Own = delivery.ModuleComposed{Module: delivery.HaveNodeWord, State: delivery.ModuleHeldState, + Words: "the node composes no composition of its own"} + } + refused = refused || a.Own.State == delivery.ModuleRefusedState + a.Verdict = delivery.ComposeCandidateComposed + if refused { + a.Verdict = delivery.ComposeCandidateRefused + } + a.Took = time.Since(began).Milliseconds() + return a +} + +// scratch is a composer over a copy of what c holds of its inputs, for composing in shadow without moving c. +func (c *Composer) scratch() *Composer { + out := New(c.env) + c.mu.Lock() + defer c.mu.Unlock() + out.placements = maps.Clone(c.placements) + out.records = maps.Clone(c.records) + out.inOwn = maps.Clone(c.inOwn) + out.modules = maps.Clone(c.modules) + out.values = maps.Clone(c.values) + out.manifests = maps.Clone(c.manifests) + out.archives = maps.Clone(c.archives) + out.builds = maps.Clone(c.builds) + out.sealedOwn = maps.Clone(c.sealedOwn) + out.ownHanded = maps.Clone(c.ownHanded) + out.certified = c.certified + out.placementKeys = c.placementKeys + return out +} + +// layCandidate lays the candidate's modules over the indexer's: each at its fingerprint, with its manifest's digest and +// the indexer's verdict on it (read from its verdicts, none read taken as accepted); a module the candidate removes goes. +// It answers the modules laid. +func (c *Composer) layCandidate(ctx context.Context, cand delivery.Candidate) (map[string]bool, error) { + touched := map[string]bool{} + for _, pull := range cand.Pulls { + for module, cm := range pull.Fps { + if cm.None2 == delivery.CandidateNoneNoModule { + c.mu.Lock() + delete(c.modules, module) + c.mu.Unlock() + continue + } + if cm.FP2 == "" { + continue // no fingerprint at the preview: composed as the trunk holds it + } + m := delivery.Module{FP2: cm.FP2, Manifest: cm.Manifest, Repository: pull.Repository, Commit: pull.Head.Head} + if cm.Manifest != "" { + key, err := delivery.VerdictsKey(cm.Manifest) + if err != nil { + return nil, err + } + raw, err := c.get(ctx, delivery.VerdictsBucket, key) + if err != nil { + return nil, fmt.Errorf("the indexer's verdict on %s's manifest cannot be read: %w", module, err) + } + if raw != nil { + v, err := delivery.DecodeManifestVerdict(raw) + if err != nil { + return nil, fmt.Errorf("the indexer's verdict on %s's manifest cannot be read: %w", module, err) + } + m.Verdict, m.Problems, m.Warnings = v.State, v.Problems, v.Warnings + } + } + c.mu.Lock() + c.modules[module] = m + c.mu.Unlock() + touched[module] = true + } + } + return touched, nil +} + +// composeShadow composes this node from what c holds, as the shadow composes it (shadowBuilds): each module at the build +// it runs, one placed and not run at the indexer's, and one the candidate laid (laid) at the candidate's. input, when +// given, is handed the scope. +func (c *Composer) composeShadow(ctx context.Context, running []delivery.Composed, laid map[string]bool, input func(frombus.Scope)) (map[string]delivery.Composed, map[string]string, error) { + node := c.env.Node + c.mu.Lock() + placement, placed := c.placements[node] + modules := maps.Clone(c.modules) + inOwn := c.inOwnOf(placement) + c.mu.Unlock() + if !placed { + return nil, nil, errors.New("this node's key in the seat's placement state is gone: it composes nothing of its own") + } + fps := shadowBuilds(running, placement, modules, inOwn) + // A module the laid candidate changes is composed at the candidate's build, though the node runs another. + for module := range fps { + if mod, ok := modules[module]; ok && laid[module] { + fps[module] = mod.FP2 + } + } + kept := map[string]string{} + for module, fp := range fps { + if mod, ok := modules[module]; ok && mod.FP2 == fp { + continue + } + kept[module] = fp + } + wants := map[string]delivery.Want{agent.NodeModule: {}} + for module, fp := range fps { + wants[module] = delivery.Want{FP: fp} + } + c.mu.Lock() + c.wants = wants + c.mu.Unlock() + scope := frombus.Scope{Node: node, KeptAt: kept, InOwn: inOwn, Running: running, SameValue: c.sameValue} + if input != nil { + input(scope) + } + comps, _, held, err := frombus.Compose(ctx, reader{c: c}, c.env.Mesh, scope, fps) + if err != nil { + return nil, nil, err + } + for key, comp := range comps { + comps[key] = c.keepRunning(comp, running) + } + return comps, held, nil +} diff --git a/internal/selfcompose/candidate_test.go b/internal/selfcompose/candidate_test.go new file mode 100644 index 0000000..a01ab32 --- /dev/null +++ b/internal/selfcompose/candidate_test.go @@ -0,0 +1,125 @@ +package selfcompose + +import ( + "context" + "strings" + "sync/atomic" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-host/internal/agent" +) + +// novox/hq ADR 0593 §7: the verb compose-candidate composes a check's candidate on this node in shadow. A candidate that +// changes a placed module composes it at the candidate's build, changed; one whose manifest the indexer refused is +// answered refused, naming the module and the judge's words, and the verdict is refused; a candidate the indexer does not +// hold is refused in words. Nothing is applied, nothing is written, no placement answer is handed, and the answer comes +// within the bound (an ended bound answers at once, refused). +func TestTheNodeComposesACandidateOnAsk(t *testing.T) { + b := newFakeBus() + fp := delivery.SourcePrefix2 + strings.Repeat("a", 64) + fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64) + digest := delivery.ManifestDigest([]byte(lemurs)) + lemurs2 := strings.Replace(lemurs, `"version":"1"`, `"version":"2"`, 1) + digest2 := delivery.ManifestDigest([]byte(lemurs2)) + for d, body := range map[string]string{digest: lemurs, digest2: lemurs2} { + man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: body}) + key, _ := delivery.ManifestsKey(d) + b.kv[delivery.ManifestsBucket+"/"+key] = man + } + candidateOf := func(manifest string) string { + head := delivery.Head{Repository: "novox/x", Number: 7, Head: "c0ffee" + manifest[len(manifest)-6:]} + heads, _ := delivery.HeadsHash([]delivery.Head{head}) + pulls := []delivery.CandidatePull{{Head: head, Modules: []string{"lemurs"}, + Fps: map[string]delivery.CandidateModule{"lemurs": {FP2: fpNew, Manifest: manifest}}}} + raw, err := delivery.EncodeCandidate(delivery.Candidate{Group: "novox/x#7", HeadsHash: heads, Pulls: pulls, + FPSet: delivery.FingerprintSet(pulls)}) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.CompositionKey("novox/x#7", heads) + b.kv[delivery.CandidatesBucket+"/"+key] = raw + return key + } + accepted, refusedKey := candidateOf(digest2), "" + { + // The same change, its manifest refused by the indexer's judge. + refusedMan := strings.Replace(lemurs2, `"version":"2"`, `"version":"3"`, 1) + d3 := delivery.ManifestDigest([]byte(refusedMan)) + man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: refusedMan}) + key, _ := delivery.ManifestsKey(d3) + b.kv[delivery.ManifestsBucket+"/"+key] = man + vkey, _ := delivery.VerdictsKey(d3) + v, err := delivery.EncodeManifestVerdict(delivery.ManifestVerdict{Module: "lemurs", State: delivery.VerdictRefused, + Problems: []string{"a field the judge refuses"}, Judge: "src2:j", At: time.Now()}) + if err != nil { + t.Fatal(err) + } + b.kv[delivery.VerdictsBucket+"/"+vkey] = v + refusedKey = candidateOf(d3) + } + + env := b.env() + state := &fakeComposed{kv: map[string]ComposedEntry{}} + env.ComposedState = state + var applied atomic.Int64 + env.Refused = func(string, map[string]string) { applied.Add(1) } + var handed atomic.Int64 + s := NewShadow(env, func(context.Context, map[string]delivery.CandidateAnswer) { handed.Add(1) }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = s.Run(ctx) }() + if a := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted}); !strings.Contains(a.Refused, "not read every input") { + t.Errorf("a shadow not ready answered %+v", a) + } + place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}, + Modules: []delivery.PlacedModule{{Module: "lemurs"}}}) + have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}) + mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}) + deadline := time.Now().Add(5 * time.Second) + watching := func() int { b.mu.Lock(); defer b.mu.Unlock(); return len(b.watches) } + for watching() < 6 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + b.put(t, delivery.PlacementBucket, "placement.n1", place) + b.put(t, delivery.HaveBucket, "n1.node", have) + b.put(t, delivery.ModulesBucket, "lemurs", mod) + for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} { + b.send(bucket, agent.Update{Caught: true}) + } + for !s.ready.Load() && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + + a := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted}) + if _, err := delivery.EncodeComposeCandidateAnswer(a); err != nil { + t.Fatalf("the answer does not fit the seat's shape: %v (%+v)", err, a) + } + if a.Verdict != delivery.ComposeCandidateComposed || len(a.Modules) != 1 || a.Modules[0].Module != "lemurs" || + a.Modules[0].State != delivery.ModuleComposedState || a.Modules[0].FP != fpNew || !a.Modules[0].Changed || + a.Own.Module != delivery.HaveNodeWord || a.Took > delivery.ComposeAnswerBound.Milliseconds() { + t.Errorf("an accepted candidate answered %+v", a) + } + r := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: refusedKey}) + if r.Verdict != delivery.ComposeCandidateRefused || len(r.Modules) != 1 || r.Modules[0].State != delivery.ModuleRefusedState || + !strings.Contains(r.Modules[0].Words, "a field the judge refuses") { + t.Errorf("a candidate whose manifest the indexer refused answered %+v", r) + } + if _, err := delivery.EncodeComposeCandidateAnswer(r); err != nil { + t.Errorf("the refused answer does not fit the seat's shape: %v", err) + } + missing, _ := delivery.CompositionKey("novox/x#8", "0123456789ab") + if m := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: missing}); !strings.Contains(m.Refused, "holds no candidate") { + t.Errorf("a candidate the indexer does not hold answered %+v", m) + } + ended, end := context.WithCancel(ctx) + end() + if e := s.ComposeCandidate(ended, delivery.ComposeCandidateAsk{Candidate: accepted}); e.Verdict != delivery.ComposeCandidateRefused { + t.Errorf("an ended bound answered %+v", e) + } + if len(state.written()) != 0 || applied.Load() != 0 || handed.Load() != 0 { + t.Errorf("composing candidates wrote %v, marked %d have(s), handed %d answer(s); want nothing", state.written(), applied.Load(), handed.Load()) + } +} diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index 67804d4..4feed7c 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -7,6 +7,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -54,6 +55,8 @@ type Shadow struct { // after a bound (laterAgain), whether or not an input moves (issue 1126, 2026-10-28: ace kept "the vault's own: not // yet ... no responders" in its answer to seed-7). owed bool + // ready says this run read every input once: the verb compose-candidate composes only then (ComposeCandidate). + ready atomic.Bool } // answeredAt is what one proposal's answer was composed at: its keys' revisions, by node, and the changes of each input. @@ -146,6 +149,8 @@ func (s *Shadow) Run(ctx context.Context) error { s.retry = map[string]bool{} s.c = New(s.c.env) s.mu.Unlock() + s.ready.Store(false) + defer s.ready.Store(false) c := s.c node := c.env.Node // In inputWords' order. @@ -238,6 +243,7 @@ func (s *Shadow) Run(ctx context.Context) error { if ready = all(caught); !ready { continue } + s.ready.Store(true) dirty = true } if dirty {