package check import ( "context" "errors" "fmt" "sort" "strings" "sync" "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) // DefaultBound is how long a check may run before it is ended failed naming the step it was stuck at (ADR 0484, issue 1678): // none since R4g. A check's asks wait in the build seats' streams as long as the backlog is (ADR 0525 §9), and each ask's // step is bounded from its start on the node that took it (buildrun.Asker.Bound, DefaultStepBound); a bound set by // MESH_CHECKER_BOUND still ends a check that ran past it. const DefaultBound time.Duration = 0 // UnheardWords end the Words of a failure because a part of the mesh the check reads was not heard (the forge, issue 1624): // like a failure of the nodes, it is not the change's, and is judged again when the nodes change. const UnheardWords = "not heard; judged again when the nodes change" // The events the checker says on its stream checks (module.json): a preview or a status was written, so a part that reads // the bucket, which is made when the checker is assigned and may not exist when that part starts, finds it (ADR 0437 §8: no // timer). The body is {"key": ...}. const ( EventPreviewWritten = "preview.written" EventOwedWritten = "owed.written" ) // Announcer says an event of the checker's stream, with the key written. type Announcer func(ctx context.Context, event, key string) // Outcome is what judging a case gives: the verdict (state, failure, levels, times, composition; the checker adds the // group, the heads and the trunk) and, for the line, what the check knows of the builds of the modules it touches. type Outcome struct { Verdict delivery.Verdict Builds []BuildNote // Slowest names the slowest tests of a level that went over its bound ("unit pkg.Test 1.2s"). Slowest []string // Judged says, by pull request (pullKey), which program judged a change to a build module (ADR 0535 §5); a pull request // that changes none is not named. Judged map[string]string // Beside are the siblings the check read at their trunk's head, by repository (ADR 0548 §6): the verdict names them, and // a move of one is carried. Beside map[string]string } // BuildNote is the state of a candidate's module build, by module: built, failed or waiting (ADR 0484 clause 7). type BuildNote struct { Module string State string } // Judge runs the check of a case: the merge, the levels. It writes what it is doing to c.Progress. It answers an error // only when it could not judge at all; a failure of the change is a Verdict. type Judge interface { Judge(ctx context.Context, c *Case) (Outcome, error) } // Checker judges the candidates the indexer writes, dark (ADR 0453 clause 4, ADR 0484): for each candidate it claims a // verdict key, runs the check through its Judge, writes the verdict and, for each pull request judged, the owed status // `mesh/check` the forge bridge performs. It holds no timer; a check starts from a candidate or from a change of the // nodes, the trunk or the controller's plan, and ends when its judge does or its bound passes. type Checker struct { Verdicts KV 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 { Holding(repositories []string, heads string) (string, delivery.TurnWait, bool) } // Bound is how long one check may run; none when 0 (DefaultBound). Bound time.Duration // Link is the base of the address a status points at; the verdict key follows. Link string // Announce says a status was written (EventOwedWritten); nil says nothing. Announce Announcer Now func() time.Time Logf func(format string, args ...any) // Settle is how long a group's heads must stay the same after the candidate last changed them before its check starts // (issue 1890): the members of a group are announced one after another, and a check of each partial set is thrown // away at the next. None when 0; DefaultSettle in the module. Settle time.Duration once sync.Once wg sync.WaitGroup mu sync.Mutex known map[string]delivery.Candidate // by candidates key inflight map[string]bool // by verdict key again map[string]bool // by verdict key: asked again while it ran tried map[string]bool // by verdict key: looked at since this process started epoch uint64 // changes of the nodes seen since this process started nodesFail map[string]uint64 // by verdict key: the epoch a failure of the nodes was written at held map[string]string // by verdict key: the turn that holds its check off, as last said live map[string]liveCheck // by verdict key: the check running in this process, ended when its heads go settling map[string]*window // by candidates key: the window its heads settle in } // liveCheck is a check running in this process: the candidates key of its group and the end of its run. type liveCheck struct { group string end context.CancelCauseFunc } // window is a group's settle window: aborted with errUnsettled when its heads change again or the group goes, and with // errSettled to end it at once. type window struct{ abort context.CancelCauseFunc } var ( errUnsettled = errors.New("the group's heads changed again") errSettled = errors.New("the group's heads settled") ) // settleNow ends the settle window of every group at once, as if Settle had passed. func (ch *Checker) settleNow() { ch.mu.Lock() defer ch.mu.Unlock() for _, w := range ch.settling { w.abort(errSettled) } } // DefaultSettle is the module's Settle (issue 1890). On 2026-10-26 the three members of a group were announced within 2.8 // seconds of each other (20:31:52.3, :53.7, :55.1 UTC), a session opening or pushing them one API call after another; // fifteen seconds holds five times that, and costs a check of 10 to 18 minutes under 2.5 % of its time. A member that comes // later still is no waste beyond its asks already taken: the check of the heads it replaced is ended at once. const DefaultSettle = 15 * time.Second // EndedWords open the words of a verdict the checker ended without judging it, its heads no candidate's any more (issue // 1890): never a failure of the change, and judged whole if a candidate holds those heads again. const EndedWords = "ended: " // IsEnded says whether a verdict was ended without being judged (EndedWords). func IsEnded(v delivery.Verdict) bool { return v.State == delivery.VerdictFail && v.Failed.Machine == "checker" && strings.HasPrefix(v.Failed.Words, EndedWords) } // superseded is the cause a running check is ended with: the heads that replaced its own, or none when its group went. type superseded struct{ by string } func (s superseded) Error() string { if s.by == "" { return EndedWords + "no candidate holds these heads any more: the group was closed or merged" } return EndedWords + "superseded by " + s.by } func (ch *Checker) init() { ch.once.Do(func() { ch.known, ch.inflight, ch.again, ch.tried = map[string]delivery.Candidate{}, map[string]bool{}, map[string]bool{}, map[string]bool{} ch.nodesFail, ch.held = map[string]uint64{}, map[string]string{} ch.live, ch.settling = map[string]liveCheck{}, map[string]*window{} }) } // endOthers ends every check of the group running in this process whose verdict key is not keep: no candidate holds its // heads any more (issue 1890). Called with ch.mu held. func (ch *Checker) endOthers(group, keep, by string) { for k, l := range ch.live { if l.group == group && k != keep { l.end(superseded{by: by}) } } } func (ch *Checker) logf(format string, args ...any) { if ch.Logf != nil { ch.Logf(format, args...) } } func (ch *Checker) now() time.Time { if ch.Now != nil { return ch.Now() } return time.Now() } func (ch *Checker) bound() time.Duration { if ch.Bound > 0 { return ch.Bound } return DefaultBound } // Wait returns when every check started so far has ended. func (ch *Checker) Wait() { ch.wg.Wait() } // Candidate takes one value of the indexer's candidates bucket in (the key is the group) and checks it if it is to be // checked. A value that is not a candidate is said and left. func (ch *Checker) Candidate(ctx context.Context, key string, value []byte) { ch.init() c, err := delivery.DecodeCandidate(value) if err != nil { ch.logf("candidates.%s holds what is not a candidate, left: %v", key, err) return } vkey, _ := delivery.CheckerVerdictKey(c.Group, c.HeadsHash) ch.mu.Lock() prev, had := ch.known[key] ch.known[key] = c // A check of heads this candidate no longer holds is ended at once (issue 1890): a member's new head, a member joining // or leaving. ch.endOthers(key, vkey, GateHeads(c)) moved := !had || prev.HeadsHash != c.HeadsHash if ch.Settle > 0 && (moved || ch.settling[key] != nil) { // The group's heads changed: its check waits until they have been the same for Settle, so a group whose members // are announced one after another is checked once, whole. A later change starts the window again. The wait is a // context's bound, no timer (ADR 0437 §8). if w := ch.settling[key]; w != nil { w.abort(errUnsettled) } base, abort := context.WithCancelCause(ctx) w := &window{abort: abort} ch.settling[key] = w ch.wg.Add(1) go func() { defer ch.wg.Done() wait, done := context.WithTimeout(base, ch.Settle) defer done() <-wait.Done() settled := errors.Is(wait.Err(), context.DeadlineExceeded) || errors.Is(context.Cause(wait), errSettled) ch.mu.Lock() if !settled || ch.settling[key] != w { ch.mu.Unlock() return // changed again, forgotten, or the checker stops } delete(ch.settling, key) now, ok := ch.known[key] ch.mu.Unlock() abort(nil) if ok { ch.consider(ctx, key, now) } }() ch.mu.Unlock() return } ch.mu.Unlock() ch.consider(ctx, key, c) } // Forget takes a candidate out (its key was deleted): a check of its heads still running is ended, as no candidate holds // them any more. func (ch *Checker) Forget(key string) { ch.init() ch.mu.Lock() delete(ch.known, key) if w := ch.settling[key]; w != nil { w.abort(errUnsettled) delete(ch.settling, key) } ch.endOthers(key, "", "") ch.mu.Unlock() } // NodesChanged is Reconsider after a change of the nodes: one came online or went, a build module became healthy or // stopped being, or a toolchain moved. A check that failed because no node qualified is run again only after such a change. func (ch *Checker) NodesChanged(ctx context.Context) { ch.init() ch.mu.Lock() ch.epoch++ ch.mu.Unlock() ch.Reconsider(ctx) } // Reconsider looks at every candidate again: the trunk moved, the nodes changed or a delivery ended. A candidate whose // verdict stands for the trunk it was judged against is left. func (ch *Checker) Reconsider(ctx context.Context) { ch.init() ch.mu.Lock() known := make(map[string]delivery.Candidate, len(ch.known)) for k, c := range ch.known { known[k] = c } ch.mu.Unlock() for k, c := range known { ch.consider(ctx, k, c) } } func (ch *Checker) heads() map[string]string { if ch.Trunks == nil { return nil } return ch.Trunks.Heads() } func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidate) { cs := CaseOf(c, ch.heads()) if cs == nil { return // nothing to judge: no trunk known } cs.Queued = ch.now() ch.mu.Lock() if ch.settling[key] != nil { ch.mu.Unlock() return // its heads are settling: the window's end considers it (Settle) } if ch.inflight[cs.Key] { ch.again[cs.Key] = true // looked at once more when the check ends, so a change of the trunk is never lost ch.mu.Unlock() return } ch.inflight[cs.Key] = true run, end := context.WithCancelCause(ctx) ch.live[cs.Key] = liveCheck{group: key, end: end} ch.mu.Unlock() ch.wg.Add(1) go func() { defer ch.wg.Done() vkey := cs.Key defer func() { ch.mu.Lock() delete(ch.inflight, vkey) delete(ch.again, vkey) delete(ch.live, vkey) ch.mu.Unlock() end(nil) }() for { // Every check is asked at once: the build seats' streams hold the backlog, a new head's check and a check after a // failure of the nodes on check, a re-judge after a trunk move on recheck, pulled after them (ADR 0525 §4, §9; // ADR 0536 §4). The checker holds no place and no capacity of its own. ch.prepare(ctx, cs) ch.try(ctx, run, key, cs) ch.mu.Lock() more := ch.again[vkey] delete(ch.again, vkey) ch.mu.Unlock() if !more || run.Err() != nil { return } // Looked at again for the same heads only: a candidate that moved to other heads has its own check (issue 1890). if next := CaseOf(ch.candidate(key, c), ch.heads()); next != nil { if next.Key != vkey { return } cs = next } cs.Queued = ch.now() } }() } // The kinds of a check (ADR 0536 §4): a head never judged and a head asked again after a failure of the nodes are asked on // the build seats' check stream; a carry, a verdict moved to a newer trunk or sibling commit by its package digests (ADR 0548 // §5), on the recheck stream, pulled after. const ( KindNew = "new" KindNodes = "nodes" KindTrunk = "trunk" ) // prepare reads the case's verdict key and says what its check is (ADR 0548 §1, §5, §7): none (or one left running) is a // head never judged, a whole check; a failure of the nodes is asked again after the nodes changed; a verdict judged on trunk // or sibling commits that moved since is carried, its packages run only where the move changed their digest, and a pass's // commits are named as the carry's CarriedFrom. A verdict that stands is ranked with the carries: its claim finds nothing // to do. func (ch *Checker) prepare(ctx context.Context, cs *Case) { cs.Kind, cs.CarriedFrom, cs.Composition = KindNew, nil, nil value, _, found, err := ch.Verdicts.Get(ctx, cs.Key) if err != nil || !found { return } v, err := delivery.DecodeVerdict(value) switch { case err != nil || v.State == delivery.VerdictRunning: return case IsNodesFailure(v): cs.Kind = KindNodes return case IsConflictFailure(v): return // a head git refused: its next check is whole (ADR 0548 §7) case IsEnded(v): return // ended unjudged when its heads went (issue 1890): held again, they are checked whole } cs.Kind = KindTrunk if v.State == delivery.VerdictPass && ch.stale(v, cs) { cs.CarriedFrom = judgedOn(v) if c := v.Composition; c != nil && c.State == delivery.CompositionPass && c.FPSet != "" && c.FPSet == cs.FPSet { kept := *c cs.Composition = &kept } } } // 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.besideNow()) } // 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 { return besideAt(ch.heads(), ch.Runs) } // besideAt places the commits a sibling is read at: the trunk heads, with the node-engine at the commit the mesh runs when // runs names one (ADR 0548 §6). The checker's decision and the status read the same commits through it. func besideAt(now map[string]string, runs interface{ CommitOf(module string) string }) map[string]string { if runs != nil { if c := runs.CommitOf(nodeEngineModule); c != "" { out := make(map[string]string, len(now)+1) for r, h := range now { out[r] = h } out[nodeEngineRepo] = c return out } } 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 := now[r]; h != "" && h != c { return true } } return false } // judgedOn are the trunk and sibling commits a verdict was judged on, by repository. func judgedOn(v delivery.Verdict) map[string]string { out := make(map[string]string, len(v.Trunk)+len(v.Beside)) for r, c := range v.Beside { out[r] = c } for r, c := range v.Trunk { out[r] = c } return out } // carried is what a carry's verdict names as carried from (ADR 0548 §5): the commits it was judged on before, of the // repositories the new verdict still judges or reads; nil when none of them moved. func carried(from map[string]string, v delivery.Verdict) map[string]string { out, moved := map[string]string{}, false for r, c := range from { now, ok := v.Trunk[r] if !ok { now, ok = v.Beside[r] } if ok && now != "" && c != "" { out[r] = c moved = moved || now != c } } if !moved { return nil } return out } // moveWords say a carry's moves, for the words of a failure: "carried from to of ", each moved // repository, sorted. func moveWords(from map[string]string, v delivery.Verdict) string { var moves []string for r, c := range from { now, ok := v.Trunk[r] if !ok { now = v.Beside[r] } if now != "" && now != c { moves = append(moves, fmt.Sprintf("%s %s to %s", r, short12(c), short12(now))) } } sort.Strings(moves) return strings.Join(moves, ", ") } func (ch *Checker) candidate(key string, fallback delivery.Candidate) delivery.Candidate { ch.mu.Lock() defer ch.mu.Unlock() if c, ok := ch.known[key]; ok { return c } return fallback } // claimed is a verdict key the checker holds: the revision the next write is made at. type claimed struct{ rev uint64 } // claim takes the check of the case, or answers nil when it is not this checker's to run now: // - no key: it is created running; a create that finds the key made meanwhile gets nothing; // - a key left running is taken over as it stands when this process looks at it for the first time (it was left by the // checker that stopped; there is one instance, as there is one builder per node, ADR 0478 clause 6), and left otherwise; // - a pass or a failure for the trunk heads the case has stands; one for other trunk heads (the trunk moved) is taken // again by compare-and-set on the same key, and so is a failure of the nodes after the nodes changed. func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64) (*claimed, error) { running := delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictRunning} data, err := delivery.EncodeVerdict(running) if err != nil { return nil, err } value, rev, found, err := ch.Verdicts.Get(ctx, cs.Key) if err != nil { return nil, fmt.Errorf("%s: %w", cs.Key, err) } if !found { if ch.heldByTurn(cs) { return nil, nil } rev, err := ch.Verdicts.Create(ctx, cs.Key, data) if errors.Is(err, ErrConflict) { return nil, nil // another checker made it first } if err != nil { return nil, fmt.Errorf("%s: %w", cs.Key, err) } return &claimed{rev}, nil } cur, err := delivery.DecodeVerdict(value) if err != nil { return nil, fmt.Errorf("%s holds what is not a verdict: %w", cs.Key, err) } retake := false switch { case cur.State == delivery.VerdictRunning: if first { return &claimed{rev}, nil } return nil, nil case IsEnded(cur), ch.stale(cur, cs): retake = true // heads held again after their check was ended unjudged (issue 1890), or a trunk that moved case cur.State == delivery.VerdictFail && IsNodesFailure(cur): ch.mu.Lock() at, ok := ch.nodesFail[cs.Key] ch.mu.Unlock() retake = !ok || epoch > at } if !retake || ch.heldByTurn(cs) { return nil, nil } rev, err = ch.Verdicts.Update(ctx, cs.Key, data, rev) if errors.Is(err, ErrConflict) { return nil, nil } if err != nil { return nil, fmt.Errorf("%s: %w", cs.Key, err) } return &claimed{rev}, nil } // heldByTurn says whether a merge turn holds the case's check off (ADR 0548 §3): an approved merge waits on one of the // repositories it merges heads into, and it is not the merge of the case's own heads. The check is claimed when the turn's // key goes (Reconsider on the turns watch). Said once per turn in the log. A check already claimed is never held: one // claimed in the moment before the gate wrote its turn is judged on the commit it read and carried once after. func (ch *Checker) heldByTurn(cs *Case) bool { if ch.Turns == nil { return false } repos := make([]string, 0, len(cs.Trunk)) for r := range cs.Trunk { repos = append(repos, r) } repo, w, held := ch.Turns.Holding(repos, cs.Heads) why := "" if held { why = repo + " " + w.Group + " " + w.Heads } ch.mu.Lock() said := ch.held[cs.Key] if why == "" { delete(ch.held, cs.Key) } else { ch.held[cs.Key] = why } ch.mu.Unlock() if held && said != why { ch.logf("%s: not claimed: the approved merge of %s (heads %s) holds a turn on %s, so its check waits for that merge (ADR 0548 §3)", cs.Key, w.Group, w.Heads, repo) } return held } // Held is every check a merge turn holds off now, by verdict key, with the turn's repository, group and heads. func (ch *Checker) Held() map[string]string { ch.init() ch.mu.Lock() defer ch.mu.Unlock() out := make(map[string]string, len(ch.held)) for k, v := range ch.held { out[k] = v } return out } // ConflictWords are what the words of a head that git would not merge onto its trunk say before the files in conflict. const ConflictWords = "does not merge into the trunk: files in conflict: " // IsConflictFailure says whether a verdict failed because its head did not merge onto its trunk (ADR 0548 §7): it has no // pass to carry, so its next check, at a push or at a later trunk move, is a whole check of its head. func IsConflictFailure(v delivery.Verdict) bool { return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, ConflictWords) } // IsNodesFailure says whether a verdict failed because a part of the mesh the check reads was not heard (UnheardWords): // it is judged again when the nodes change. No check fails for want of a build node any more: its asks wait in the build // seats' streams (ADR 0525 §6, issue 1912). func IsNodesFailure(v delivery.Verdict) bool { return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, UnheardWords) } // try claims the check, judges it within its bound under run and writes the verdict and the owed statuses. A check that ends // because ctx did is left running: the checker takes it again when it starts. A check whose run was ended because no // candidate holds its heads any more (issue 1890) is ended: its verdict says so, and no status is written. func (ch *Checker) try(ctx, run context.Context, key string, cs *Case) { if run.Err() != nil { return // ended before it was claimed: nothing to write } ch.mu.Lock() first := !ch.tried[cs.Key] ch.tried[cs.Key] = true epoch := ch.epoch ch.mu.Unlock() cl, err := ch.claim(ctx, cs, first, epoch) if err != nil { ch.logf("%s: %v", cs.Key, err) return } if cl == nil { return } began := ch.now() ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{}) out := ch.judge(run, cs) // An ask the build seat gave up on judged nothing (issue 734): the check is asked again under the same claim, its status // pending, and after LostAskTries it fails as not heard, judged again when the nodes change. for try := 1; IsLostAsk(out.Verdict) && run.Err() == nil; try++ { if try >= LostAskTries { out.Verdict.Failed.Words = strings.TrimSpace(out.Verdict.Failed.Words + "\nasked " + fmt.Sprint(LostAskTries) + " times and given up each time: " + UnheardWords) break } ch.logf("%s: %s; asked again (%d of %d)", cs.Key, out.Verdict.Failed.Words, try+1, LostAskTries) out = ch.judge(run, cs) } if ctx.Err() != nil { return } var why superseded if errors.As(context.Cause(run), &why) { ch.end(ctx, cs, cl, why) return } v := out.Verdict v.Group, v.Heads, v.Trunk, v.Beside = cs.Group, cs.Heads, cs.Trunk, nil for r, c := range out.Beside { if _, merged := cs.Trunk[r]; !merged && c != "" { if v.Beside == nil { v.Beside = map[string]string{} } v.Beside[r] = c } } if cs.CarriedFrom != nil { // A carry (ADR 0548 §5): a pass names the commits it was carried from; a failure names the move it failed on. switch v.State { case delivery.VerdictPass: v.CarriedFrom = carried(cs.CarriedFrom, v) case delivery.VerdictFail: if moves := moveWords(cs.CarriedFrom, v); moves != "" { v.Failed.Words = strings.TrimSpace(v.Failed.Words + "\nfailed on the carry over the move of " + moves + ": the move changed what these tests read") } } } if v.Took.Queue == 0 && !cs.Queued.IsZero() && began.After(cs.Queued) { v.Took.Queue = began.Sub(cs.Queued).Milliseconds() } data, err := delivery.EncodeVerdict(v) if err != nil { ch.logf("%s: the verdict cannot be written: %v", cs.Key, err) v = delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictFail, Failed: delivery.FailedFirst{Machine: "checker", Words: "the verdict could not be written: " + err.Error()}} if data, err = delivery.EncodeVerdict(v); err != nil { return } } if _, err := ch.Verdicts.Update(ctx, cs.Key, data, cl.rev); err != nil { ch.logf("%s: the verdict could not be written: %v", cs.Key, err) return } if IsNodesFailure(v) { ch.mu.Lock() ch.nodesFail[cs.Key] = epoch ch.mu.Unlock() } 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) ch.logf("%s: %s on a trunk that moved while it ran; the status stays pending until it is carried", cs.Key, v.State) return } ch.writeOwed(ctx, cs, v, out) ch.logf("%s: %s", cs.Key, v.State) } // end writes the verdict of a check whose heads no candidate holds any more (issue 1890), by compare-and-set on its claim, // and says it once in the log with the asks it had put: they stay in the build seats' streams, since an ask once put cannot // be taken back (buildrun.Asker has no withdraw), and are no longer waited for. No status is written: the heads that // replaced these have their own check, whose pending status stands. func (ch *Checker) end(ctx context.Context, cs *Case, cl *claimed, why superseded) { v := delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictFail, Failed: delivery.FailedFirst{Machine: "checker", Words: why.Error()}} data, err := delivery.EncodeVerdict(v) if err == nil { _, err = ch.Verdicts.Update(ctx, cs.Key, data, cl.rev) } if err != nil { ch.logf("%s: %s, but the verdict could not say so: %v", cs.Key, why.Error(), err) return } stage, _ := cs.Progress.Now() asks := cs.Asks() ch.logf("%s: %s; it was at %q with %d asks put %v, which are no longer waited for", cs.Key, why.Error(), stage, len(asks), asks) } // judge runs the case within the bound. A judge that fails to judge, and a check that ran past its bound, are verdicts of // failure naming the machine and the step. func (ch *Checker) judge(ctx context.Context, cs *Case) Outcome { out, _ := judgeWithin(ctx, ch.Judge, cs, ch.bound()) return out } var errBound = errors.New("the check ran past its bound") func failure(machine, words string) delivery.Verdict { return delivery.Verdict{State: delivery.VerdictFail, Failed: delivery.FailedFirst{Machine: machine, Words: words}} } // writeOwed writes, for each pull request judged, the status the forge bridge performs: pending while the check runs, success // or failure at its end (ADR 0484 clause 2). A key already holding the same value is not written again, so the bridge // performs one act for each change; an older head's key is never touched. func (ch *Checker) writeOwed(ctx context.Context, cs *Case, v delivery.Verdict, out Outcome) { for _, p := range cs.Pulls { key, err := delivery.CheckerOwedKey(p.Repository, p.Number, p.Head) if err != nil { ch.logf("%s: %v", cs.Key, err) continue } state := map[string]string{delivery.VerdictRunning: "pending", delivery.VerdictPass: "success", delivery.VerdictFail: "failure"}[v.State] o := delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, Target: strings.TrimSuffix(ch.Link, "/") + "/" + cs.Key, Text: Line(v, cs, p, out.Builds, out.Slowest, out.Judged)} data, err := delivery.EncodeCheckerOwed(o) if err != nil { ch.logf("%s: the status of %s cannot be written: %v", cs.Key, key, err) continue } wrote, err := putOnce(ctx, ch.Owed, key, data, func(cur []byte) bool { held, err := delivery.DecodeCheckerOwed(cur) o.V = delivery.Version return err == nil && delivery.SameOwed(held, o) }) if err != nil { ch.logf("%s: the status of %s cannot be written: %v", cs.Key, key, err) } else if wrote && ch.Announce != nil { ch.Announce(ctx, EventOwedWritten, key) } } } // putOnce writes data under key by create, or by compare-and-set when the key exists, and writes nothing when same says the // key holds the value already. func putOnce(ctx context.Context, kv KV, key string, data []byte, same func(cur []byte) bool) (bool, error) { for try := 0; try < 4; try++ { cur, rev, found, err := kv.Get(ctx, key) if err != nil { return false, err } if found && same(cur) { return false, nil } if !found { _, err = kv.Create(ctx, key, data) } else { _, err = kv.Update(ctx, key, data, rev) } if errors.Is(err, ErrConflict) { continue } return err == nil, err } return false, fmt.Errorf("%s changed under every one of 4 tries", key) } // Sweep closes, at start, every verdict left running for a candidate that no longer holds those heads: nothing will check // them again, and the key would stay running for ever (the builder's Sweep, issue 1660). Call it once the candidates have been // read. A key of heads a candidate holds now is not touched: the check takes it over. func (ch *Checker) Sweep(ctx context.Context) error { ch.init() keys, err := ch.Verdicts.Keys(ctx) if err != nil { return err } ch.mu.Lock() current := map[string]bool{} for _, c := range ch.known { if k, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash); err == nil { current[k] = true } } ch.mu.Unlock() for _, key := range keys { if current[key] { continue } value, rev, found, err := ch.Verdicts.Get(ctx, key) if err != nil || !found { continue } cur, err := delivery.DecodeVerdict(value) if err != nil || cur.State != delivery.VerdictRunning { continue } cur.State = delivery.VerdictFail cur.Failed = delivery.FailedFirst{Machine: "checker", Words: EndedWords + "closed at start: left running, and no candidate holds these heads any more, so nothing will check them"} data, err := delivery.EncodeVerdict(cur) if err != nil { return err } if _, err := ch.Verdicts.Update(ctx, key, data, rev); err != nil && !errors.Is(err, ErrConflict) { return fmt.Errorf("%s: %w", key, err) } ch.logf("%s: closed, left running for heads no candidate holds", key) } return nil } // TrunkMoved is Reconsider after a move of a trunk, which first owes mesh/check pending on every open pull request of a // candidate whose verdict was judged on other trunk heads, every member's for a group (the lead's decision L4, the checker's // part of issue 1302): the forge's success of the old trunk never stands while the check waits for a place or for a delivery // that holds it, so the gate never reads it as the head's. func (ch *Checker) TrunkMoved(ctx context.Context) { ch.init() ch.mu.Lock() known := make(map[string]delivery.Candidate, len(ch.known)) for k, c := range ch.known { known[k] = c } ch.mu.Unlock() heads := ch.heads() for _, c := range known { cs := CaseOf(c, heads) if cs == nil { continue } value, _, found, err := ch.Verdicts.Get(ctx, cs.Key) if err != nil || !found { continue // never judged: no status to put out of date } v, err := delivery.DecodeVerdict(value) if err == nil && !ch.stale(v, cs) { continue } if err == nil && v.State == delivery.VerdictPass { cs.CarriedFrom = judgedOn(v) // the status says it is being carried to the new commit (ADR 0548 §6) } ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{}) } ch.Reconsider(ctx) } // owePending owes mesh/check pending on the heads of a verdict judged on commits that moved while it ran: a pass says it is // being carried to the trunk's head now (ADR 0548 §6), anything else that it is checked again. func (ch *Checker) owePending(ctx context.Context, judged *Case, v delivery.Verdict) { next := judged ckey, err := delivery.CandidatesKey(v.Group) ch.mu.Lock() c, ok := ch.known[ckey] ch.mu.Unlock() if err == nil && ok { if now := CaseOf(c, ch.heads()); now != nil && now.Key == judged.Key { next = now } } if v.State == delivery.VerdictPass { next.CarriedFrom = judgedOn(v) } ch.writeOwed(ctx, next, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{}) } // movedSince says whether a trunk the case was judged on is no longer the trunk's head. func (ch *Checker) movedSince(cs *Case) bool { now := ch.heads() for repo, head := range cs.Trunk { if now[repo] != head { return true } } return false }