From d95e9f3229139fe12e854c73ddcffac3770a9a27 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 21:44:02 +0100 Subject: [PATCH] End a check whose heads no candidate holds, and start a group's check once its heads settle (issue 1890) --- .../mesh-delivery-checker/integration_test.go | 3 +- .../cmd/mesh-delivery-checker/main.go | 5 + .../internal/check/checker.go | 193 ++++++++++++++++-- .../internal/check/checker_test.go | 87 +++++++- 4 files changed, 268 insertions(+), 20 deletions(-) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index 291e449..3e01724 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go @@ -427,7 +427,8 @@ func (b *bed) start(t *testing.T, env map[string]string) { if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } - e := map[string]string{"MESH_BROKER_FILE": broker} + // The settle window of a group's heads (check.DefaultSettle) is cut short: each bed's candidate is written whole. + e := map[string]string{"MESH_BROKER_FILE": broker, "MESH_CHECKER_SETTLE": "10ms"} for k, v := range env { e[k] = v } 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 767a70e..c4d3902 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -189,6 +189,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + settleWindow, err := durationEnv(getenv, "MESH_CHECKER_SETTLE", check.DefaultSettle) + if err != nil { + return err + } link := strings.TrimSpace(getenv("MESH_CHECKER_LINK")) if link == "" { link = defaultLink @@ -221,6 +225,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Bound: checkBound, Link: link, Logf: logger.Printf, + Settle: settleWindow, } // The builder's builds say how the candidates' fingerprints stand (issue 1684); where its bucket is not there yet the line // says nothing of the builds, until the checker starts again. diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index e74f944..940f434 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -85,6 +85,10 @@ type Checker struct { 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 @@ -96,15 +100,77 @@ type Checker struct { 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...) @@ -137,17 +203,62 @@ func (ch *Checker) Candidate(ctx context.Context, key string, value []byte) { 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). +// 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() } @@ -190,32 +301,49 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat } 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, key, cs) + ch.try(ctx, run, key, cs) ch.mu.Lock() - more := ch.again[cs.Key] - delete(ch.again, cs.Key) - if !more || ctx.Err() != nil { - delete(ch.inflight, cs.Key) - ch.mu.Unlock() + more := ch.again[vkey] + delete(ch.again, vkey) + ch.mu.Unlock() + if !more || run.Err() != nil { return } - ch.mu.Unlock() + // 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() @@ -252,6 +380,8 @@ func (ch *Checker) prepare(ctx context.Context, cs *Case) { 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) { @@ -402,8 +532,8 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64 return &claimed{rev}, nil } return nil, nil - case ch.stale(cur, cs): - retake = true + 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] @@ -483,9 +613,13 @@ 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 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. -func (ch *Checker) try(ctx context.Context, key string, cs *Case) { +// 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 @@ -501,21 +635,26 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) { } began := ch.now() ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{}) - out := ch.judge(ctx, cs) + 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) && ctx.Err() == nil; try++ { + 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(ctx, cs) + 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 { @@ -569,6 +708,26 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) { 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 { @@ -667,7 +826,7 @@ func (ch *Checker) Sweep(ctx context.Context) error { continue } cur.State = delivery.VerdictFail - cur.Failed = delivery.FailedFirst{Machine: "checker", Words: "closed at start: left running, and no candidate holds these heads any more, so nothing will check them"} + 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 diff --git a/modules/mesh-delivery-checker/internal/check/checker_test.go b/modules/mesh-delivery-checker/internal/check/checker_test.go index e8126c9..bc5820b 100644 --- a/modules/mesh-delivery-checker/internal/check/checker_test.go +++ b/modules/mesh-delivery-checker/internal/check/checker_test.go @@ -3,6 +3,7 @@ package check import ( "context" "encoding/json" + "fmt" "strings" "sync" "testing" @@ -333,7 +334,10 @@ func TestTheStatusIsPendingWhileTheCheckRunsAndEndsSuccessOrFailure(t *testing.T } } -func TestSweepClosesARunningKeyNoCandidateHoldsAndLeavesOneItDoes(t *testing.T) { +// No check of heads no candidate holds is left running (issue 1890): at start the sweep ends a key left running; while the +// checker runs, a check whose group moved to other heads is ended at once, unjudged, and writes no status; heads held again +// are checked whole; and a group's check starts only once its heads settled, so a partial set of members is never checked. +func TestHeadsNoCandidateHoldsAreNeverLeftRunning(t *testing.T) { ctx, j := context.Background(), passJudge() r := newRig(t, j) cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) @@ -351,12 +355,91 @@ func TestSweepClosesARunningKeyNoCandidateHoldsAndLeavesOneItDoes(t *testing.T) if err := r.ch.Sweep(ctx); err != nil { t.Fatal(err) } - if v := r.verdict(t, dead); v.State != delivery.VerdictFail || !strings.Contains(v.Failed.Words, "closed at start") { + if v := r.verdict(t, dead); !IsEnded(v) || !strings.Contains(v.Failed.Words, "closed at start") { t.Errorf("the dead key = %+v", v) } if v := r.verdict(t, live); v.State != delivery.VerdictRunning { t.Errorf("the live key = %+v, want it left for the check to take over", v) } + + // A member joins while the check of the lone member's heads runs: that check is ended, its judge's run cancelled. + started, ended := make(chan struct{}), make(chan error, 1) + j = &judgeFunc{f: func(run context.Context, _ *Case, call int) (Outcome, error) { + if call == 1 { + close(started) + <-run.Done() + ended <- context.Cause(run) + return Outcome{}, run.Err() + } + return Outcome{Verdict: passing()}, nil + }} + r = newRig(t, j) + var logged []string + var logMu sync.Mutex + r.ch.Logf = func(f string, a ...any) { logMu.Lock(); logged = append(logged, fmt.Sprintf(f, a...)); logMu.Unlock() } + alone, both := cand, candidateOf(t, "feat/x", pullOf(repoA, 7, head1), pullOf(repoSDK, 3, head2)) + r.ch.Candidate(ctx, "feat%2Fx", alone) + <-started + r.ch.Candidate(ctx, "feat%2Fx", both) + var bothC delivery.Candidate + _ = json.Unmarshal(both, &bothC) + select { + case err := <-ended: + if err == nil || !strings.Contains(err.Error(), "superseded by "+GateHeads(bothC)) { + t.Errorf("the old check's run ended with %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("the check of the heads the candidate left still runs") + } + r.ch.Wait() + if v := r.verdict(t, live); !IsEnded(v) || v.Failed.Words != EndedWords+"superseded by "+GateHeads(bothC) { + t.Errorf("the superseded verdict = %+v", v.Failed) + } + if v := r.verdict(t, keyOf(t, "feat/x", both)); v.State != delivery.VerdictPass { + t.Errorf("the group's verdict = %s", v.State) + } + okey, _ := delivery.CheckerOwedKey(repoA, 7, head1) + if raw, _, _, _ := r.owed.Get(ctx, okey); !strings.Contains(string(raw), `"success"`) { + t.Errorf("the head's status = %s, want the group's success, never the ended check's", raw) + } + logMu.Lock() + said := 0 + for _, l := range logged { + if strings.Contains(l, "ended: superseded") { + said++ + } + } + logMu.Unlock() + if said != 1 { + t.Errorf("the end was said %d times: %v", said, logged) + } + // The member leaves again: the ended heads are held again and checked whole. + r.ch.Candidate(ctx, "feat%2Fx", alone) + r.ch.Wait() + if v := r.verdict(t, live); v.State != delivery.VerdictPass || j.count() != 3 { + t.Errorf("the heads held again = %s after %d judgings, want a pass of a third", v.State, j.count()) + } + + // Members announced one after another: only the settled, whole set is checked. + j = passJudge() + r = newRig(t, j) + r.ch.Settle = time.Hour + sdk := candidateOf(t, "feat/y", pullOf(repoSDK, 4, head2)) + whole := candidateOf(t, "feat/y", pullOf(repoSDK, 4, head2), pullOf(repoA, 8, head1)) + r.ch.Candidate(ctx, "feat%2Fy", sdk) + r.ch.Candidate(ctx, "feat%2Fy", whole) + r.ch.Reconsider(ctx) // a trunk move or a turn while the heads settle starts nothing + if j.count() != 0 { + t.Fatalf("judged %d times before the heads settled", j.count()) + } + r.ch.settleNow() + r.ch.Wait() + if _, _, found, _ := r.verdicts.Get(ctx, keyOf(t, "feat/y", sdk)); found || j.count() != 1 { + t.Errorf("the partial set has a verdict (%v) or the judge ran %d times, want the whole set alone", found, j.count()) + } + if v := r.verdict(t, keyOf(t, "feat/y", whole)); v.State != delivery.VerdictPass { + t.Errorf("the whole set = %s", v.State) + } } func TestTheForgeBridgeIsToldOnceForEachChangeOfAStatus(t *testing.T) { // 1683 (the checker's half of AC5) -- 2.54.0