mesh-delivery-checker: stop a check whose heads its group no longer holds and end it failed, superseded; judge a group's open members by the verdict it had once others merged, so their merge is not refused (issue 1921, issue 1791, try speed S9)

modules/mesh-delivery-checker/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-checker
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024214154-bc16740feb0d
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024231716-571ac521f053
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-checker/go.sum

@@ -1,5 +1,7 @@
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024214154-bc16740feb0d h1:oJREw0ggKqJWJ5DEISSL0i2ZC5XqdXAidxo9NJ74TCc=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024214154-bc16740feb0d/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024231716-571ac521f053 h1:8Kbp3aJG3jGndDOfmGdMSDgJVEcc1hyu921hVnnlnlQ=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024231716-571ac521f053/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=

modules/mesh-delivery-checker/internal/check/case.go

@@ -82,6 +82,9 @@ func CaseOf(c delivery.Candidate, trunk map[string]string) *Case {
 	}
 	cs := &Case{Key: key, Group: c.Group, HeadsHash: c.HeadsHash, Heads: GateHeads(c), Trunk: map[string]string{}, Progress: &Progress{}}
 	for _, p := range c.Pulls {
+		if p.Merged {
+			continue // merged while the group waits for its other members: judged by the verdict the group had (issue 1791)
+		}
 		switch head, known := trunk[p.Repository]; {
 		case !known || head == "":
 			cs.Left = append(cs.Left, fmt.Sprintf("%s#%d: the trunk head of %s is not known", p.Repository, p.Number, p.Repository))
@@ -119,7 +122,18 @@ func GateHeads(c delivery.Candidate) string {
 }
 
 // sameTrunk says whether two verdicts' trunk heads are the same.
-func sameTrunk(a, b map[string]string) bool { return maps.Equal(a, b) }
+// sameTrunk says whether a verdict judged on the trunk heads judged stands for a case on the trunk heads now: every
+// repository of the case at the head it was judged on. A repository the verdict names and the case does not is a member
+// that merged since (issue 1791): its own merge moved that trunk, and the members left are judged by the verdict the
+// group had.
+func sameTrunk(judged, now map[string]string) bool {
+	for repo, head := range now {
+		if judged[repo] != head {
+			return false
+		}
+	}
+	return len(now) > 0 || len(judged) == 0
+}
 
 // Trunks holds the trunk head of each repository, as the gate's source.main says (a watch puts it in).
 type Trunks struct {

modules/mesh-delivery-checker/internal/check/checker.go

@@ -83,12 +83,24 @@ type Checker struct {
 	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
+	// current is, by candidates key, the verdict key of the heads the group holds now; running holds, by candidates key,
+	// the stop of each check that runs, by its verdict key (issue 1791; issue 1921, try speed S9).
+	current map[string]string
+	running map[string]map[string]context.CancelCauseFunc
+}
+
+// superseded is why a check whose heads the group no longer holds was stopped: the heads it holds now.
+type superseded struct{ by string }
+
+func (s superseded) Error() string {
+	return "superseded: the group holds other heads now (" + s.by + "), whose check judges it"
 }
 
 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 = map[string]uint64{}
+		ch.current, ch.running = map[string]string{}, map[string]map[string]context.CancelCauseFunc{}
 	})
 }
 
@@ -126,6 +138,15 @@ func (ch *Checker) Candidate(ctx context.Context, key string, value []byte) {
 	}
 	ch.mu.Lock()
 	ch.known[key] = c
+	if vk, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash); err == nil {
+		ch.current[key] = vk
+		// A check of heads the group no longer holds is stopped: its asks are not put, and it ends superseded.
+		for running, stop := range ch.running[key] {
+			if running != vk {
+				stop(superseded{by: c.HeadsHash})
+			}
+		}
+	}
 	ch.mu.Unlock()
 	ch.consider(ctx, key, c)
 }
@@ -329,12 +350,48 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
 	if cl == nil {
 		return
 	}
+	run, stop := context.WithCancelCause(ctx)
+	defer stop(nil)
+	ch.mu.Lock()
+	now := ch.current[key]
+	if ch.running[key] == nil {
+		ch.running[key] = map[string]context.CancelCauseFunc{}
+	}
+	ch.running[key][cs.Key] = stop
+	ch.mu.Unlock()
+	defer func() {
+		ch.mu.Lock()
+		delete(ch.running[key], cs.Key)
+		if len(ch.running[key]) == 0 {
+			delete(ch.running, key)
+		}
+		ch.mu.Unlock()
+	}()
+	if now != "" && now != cs.Key {
+		stop(superseded{by: now}) // the group moved on before the check began
+	}
 	began := ch.now()
-	ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{})
-	out := ch.judge(ctx, cs)
+	if run.Err() == nil {
+		ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{})
+	}
+	out := ch.judge(run, cs)
 	if ctx.Err() != nil {
 		return
 	}
+	var sup superseded
+	if errors.As(context.Cause(run), &sup) {
+		// Stopped for newer heads: the verdict ends failed, not running, and no status is put, as the newer check owns the
+		// status of the heads they share.
+		v := delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictFail,
+			Failed: delivery.FailedFirst{Machine: "checker", Words: sup.Error()}}
+		if data, err := delivery.EncodeVerdict(v); err == nil {
+			if _, err := ch.Verdicts.Update(ctx, cs.Key, data, cl.rev); err != nil {
+				ch.logf("%s: the superseded verdict could not be written: %v", cs.Key, err)
+			}
+		}
+		ch.logf("%s: %s", cs.Key, sup.Error())
+		return
+	}
 	v := out.Verdict
 	v.Group, v.Heads, v.Trunk = cs.Group, cs.Heads, cs.Trunk
 	if v.Took.Queue == 0 && !cs.Queued.IsZero() && began.After(cs.Queued) {

modules/mesh-delivery-checker/internal/check/judge.go

@@ -400,9 +400,13 @@ func (st *state) target(ctx context.Context, p Pull, bm string) *target {
 	return t
 }
 
-// put puts the target's check ask on its build seat's stream.
+// put puts the target's check ask on its build seat's stream; a check that was stopped (its group holds other heads now,
+// issue 1791) puts none.
 func (st *state) put(ctx context.Context, t *target) *delivery.Verdict {
 	p := t.pull
+	if err := context.Cause(ctx); err != nil {
+		return fail(t.module, fmt.Sprintf("the check of %s#%d was not asked: %v", p.Repository, p.Number, err))
+	}
 	job := &delivery.CheckJob{Fetch: t.fetchArgs, Levels: st.levelsOf(), Kept: true}
 	if t.runner != "" {
 		job.Runner, job.RunnerWhen = t.runner, buildGoDir

modules/mesh-delivery-checker/internal/check/stale_test.go

@@ -0,0 +1,102 @@
+package check
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+const repoC = "novox/mesh-controller"
+
+func statusOf(t *testing.T, r *rig, repo string, n int, head string) string {
+	t.Helper()
+	key, _ := delivery.CheckerOwedKey(repo, n, head)
+	raw, _, found, _ := r.owed.Get(context.Background(), key)
+	if !found {
+		return ""
+	}
+	o, err := delivery.DecodeCheckerOwed(raw)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return o.State
+}
+
+// 1791 note 9 (issue 1921, try speed S9): two of a group's three members merge. The candidate keeps them, marked merged,
+// with the same heads; their trunks moved by their own merges. No new check is asked, and the member left keeps its
+// passed status, so the merge of the rest is not refused.
+func TestTwoMembersMergingAskNoNewCheckAndTheThirdKeepsItsPass(t *testing.T) {
+	ctx, j := context.Background(), passJudge()
+	r := newRig(t, j)
+	r.setTrunks(t, map[string]string{repoA: trunkA, repoSDK: trunkS, repoC: trunkA})
+	a, s, c := pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2), pullOf(repoC, 3, head1)
+	r.ch.Candidate(ctx, "feat%2Fx", candidateOf(t, "feat/x", a, s, c))
+	r.ch.Wait()
+	if j.count() != 1 || statusOf(t, r, repoC, 3, head1) != "success" {
+		t.Fatalf("judged %d times, the third's status %q", j.count(), statusOf(t, r, repoC, 3, head1))
+	}
+	a.Merged, s.Merged = true, true
+	r.ch.Candidate(ctx, "feat%2Fx", candidateOf(t, "feat/x", a, s, c))
+	r.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS2, repoC: trunkA})
+	r.ch.TrunkMoved(ctx)
+	r.ch.Wait()
+	if j.count() != 1 {
+		t.Errorf("a new check was asked after two members merged: judged %d times", j.count())
+	}
+	if got := statusOf(t, r, repoC, 3, head1); got != "success" {
+		t.Errorf("the third member's status is %q, want success", got)
+	}
+	// The trunk of the member left moves: that is a re-judge as before.
+	r.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS2, repoC: trunkA2})
+	r.ch.TrunkMoved(ctx)
+	r.ch.Wait()
+	if j.count() != 2 {
+		t.Errorf("the member left's trunk moved and it was judged %d times, want again", j.count())
+	}
+}
+
+// 1791 note 8 (issue 1921, try speed S9): a group opened member by member writes a candidate for each partial set. When
+// the group's candidate moves to new heads, the check of the older heads is stopped and ends failed, superseded, instead
+// of running on or staying running; it puts no status, as the newer check owns the heads they share.
+func TestACheckWhoseHeadsTheGroupNoLongerHoldsIsStoppedAndEndsSuperseded(t *testing.T) {
+	ctx := context.Background()
+	started := make(chan struct{})
+	j := &judgeFunc{f: func(ctx context.Context, c *Case, call int) (Outcome, error) {
+		if len(c.Pulls) == 1 {
+			close(started)
+			select {
+			case <-ctx.Done():
+				return Outcome{Verdict: failure("checker", ctx.Err().Error())}, nil
+			case <-time.After(3 * time.Second):
+				return Outcome{Verdict: passing()}, nil // never stopped
+			}
+		}
+		return Outcome{Verdict: passing()}, nil
+	}}
+	r := newRig(t, j)
+	partial := candidateOf(t, "feat/x", pullOf(repoA, 7, head1))
+	whole := candidateOf(t, "feat/x", pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2))
+	r.ch.Candidate(ctx, "feat%2Fx", partial)
+	<-started
+	r.ch.Candidate(ctx, "feat%2Fx", whole)
+	done := make(chan struct{})
+	go func() { r.ch.Wait(); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("the check of the older heads was not stopped")
+	}
+	old := r.verdict(t, keyOf(t, "feat/x", partial))
+	if old.State != delivery.VerdictFail || !strings.Contains(old.Failed.Words, "superseded") {
+		t.Errorf("the older heads' verdict = %+v, want failed, superseded", old)
+	}
+	if v := r.verdict(t, keyOf(t, "feat/x", whole)); v.State != delivery.VerdictPass {
+		t.Errorf("the newer heads' verdict = %+v", v)
+	}
+	if got := statusOf(t, r, repoA, 7, head1); got != "success" {
+		t.Errorf("the shared head's status is %q, want the newer check's success", got)
+	}
+}