diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index 6540a88..faec5b8 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/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 ) diff --git a/modules/mesh-delivery-checker/go.sum b/modules/mesh-delivery-checker/go.sum index e514275..c5c11c1 100644 --- a/modules/mesh-delivery-checker/go.sum +++ b/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= diff --git a/modules/mesh-delivery-checker/internal/check/case.go b/modules/mesh-delivery-checker/internal/check/case.go index 1c99029..ddc5eca 100644 --- a/modules/mesh-delivery-checker/internal/check/case.go +++ b/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 { diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index d116005..b4ff70f 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/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) { diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index 36033b6..9b7ff24 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/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 diff --git a/modules/mesh-delivery-checker/internal/check/stale_test.go b/modules/mesh-delivery-checker/internal/check/stale_test.go new file mode 100644 index 0000000..26f1224 --- /dev/null +++ b/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) + } +}