diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go index 8d31c6e..acac211 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go @@ -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) diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index 78d6ba2..33f795e 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -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() diff --git a/modules/mesh-delivery-checker/internal/check/rank_test.go b/modules/mesh-delivery-checker/internal/check/rank_test.go new file mode 100644 index 0000000..0de4329 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/rank_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/read.go b/modules/mesh-delivery-checker/internal/check/read.go index 165e406..7fd9055 100644 --- a/modules/mesh-delivery-checker/internal/check/read.go +++ b/modules/mesh-delivery-checker/internal/check/read.go @@ -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) + } } } }