diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/lazyturns.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/lazyturns.go new file mode 100644 index 0000000..4d84c22 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/lazyturns.go @@ -0,0 +1,50 @@ +package main + +import ( + "sync" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-checker/internal/check" +) + +// lazyTurns is the gate's merge turns as the checker reads them (novox/hq ADR 0548 §3), opened when the gate's bucket is +// there: before each claim it is opened again until it is, so a checker that started before the gate made the bucket holds +// checks from the gate's first turn on, with no timer and no restart. Until then it holds nothing, and says so once, loudly, +// and once again when it opens. +type lazyTurns struct { + turns *check.Turns + open func() error + logf func(format string, args ...any) + + mu sync.Mutex + opened bool + said bool +} + +// ensure opens the bucket once it is there; a failure is said once. +func (l *lazyTurns) ensure() { + l.mu.Lock() + defer l.mu.Unlock() + if l.opened || l.open == nil { + return + } + if err := l.open(); err != nil { + if !l.said { + l.said = true + l.logf("NO MERGE TURN HOLDS A CHECK: the gate's bucket %s cannot be read (%v); every check is claimed as if no approved merge waited, "+ + "and the bucket is opened again before each claim until it is there", delivery.TurnsBucket, err) + } + return + } + l.opened = true + if l.said { + l.logf("the gate's merge turns are read from %s from now on: an approved merge's turn holds checks off its repositories again", delivery.TurnsBucket) + } +} + +// Holding opens the bucket when it is not yet, then answers what the turns hold. +func (l *lazyTurns) Holding(repositories []string, heads string) (string, delivery.TurnWait, bool) { + l.ensure() + return l.turns.Holding(repositories, heads) +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index a8c0ddc..851b2d5 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -194,6 +194,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro presence := &buildrun.PresenceSet{} trunks := &check.Trunks{} turns := &check.Turns{} + lazy := &lazyTurns{turns: turns, logf: logger.Printf} // A check is one ask per pull request and build module on the build seats' check stream, pulled by a build node with a // free slot (novox/hq ADR 0525 §1, §3): the checker chooses no node and starts no step. asks := buildrun.JetStreamAsks{JS: js} @@ -201,7 +202,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]}, Owed: kvStore{buckets[delivery.CheckerOwedBucket]}, Trunks: trunks, - Turns: turns, + Turns: lazy, Bound: checkBound, Link: link, Logf: logger.Printf, @@ -238,6 +239,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } } repos := &repoModules{manifests: kvStore{buckets[delivery.ManifestsBucket]}, logf: logger.Printf} + ch.Runs = repos // the node-engine a check reads beside at its running commit (ADR 0548 §6) ch.Judge = newJudge(judgeDeps{ asks: buildrun.Asker{Publisher: asks, Waiter: asks, Bound: stepBound}, tools: tools, trunks: trunks, kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]}, @@ -317,28 +319,40 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // The gate's merge turns (novox/hq ADR 0548 §3): no check is claimed of a repository an approved merge waits on, and the // delete of its turn claims it. Read before the candidates, so no check starts past a turn written before this start. - // Where the gate's bucket is not there yet it is read as empty: no turn holds a check until the checker starts again. - if kv := process.OptionalBucket(ctx, js, delivery.TurnsBucket, "no merge turn holds a check until the checker starts again", logger.Printf); kv != nil { + // Where the gate's bucket is not there yet it is read as empty, and opened again before each claim until it is there + // (lazyTurns), so a gate delivered after the checker started holds checks from its first turn on. + lazy.open = func() error { + kv, err := js.KeyValue(ctx, delivery.TurnsBucket) + if err != nil { + return err + } tw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { turns.Put(e.Key(), e.Value(), !opOf(e)) }) if err != nil { - process.NotWatched(logger.Printf, delivery.TurnsBucket, err, "no merge turn holds a check until the checker starts again") - } else { - go func() { - errs <- follow(ctx, tw, func(e jetstream.KeyValueEntry) { - if turns.Put(e.Key(), e.Value(), !opOf(e)) { - ch.Reconsider(ctx) // a turn went or changed: a check it held is claimed now - } - }) - }() + return err } + go func() { + errs <- follow(ctx, tw, func(e jetstream.KeyValueEntry) { + if turns.Put(e.Key(), e.Value(), !opOf(e)) { + ch.Reconsider(ctx) // a turn went or changed: a check it held is claimed now + } + }) + }() + return nil } + lazy.ensure() // The modules say which repository holds a TypeScript module, so the Node build module judges it too. mw, err := settle(ctx, buckets[delivery.ModulesBucket], func(e jetstream.KeyValueEntry) { repos.Put(ctx, e.Key(), e.Value(), !opOf(e)) }) if err != nil { return err } go func() { - errs <- follow(ctx, mw, func(e jetstream.KeyValueEntry) { repos.Put(ctx, e.Key(), e.Value(), !opOf(e)) }) + errs <- follow(ctx, mw, func(e jetstream.KeyValueEntry) { + before := repos.CommitOf(check.NodeEngineModule) + repos.Put(ctx, e.Key(), e.Value(), !opOf(e)) + if after := repos.CommitOf(check.NodeEngineModule); after != before { + ch.TrunkMoved(ctx) // the node-engine the mesh runs moved: a verdict that read it beside is carried (ADR 0548 §6) + } + }) }() each := func(e jetstream.KeyValueEntry) { if opOf(e) { diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index b765475..271e3b0 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026102507-f5c233a9288c + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026111253-9c61918ac36f github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/go.sum b/modules/mesh-delivery-checker/go.sum index 7095bff..346bdcc 100644 --- a/modules/mesh-delivery-checker/go.sum +++ b/modules/mesh-delivery-checker/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026102507-f5c233a9288c h1:nciBayWNL5wOUqqgTIK5XxQdd2R+Ce4HuHU2Fwv2AVo= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026102507-f5c233a9288c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026111253-9c61918ac36f h1:9+TZpNq8Ifj1+qmbidE6l/7qkbua4yLdzRMrezNvf1A= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026111253-9c61918ac36f/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/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-checker/internal/check/adr548_test.go b/modules/mesh-delivery-checker/internal/check/adr548_test.go index 7f2ff65..87d7d88 100644 --- a/modules/mesh-delivery-checker/internal/check/adr548_test.go +++ b/modules/mesh-delivery-checker/internal/check/adr548_test.go @@ -264,6 +264,57 @@ func TestOneCheckPerDelivery(t *testing.T) { t.Errorf("the pending line of a carry = %q", pending) } }, + // §3 (review round 1): two approved merges in one turn whose trunk moved each get their carry; neither holds the other's. + "two merges waiting in one turn each get their carry": func(t *testing.T) { + var r *rig + j := &judgeFunc{f: func(context.Context, *Case, int) (Outcome, error) { return Outcome{Verdict: passing()}, nil }} + r = newRig(t, j) + candA, candC := candidateOf(t, "feat/a", pullOf(repoA, 7, head1)), candidateOf(t, "feat/c", pullOf(repoA, 8, head2)) + r.ch.Candidate(ctx, "feat%2Fa", candA) + r.ch.Candidate(ctx, "feat%2Fc", candC) + r.ch.Wait() + a, _ := delivery.DecodeCandidate(candA) + c, _ := delivery.DecodeCandidate(candC) + at := time.Date(2026, 10, 26, 9, 0, 0, 0, time.UTC) + turn, err := delivery.EncodeTurn(delivery.Turn{Repository: repoA, Waiting: []delivery.TurnWait{ + {Group: "feat/a", Heads: GateHeads(a), Approved: at, Since: at}, {Group: "feat/c", Heads: GateHeads(c), Approved: at.Add(time.Minute), Since: at}}}) + if err != nil { + t.Fatal(err) + } + turns := &Turns{} + turns.Put(turnKey(t, repoA), turn, false) + r.ch.Turns = turns + r.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS}) // a third merge moves the trunk under both + r.ch.TrunkMoved(ctx) + r.ch.Wait() + for _, cand := range [][]byte{candA, candC} { + group := map[bool]string{true: "feat/a", false: "feat/c"}[string(cand) == string(candA)] + if v := r.verdict(t, keyOf(t, group, cand)); v.Trunk[repoA] != trunkA2 || v.CarriedFrom[repoA] != trunkA { + t.Errorf("%s: verdict on %v carried from %v; want its carry claimed and done", group, v.Trunk, v.CarriedFrom) + } + } + }, + // §6 (review round 1): the node-engine read at the commit the mesh runs is named, and a new running commit is carried. + "a new running node-engine is carried": func(t *testing.T) { + runs := runsAt{nodeEngineModule: trunkS} + j := &judgeFunc{f: func(_ context.Context, c *Case, _ int) (Outcome, error) { + return Outcome{Verdict: passing(), Beside: map[string]string{nodeEngineRepo: runs[nodeEngineModule]}}, nil + }} + r := newRig(t, j) + r.ch.Runs = runs + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) + r.ch.Candidate(ctx, "feat%2Fx", cand) + r.ch.Wait() + r.ch.TrunkMoved(ctx) // nothing moved: nothing is carried + r.ch.Wait() + runs[nodeEngineModule] = trunkS2 + r.ch.TrunkMoved(ctx) // what the indexer's watch does when mesh-host's running commit changes + r.ch.Wait() + v := r.verdict(t, keyOf(t, "feat/x", cand)) + if j.count() != 2 || v.CarriedFrom[nodeEngineRepo] != trunkS || v.Beside[nodeEngineRepo] != trunkS2 { + t.Errorf("judged %d times, verdict beside %v carried from %v; want one carry over the node-engine's new running commit", j.count(), v.Beside, v.CarriedFrom) + } + }, // §5: a carry that fails names the move it failed on. "a carry that fails names the move": func(t *testing.T) { j := &judgeFunc{f: func(_ context.Context, c *Case, call int) (Outcome, error) { diff --git a/modules/mesh-delivery-checker/internal/check/beside.go b/modules/mesh-delivery-checker/internal/check/beside.go index 042d178..fd5ab99 100644 --- a/modules/mesh-delivery-checker/internal/check/beside.go +++ b/modules/mesh-delivery-checker/internal/check/beside.go @@ -16,8 +16,14 @@ type BesideRef struct { // carried (novox/hq ADR 0548 §6). Never sent: the node-engine at the commit the mesh runs, mesh-sdk at a pin and a // group's member at its head do not move with a trunk. atTrunk bool + // runs says the sibling was read at the commit the mesh runs (the node-engine, the indexer's record of mesh-host): a new + // running commit puts the verdict out of date and is carried (ADR 0548 §6). Never sent. + runs bool } +// NodeEngineModule is the module whose running commit the node-engine sibling is read at. +const NodeEngineModule = nodeEngineModule + const ( catalogueRepository = "novox/mesh-catalog" nodeEngineRepo = "novox/mesh-host" @@ -62,11 +68,14 @@ func (j *Examiner) besideFor(ctx context.Context, c *Case, p Pull) ([]BesideRef, if j.Runs != nil { commit = j.Runs.CommitOf(nodeEngineModule) } + atTrunk := false if commit == "" { - commit = heads[repo] + commit, atTrunk = heads[repo], true } if commit != "" { - out = append(out, BesideRef{Repository: repo, Commit: commit}) + // Named in the verdict either way (ADR 0548 §6, issue 568's own example): at the commit the mesh runs, a new + // running commit puts the verdict out of date; at the trunk, a trunk move does (Checker.besideNow). + out = append(out, BesideRef{Repository: repo, Commit: commit, atTrunk: atTrunk, runs: !atTrunk}) } case repo == SDKRepository && p.Repository == controllerRepo && j.Forge != nil: text, err := j.Forge.File(ctx, p.Repository, "go.mod", p.Head) @@ -117,7 +126,7 @@ func refOf(number int) string { // verdict names as Beside (ADR 0548 §6). func besideAtTrunk(c *Case, refs []BesideRef, into map[string]string) { for _, r := range refs { - if _, merged := c.Trunk[r.Repository]; r.atTrunk && !merged { + if _, merged := c.Trunk[r.Repository]; (r.atTrunk || r.runs) && !merged { into[r.Repository] = r.Commit } } diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index 3cfc970..e168211 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -69,6 +69,9 @@ type Checker struct { Owed KV Judge Judge Trunks interface{ Heads() map[string]string } + // Runs says which commit the mesh runs of a module (the indexer's modules): the node-engine a verdict read beside at its + // running commit is out of date when that commit moves (ADR 0548 §6); nil reads every sibling at its trunk. + Runs interface{ CommitOf(module string) string } // Turns are the gate's merge turns (ADR 0548 §3): no check is claimed of a repository a turn names, but the check of the // waiting delivery's own heads; nil holds none. Turns interface { @@ -263,13 +266,28 @@ func (ch *Checker) prepare(ctx context.Context, cs *Case) { // stale says whether a verdict was judged on commits that are no longer the case's: a trunk it merged moved, or a sibling it // read at its trunk's head (ADR 0548 §6). func (ch *Checker) stale(v delivery.Verdict, cs *Case) bool { - return !sameTrunk(v.Trunk, cs.Trunk) || besideMoved(v.Beside, ch.heads()) + return !sameTrunk(v.Trunk, cs.Trunk) || besideMoved(v.Beside, ch.besideNow()) } -// besideMoved says whether the trunk of a sibling read at these commits moved: its head now is known and another. -func besideMoved(beside, heads map[string]string) bool { +// besideNow are the commits a sibling is read at now, by repository: its trunk head, and for the node-engine the commit the +// mesh runs when the indexer names one, as besideFor places them. +func (ch *Checker) besideNow() map[string]string { + now := ch.heads() + if ch.Runs != nil { + if c := ch.Runs.CommitOf(nodeEngineModule); c != "" { + if now == nil { + now = map[string]string{} + } + now[nodeEngineRepo] = c + } + } + return now +} + +// besideMoved says whether a sibling read at these commits moved: the commit it is read at now is known and another. +func besideMoved(beside, now map[string]string) bool { for r, c := range beside { - if h := heads[r]; h != "" && h != c { + if h := now[r]; h != "" && h != c { return true } } @@ -522,7 +540,7 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) { ch.nodesFail[cs.Key] = epoch ch.mu.Unlock() } - if ch.movedSince(cs) || besideMoved(v.Beside, ch.heads()) { + if ch.movedSince(cs) || besideMoved(v.Beside, ch.besideNow()) { // Judged on a trunk that moved while it ran: the verdict is kept as judged, and the status stays pending until the // check of the new trunk, which Reconsider starts, ends. ch.owePending(ctx, cs, v) diff --git a/modules/mesh-delivery-checker/internal/check/pins_test.go b/modules/mesh-delivery-checker/internal/check/pins_test.go index a2ddebc..727903d 100644 --- a/modules/mesh-delivery-checker/internal/check/pins_test.go +++ b/modules/mesh-delivery-checker/internal/check/pins_test.go @@ -240,7 +240,12 @@ func TestTheSiblingsAreWhereTheTestsStateThem(t *testing.T) { // review D2, ADR catalogue + " go.mod " + head2: "module git.novox.be/novox/mesh-catalog\n"}, onTrunk: map[string]bool{"cb54756c3865": true}} b.ex.Forge = forge // A group of the controller and the catalogue: the controller's tests read the catalogue's trunk, not the group's head. - b.judge(t, caseOf(t, pullOf(controller, 7, head1), pullOf(catalogue, 12, head2))) + out := b.judge(t, caseOf(t, pullOf(controller, 7, head1), pullOf(catalogue, 12, head2))) + // ADR 0548 §6, issue 568's own example: the verdict names the node-engine at the commit the mesh runs, and mesh-sdk at the + // trunk the catalogue's tests read it at; the catalogue, a member here, is not named. + if out.Beside[host] != hostRuns || out.Beside[catalogue] != "" || out.Beside[SDKRepository] != "9999999999999999999999999999999999999999" { + t.Errorf("the siblings named for the verdict = %v, want the node-engine at %s and mesh-sdk at its trunk", out.Beside, hostRuns[:12]) + } got := besideOf(t, b, controller) if c := got[catalogue]; c.Commit != trunkS || c.Ref != "" { t.Errorf("the catalogue beside the controller = %+v, want its trunk %s", c, trunkS)