A try says its end with the state it ended in, kept in the try's own state, so an end after a checker restart is still said; only the checker publishes try events (issue 1921)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i

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

@@ -81,6 +81,9 @@ type Try struct {
 	Tests string `json:"tests,omitempty"`
 	Level string `json:"level,omitempty"`
 	State string `json:"state"`
+	// 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"`
 	// 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.
@@ -519,7 +522,7 @@ type Tryer struct {
 	Say func(ctx context.Context, event string, body tryevent.Event)
 
 	mu       sync.Mutex
-	told     map[string]bool // whole tries whose start was said and whose end was not
+	ending   map[string]bool // tries whose end is being said now
 	wg       sync.WaitGroup
 	inflight map[string]context.CancelFunc
 	seen     map[string]bool // looked at since this process started: a try left running is taken over the first time
@@ -558,12 +561,8 @@ func (tr *Tryer) Put(ctx context.Context, key string, value []byte) {
 		if stop := tr.inflight[key]; stop != nil {
 			stop()
 		}
-		said := tr.told[key]
-		delete(tr.told, key)
 		tr.mu.Unlock()
-		if said { // ended by hand while it ran: the indexer drops its candidate
-			tr.say(ctx, tryevent.Ended, tryevent.Event{ID: key, State: t.State})
-		}
+		tr.sendEnd(ctx, key) // ended by hand while it ran, or its end was written and not said: the indexer drops its candidate
 		return
 	}
 	if tr.inflight[key] != nil || (t.State == TryRunning && tr.seen[key]) {
@@ -619,6 +618,7 @@ func (tr *Tryer) run(ctx context.Context, key string) {
 		return
 	}
 	t.State, t.Started = TryRunning, tr.now().UTC()
+	t.Told = (&TryOptions{Level: t.Level, Run: t.Tests}).Whole() // its start is said below; its end is owed from now on
 	data, err := encodeTry(t)
 	if err != nil {
 		return
@@ -630,17 +630,11 @@ func (tr *Tryer) run(ctx context.Context, key string) {
 		return
 	}
 	progress := &tryWriter{tr: tr, ctx: ctx, key: key, t: t, rev: rev}
-	if (&TryOptions{Level: t.Level, Run: t.Tests}).Whole() {
+	if t.Told {
 		started := tryevent.Event{ID: key}
 		for _, m := range t.MembersOf() {
 			started.Members = append(started.Members, tryevent.Member{Repository: m.Repository, Head: m.Head, Trunk: m.Trunk})
 		}
-		tr.mu.Lock()
-		if tr.told == nil {
-			tr.told = map[string]bool{}
-		}
-		tr.told[key] = true
-		tr.mu.Unlock()
 		tr.say(ctx, tryevent.Started, started)
 	}
 	cs := TryCase(key, t, progress.heard)
@@ -653,13 +647,6 @@ 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()
-	tr.mu.Lock()
-	said := tr.told[key]
-	delete(tr.told, key)
-	tr.mu.Unlock()
-	if said {
-		defer tr.say(ctx, tryevent.Ended, tryevent.Event{ID: key})
-	}
 	v := out.Verdict
 	t.Ended = tr.now().UTC()
 	t.Took = t.Ended.Sub(t.Started).Round(time.Second).String()
@@ -707,6 +694,41 @@ func (tr *Tryer) run(ctx context.Context, key string) {
 		return
 	}
 	tr.logf("%s: %s", key, t.State)
+	tr.sendEnd(ctx, key)
+}
+
+// sendEnd says the end of an ended try whose start was said (Told) with the state it ended in, then clears Told in the try's
+// own state; a checker that stopped between the two says it again at its start. Every end is said so: pass, fail, error and
+// by hand.
+func (tr *Tryer) sendEnd(ctx context.Context, key string) {
+	tr.mu.Lock()
+	if tr.ending[key] {
+		tr.mu.Unlock()
+		return
+	}
+	if tr.ending == nil {
+		tr.ending = map[string]bool{}
+	}
+	tr.ending[key] = true
+	tr.mu.Unlock()
+	defer func() { tr.mu.Lock(); delete(tr.ending, key); tr.mu.Unlock() }()
+	raw, rev, found, err := tr.Tries.Get(ctx, key)
+	if err != nil || !found {
+		return
+	}
+	t, err := DecodeTry(raw)
+	if err != nil || !t.Over() || !t.Told {
+		return
+	}
+	tr.say(ctx, tryevent.Ended, tryevent.Event{ID: key, State: t.State})
+	t.Told = false
+	data, err := encodeTry(t)
+	if err == nil {
+		_, err = tr.Tries.Update(ctx, key, data, rev)
+	}
+	if err != nil && !errors.Is(err, ErrConflict) {
+		tr.logf("%s: its end was said and Told could not be cleared: %v", key, err)
+	}
 }
 
 // say publishes a try event; a failure is said in the log (the indexer finds the open tries again at its start).

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

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -181,16 +182,19 @@ func TestATryIsJudgedAsAGroupsCheckAndOwesTheForgeNothing(t *testing.T) { // ADR
 	written := &writtenTries{memKV: kv}
 	var said []string
 	tr := &Tryer{Tries: written, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) {
-		said = append(said, fmt.Sprintf("%s %s %d", event, e.ID, len(e.Members)))
+		said = append(said, fmt.Sprintf("%s %s %d %s", event, e.ID, len(e.Members), e.State))
 	}}
 	raw, _, _, _ := kv.Get(ctx, id)
 	tr.Put(ctx, id, raw)
 	tr.Wait()
 	// The indexer hears the start with the members at their commits, and the end (it writes and deletes the group's candidate).
-	if len(said) != 2 || said[0] != "try.started "+id+" 2" || said[1] != "try.ended "+id+" 0" {
+	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)
 	}
 	got := readTry(t, kv, id)
+	if got.Told { // the end was said: nothing is owed
+		t.Errorf("the ended try still owes its end: %+v", got)
+	}
 	if open, err := OpenTries(memTries{kv}); err != nil || len(open) != 0 {
 		t.Errorf("open tries after the end = %v, %v", open, err)
 	}
@@ -282,26 +286,24 @@ func TestATryOfOneLevelAndSomeTestsRunsThemAloneAndKeepsNothing(t *testing.T) {
 	if got := readTry(t, kv, id); got.State != TryPass || !strings.Contains(got.Said, "only the tests TestBus|TestStore") {
 		t.Errorf("try = %+v", got)
 	}
-}
-
-func TestATryOfOneLevelKeepsOnlyThatLevel(t *testing.T) {
-	ctx := context.Background()
-	kv := newMemKV()
-	id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1, Level: delivery.LevelUnit}, time.Now())
-	b := newBench(t)
-	tr := &Tryer{Tries: kv, Judge: b.ex}
-	raw, _, _, _ := kv.Get(ctx, id)
-	tr.Put(ctx, id, raw)
-	tr.Wait()
-	keys, _ := b.kept.Keys(ctx)
-	if got := readTry(t, kv, id); got.State != TryPass || len(keys) == 0 {
-		t.Fatalf("try = %+v, kept %v", got, keys)
-	}
-	for _, k := range keys {
-		if !strings.HasPrefix(k, delivery.LevelUnit+".") {
-			t.Errorf("a try of the unit level kept %s", k)
+	t.Run("a try of one level keeps only that level", func(t *testing.T) {
+		kv := newMemKV()
+		id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1, Level: delivery.LevelUnit}, time.Now())
+		b := newBench(t)
+		tr := &Tryer{Tries: kv, Judge: b.ex}
+		raw, _, _, _ := kv.Get(ctx, id)
+		tr.Put(ctx, id, raw)
+		tr.Wait()
+		keys, _ := b.kept.Keys(ctx)
+		if got := readTry(t, kv, id); got.State != TryPass || len(keys) == 0 {
+			t.Fatalf("try = %+v, kept %v", got, keys)
 		}
-	}
+		for _, k := range keys {
+			if !strings.HasPrefix(k, delivery.LevelUnit+".") {
+				t.Errorf("a try of the unit level kept %s", k)
+			}
+		}
+	})
 }
 
 func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T) {
@@ -340,10 +342,18 @@ func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T)
 			comps.v.State = delivery.CompositionPass
 		}
 		id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
-		tr := &Tryer{Tries: kv, Judge: b.ex}
+		var ends []string
+		tr := &Tryer{Tries: kv, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) {
+			if event == tryevent.Ended {
+				ends = append(ends, e.State)
+			}
+		}}
 		raw, _, _, _ := kv.Get(ctx, id)
 		tr.Put(ctx, id, raw)
 		tr.Wait()
+		if len(ends) != 1 || ends[0] != row.state { // every end says its state: error as well as fail
+			t.Errorf("%s: ends said = %v, want once %s", row.name, ends, row.state)
+		}
 		if got := readTry(t, kv, id); got.State != row.state || !strings.Contains(strings.Join(got.Failed, " | ")+got.Said, row.said) {
 			t.Errorf("%s: try = %s, failed %v, said %q, want %s naming %q", row.name, got.State, got.Failed, got.Said, row.state, row.said)
 		}
@@ -353,22 +363,56 @@ func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T)
 
 func TestATryLeftRunningIsTakenOver(t *testing.T) {
 	ctx := context.Background()
-	kv := newMemKV()
-	id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
-	// Left running by a checker that stopped.
-	raw, rev, _, _ := kv.Get(ctx, id)
-	cur, _ := DecodeTry(raw)
-	cur.State = TryRunning
-	data, _ := encodeTry(cur)
-	if _, err := kv.Update(ctx, id, data, rev); err != nil {
-		t.Fatal(err)
-	}
-	b := newBench(t)
-	wholeBench(b, delivery.Candidate{})
-	tr := &Tryer{Tries: kv, Judge: b.ex}
-	tr.Reconsider(ctx)
-	tr.Wait()
-	if got := readTry(t, kv, id); got.State != TryPass {
-		t.Errorf("the try left running = %+v, want taken over and passed", got)
+	// 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: "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) {
+			if _, err := EndTry(endable{kv}, id, "stuck", time.Now()); err != nil {
+				t.Fatal(err)
+			}
+		}},
+	} {
+		t.Run(row.name, func(t *testing.T) {
+			kv := newMemKV()
+			id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
+			// Left running by a checker that stopped.
+			raw, rev, _, _ := kv.Get(ctx, id)
+			cur, _ := DecodeTry(raw)
+			cur.State, cur.Told = TryRunning, row.told
+			data, _ := encodeTry(cur)
+			if _, err := kv.Update(ctx, id, data, rev); err != nil {
+				t.Fatal(err)
+			}
+			if row.after != nil {
+				row.after(t, kv, id)
+			}
+			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) {
+				if event == tryevent.Ended {
+					ends = append(ends, e.State)
+				}
+			}}
+			tr.Reconsider(ctx)
+			tr.Wait()
+			tr.Reconsider(ctx) // said once: the second look finds nothing owed
+			tr.Wait()
+			if row.after == nil {
+				if got := readTry(t, kv, id); got.State != TryPass {
+					t.Errorf("the try left running = %+v, want taken over and passed", got)
+				}
+			}
+			if !slices.Equal(ends, row.ends) || readTry(t, kv, id).Told {
+				t.Errorf("ends said = %v, want %v, and nothing owed after", ends, row.ends)
+			}
+		})
 	}
 }

modules/mesh-delivery-checker/manifest_test.go

@@ -134,6 +134,32 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
 	if want := []stream{{Name: "checks", Events: []string{"preview.**", "owed.**", "try.**"}}}; !reflect.DeepEqual(m.Streams, want) {
 		t.Errorf("streams = %+v, want %+v", m.Streams, want)
 	}
+	// Only the checker's account publishes try.* on its stream checks: the stream is declared by its owner alone, and no other
+	// module of the repository declares a stream carrying a try event (the indexer and the others only consume them).
+	siblings, err := filepath.Glob("../*/module.json")
+	if err != nil || len(siblings) < 2 {
+		t.Fatalf("the sibling manifests = %v, %v", siblings, err)
+	}
+	for _, f := range siblings {
+		raw, err := os.ReadFile(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var other struct {
+			Module  string   `json:"module"`
+			Streams []stream `json:"streams"`
+		}
+		if err := json.Unmarshal(raw, &other); err != nil {
+			t.Fatalf("%s: %v", f, err)
+		}
+		for _, st := range other.Streams {
+			for _, e := range st.Events {
+				if strings.HasPrefix(e, "try.") && other.Module != m.Module {
+					t.Errorf("%s declares the stream %s with %s: only the checker's account publishes try events", other.Module, st.Name, e)
+				}
+			}
+		}
+	}
 	// The read tools (issue 1686) are served by the tools bundle, not by the checker's process.
 	// The try's tools (ADR 0501 §5; the end by hand, issue 734) beside them: mesh-controller's try calls them. A lost check
 	// is asked again by hand (checker_ask_again).