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

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})

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

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())

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

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")
+		}
+	})
 }

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)

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())
+			}
 		})
 	}
 }

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
 	}