Assigner: its one pull of its durable on the shared events stream is the module's event hub: the engines' reads changed and the indexer's views are handlers of it (B7c finding 1; issue 1966)

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

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

@@ -401,9 +401,13 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
 			return err
 		}
 	}
+	// The assigner's one pull of its durable on the shared events stream: the indexer's views and the engines' reads changed
+	// are handlers of it, started once both are registered (process.EventHub).
+	hub := process.NewEventHub(logger.Printf)
 	// The indexer's manifests, modules and module graph, through its first views and views family, never its buckets
 	// (DLV-F2r, ADR 0631): the first pass waits for the three answers, as it waits for every watch to hand what stands.
 	if iv != nil {
+		iv.hub = hub
 		started += iv.start(ctx, w, wake, settled)
 	}
 	// The seat's states arrive with the same delivery as this assigner (ADR 0567 §1): one not on the bus at this start is
@@ -451,7 +455,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
 	// Heard on the assigner's durable consumer on EVENTS, which the grants part makes from the manifest's consumes; where it
 	// is not there yet, a node is asked again only when it comes back online.
 	if js != nil {
-		hearReadsChanged(ctx, js, reads, findings, logger.Printf)
+		hearReadsChanged(ctx, js, hub, reads, findings, logger.Printf)
 	}
 	if bus != nil && kvs[delivery.AssignerPutsBucket] != nil {
 		p.acts = newPutActs(bus, wake, logger.Printf)
@@ -518,18 +522,13 @@ func sayVia(ctx context.Context, js jetstream.JetStream) func(oc.Condition) erro
 	}
 }
 
-// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's durable consumer on EVENTS,
-// and asks the node it names again. Any other event the consumer hands is acknowledged and left.
-func hearReadsChanged(ctx context.Context, js jetstream.JetStream, reads *enginereads.Keeper, findings assign.Findings, logf func(string, ...any)) {
+// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's one hub of its durable
+// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of
+// indexerViews, and starts once both are registered.
+func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, logf func(string, ...any)) {
 	account := process.Account{Node: thisNode, Module: delivery.AssignerModule}
 	w := notHeardWriter(findings, logf)
-	cons, err := account.EventConsumer(ctx, js, process.EventsStream)
-	if err != nil {
-		w.NotHeard(process.EventsStream, enginereads.Event+" is not heard until the assigner starts again (a node is asked again when it comes back online)",
-			fmt.Errorf("the consumer %s on %s does not exist yet: %w", account.EventConsumerName(), process.EventsStream, err))
-		return
-	}
-	cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
+	handle := func(m jetstream.Msg) {
 		if err := m.Ack(); err != nil {
 			logf("acknowledging %s: %v", m.Subject(), err)
 		}
@@ -542,14 +541,18 @@ func hearReadsChanged(ctx context.Context, js jetstream.JetStream, reads *engine
 		default:
 			reads.AskNode(node, enginereads.Event)
 		}
-	}, logf)
+	}
+	err := hub.Handle(enginereads.SubjectPrefix+"*", handle)
+	if err == nil {
+		err = hub.Start(ctx, js, account)
+	}
 	if err != nil {
-		w.NotHeard(process.EventsStream, enginereads.Event+" is not heard until the assigner starts again", err)
+		w.NotHeard(process.EventsStream, enginereads.Event+" is not heard until the assigner starts again (a node is asked again when it comes back online)", err)
 		return
 	}
 	w.Heard(process.EventsStream)
 	go func() {
 		<-ctx.Done()
-		cc.Stop()
+		hub.Stop()
 	}()
 }

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

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"slices"
 	"time"
 
@@ -132,6 +131,7 @@ type indexerViews struct {
 	findings assign.Findings
 	status   *process.AwaitedBucket
 	logf     func(string, ...any)
+	hub      *process.EventHub
 }
 
 // indexerFirstViews are the indexer's states the assigner takes by first view and the views family, each by the name of
@@ -195,15 +195,10 @@ func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settle
 			}
 		})
 	}
-	// Not heard, the indexer's views family is the finding family-not-heard until it binds (ADR 0615 §5).
-	stream := process.EventsStream
-	_ = notHeardWriter(iv.findings, iv.logf).Bind(stream, "the indexer's manifests, modules and graph are the first views' alone until the assigner starts again", func() error {
-		cons, err := iv.account.EventConsumer(ctx, iv.js, stream)
-		if err != nil {
-			return fmt.Errorf("the consumer %s on %s does not exist yet: %w", iv.account.EventConsumerName(), stream, err)
-		}
-		_, err = process.HearEvents(cons, views.Hear(ctx, followers, iv.logf), iv.logf)
-		return err
+	// Not heard, the indexer's views family is the finding family-not-heard until it binds (ADR 0615 §5): the handler is
+	// registered on the module's hub, which the assigner starts once every handler is.
+	_ = notHeardWriter(iv.findings, iv.logf).Bind(process.EventsStream, "the indexer's manifests, modules and graph are the first views' alone until the assigner starts again", func() error {
+		return iv.hub.Handle("mesh.mod."+delivery.IndexerModule+".event."+views.EventPrefix+"*", views.Hear(ctx, followers, iv.logf))
 	})
 	return len(all)
 }