A try's end is said on a context its own stop does not cancel; the checker's test bed holds the streams its manifest declares (issue 1921)

The end a run writes comes back on the key's watch, where Put stops the run's context and leaves the saying to the run, so the end was never said and the indexer kept the try's candidate. And the bed's stream checks lacked try.**, so every whole try's say waited out ten retries (10 s each), which on a loaded check ran the test past its 20 s bound ("the held try's ask was never taken"). The bed now makes the streams from module.json, and the test reads both events on the stream.

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

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go

@@ -41,8 +41,10 @@ import (
 	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg"
 	"git.novox.be/novox/mesh-delivery/sdk/buildstep"
 	"git.novox.be/novox/mesh-sdk/go/announce"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/keys"
 	"git.novox.be/novox/mesh-sdk/go/process"
 	"git.novox.be/novox/mesh-sdk/go/views/viewstest"
@@ -434,15 +436,30 @@ func startBed(t *testing.T) *bed {
 	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: buildrun.DeadLettersStream, Subjects: []string{"mesh.events.dead.>"}}); err != nil {
 		t.Fatal(err)
 	}
-	// The checker's streams as its manifest declares them (ADR 0619, ADR 0620): checks, which the indexer and the forge
-	// bridge hear, and asks, with verdict and tag-wanted at the subjects the mesh-delivery-checker definition names.
-	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHECKS", Subjects: []string{meshdeliverychecker.HolderSubjectPreviewWritten(check.CheckerModule)}}); err != nil {
+	// The checker's streams as its manifest declares them (ADR 0619, ADR 0620), read from it so the bed holds every event
+	// it declares: checks, which the indexer and the forge bridge hear (preview, owed and a whole try's start and end, whose
+	// say would otherwise wait out its retries on a subject no stream holds), and asks, with verdict and tag-wanted.
+	raw, err := os.ReadFile("../../module.json")
+	if err != nil {
 		t.Fatal(err)
 	}
-	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASKS", Subjects: []string{meshdeliverychecker.HolderSubjectVerdict(check.CheckerModule),
-		meshdeliverychecker.HolderSubjectTagWanted(check.CheckerModule)}}); err != nil {
+	m, err := catalogue.Decode(raw)
+	if err != nil {
 		t.Fatal(err)
 	}
+	for _, s := range m.Streams {
+		var subjects []string
+		for _, e := range s.Events {
+			subjects = append(subjects, process.ModuleStreamSubject(check.CheckerModule, strings.ReplaceAll(e, "**", ">")))
+		}
+		if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: s.BusName(), Subjects: subjects}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if !slices.ContainsFunc(m.Streams, func(s catalogue.StreamDeclaration) bool { return s.Name == "checks" }) ||
+		meshdeliverychecker.HolderSubjectVerdict(check.CheckerModule) != process.ModuleStreamSubject(check.CheckerModule, "verdict") {
+		t.Fatalf("the checker's manifest declares the streams %+v, want checks, and asks at the definition's subjects", m.Streams)
+	}
 	// The shared events stream with the seats' and modules' events (the package registry seats' published among them), and the checker's durable consumer of it.
 	if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil {
 		t.Fatal(err)
@@ -1032,6 +1049,22 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) {
 	if b.fake.count("mesh-build-go.build_go_fetch") == 0 {
 		t.Errorf("the try fetched nothing")
 	}
+	// A whole try's start and end are said on the stream checks, which the indexer hears to write and delete the
+	// candidate of its group: said once each, never waited out on a subject no stream holds.
+	checks, err := b.js.Stream(b.ctx, "CHECKS")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, event := range []string{checkermsg.Started, checkermsg.Ended} { // the end is said once the try's end is written
+		said, err := b.js.OrderedConsumer(b.ctx, checks.CachedInfo().Config.Name, jetstream.OrderedConsumerConfig{
+			FilterSubjects: []string{process.ModuleStreamSubject(check.CheckerModule, event)}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if m, err := said.Next(jetstream.FetchMaxWait(5 * time.Second)); err != nil || !strings.Contains(string(m.Data()), id) {
+			t.Errorf("the stream checks holds no %s of the try %s: %v", event, id, err)
+		}
+	}
 	for _, name := range []string{check.CheckerOwedBucket, check.CheckerVerdictsBucket} {
 		if keys, _ := b.buckets[name].Keys(b.ctx); len(keys) != 0 {
 			t.Errorf("a try wrote %s: %v", name, keys)

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

@@ -711,7 +711,10 @@ func (tr *Tryer) run(ctx context.Context, key string) {
 		return
 	}
 	tr.logf("%s: %s", key, t.State)
-	tr.sendEnd(ctx, key)
+	// The end just written comes back on the key's watch, and Put then stops this run's context (it stops judging a try
+	// ended by hand) while this run says the end, which Put's own say leaves to it: the end is said on a context that
+	// stop does not cancel, or it is never said and the indexer keeps the candidate of the try's group.
+	tr.sendEnd(context.WithoutCancel(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