From f9578cb7b41893e1c3b1e69738722a917c1d8425 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 19:24:45 +0100 Subject: [PATCH] A try's end is said again until the stream acknowledges it; a waiting try says why; a hand recovery never overlaps a try event (issue 1921, ADR 0639) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-checker/main.go | 7 +- .../internal/check/case.go | 2 + .../internal/check/judge.go | 4 + .../internal/check/try.go | 34 ++++++-- .../internal/check/try_test.go | 79 +++++++++++++++++-- .../internal/index/candidates.go | 1 + .../internal/index/previews_test.go | 17 ++++ .../internal/index/trycandidates.go | 15 +++- 8 files changed, 141 insertions(+), 18 deletions(-) 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 63bcfcae..d464578d 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -192,11 +192,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The tries (ADR 0501 §5) are judged by the same judge, held by the same deliveries. tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, NotTaken: stepBound, Logf: logger.Printf, // A whole try's start and end are said on the stream checks, for the indexer to write and delete the candidate of its group. - Say: func(ctx context.Context, event string, body tryevent.Event) { + Say: func(ctx context.Context, event string, body tryevent.Event) error { data, _ := json.Marshal(body) - if _, err := js.Publish(ctx, eventSubject(event), data, jetstream.WithRetryAttempts(10), jetstream.WithRetryWait(time.Second)); err != nil { - logger.Printf("%s of %s could not be said: %v (the indexer finds the open tries at its start)", event, body.ID, err) - } + _, err := js.Publish(ctx, eventSubject(event), data, jetstream.WithRetryAttempts(10), jetstream.WithRetryWait(time.Second)) + return err }} announce := func(ctx context.Context, event, key string) { body, _ := json.Marshal(map[string]string{"key": key}) diff --git a/modules/mesh-delivery-checker/internal/check/case.go b/modules/mesh-delivery-checker/internal/check/case.go index 3f80136f..050bd601 100644 --- a/modules/mesh-delivery-checker/internal/check/case.go +++ b/modules/mesh-delivery-checker/internal/check/case.go @@ -55,6 +55,8 @@ type Case struct { Watch func(askID string, step *delivery.Step) // Progress says what the check is doing, so a check that ran past its bound names the step it was stuck at. Progress *Progress + // Wait says what a whole try waits for now ("" when nothing); the try keeps it for try-read. Nil for a pull request's check. + Wait func(why string) // Kind is what asked the check (KindNew, KindNodes, KindTrunk): a carry after a trunk or sibling move is asked on the // build seats' recheck stream, every other check on check (ADR 0536 §4, ADR 0548 §5). Kind string diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index 267f40c0..391d9f27 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -283,6 +283,10 @@ func (st *state) takeTryCandidate(ctx context.Context) { no("no preview of a merge of " + strings.Join(st.unpreviewed, ", ") + " (its head does not merge or its fetch did not answer)") return } + if st.c.Wait != nil { + st.c.Wait("the indexer's candidate of the group at its heads (up to " + st.j.buildWait().String() + ")") + defer st.c.Wait("") + } cand, err := st.j.Candidates.TryCandidate(ctx, st.c.Group, st.c.HeadsHash, st.j.buildWait()) if err != nil { no(err.Error()) diff --git a/modules/mesh-delivery-checker/internal/check/try.go b/modules/mesh-delivery-checker/internal/check/try.go index b2afc02c..16b64403 100644 --- a/modules/mesh-delivery-checker/internal/check/try.go +++ b/modules/mesh-delivery-checker/internal/check/try.go @@ -84,6 +84,9 @@ type Try struct { // Told is true from the moment a whole try's start is said on the checker's stream until its end is said there: kept in the // try's own state, so an end that is written and not said (the checker stopped between) is said when the checker starts. Told bool `json:"told,omitempty"` + // Waiting is what the running try waits for now (its group's candidate from the indexer, ADR 0639 §5), empty when it waits for nothing; + // try-read gives it, so a try that sits still says why. + Waiting string `json:"waiting,omitempty"` // Said is the try's report in words: what failed first, the tests run, kept and skipped per level, the time of each part. Said string `json:"said,omitempty"` // Took is the time from its start to its end, as Go writes a duration. @@ -180,6 +183,17 @@ func (w *tryWriter) heard(askID string, step *delivery.Step) { } } +// waiting writes what the try waits for now ("" when it waits for nothing). +func (w *tryWriter) waiting(why string) { + w.mu.Lock() + defer w.mu.Unlock() + if w.stopped || w.t.Waiting == why { + return + } + w.t.Waiting = why + w.write() +} + // write writes the record at the revision written last; a refusal stops the writes. func (w *tryWriter) write() { data, err := encodeTry(w.t) @@ -519,7 +533,8 @@ type Tryer struct { // Say publishes an event of the checker's stream (tryevent.Started, tryevent.Ended) with its body; nil says nothing. The // indexer, which hears them, writes and deletes the candidate of a whole try's group. - Say func(ctx context.Context, event string, body tryevent.Event) + // It answers the publish's error, nil only when the stream acknowledged it: an end not acknowledged is said again. + Say func(ctx context.Context, event string, body tryevent.Event) error mu sync.Mutex ending map[string]bool // tries whose end is being said now @@ -638,6 +653,7 @@ func (tr *Tryer) run(ctx context.Context, key string) { tr.say(ctx, tryevent.Started, started) } cs := TryCase(key, t, progress.heard) + cs.Wait = progress.waiting judging := ctx if tr.NotTaken > 0 { judging = buildrun.NotTakenBound(ctx, tr.NotTaken) @@ -647,6 +663,7 @@ func (tr *Tryer) run(ctx context.Context, key string) { return // left running: the checker takes it over when it starts; or ended by hand, which wrote its end } t, rev = progress.end() + t.Waiting = "" v := out.Verdict t.Ended = tr.now().UTC() t.Took = t.Ended.Sub(t.Started).Round(time.Second).String() @@ -720,7 +737,9 @@ func (tr *Tryer) sendEnd(ctx context.Context, key string) { if err != nil || !t.Over() || !t.Told { return } - tr.say(ctx, tryevent.Ended, tryevent.Event{ID: key, State: t.State}) + if tr.say(ctx, tryevent.Ended, tryevent.Event{ID: key, State: t.State}) != nil { + return // not acknowledged: Told stays, and the next Put or Reconsider says the end again + } t.Told = false data, err := encodeTry(t) if err == nil { @@ -732,10 +751,15 @@ func (tr *Tryer) sendEnd(ctx context.Context, key string) { } // say publishes a try event; a failure is said in the log (the indexer finds the open tries again at its start). -func (tr *Tryer) say(ctx context.Context, event string, body tryevent.Event) { - if tr.Say != nil { - tr.Say(ctx, event, body) +func (tr *Tryer) say(ctx context.Context, event string, body tryevent.Event) error { + if tr.Say == nil { + return nil } + err := tr.Say(ctx, event, body) + if err != nil { + tr.logf("%s of %s could not be said: %v", event, body.ID, err) + } + return err } // OpenTries are every try that waits or runs now: what checker_tries answers, so a part that missed an event (the indexer at diff --git a/modules/mesh-delivery-checker/internal/check/try_test.go b/modules/mesh-delivery-checker/internal/check/try_test.go index 9aabfd73..74a4681f 100644 --- a/modules/mesh-delivery-checker/internal/check/try_test.go +++ b/modules/mesh-delivery-checker/internal/check/try_test.go @@ -152,9 +152,14 @@ type fakeTryCandidates struct { cand delivery.Candidate err error asked []string + // during is called while the try waits for the candidate. + during func() } func (f *fakeTryCandidates) TryCandidate(_ context.Context, group, heads string, _ time.Duration) (delivery.Candidate, error) { + if f.during != nil { + f.during() + } f.asked = append(f.asked, group+"."+heads) return f.cand, f.err } @@ -180,9 +185,12 @@ func TestATryIsJudgedAsAGroupsCheckAndOwesTheForgeNothing(t *testing.T) { // ADR b := newBench(t) comps, cands := wholeBench(b, delivery.Candidate{FPSet: "fps-of-the-try"}) written := &writtenTries{memKV: kv} + var waiting string + cands.during = func() { waiting = readTry(t, kv, id).Waiting } // ADR 0639 §5: the stored try says what it waits for var said []string - tr := &Tryer{Tries: written, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) { + tr := &Tryer{Tries: written, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) error { said = append(said, fmt.Sprintf("%s %s %d %s", event, e.ID, len(e.Members), e.State)) + return nil }} raw, _, _, _ := kv.Get(ctx, id) tr.Put(ctx, id, raw) @@ -191,7 +199,25 @@ func TestATryIsJudgedAsAGroupsCheckAndOwesTheForgeNothing(t *testing.T) { // ADR if len(said) != 2 || said[0] != "try.started "+id+" 2 " || said[1] != "try.ended "+id+" 0 "+TryPass { t.Errorf("events = %v, want the start with 2 members and the end", said) } + // What the indexer reads: tryevent.Decode accepts the events as said and refuses a body that is no try's. + for _, row := range []struct { + name, event, body string + ok bool + }{ + {"a start with its members", tryevent.Started, `{"id":"` + id + `","members":[{"repository":"` + repoA + `","head":"` + head1 + `","trunk":"` + trunkA + `"}]}`, true}, + {"an end with its state", tryevent.Ended, `{"id":"` + id + `","state":"pass"}`, true}, + {"an id that is no try's", tryevent.Ended, `{"id":"pull-7","state":"pass"}`, false}, + {"a start with no member", tryevent.Started, `{"id":"` + id + `"}`, false}, + {"a body that is no JSON", tryevent.Ended, `try-7`, false}, + } { + if e, err := tryevent.Decode(row.event, []byte(row.body)); (err == nil) != row.ok || (row.ok && e.ID != id) { + t.Errorf("Decode of %s = %+v, %v, want ok=%v", row.name, e, err, row.ok) + } + } got := readTry(t, kv, id) + if !strings.Contains(waiting, "candidate") || got.Waiting != "" { + t.Errorf("waiting while the candidate was awaited = %q, after the end %q: want the reason, then none", waiting, got.Waiting) + } if got.Told { // the end was said: nothing is owed t.Errorf("the ended try still owes its end: %+v", got) } @@ -264,8 +290,9 @@ func TestATryOfOneLevelAndSomeTestsRunsThemAloneAndKeepsNothing(t *testing.T) { t.Fatal(err) } b := newBench(t) - tr := &Tryer{Tries: kv, Judge: b.ex, Say: func(context.Context, string, tryevent.Event) { + tr := &Tryer{Tries: kv, Judge: b.ex, Say: func(context.Context, string, tryevent.Event) error { t.Error("a try of one level and some tests said an event: it has no candidate") + return nil }} raw, _, _, _ := kv.Get(ctx, id) tr.Put(ctx, id, raw) @@ -343,10 +370,11 @@ func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T) } id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now()) var ends []string - tr := &Tryer{Tries: kv, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) { + tr := &Tryer{Tries: kv, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) error { if event == tryevent.Ended { ends = append(ends, e.State) } + return nil }} raw, _, _, _ := kv.Get(ctx, id) tr.Put(ctx, id, raw) @@ -366,11 +394,13 @@ func TestATryLeftRunningIsTakenOver(t *testing.T) { // Whether the start was said is in the try's own state (Told), so a checker that starts later says the end of a try it did // not start, whatever way the try ended. for _, row := range []struct { - name string - told bool - after func(t *testing.T, kv *memKV, id string) - ends []string + name string + told bool + failFirst bool // the stream does not acknowledge the first end said + after func(t *testing.T, kv *memKV, id string) + ends []string }{ + {name: "the first publish of the end is not acknowledged", told: true, failFirst: true, ends: []string{TryPass}}, {name: "left running after its start was said", told: true, ends: []string{TryPass}}, {name: "left running with no start said", told: false, ends: []string{TryPass}}, {name: "ended by hand after the start was said", told: true, ends: []string{TryError}, after: func(t *testing.T, kv *memKV, id string) { @@ -396,13 +426,22 @@ func TestATryLeftRunningIsTakenOver(t *testing.T) { b := newBench(t) wholeBench(b, delivery.Candidate{}) var ends []string - tr := &Tryer{Tries: kv, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) { + failed := false + tr := &Tryer{Tries: kv, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) error { if event == tryevent.Ended { + if row.failFirst && !failed { + failed = true + return errors.New("the stream did not acknowledge") + } ends = append(ends, e.State) } + return nil }} tr.Reconsider(ctx) tr.Wait() + if row.failFirst && !readTry(t, kv, id).Told { + t.Errorf("an end the stream did not acknowledge cleared Told: it is never said again") + } tr.Reconsider(ctx) // said once: the second look finds nothing owed tr.Wait() if row.after == nil { @@ -415,4 +454,28 @@ func TestATryLeftRunningIsTakenOver(t *testing.T) { } }) } + // An ended try is kept TryKeep and removed by the next Reconsider after. + t.Run("an ended try is kept seven days", func(t *testing.T) { + kv := newMemKV() + id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now()) + raw, rev, _, _ := kv.Get(ctx, id) + cur, _ := DecodeTry(raw) + ended := time.Now() + cur.State, cur.Ended = TryPass, ended + data, _ := encodeTry(cur) + if _, err := kv.Update(ctx, id, data, rev); err != nil { + t.Fatal(err) + } + now := ended.Add(TryKeep - time.Hour) + tr := &Tryer{Tries: kv, Now: func() time.Time { return now }} + tr.Reconsider(ctx) + if _, _, found, _ := kv.Get(ctx, id); !found { + t.Fatal("the try was removed before TryKeep") + } + now = ended.Add(TryKeep + time.Hour) + tr.Reconsider(ctx) + if _, _, found, _ := kv.Get(ctx, id); found { + t.Error("the try was kept past TryKeep") + } + }) } diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 137609cf..b8bfc4df 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -147,6 +147,7 @@ type Indexer struct { OpenTries func(ctx context.Context) ([]tryevent.Open, error) // TryRecovery keeps how the last recovery of the open tries went (trycandidates.go); nil keeps none. TryRecovery Store + tryMu sync.Mutex // one try event or recovery at a time (trycandidates.go) recoveryFailed atomic.Bool // the last recovery of the open tries failed (trycandidates.go) tries map[string][]TryMember // the whole tries that wait or run, by try id (trycandidates.go); guarded by mu pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go) diff --git a/modules/mesh-delivery-indexer/internal/index/previews_test.go b/modules/mesh-delivery-indexer/internal/index/previews_test.go index 6dfe4310..3edcb9ef 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews_test.go +++ b/modules/mesh-delivery-indexer/internal/index/previews_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "os" + "runtime" "slices" "strings" "testing" @@ -385,6 +386,22 @@ func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsW if err := ix.EndTry(ctx, "try-7"); err != nil || len(held()) != 0 { t.Errorf("an ended try: %v, candidates %v, want none", err, held()) } + // The hand recovery never overlaps a try event: a start heard while the call to the checker is out writes its candidate + // before or after the sweep, never between the call and the sweep (which would delete it, the try not being listed). + startDone := make(chan error, 1) + ix.OpenTries = func(context.Context) ([]tryevent.Open, error) { + go func() { startDone <- ix.StartTry(ctx, "try-12", members) }() + for i := 0; i < 2000 && len(held()) == 0; i++ { + runtime.Gosched() // lets the start write if nothing holds it back + } + return nil, nil + } + if r := ix.RecoverTriesByHand(ctx, "lead", "a test"); !r.Stored { + t.Fatalf("the hand recovery = %+v", r) + } + if err := <-startDone; err != nil || len(held()) != 1 || held()[0] != "try-12" { + t.Errorf("a start heard during the hand recovery: %v, candidates %v, want try-12 kept", err, held()) + } }) } } diff --git a/modules/mesh-delivery-indexer/internal/index/trycandidates.go b/modules/mesh-delivery-indexer/internal/index/trycandidates.go index 75a71b51..0cec912b 100644 --- a/modules/mesh-delivery-indexer/internal/index/trycandidates.go +++ b/modules/mesh-delivery-indexer/internal/index/trycandidates.go @@ -57,6 +57,8 @@ type TryMember struct{ Repository, Head, Trunk string } // StartTry takes a whole try in, as the checker said it started (tryevent.Started): its members at their commits are kept, and // its candidate is written once every member's preview is held. The checker's state is never read (ADR 0596). func (ix *Indexer) StartTry(ctx context.Context, id string, members []TryMember) error { + ix.tryMu.Lock() + defer ix.tryMu.Unlock() ix.retryRecovery(ctx) return ix.startTry(ctx, id, members) } @@ -64,7 +66,7 @@ func (ix *Indexer) StartTry(ctx context.Context, id string, members []TryMember) // retryRecovery runs the recovery again on a try event when the last one failed: the failure stays said until one succeeds. func (ix *Indexer) retryRecovery(ctx context.Context) { if ix.recoveryFailed.Load() && ix.OpenTries != nil { - _ = ix.RecoverTries(ctx) + _ = ix.recoverNow(ctx) } } @@ -83,6 +85,8 @@ func (ix *Indexer) startTry(ctx context.Context, id string, members []TryMember) // EndTry takes a try out, as the checker said it ended (tryevent.Ended), and deletes its candidate. func (ix *Indexer) EndTry(ctx context.Context, id string) error { + ix.tryMu.Lock() + defer ix.tryMu.Unlock() ix.retryRecovery(ctx) return ix.endTry(ctx, id) } @@ -103,7 +107,16 @@ func (ix *Indexer) endTry(ctx context.Context, id string) error { // // A recovery that fails is said loudly and kept (the try-recovery state, read by indexer_try_candidates) until one succeeds: // the events are still heard after it, and indexer_recover_tries runs it again by hand. +// +// A recovery and a try event never overlap (tryMu): a sweep that deleted the candidate a start event wrote after the call to the +// checker would leave a running try with none. func (ix *Indexer) RecoverTries(ctx context.Context) error { + ix.tryMu.Lock() + defer ix.tryMu.Unlock() + return ix.recoverNow(ctx) +} + +func (ix *Indexer) recoverNow(ctx context.Context) error { if ix.OpenTries == nil || ix.Store == nil { return nil }