checker: a check is never held by a turn its own heads wait in, the node-engine read at its running commit is named and carried when it moves, and a gate bucket made after the start is opened before the next claim (ADR 0548 §3, §6; issue 568, review round 1)
mesh/delivery merged: source.main moved to novox/mesh-delivery@848c6fdc5648
mesh/delivery-group group feat/548-one-check merged: every member is merged
mesh/check failed: ace: the check of novox/mesh-delivery#65 refused: the installed go version could not be read: `go version` with GOTOOLCHAIN=local f…
mesh/delivery merged: source.main moved to novox/mesh-delivery@848c6fdc5648
mesh/delivery-group group feat/548-one-check merged: every member is merged
mesh/check failed: ace: the check of novox/mesh-delivery#65 refused: the installed go version could not be read: `go version` with GOTOOLCHAIN=local f…
This commit is contained in:
@@ -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)
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user