diff --git a/cmd/mesh-host/engineseat.go b/cmd/mesh-host/engineseat.go index 443f0fc..f7c8b88 100644 --- a/cmd/mesh-host/engineseat.go +++ b/cmd/mesh-host/engineseat.go @@ -363,7 +363,8 @@ func packagesFrom(node string, session *link.Session, composer *atomic.Pointer[s } // 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 +// node in shadow by the run's shadow, applying and writing nothing, within the shadow's own bound (shorter than +// delivery.ComposeAnswerBound, selfcompose candidateWithin). The answer is // the seat's shape (delivery.ComposeCandidateAnswer), checked as it is encoded; a node whose shadow does not run yet // answers unready, which the checker takes as no answer. func composeCandidateFrom(node string, shadow *atomic.Pointer[selfcompose.Shadow]) func(candidate string, input bool) (any, error) { diff --git a/internal/composer/composer.go b/internal/composer/composer.go index d5bba7c..9204bc1 100644 --- a/internal/composer/composer.go +++ b/internal/composer/composer.go @@ -134,10 +134,28 @@ func runCase(r Request, c Case, depth string) (CaseVerdict, error) { if err != nil { return CaseVerdict{}, fmt.Errorf("composer: the case %q: %w", c.Name, err) } + // The module the mesh provides itself (the private network) has no repository the indexer reads, so no case carries + // it: its manifest comes with mesh-sdk and always wins, as the bus reader has it (frombus readBase). Without it a node + // it is assigned to composed nothing (issue 1126, check of mesh-host #200: "nothing provides mesh-wireguard"). + for _, name := range delivery.ProvidedModules() { + m, _, err := delivery.ProvidedManifest(name) + if err != nil { + return CaseVerdict{}, fmt.Errorf("composer: %w", err) + } + shelf[name] = m + } + // A module the input carries as its node-engine took it in from its build (SentInput) is composed so: the bus + // reader composes it from its build, never from the indexer's manifest a case gives. + for name, m := range base.Catalogue { + shelf[name] = m + } in.Catalogue = shelf in.Nodes = slices.Clone(in.Nodes) in.Holdings = slices.Clone(in.Holdings) for i := range in.Nodes { + if len(c.Kept[in.Nodes[i].Name]) == 0 { + continue // what the input keeps of this node, as its engine took it in, stands + } in.Nodes[i].Kept = nil if kept := c.Kept[in.Nodes[i].Name]; len(kept) > 0 { k, err := parse(kept, nil, nil) @@ -234,3 +252,38 @@ const Package = "github.com/novox/mesh-host/cmd/mesh-compose" // Decode reads a request (sdk.Decode); it refuses one that is not JSON or of another version. func Decode(data []byte) (Request, error) { return sdk.Decode(data) } + +// SentInput is the composition input a node-engine answers for the sandbox (delivery.ComposeCandidateAsk.Input), from the +// input it composed node from: the one shape both read (issue 1126, checks of mesh-host #200). Every other node's modules +// are left to the cases (the indexer's manifests, as the bus reader reads them too); node's own placed modules stay as the +// bus reader took them in from their builds (their artifacts, the texts of their templates archive, what is handed to the +// provisioner), with what node keeps at another build, which no case can give: the sandbox has no build or archive to +// read. Without them the sandbox composed another node than the engine (restic's templates and artifacts unread, postgres's +// secret reader handed and the module not). +func SentInput(in compose.Input, node string) compose.Input { + own := map[string]bool{} + for _, n := range in.Nodes { + if n.Name == node { + for _, m := range n.Assigned { + own[m] = true + } + } + } + taken := in.Catalogue + in.Catalogue = nil + for name, m := range taken { + if own[name] { + if in.Catalogue == nil { + in.Catalogue = map[string]catalogue.Manifest{} + } + in.Catalogue[name] = m + } + } + in.Nodes = slices.Clone(in.Nodes) + for i := range in.Nodes { + if in.Nodes[i].Name != node { + in.Nodes[i].Kept = nil + } + } + return in +} diff --git a/internal/composer/composer_test.go b/internal/composer/composer_test.go index dcce0ac..281c310 100644 --- a/internal/composer/composer_test.go +++ b/internal/composer/composer_test.go @@ -1,14 +1,18 @@ package composer import ( + "context" "encoding/json" "os" + "reflect" "strings" "testing" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/novox/mesh-host/internal/compose" + "github.com/novox/mesh-host/internal/compose/frombus" + "github.com/novox/mesh-host/internal/compose/frombus/frombustest" ) const demo = `{"module": "demo", "version": "1", "claims": [{"name": "node-files"}]}` @@ -60,6 +64,50 @@ func TestARequestItCannotRunIsRefused(t *testing.T) { if _, err := Run(placed); err == nil || !strings.Contains(err.Error(), "holds no node for") { t.Errorf("a case placing a node the input lacks: %v", err) } + // Issue 1126 (check of mesh-host #200): the module the mesh provides itself (the private network) comes from mesh-sdk, + // a second source beside the indexer's manifests the cases carry, and the input carries no catalogue: a node it is + // assigned to still composes, and what it provides is provided. + wg := delivery.ProvidedModules()[0] + provided := withInput(t, Request{V: Version, Machines: []string{"a"}, Depth: DepthResolve, + Cases: []Case{{Name: "change", Manifests: map[string]Manifest{"demo": {Raw: json.RawMessage(demo)}}}}}, + compose.Input{Nodes: []compose.Node{{Name: "a", Assigned: []string{wg, "demo"}}}}) + if v, err := Run(provided); err != nil || !v.Cases[0].Machines["a"].Composes { + t.Errorf("a node assigned the mesh's own module: %+v, %v", v, err) + } + // Issue 1126 (checks of mesh-host #200, fix-a-bug-met-twice): the sandbox composes one node from the same inputs as + // the bus reader, through one path. A mesh whose restic is built (its files from an archive) and handed to the + // provisioner composes n1 through frombus, and the request a node-engine answers for it (the input as the engine + // sends it, the indexer's manifests in the case) declares the same paths for n1. + t.Run("the sandbox composes a node as the bus reader does", func(t *testing.T) { + b, err := frombustest.Provisioned() + if err != nil { + t.Fatal(err) + } + ctx, scope := context.Background(), frombus.Scope{Node: frombustest.Node} + read, err := frombus.Input(ctx, b, frombus.Mesh{}, scope) + if err != nil { + t.Fatal(err) + } + d, problems, err := compose.Compose(read, frombustest.Node) + if err != nil || len(problems) > 0 { + t.Fatalf("the bus reader's input composes n1: %v %v", err, problems) + } + want := pathsOf(d) + sent := SentInput(read, frombustest.Node) + cases := Case{Name: "change", Manifests: map[string]Manifest{}} + for name, m := range b.Module { + cases.Manifests[name] = Manifest{Raw: b.Manifests[m.Manifest]} + } + r := withInput(t, Request{V: Version, Machines: []string{frombustest.Node}, Declarations: true, Cases: []Case{cases}}, sent) + v, err := Run(r) + if err != nil { + t.Fatal(err) + } + got := v.Cases[0].Machines[frombustest.Node] + if !got.Composes || !reflect.DeepEqual(got.Declared, want) { + t.Errorf("the sandbox composed n1 %+v; the bus reader declared %+v", got, want) + } + }) v, err := Run(Request{V: Version, Cases: []Case{{Name: "only-read", Manifests: map[string]Manifest{"demo": {Raw: json.RawMessage(demo)}}}}}) if err != nil || len(v.Cases) != 1 || !v.Cases[0].Manifests["demo"].Accepted || v.Cases[0].Machines != nil || v.SDK == "" { t.Fatalf("a request with no machines: %+v, %v", v, err) diff --git a/internal/link/engineseat.go b/internal/link/engineseat.go index 7261559..c51d0e9 100644 --- a/internal/link/engineseat.go +++ b/internal/link/engineseat.go @@ -116,11 +116,10 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) { if err != nil { return } + apart := make(chan struct{}, apartAtOnce) for _, v := range catalogue.EngineVerbs() { verb := v.Name - answer := micro.HandlerFunc(func(r micro.Request) { - _ = r.Respond(AnswerEngineCall(handle, verb, l.node, r.Subject(), r.Data())) - }) + answer := EngineAnswer(handle, verb, l.node, apart) meta := EngineEndpointMetadata(v, l.node) _ = svc.AddEndpoint(verb, answer, micro.WithEndpointSubject(EngineToolSubject(verb, l.node)), micro.WithEndpointMetadata(meta)) @@ -130,6 +129,40 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) { l.engine = svc } +// answeredApart are the verbs answered off the subscription that delivered them: each endpoint's subscription hands its +// asks over one at a time, so a second ask of a verb that takes long waited for the first, and its bound counted from +// when the first ended (issue 1126, review of 2026-11-03: two checks asking compose-candidate of one node at once, the +// second answered at about 100 s, past the bus's one-minute reply window). Each is answered in a goroutine of its own, +// its bound counted from its arrival. +var answeredApart = map[string]bool{ActComposeCandidate: true} + +// apartAtOnce bounds the asks answered apart at once on this node; one more is refused in words at once. +const apartAtOnce = 4 + +// EngineAnswer is the handler of one verb: answered on the subscription's goroutine, or, for a verb answered apart, in +// its own while fewer than apart's capacity run, and refused at once past it. +func EngineAnswer(handle EngineFunc, verb, node string, apart chan struct{}) micro.HandlerFunc { + return func(r micro.Request) { + if !answeredApart[verb] { + _ = r.Respond(AnswerEngineCall(handle, verb, node, r.Subject(), r.Data())) + return + } + select { + case apart <- struct{}{}: + default: + raw, _ := json.Marshal(map[string]any{"node": node, "error": fmt.Sprintf( + "this node answers %d asks of %s at once now: ask again when one is answered", cap(apart), verb)}) + _ = r.Respond(raw) + return + } + subject, data := r.Subject(), r.Data() + go func() { + defer func() { <-apart }() + _ = r.Respond(AnswerEngineCall(handle, verb, node, subject, data)) + }() + } +} + // EngineModule is the module whose code answers the seat, as discovery names it: the node-engine's manifest's. const EngineModule = "mesh-host" diff --git a/internal/link/handover_test.go b/internal/link/handover_test.go index be1b363..2c46bfe 100644 --- a/internal/link/handover_test.go +++ b/internal/link/handover_test.go @@ -7,11 +7,13 @@ import ( "errors" "sort" "strings" + "sync" "testing" "time" stdio "git.novox.be/novox/mesh-sdk/go" "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/micro" ) // The hand-over's ask and answer hold these field names; the controller's side holds the same list @@ -265,8 +267,63 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) { t.Errorf("the judgements are recorded as %+v", got) } }) + // Issue 1126 (review of 2026-11-03): two asks of compose-candidate at once, handed over one at a time as an + // endpoint's subscription hands them, are both answered while the other runs: neither waits for the other, so each + // bound counts from its own arrival. A verb answered on the subscription answers in order, as before. + t.Run("two asks of compose-candidate at once are answered side by side", func(t *testing.T) { + var mu sync.Mutex + arrived, both := 0, make(chan struct{}) + handle := func(c EngineCall) (any, error) { + mu.Lock() + if arrived++; arrived == 2 { + close(both) + } + mu.Unlock() + select { + case <-both: + return "together", nil + case <-time.After(2 * time.Second): // only when the asks are answered one after the other + return "alone", nil + } + } + answered := make(chan string, 2) + answer := EngineAnswer(handle, ActComposeCandidate, "n1", make(chan struct{}, 4)) + for range 2 { + answer(fakeRequest{subject: EngineToolSubject(ActComposeCandidate, "n1"), answered: answered}) + } + for range 2 { + if got := <-answered; !strings.Contains(got, `"result":"together"`) { + t.Errorf("an ask answered %s; want both answered side by side", got) + } + } + full := make(chan struct{}, 1) + full <- struct{}{} + EngineAnswer(handle, ActComposeCandidate, "n1", full)(fakeRequest{subject: "s", answered: answered}) + if got := <-answered; !strings.Contains(got, "asks of compose-candidate at once") { + t.Errorf("an ask past the bound answered %s", got) + } + }) } +// fakeRequest is a seat call as the subscription hands it over, its answer sent to answered. +type fakeRequest struct { + subject string + answered chan string +} + +func (r fakeRequest) Respond(b []byte, _ ...micro.RespondOpt) error { + r.answered <- string(b) + return nil +} +func (r fakeRequest) RespondJSON(any, ...micro.RespondOpt) error { return nil } +func (r fakeRequest) Error(string, string, []byte, ...micro.RespondOpt) error { + return nil +} +func (r fakeRequest) Data() []byte { return []byte(`{}`) } +func (r fakeRequest) Headers() micro.Headers { return nil } +func (r fakeRequest) Subject() string { return r.subject } +func (r fakeRequest) Reply() string { return "_INBOX.x" } + // **Against a real server**: a signed request on the node's own subject is answered; one on another node's is // not heard; a message with no reply is answered by nobody. func TestNatsAHandOverIsAskedOfTheNode(t *testing.T) { diff --git a/internal/selfcompose/candidate.go b/internal/selfcompose/candidate.go index 6c6fe44..9d8b091 100644 --- a/internal/selfcompose/candidate.go +++ b/internal/selfcompose/candidate.go @@ -6,13 +6,16 @@ import ( "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" + "github.com/novox/mesh-host/internal/composer" ) // **A check's candidate, composed in shadow on a call** (novox/hq ADR 0593 §7): the node seat's verb compose-candidate. @@ -28,17 +31,102 @@ import ( // 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). +// 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 + +// maxSentInput is the most composition input an answer carries: the bus carries 1 MiB in one message (its max_payload), +// and the answer's own fields and envelope come beside it. +const maxSentInput = 900 << 10 + +// 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 { - began := time.Now() 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 = time.Since(began).Milliseconds() + 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 @@ -58,12 +146,10 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi if err != nil { return refuse(err.Error()) } - ctx, cancel := context.WithTimeout(ctx, delivery.ComposeAnswerBound) - defer cancel() if !s.ready.Load() { return unready(errNotReady.Error()) } - base := c.scratch() + 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)) @@ -80,17 +166,21 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi // 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)) } - laid := c.scratch() - touched, err := laid.layCandidate(ctx, cand) - if err != nil { - return unready(err.Error()) - } + // 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 @@ -99,17 +189,21 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi 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 { + // As the sandbox reads it (composer.SentInput, the one shape): this node's own modules as taken in from their + // builds, every other module's manifest left to the cases. + if body, err := json.Marshal(composer.SentInput(in, laid.env.Node)); err == nil { a.Input = body } }) + if len(a.Input) > maxSentInput { + // Past what the bus carries in one answer, the checker would hear nothing: refused, saying so. + size := len(a.Input) + a.Input = nil + return refuse(fmt.Sprintf("the composition input this node composed from is %d bytes, more than the %d an answer on the bus carries", size, maxSentInput)) + } if ctx.Err() != nil { - return unready(fmt.Sprintf("the bound of %s passed before the node composed the candidate", delivery.ComposeAnswerBound)) + _, 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): @@ -118,7 +212,7 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi // 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() + a.Took = c.now().Sub(began).Milliseconds() return a case laidErr != nil: return refuse(laidErr.Error()) @@ -193,7 +287,7 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi if refused { a.Verdict = delivery.ComposeCandidateRefused } - a.Took = time.Since(began).Milliseconds() + a.Took = c.now().Sub(began).Milliseconds() return a } @@ -214,9 +308,63 @@ func (c *Composer) scratch() *Composer { 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. diff --git a/internal/selfcompose/candidate_test.go b/internal/selfcompose/candidate_test.go index e935430..2efbfbb 100644 --- a/internal/selfcompose/candidate_test.go +++ b/internal/selfcompose/candidate_test.go @@ -132,4 +132,121 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) { 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()) } + // Issue 1126, 2026-11-03: off the control-node an ask of compose-candidate was not answered within a minute, and said + // nothing of why. Each of its two compositions read every manifest in the mesh one by one, on copies of the shadow + // that kept nothing, so every ask read them all twice again. Each manifest is now read once, before either + // composition, into the shadow, and a later ask reads none; an ask whose read does not answer is answered unready at + // its bound, naming what the node waits on; and every answer is said in one line with what it waited on. + t.Run("each manifest is read once, and the answer comes within its bound", func(t *testing.T) { + fp := delivery.SourcePrefix2 + strings.Repeat("a", 64) + fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64) + lemurs2 := strings.Replace(lemurs, `"version":"1"`, `"version":"2"`, 1) + other := strings.NewReplacer(`"module":"lemurs"`, `"module":"other"`, "lemurs.service", "other.service").Replace(lemurs) + digest, digest2, digestOther := delivery.ManifestDigest([]byte(lemurs)), delivery.ManifestDigest([]byte(lemurs2)), + delivery.ManifestDigest([]byte(other)) + otherKey, _ := delivery.ManifestsKey(digestOther) + + // shadow is a ready shadow of n1 over b: lemurs placed on it, other placed nowhere (its manifest read for the mesh's + // catalogue alone), and a candidate laying lemurs at a new build; its clock moves a second at each reading. + shadow := func(b *fakeBus) (*Shadow, string) { + for d, body := range map[string]string{digest: lemurs, digest2: lemurs2, digestOther: other} { + man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "x", JSON: body}) + key, _ := delivery.ManifestsKey(d) + b.kv[delivery.ManifestsBucket+"/"+key] = man + } + head := delivery.Head{Repository: "novox/x", Number: 7, Head: "c0ffee"} + heads, _ := delivery.HeadsHash([]delivery.Head{head}) + pulls := []delivery.CandidatePull{{Head: head, Modules: []string{"lemurs"}, + Fps: map[string]delivery.CandidateModule{"lemurs": {FP2: fpNew, Manifest: digest2}}}} + raw, _ := delivery.EncodeCandidate(delivery.Candidate{Group: "fix/1126", HeadsHash: heads, Pulls: pulls, + FPSet: delivery.FingerprintSet(pulls)}) + ckey, _ := delivery.CandidatesKey("fix/1126") + b.kv[delivery.CandidatesBucket+"/"+ckey] = raw + env := b.env() + var tick atomic.Int64 + env.Now = func() time.Time { return time.Unix(0, 0).Add(time.Duration(tick.Add(1)) * time.Second) } + s := NewShadow(env, nil) + s.c.placements["n1"] = delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}, + Modules: []delivery.PlacedModule{{Module: "lemurs"}}} + s.c.records["n1"] = delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}} + s.c.modules["lemurs"] = delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"} + s.c.modules["other"] = delivery.Module{FP2: fp, Manifest: digestOther, Repository: "novox/x", Commit: "c"} + s.ready.Store(true) + key, _ := delivery.CompositionKey("fix/1126", heads) + return s, key + } + readsOf := func(b *fakeBus, key string) (n int) { + b.mu.Lock() + defer b.mu.Unlock() + for _, r := range b.reads { + if r == [2]string{delivery.ManifestsBucket, key} { + n++ + } + } + return n + } + saidLast := func(b *fakeBus) string { + b.mu.Lock() + defer b.mu.Unlock() + if len(b.said) == 0 { + return "" + } + return b.said[len(b.said)-1] + } + + for _, tc := range []struct { + name string + // stuck: the read of other's manifest does not answer until the ask's bound ends. + stuck bool + asks int + verdict string + reads int // of other's manifest, over every ask + words string // in the answer's words + said string // in the line said of the last answer + }{ + {name: "one ask reads each manifest once", asks: 1, verdict: delivery.ComposeCandidateComposed, reads: 1, + said: "compose-candidate fix/1126."}, + {name: "a later ask reads none again", asks: 2, verdict: delivery.ComposeCandidateComposed, reads: 1, + said: "read(s) waited on"}, + {name: "a read that does not answer is named at the bound", stuck: true, asks: 1, verdict: delivery.ComposeCandidateUnready, + words: "waits on", said: "answered unready"}, + } { + t.Run(tc.name, func(t *testing.T) { + b := newFakeBus() + s, key := shadow(b) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if tc.stuck { + get := s.c.env.Get + s.c.env.Get = func(rctx context.Context, bucket, k string) ([]byte, error) { + if bucket == delivery.ManifestsBucket && k == otherKey { + b.mu.Lock() + b.reads = append(b.reads, [2]string{bucket, k}) + b.mu.Unlock() + cancel() // the ask's bound ends while the read waits + <-rctx.Done() + return nil, rctx.Err() + } + return get(rctx, bucket, k) + } + } + var a delivery.ComposeCandidateAnswer + for range tc.asks { + a = s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: key}) + } + if a.Verdict != tc.verdict || !strings.Contains(a.Refused, tc.words) { + t.Errorf("answered %s %q; want %s with %q", a.Verdict, a.Refused, tc.verdict, tc.words) + } + if tc.stuck && !strings.Contains(a.Refused, "manifest") { + t.Errorf("answered %q; want the manifest it waits on named", a.Refused) + } + if got := readsOf(b, otherKey); tc.reads > 0 && got != tc.reads { + t.Errorf("other's manifest was read %d time(s); want %d", got, tc.reads) + } + if line := saidLast(b); !strings.Contains(line, tc.said) { + t.Errorf("said %q; want a line with %q", line, tc.said) + } + }) + } + }) } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 41c4203..3019186 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -97,6 +97,8 @@ type Env struct { // ComposedState is this node's composition state `composed-` (novox/hq ADR 0593 §6), where every composition // a pass makes is published for its readers (publishComposed). Nil keeps none. ComposedState ComposedStore + // Now is the clock what the composer waits on is timed by; nil is time.Now. + Now func() time.Time } // VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value. @@ -141,10 +143,16 @@ type Composer struct { held []string // doing is what the composer waits on now (waiting), for a running answer to say (issue 1126, T2); its own lock, so // it is read while c.mu is held. - doingMu sync.Mutex - doing string - caught bool - rev uint64 + doingMu sync.Mutex + doing string + doingSince time.Time + // waits is what the composer waited on, under doingMu: for compose-candidate's line (issue 1126). + waits waitLog + // origin is the composer a scratch copy was taken of (scratch): what is read once by its digest or key and never + // changes (a manifest, a templates archive, a done build) is kept there too, so the next copy reads it no more. + origin *Composer + caught bool + rev uint64 // missed counts the reads the bus did not answer (BusFailed), for a pass or an answer to tell whether it met one: // a read that failed on the bus is tried again after a bound, never kept as an answer (issue 1126, 2026-10-28). missed atomic.Uint64 @@ -1060,19 +1068,56 @@ func (c *Composer) keepRunning(comp delivery.Composed, running []delivery.Compos return delivery.KeepSeals(comp, kept, c.sameValue) } -// waiting notes what the composer waits on until the returned func is called, which restores what it waited on before. +// waitLog is what a composer waited on: how many waits ended, the longest and what it was, and the last. +type waitLog struct { + n int + longest time.Duration + longestOn string + lastOn string +} + +// now is the composer's clock (Env.Now). +func (c *Composer) now() time.Time { + if c.env.Now != nil { + return c.env.Now() + } + return time.Now() +} + +// waiting notes what the composer waits on until the returned func is called, which restores what it waited on before +// and counts the wait, timed, in waits. func (c *Composer) waiting(what string) func() { + began := c.now() c.doingMu.Lock() - before := c.doing - c.doing = what + before, beforeSince := c.doing, c.doingSince + c.doing, c.doingSince = what, began c.doingMu.Unlock() return func() { + took := c.now().Sub(began) c.doingMu.Lock() - c.doing = before + c.doing, c.doingSince = before, beforeSince + c.waits.n++ + c.waits.lastOn = what + if took > c.waits.longest || c.waits.longestOn == "" { + c.waits.longest, c.waits.longestOn = took, what + } c.doingMu.Unlock() } } +// keepRead keeps, in c and in the composer c is a scratch copy of, what set writes: a read that never changes for its +// digest or key. Each under its own lock, one after the other. +func (c *Composer) keepRead(set func(*Composer)) { + c.mu.Lock() + set(c) + c.mu.Unlock() + if o := c.origin; o != nil { + o.mu.Lock() + set(o) + o.mu.Unlock() + } +} + // waitingOn is what the composer waits on now, in words. func (c *Composer) waitingOn() string { c.doingMu.Lock() @@ -1303,9 +1348,7 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) { if err != nil { return nil, err } - r.c.mu.Lock() - r.c.manifests[digest] = []byte(m.JSON) - r.c.mu.Unlock() + r.c.keepRead(func(k *Composer) { k.manifests[digest] = []byte(m.JSON) }) return []byte(m.JSON), nil } @@ -1330,9 +1373,7 @@ func (r reader) Build(ctx context.Context, module, fp string) (*delivery.Builder return nil, fmt.Errorf("the builder's %s: %w", key, err) } if b.State == delivery.BuildDone { - r.c.mu.Lock() - r.c.builds[key] = b - r.c.mu.Unlock() + r.c.keepRead(func(k *Composer) { k.builds[key] = b }) } return &b, nil } @@ -1363,9 +1404,7 @@ func (r reader) Archive(ctx context.Context, url, digest string) ([]byte, error) if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest { return nil, fmt.Errorf("%s was recorded as %s and what arrived is %s", url, digest, got) } - r.c.mu.Lock() - r.c.archives[digest] = body - r.c.mu.Unlock() + r.c.keepRead(func(k *Composer) { k.archives[digest] = body }) return body, nil }