package selfcompose import ( "context" "encoding/json" "errors" "fmt" "maps" "slices" "sort" "sync" "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 check's composition key (`.`), it reads the group's candidate from the indexer // (delivery.CandidatesBucket, keyed by the group alone, delivery.CandidatesKey) at those heads, and 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") // candidateWithin is the node's own bound on an answer to compose-candidate: shorter than the checker's // (delivery.ComposeAnswerBound), because the bus lets the engine answer an ask only within a minute of it (the grants // part's allow_responses ttl, mesh-grants ResponseTTL): an answer at the checker's 60 s bound was refused by the bus // (issue 1126, 2026-11-03 01:17:04 on shanks, "Permissions Violation for Publish to _INBOX..."), so the checker heard // nothing at all. A node not done by then answers unready, naming what it waits on. var candidateWithin = delivery.ComposeAnswerBound - 10*time.Second // warmAtOnce is how many of the indexer's manifests the shadow reads at once before it composes a candidate. const warmAtOnce = 8 // candidateRun is what one ask's compositions wait on: each scratch composer it made, the last the one composing now. type candidateRun struct { mu sync.Mutex parts []*Composer } func (r *candidateRun) add(c *Composer) *Composer { r.mu.Lock() defer r.mu.Unlock() r.parts = append(r.parts, c) return c } // said is the run's waits in words: how many reads it waited on the bus or the vault for, and the longest; and, while // one is still running (now), what it waits on and since when. func (r *candidateRun) said(now time.Time) (waits string, on string) { r.mu.Lock() parts := slices.Clone(r.parts) r.mu.Unlock() n, longest, longestOn := 0, time.Duration(0), "" for _, c := range parts { c.doingMu.Lock() n += c.waits.n if c.waits.longest > longest { longest, longestOn = c.waits.longest, c.waits.longestOn } if c.doing != "" { on = fmt.Sprintf("%s (for %s)", c.doing, now.Sub(c.doingSince).Round(time.Millisecond)) } else if on == "" && c.waits.lastOn != "" { on = c.waits.lastOn + " (the last read)" } c.doingMu.Unlock() } waits = fmt.Sprintf("%d read(s) waited on", n) if longestOn != "" { waits += fmt.Sprintf(", the longest %s on %s", longest.Round(time.Millisecond), longestOn) } if on == "" { on = "nothing it reads (composing)" } return waits, on } // ComposeCandidate is the node's answer to ask, within candidateWithin (or ctx's deadline, if sooner). Each answer is // said in one line, with what it waited on and for how long (issue 1126: ace, g14 and shanks answered nothing for a // minute while nothing said why). An answer not composed within the bound is unready, naming what the node waits on. func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandidateAsk) delivery.ComposeCandidateAnswer { s.mu.Lock() c := s.c s.mu.Unlock() began := c.now() ctx, cancel := context.WithTimeout(ctx, candidateWithin) defer cancel() run := &candidateRun{} done := make(chan delivery.ComposeCandidateAnswer, 1) go func() { done <- s.composeCandidate(ctx, c, ask, began, run) }() var a delivery.ComposeCandidateAnswer select { case a = <-done: case <-ctx.Done(): _, on := run.said(c.now()) a = delivery.ComposeCandidateAnswer{Node: c.env.Node, Candidate: ask.Candidate, Verdict: delivery.ComposeCandidateUnready, Refused: fmt.Sprintf("the node did not compose the candidate within %s: it waits on %s", candidateWithin, on), Took: c.now().Sub(began).Milliseconds()} } waits, _ := run.said(c.now()) words := "" if a.Refused != "" { words = ": " + a.Refused } c.env.Say(fmt.Sprintf("compose-candidate %s answered %s in %s (%s)%s", ask.Candidate, a.Verdict, (time.Duration(a.Took) * time.Millisecond).String(), waits, words)) return a } // composeCandidate composes ask on scratch copies of c, each added to run as it starts. func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask delivery.ComposeCandidateAsk, began time.Time, run *candidateRun) delivery.ComposeCandidateAnswer { 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 = c.now().Sub(began).Milliseconds() return a } // A node that cannot compose now (its shadow not ready, a read the bus did not answer, the bound passed) answers // unready: the checker takes it as no answer (ADR 0593 §9) by the verdict, the words only saying why. unready := func(why string) delivery.ComposeCandidateAnswer { a = refuse(why) a.Verdict = delivery.ComposeCandidateUnready return a } group, heads, err := delivery.ParseCompositionKey(ask.Candidate) if err != nil { return refuse(err.Error()) } // The ask names the check (the composition key, `.`); the indexer keys the group's candidate by // the group alone (delivery.CandidatesKey) and holds its heads hash in the value. ckey, err := delivery.CandidatesKey(group) if err != nil { return refuse(err.Error()) } if !s.ready.Load() { return unready(errNotReady.Error()) } base := run.add(c.scratch()) raw, err := base.get(ctx, delivery.CandidatesBucket, ckey) if err != nil { return unready(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)) } if cand.HeadsHash != heads { // The group moved to other heads: the check asked is replaced, and this node composes not a candidate it was // not asked about. No answer (ADR 0593 §9), never a verdict on the wrong heads. return unready(fmt.Sprintf("the indexer's candidate of the group %s is at heads %s, not %s: the check asked is replaced", group, cand.HeadsHash, heads)) } // Every module's manifest is read before either composition, several at once, into the shadow's own: a composition // reads every manifest in the mesh one by one, which off the control-node took about 35 s each (issue 1126). base.warm(ctx, c, cand) var running []delivery.Composed if c.env.Running != nil { running = c.env.Running() } baseComps, baseHeld, baseErr := base.composeShadow(ctx, running, nil, nil) // The candidate laid over a copy taken after the composition without it, so what that read is not read again. laid := run.add(c.scratch()) touched, err := laid.layCandidate(ctx, cand) if err != nil { return unready(err.Error()) } 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 { _, on := run.said(c.now()) return unready(fmt.Sprintf("the node did not compose the candidate within %s: it waits on %s", candidateWithin, on)) } switch { case laidErr != nil && BusFailed(laidErr): return unready("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 = c.now().Sub(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 = c.now().Sub(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 out.origin = c return out } // warm reads, several at once, each manifest the indexer's modules and the candidate name that the shadow (origin) does // not hold yet, keeping each in origin and in c; one not read is left to the composition, which reads it and says why. // Its reads are not noted as the bus failing: the composition that reads again notes its own. func (c *Composer) warm(ctx context.Context, origin *Composer, cand delivery.Candidate) { origin.mu.Lock() want := map[string]bool{} for _, m := range origin.modules { if _, ok := origin.manifests[m.Manifest]; m.Manifest != "" && !ok { want[m.Manifest] = true } } for _, pull := range cand.Pulls { for _, cm := range pull.Fps { if _, ok := origin.manifests[cm.Manifest]; cm.Manifest != "" && !ok { want[cm.Manifest] = true } } } origin.mu.Unlock() if len(want) == 0 { return } defer c.waiting(fmt.Sprintf("%d of the indexer's manifests (%s)", len(want), delivery.ManifestsBucket))() slots := make(chan struct{}, warmAtOnce) var wg sync.WaitGroup for digest := range want { key, err := delivery.ManifestsKey(digest) if err != nil { continue } wg.Add(1) go func() { defer wg.Done() select { case slots <- struct{}{}: case <-ctx.Done(): return } defer func() { <-slots }() value, err := c.env.Get(ctx, delivery.ManifestsBucket, key) if err != nil || value == nil { return } m, err := delivery.DecodeManifest(value) if err != nil { return } c.keepRead(func(k *Composer) { k.manifests[digest] = []byte(m.JSON) }) }() } wg.Wait() } // 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 }