Checker: its one pull of its durable on the shared events stream is the module's event hub: the registry seats' published and the views reader are two handlers of it (B7c finding 1; issue 1966)

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

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

@@ -27,7 +27,6 @@ import (
 	"git.novox.be/novox/mesh-contracts/go/seats/gopackageregistry"
 	"git.novox.be/novox/mesh-contracts/go/seats/meshdeliverychecker"
 	ne "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
-	"git.novox.be/novox/mesh-contracts/go/seats/npmpackageregistry"
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
 	"git.novox.be/novox/mesh-sdk/go/announce"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -425,8 +424,8 @@ func startBed(t *testing.T) *bed {
 		meshdeliverychecker.HolderSubjectTagWanted(delivery.CheckerModule)}}); err != nil {
 		t.Fatal(err)
 	}
-	// The shared events stream with the package registry seats' published, and the checker's durable consumer of it.
-	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{gopackageregistry.SubjectPublished, npmpackageregistry.SubjectPublished}}); err != nil {
+	// 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)
 	}
 	if _, err := js.CreateOrUpdateConsumer(ctx, process.EventsStream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-checker", AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
@@ -1057,11 +1056,6 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) {
 // the readable ones, which is what grants the forge module its read of owed (ADR 0495 §2-§3, ADR 0521; R4e finding R2-1).
 func TestTheCheckerSaysOwedReadableOnTheBus(t *testing.T) { // R2-1
 	b := startBed(t)
-	// The bed's shared events stream carries the registry seats' published; the module events go on it too.
-	if _, err := b.js.UpdateStream(b.ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>",
-		gopackageregistry.SubjectPublished, npmpackageregistry.SubjectPublished}}); err != nil {
-		t.Fatal(err)
-	}
 	b.start(t, nil)
 	events, err := b.js.Stream(b.ctx, "EVENTS")
 	if err != nil {

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

@@ -142,7 +142,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// first view of their read tools and then their views families, never their buckets (ADR 0615 §1, ADR 0596, DLV-M2).
 	// An owner that is down never holds the start: the checks start once every first view is answered, and until then the
 	// unanswered first view is the checker's finding.
-	reader := &views.ProcessReader{Module: delivery.CheckerModule, Call: caller.Call, Logf: logger.Printf}
+	// The module has one durable consumer of EVENTS: one hub pulls it for the views reader and for the registry seats'
+	// published, started once both are registered.
+	hub := process.NewEventHub(logger.Printf)
+	reader := &views.ProcessReader{Module: delivery.CheckerModule, Call: caller.Call, Logf: logger.Printf, Hub: hub}
 	findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf)
 	reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) }
 	// Each by the owner's read tool and views family as the owner's manifest names them.
@@ -260,20 +263,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// not there yet a group's check asks and waits in vain until its bound, said in its verdict.
 	tags := tagAsks{js: js, node: cred.Node, heard: &check.PublishedHeard{}, logf: logger.Printf, now: time.Now}
 	ch.Says = tags // a pull request's verdict event, on the same stream asks (design 26 *Out of step*)
-	switch cons, err := cred.EventConsumer(ctx, js, process.EventsStream); {
-	case err != nil:
-		logger.Printf("the consumer %s_%s on %s does not exist yet, so no package registry's published is heard and a group's check fails at its publish bound until the checker starts again: %v",
-			cred.Node, cred.Module, process.EventsStream, err)
-	default:
-		cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
-			if !tags.heardPublished(m) {
-				_ = m.Ack() // nothing else the checker consumes rides the shared stream
-			}
-		}, logger.Printf)
-		if err != nil {
-			return err
+	if err := hub.Handle("mesh.seat.*.event.>", func(m jetstream.Msg) {
+		if !tags.heardPublished(m) {
+			_ = m.Ack() // nothing else the checker consumes rides the shared stream
 		}
-		defer cc.Stop()
+	}); err != nil {
+		return err
 	}
 	ch.Judge = newJudge(judgeDeps{
 		asks: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound}, tools: tools, trunks: trunks,
@@ -461,6 +456,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return nil
 	}
 	reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "checker_first_views reads nothing and checker_first_view_again is not heard", logger.Printf))
+	// Every part has registered. Where the consumer is not there yet no package registry's published is heard and a group's
+	// check fails at its publish bound until the checker starts again.
+	if err := hub.Start(ctx, js, cred); err != nil {
+		logger.Printf("%v, so no package registry's published is heard and a group's check fails at its publish bound until the checker starts again", err)
+	}
+	defer hub.Stop()
 	// The checks, the tries and the hand asks, once every first view is answered (the lead's ruling for DLV-M2): a check
 	// judged without the candidates, the modules, the manifests, the published packages, the predefined keys or the
 	// placement would be another check.