Start a never-judged head's check before the re-judges a trunk move asks, oldest first (issue 1897 S1)
mesh/delivery delivered
mesh/delivery-group group feat/fastcheck delivered: every member is delivered
mesh/check checking 4ffe5f4d5243 on 85aca46348a0 | novox/mesh-delivery#25 in group feat/fastcheck

A new pull request waited 25 to 45 minutes behind every open pull request's re-judge after a
merge. The semaphore becomes a ranked admission (ADR 0536 §4); checker_status names the
waiting checks by kind.
This commit is contained in:
2026-10-23 14:09:42 +02:00
parent 2bdb8d0e00
commit 4ffe5f4d52
4 changed files with 382 additions and 17 deletions
@@ -145,7 +145,8 @@ func Tools(r check.Reader) []stdio.Tool {
{Name: check.StatusReadTool,
Description: "What holds the checker back and what runs: the capacity per node (the setting, the default one step per node), whether the controller's " +
"delivery plans in their build phase hold a key (no new check starts while one does), the steps in flight on the build modules by node, and the " +
"candidates without a verdict for their heads (a group of hq's pull requests alone is left out: merge-check.sh judges it). A key a build " +
"candidates without a verdict for their heads (a group of hq's pull requests alone is left out: merge-check.sh judges it), and the " +
"candidates whose check waits by kind (waiting_by_kind: new, nodes, and trunk, which start after the other two). A key a build " +
"module left running (started past the step bound, or its build module not running on its node) is listed apart under left_running " +
"with why, and not counted. The places the checker holds are in its process, not in a bucket, and the answer says so. Read-only.",
Input: map[string]any{"type": "object", "properties": map[string]any{}},
@@ -184,19 +185,21 @@ func main() {
Capacity: capacity, HaveState: read(stdio.State("node-engine.have")), StepBound: stepBound()}
tries := stdio.State(check.TriesState)
source := stdio.State("mesh-delivery-gate.source")
trunks := func() (map[string]string, error) {
e, err := source.Get(delivery.SourceMainKey)
if err != nil || e == nil {
return nil, err
}
s, err := delivery.DecodeSource(e.Value)
return s.Heads, err
}
r.Trunks = trunks
trying := Trying{
Tries: stateStore{StateRead: read(tries), create: func(key string, value []byte) error {
_, err := tries.Create(key, json.RawMessage(value))
return err
}},
Trunks: func() (map[string]string, error) {
e, err := source.Get(delivery.SourceMainKey)
if err != nil || e == nil {
return nil, err
}
s, err := delivery.DecodeSource(e.Value)
return s.Heads, err
},
Trunks: trunks,
}
if err := stdio.Serve("", append(Tools(r), TryTools(trying)...)); err != nil {
fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err)
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"slices"
"strings"
"sync"
"time"
@@ -68,7 +69,7 @@ type Checker struct {
Trunks interface{ Heads() map[string]string }
// Bound is how long one check may run; DefaultBound when 0.
Bound time.Duration
// Limit is how many checks run at once; two when 0.
// Limit is how many checks run at once; two when 0. A freed place goes to the first waiting check by rank (ADR 0536 §4).
Limit int
// Hold says whether a delivery builds, so no new check starts (ADR 0484 clause 12); nil never holds.
Hold func() bool
@@ -80,7 +81,10 @@ type Checker struct {
Logf func(format string, args ...any)
once sync.Once
sem chan struct{}
limit int
running int // checks holding a place
waiting []*waiter // checks waiting for a place, in the order they asked
seq uint64
wg sync.WaitGroup
mu sync.Mutex
known map[string]delivery.Candidate // by candidates key
@@ -97,7 +101,7 @@ func (ch *Checker) init() {
if limit <= 0 {
limit = 2
}
ch.sem = make(chan struct{}, limit)
ch.limit = limit
ch.known, ch.inflight, ch.again, ch.tried = map[string]delivery.Candidate{}, map[string]bool{}, map[string]bool{}, map[string]bool{}
ch.nodesFail = map[string]uint64{}
})
@@ -202,10 +206,10 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat
ch.wg.Add(1)
go func() {
defer ch.wg.Done()
ch.sem <- struct{}{}
defer func() { <-ch.sem }()
for {
ch.admit(cs, ch.kindOf(ctx, cs))
ch.try(ctx, key, cs)
ch.leave()
ch.mu.Lock()
more := ch.again[cs.Key]
delete(ch.again, cs.Key)
@@ -218,10 +222,121 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat
if next := CaseOf(ch.candidate(key, c), ch.heads()); next != nil {
cs = next
}
cs.Queued = ch.now() // asked again: it waits for a place by its rank like any other
}
}()
}
// The kinds of a check waiting for a place, by the rank they start in (ADR 0536 §4): a head never judged and a head asked
// again after a failure of the nodes first, then a head asked again only because its trunk moved.
const (
KindNew = "new"
KindNodes = "nodes"
KindTrunk = "trunk"
)
var kindRank = map[string]int{KindNew: 0, KindNodes: 0, KindTrunk: 1}
// waiter is a check waiting for a place.
type waiter struct {
group, key, kind string
queued time.Time
seq uint64
ready chan struct{}
}
// Waiting names a check waiting for a place: its group, its verdict key, its kind and when it was asked.
type Waiting struct {
Group string `json:"group"`
Key string `json:"key"`
Kind string `json:"kind"`
Queued time.Time `json:"queued"`
}
// Waiting are the checks waiting for a place, in the order they would start.
func (ch *Checker) Waiting() []Waiting {
ch.init()
ch.mu.Lock()
defer ch.mu.Unlock()
ws := slices.Clone(ch.waiting)
slices.SortStableFunc(ws, before)
out := make([]Waiting, len(ws))
for i, w := range ws {
out[i] = Waiting{Group: w.group, Key: w.key, Kind: w.kind, Queued: w.queued}
}
return out
}
// before orders waiting checks: the lower rank, then the one asked first, then the one that asked for a place first.
func before(a, b *waiter) int {
if ra, rb := kindRank[a.kind], kindRank[b.kind]; ra != rb {
return ra - rb
}
if c := a.queued.Compare(b.queued); c != 0 {
return c
}
return int(a.seq) - int(b.seq)
}
// kindOf is the kind of a case's check, read from its verdict key: none (or one left running) is a head never judged; a
// failure of the nodes is asked again after the nodes changed; a verdict on other trunk heads is a re-judge after a trunk
// move. A verdict that stands is ranked with the re-judges: its claim finds nothing to do.
func (ch *Checker) kindOf(ctx context.Context, cs *Case) string {
value, _, found, err := ch.Verdicts.Get(ctx, cs.Key)
if err != nil || !found {
return KindNew
}
v, err := delivery.DecodeVerdict(value)
switch {
case err != nil || v.State == delivery.VerdictRunning:
return KindNew
case IsNodesFailure(v):
return KindNodes
}
return KindTrunk
}
// admit returns when the check holds a place: at once when one is free and no check waits, else when a check's end hands its
// place to this one as the first by rank. No timer: a check's end wakes the next (leave).
func (ch *Checker) admit(cs *Case, kind string) {
ch.mu.Lock()
if ch.running < ch.limit && len(ch.waiting) == 0 {
ch.running++
ch.mu.Unlock()
return
}
ch.seq++
w := &waiter{group: cs.Group, key: cs.Key, kind: kind, queued: cs.Queued, seq: ch.seq, ready: make(chan struct{})}
ch.waiting = append(ch.waiting, w)
ch.handOn()
ch.mu.Unlock()
<-w.ready
}
// leave frees the check's place and hands it to the first waiting check by rank.
func (ch *Checker) leave() {
ch.mu.Lock()
ch.running--
ch.handOn()
ch.mu.Unlock()
}
// handOn gives each free place to the first waiting check by rank; it is called with mu held.
func (ch *Checker) handOn() {
for ch.running < ch.limit && len(ch.waiting) > 0 {
first := 0
for i := range ch.waiting {
if before(ch.waiting[i], ch.waiting[first]) < 0 {
first = i
}
}
w := ch.waiting[first]
ch.waiting = slices.Delete(ch.waiting, first, first+1)
ch.running++
close(w.ready)
}
}
func (ch *Checker) candidate(key string, fallback delivery.Candidate) delivery.Candidate {
ch.mu.Lock()
defer ch.mu.Unlock()
@@ -0,0 +1,217 @@
package check
import (
"context"
"fmt"
"runtime"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// The checker's ranked start of checks (ADR 0536 §4, issue 1897): a freed place goes to a head never judged (or asked again
// after a failure of the nodes) before a head asked again only because its trunk moved; among equals the oldest asked first.
// A running check is never stopped.
// rankRig is a checker of one check at a time whose judge records the order checks start in and holds the first check
// after hold is set until release is closed.
type rankRig struct {
*rig
mu sync.Mutex
started []*Case
hold bool
held chan struct{} // closed when the held check started
release chan struct{}
errs []error // ctx errors the held check saw when released
}
func newRankRig(t *testing.T) *rankRig {
rr := &rankRig{held: make(chan struct{}), release: make(chan struct{})}
j := &judgeFunc{f: func(ctx context.Context, c *Case, _ int) (Outcome, error) {
rr.mu.Lock()
rr.started = append(rr.started, c)
hold := rr.hold
rr.hold = false
rr.mu.Unlock()
if hold {
close(rr.held)
<-rr.release
rr.mu.Lock()
rr.errs = append(rr.errs, ctx.Err())
rr.mu.Unlock()
}
return Outcome{Verdict: passing()}, nil
}}
rr.rig = newRig(t, j)
rr.ch.Limit = 1
now := time.Date(2026, 10, 23, 14, 0, 0, 0, time.UTC)
var clock sync.Mutex
rr.ch.Now = func() time.Time {
clock.Lock()
defer clock.Unlock()
now = now.Add(time.Second)
return now
}
return rr
}
func group(n int) string { return fmt.Sprintf("feat/g%d", n) }
// judged puts n candidates in and judges them on the current trunk.
func (rr *rankRig) judged(t *testing.T, n int) {
for i := 1; i <= n; i++ {
rr.ch.Candidate(context.Background(), group(i), candidateOf(t, group(i), pullOf(repoA, i, head1)))
}
rr.ch.Wait()
rr.mu.Lock()
rr.started = nil
rr.mu.Unlock()
}
// waitFor returns when n checks wait for a place.
func (rr *rankRig) waitFor(t *testing.T, n int) {
t.Helper()
deadline := time.Now().Add(5 * time.Second)
for {
if w := len(rr.ch.Waiting()); w == n {
return
} else if time.Now().After(deadline) {
t.Fatalf("%d checks wait, want %d", w, n)
}
runtime.Gosched()
}
}
// moveTrunkHoldingOne moves the trunk, so every judged candidate is asked again; the first re-judge to start is held.
func (rr *rankRig) moveTrunkHoldingOne(t *testing.T, judged int) {
rr.mu.Lock()
rr.hold = true
rr.mu.Unlock()
rr.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS})
rr.ch.TrunkMoved(context.Background())
<-rr.held
rr.waitFor(t, judged-1)
}
func TestANewHeadStartsBeforeTheReJudgesOfATrunkMove(t *testing.T) { // 1897/AC1
rr := newRankRig(t)
rr.judged(t, 4)
rr.moveTrunkHoldingOne(t, 4) // one runs, three re-judges wait
rr.ch.Candidate(context.Background(), group(9), candidateOf(t, group(9), pullOf(repoA, 9, head2)))
rr.waitFor(t, 4)
close(rr.release)
rr.ch.Wait()
if len(rr.started) != 5 {
t.Fatalf("%d checks started, want 5", len(rr.started))
}
if g := rr.started[1].Group; g != group(9) {
t.Errorf("after the running check, %s started, want the new head's %s; order %v", g, group(9), groups(rr.started))
}
}
func TestTheReJudgeQueuedFirstStartsFirst(t *testing.T) { // 1897/AC2
rr := newRankRig(t)
rr.judged(t, 3)
rr.moveTrunkHoldingOne(t, 3) // one runs, two re-judges wait
close(rr.release)
rr.ch.Wait()
if len(rr.started) != 3 {
t.Fatalf("%d checks started, want 3", len(rr.started))
}
if a, b := rr.started[1], rr.started[2]; !a.Queued.Before(b.Queued) {
t.Errorf("%s (queued %s) started before %s (queued %s)", a.Group, a.Queued.Format(time.TimeOnly), b.Group, b.Queued.Format(time.TimeOnly))
}
}
func TestANewHeadDoesNotStopARunningReJudge(t *testing.T) { // 1897/AC3
rr := newRankRig(t)
rr.judged(t, 1)
rr.moveTrunkHoldingOne(t, 1) // the re-judge runs
rr.ch.Candidate(context.Background(), group(9), candidateOf(t, group(9), pullOf(repoA, 9, head2)))
rr.waitFor(t, 1)
close(rr.release)
rr.ch.Wait()
if len(rr.errs) != 1 || rr.errs[0] != nil {
t.Fatalf("the running re-judge saw %v, want it never stopped", rr.errs)
}
if len(rr.started) != 2 || rr.started[0].Group != group(1) || rr.started[1].Group != group(9) {
t.Errorf("started %v, want the re-judge to end before the new head starts", groups(rr.started))
}
key := keyOf(t, group(1), candidateOf(t, group(1), pullOf(repoA, 1, head1)))
if v := rr.verdict(t, key); v.State != delivery.VerdictPass || v.Trunk[repoA] != trunkA2 {
t.Errorf("the re-judge's verdict = %+v, want its pass on the moved trunk", v)
}
}
func TestWaitingChecksAreNamedByKind(t *testing.T) { // 1897/AC4, the checker's own view
rr := newRankRig(t)
rr.judged(t, 2)
rr.moveTrunkHoldingOne(t, 2)
rr.ch.Candidate(context.Background(), group(9), candidateOf(t, group(9), pullOf(repoA, 9, head2)))
rr.waitFor(t, 2)
kinds := map[string]string{}
for _, w := range rr.ch.Waiting() {
kinds[w.Group] = w.Kind
}
close(rr.release)
rr.ch.Wait()
running := rr.started[0].Group
other := group(1)
if running == other {
other = group(2)
}
if kinds[group(9)] != KindNew || kinds[other] != KindTrunk || len(kinds) != 2 {
t.Errorf("waiting = %v, want %s %s and %s %s", kinds, group(9), KindNew, other, KindTrunk)
}
}
func groups(cs []*Case) []string {
out := make([]string, len(cs))
for i, c := range cs {
out[i] = c.Group
}
return out
}
func TestCheckerStatusNamesTheWaitingChecksByKind(t *testing.T) { // 1897/AC4
ctx := context.Background()
j := &judgeFunc{f: func(_ context.Context, c *Case, _ int) (Outcome, error) {
if c.Group == group(2) {
return Outcome{Verdict: failure("checker", "no "+NodesWords+": every node is offline")}, nil
}
return Outcome{Verdict: passing()}, nil
}}
r := newRig(t, j)
cands := fakeState{}
put := func(n int, head string) {
cand := candidateOf(t, group(n), pullOf(repoA, n, head))
ckey, _ := delivery.CandidatesKey(group(n))
cands[ckey] = cand
r.ch.Candidate(ctx, group(n), cand)
}
put(1, head1) // judged on the old trunk: a re-judge waits
put(2, head1) // failed for the nodes: waits for them
r.ch.Wait()
r.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS})
put(4, head1) // judged on the new trunk: stands
r.ch.Wait()
cand3 := candidateOf(t, group(3), pullOf(repoA, 3, head1)) // never judged
ckey3, _ := delivery.CandidatesKey(group(3))
cands[ckey3] = cand3
verdicts := fakeState{}
for k, v := range r.verdicts.vals {
verdicts[k] = v
}
rd := Reader{CandidateState: cands.read(), VerdictState: verdicts.read(),
Trunks: func() (map[string]string, error) { return map[string]string{repoA: trunkA2, repoSDK: trunkS}, nil }}
st, err := rd.Status()
if err != nil {
t.Fatal(err)
}
want := map[string][]string{KindNew: {group(3)}, KindNodes: {group(2)}, KindTrunk: {group(1)}}
if fmt.Sprint(st.Waiting) != fmt.Sprint(want) {
t.Errorf("waiting = %v, want %v", st.Waiting, want)
}
}
@@ -40,7 +40,9 @@ type Reader struct {
Capacity func() (string, error) // the capacity setting as written, `node=steps,...`
HaveState StateRead // the node-engine's have: whether a build module is on a node and running
StepBound time.Duration // how long the checker waits for one step; DefaultStepBound when 0
Now func() time.Time
// Trunks are the trunk heads of every repository (the gate's source); nil names no re-judge after a trunk move.
Trunks func() (map[string]string, error)
Now func() time.Time
}
// VerdictRead is a verdict as stored with the full report of the status of each pull request it judged (Report), when the
@@ -280,7 +282,10 @@ type Status struct {
LeftRunning []LeftStep `json:"left_running"`
RunningByNode map[string]int `json:"running_by_node"`
Unjudged []string `json:"unjudged_candidates"`
Notes []string `json:"notes,omitempty"`
// Waiting are the candidates whose check waits, by kind (ADR 0536 §4): new (no verdict for its heads) and nodes (failed
// for the nodes, asked again when they change) start before trunk (judged on trunk heads that moved since).
Waiting map[string][]string `json:"waiting_by_kind"`
Notes []string `json:"notes,omitempty"`
}
// Status reads the capacity, the plan bucket, the build modules' steps and the candidates against their verdicts.
@@ -353,6 +358,15 @@ func (r Reader) Status() (Status, error) {
st.Notes = append(st.Notes, "left_running are keys a build module left running (it stopped, restarted or was unassigned before it wrote the step's end): the checker's wait for each ended at the step bound, so none holds a place; the build module never ends them")
}
st.Unjudged = []string{}
st.Waiting = map[string][]string{KindNew: {}, KindNodes: {}, KindTrunk: {}}
var heads map[string]string
if r.Trunks == nil {
st.Notes = append(st.Notes, "the trunk heads are not readable here, so no check is named as waiting for a trunk move")
} else if h, err := r.Trunks(); err != nil {
st.Notes = append(st.Notes, "the trunk heads cannot be read, so no check is named as waiting for a trunk move: "+err.Error())
} else {
heads = h
}
if r.CandidateState.Keys != nil && r.CandidateState.Get != nil && r.VerdictState.Get != nil {
keys, err := r.CandidateState.Keys()
if err != nil {
@@ -375,8 +389,24 @@ func (r Reader) Status() (Status, error) {
if !checksAny(c) {
continue // hq's pull requests alone: merge-check.sh judges them, the checker never does
}
if v, err := r.VerdictState.Get(vkey); err == nil && len(v) == 0 {
raw, err = r.VerdictState.Get(vkey)
if err != nil {
continue
}
if len(raw) == 0 {
st.Unjudged = append(st.Unjudged, c.Group)
st.Waiting[KindNew] = append(st.Waiting[KindNew], c.Group)
continue
}
v, err := delivery.DecodeVerdict(raw)
switch {
case err != nil || v.State == delivery.VerdictRunning:
case IsNodesFailure(v):
st.Waiting[KindNodes] = append(st.Waiting[KindNodes], c.Group)
case heads != nil:
if cs := CaseOf(c, heads); cs != nil && !sameTrunk(v.Trunk, cs.Trunk) {
st.Waiting[KindTrunk] = append(st.Waiting[KindTrunk], c.Group)
}
}
}
}