Merge pull request 'The checker ends a check whose heads no candidate holds, and starts a group's check once its heads settle (issue 1890)' (#79) from fix/1890-end-superseded-checks into main

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
 	}

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.

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]
+			more := ch.again[vkey]
-			delete(ch.again, cs.Key)
+			delete(ch.again, vkey)
-			if !more || ctx.Err() != nil {
+			ch.mu.Unlock()
-				delete(ch.inflight, cs.Key)
+			if !more || run.Err() != nil {
-				ch.mu.Unlock()
 				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):
+	case IsEnded(cur), ch.stale(cur, cs):
-		retake = true
+		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
+// 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.
+// because ctx did is left running: the checker takes it again when it starts. A check whose run was ended because no
-func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
+// 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

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)