Indexer: its one pull of its durable on the shared events stream is the module's event hub: the node-engine's reads changed, the registries' published, the builder's numbered and the assigner's views are handlers of it (B7c finding 1; issue 1966)

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

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

@@ -30,6 +30,7 @@ import (
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 
+	"git.novox.be/novox/mesh-delivery/sdk/enginereads"
 	"git.novox.be/novox/mesh-delivery/sdk/graph"
 	"git.novox.be/novox/mesh-sdk/go/process"
 
@@ -501,7 +502,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	})
 	// The placement is the assigner's: read through its first view and its views family, never its bucket (DLV-F1r).
 	placementErrs := make(chan error)
-	followPlacement(ctx, js, cred, calls.Call, gate, func(ctx context.Context) {
+	// The indexer's one pull of its durable on the shared events stream (process.EventHub): every handler is registered
+	// before it starts.
+	hub := process.NewEventHub(logger.Printf)
+	followPlacement(ctx, hub, cred, calls.Call, gate, func(ctx context.Context) {
 		if err := ix.Recompute(ctx); err != nil {
 			logger.Printf("writing the modules again after the placement of a build module changed: %v", err)
 		}
@@ -609,13 +613,33 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// The versions each package's series holds (DLV-I5): the registry seats' published and the builder's numbered, on the
 	// indexer's durable consumer of the shared events stream. Where it is not there yet, no series moves until the indexer
 	// starts again, said.
-	hear(process.EventsStream, "no package's series moves and no node's manifest-reads-changed is heard until the indexer starts again", func(m jetstream.Msg) {
-		if reads.heard(m, logger.Printf) { // the node-engine seat's manifest-reads-changed (ADR 0618 §4): its node asked again
-			return
+	if err := hub.Handle(enginereads.SubjectPrefix+"*", func(m jetstream.Msg) { // the node-engine seat's manifest-reads-changed (ADR 0618 §4): its node asked again
+		if !reads.heard(m, logger.Printf) {
+			_ = m.Ack()
 		}
+	}); err != nil {
+		return err
+	}
+	versions := func(m jetstream.Msg) {
 		if !heardVersion(ctx, ix, m, logger.Printf) {
 			_ = m.Ack() // nothing else the indexer consumes rides the shared stream
 		}
+	}
+	for subject := range publishedSeats {
+		if err := hub.Handle(subject, versions); err != nil {
+			return err
+		}
+	}
+	if err := hub.Handle(numberedSubject, versions); err != nil {
+		return err
+	}
+	_ = notHeard.Bind(process.EventsStream, "no package's series moves, no node's manifest-reads-changed and no placement view is heard until the indexer starts again", func() error {
+		if err := hub.Start(ctx, js, cred); err != nil {
+			stopped = cmp.Or(stopped, err)
+			return err
+		}
+		go func() { <-ctx.Done(); hub.Stop() }()
+		return nil
 	})
 	// The checker's verdicts carry a pull request's out-of-step files (DLV-I20): on the indexer's durable consumer of the
 	// checker's stream asks. Where it is not there yet, no pull request's out-of-step files are kept until the indexer

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"sync/atomic"
 	"time"
 
@@ -117,7 +116,7 @@ const placementTool = delivery.AssignerModule + "." + delivery.PlacementState
 // changed told whenever what the gate passes on changed. It waits for the first answer at most views.Bound, so a
 // normal start, the two owners coming up together, computes the fingerprints once; past it the indexer goes on, the gate
 // passing every report, and raises its condition naming the assigner and placement until the answer clears it.
-func followPlacement(ctx context.Context, js jetstream.JetStream, cred process.Account, call func(string, any) (json.RawMessage, error),
+func followPlacement(ctx context.Context, hub *process.EventHub, cred process.Account, call func(string, any) (json.RawMessage, error),
 	gate *buildrun.ToolchainGate, changed func(context.Context), findings, status *process.AwaitedBucket, logf func(string, ...any)) *views.Follower {
 	finding := func(s views.Status, raised bool) {
 		w := sdkfindings.Writer(ctx, findings.KV)
@@ -164,16 +163,10 @@ func followPlacement(ctx context.Context, js jetstream.JetStream, cred process.A
 		}
 		close(answered)
 	}()
-	// The assigner's views family, on the indexer's durable consumer, once the first view is asked; not heard, it is the
-	// finding family-not-heard until it binds (ADR 0615 §5).
-	stream := process.EventsStream
-	_ = notHeardWriter(ctx, findings, logf).Bind(stream, "the placement is the first view's alone, asked again by hand, until the indexer starts again", func() error {
-		cons, err := cred.EventConsumer(ctx, js, stream)
-		if err != nil {
-			return fmt.Errorf("the consumer %s on %s does not exist yet: %w", cred.EventConsumerName(), stream, err)
-		}
-		_, err = process.HearEvents(cons, views.Hear(ctx, map[string]*views.Follower{delivery.PlacementState: f}, logf), logf)
-		return err
+	// The assigner's views family, on the module's one hub (the host starts it once every handler is registered); a family the
+	// hub cannot register is the finding family-not-heard until the indexer starts again (ADR 0615 §5).
+	_ = notHeardWriter(ctx, findings, logf).Bind(process.EventsStream, "the placement is the first view's alone, asked again by hand, until the indexer starts again", func() error {
+		return hub.Handle("mesh.mod."+delivery.AssignerModule+".event."+views.EventPrefix+"*", views.Hear(ctx, map[string]*views.Follower{delivery.PlacementState: f}, logf))
 	})
 	wait, cancel := context.WithTimeout(ctx, views.Bound) // a context's bound, no timer (ADR 0437 §8)
 	defer cancel()