Grants part: its one pull of its durable on the shared events stream is the module's event hub: the owners' views, the seat mesh-catalog's verdicts and the other seats' events are handlers of it (B7c finding 1; issue 1966)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 15:32:27 +01:00
parent 2e66238a01
commit 25a704a1f5
3 changed files with 63 additions and 28 deletions
@@ -142,41 +142,40 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// dropped carry the bindings: ADR 0595 §1, §7), each on the durable consumer its own membership names; one not
 	// there yet is opened at the next change, and one opened is never opened twice.
 	opened := map[string]bool{}
+	// The module has one durable consumer of EVENTS: one hub pulls it and hands each message to the part that hears it
+	// (the owners' views events, the seat mesh-catalog's verdicts, the other seats' events the follower hears), started
+	// once all are registered.
+	hub := process.NewEventHub(logger.Printf)
+	defer hub.Stop()
+	registered := false
 	f.Hear = func(handle func(string, []byte)) error {
-		consume := func(stream, name string) error {
-			if opened[stream] {
-				return nil
+		if !registered {
+			registered = true
+			// An owner's views event goes to the follower of its state, which acknowledges it (firstviews.go).
+			if err := hub.Handle(views.EventsPattern, hearViews); err != nil {
+				return err
 			}
-			cons, err := js.Consumer(ctx, stream, name)
-			if err != nil {
-				return fmt.Errorf("the consumer %s on %s: %w", name, stream, err)
-			}
-			if _, err := cons.Consume(func(m jetstream.Msg) {
-				// An owner's views event goes to the follower of its state, which acknowledges it (firstviews.go).
-				if strings.Contains(m.Subject(), ".event."+views.EventPrefix) {
-					hearViews(m)
-					return
-				}
-				// The seat mesh-catalog's verdict events go to the verdicts' first view (ADR 0622 §2), which takes each at its revision.
-				if event, ok := strings.CutPrefix(m.Subject(), "mesh.seat."+meshcatalog.Seat+".event."); ok &&
-					(event == meshcatalog.EventManifestAccepted || event == meshcatalog.EventManifestRefused) {
-					if err := fv.HearVerdict(ctx, event, m.Data(), logger.Printf); err != nil {
+			// The seat mesh-catalog's verdict events go to the verdicts' first view (ADR 0622 §2), which takes each at its revision.
+			for _, event := range []string{meshcatalog.EventManifestAccepted, meshcatalog.EventManifestRefused} {
+				if err := hub.Handle("mesh.seat."+meshcatalog.Seat+".event."+event, func(m jetstream.Msg) {
+					if err := fv.HearVerdict(ctx, strings.TrimPrefix(m.Subject(), "mesh.seat."+meshcatalog.Seat+".event."), m.Data(), logger.Printf); err != nil {
 						logger.Printf("views: %s is not a verdict event, dropped: %v", m.Subject(), err)
 						_ = m.Term()
 						return
 					}
 					_ = m.Ack()
-					return
+				}); err != nil {
+					return err
+				}
+			}
+			// The other seats' events the part follows its passes by (Consumes).
+			for _, subject := range grants.SeatEventSubjects() {
+				if err := hub.Handle(subject, func(m jetstream.Msg) { handle(m.Subject(), m.Data()); _ = m.Ack() }); err != nil {
+					return err
 				}
-				handle(m.Subject(), m.Data())
-				_ = m.Ack()
-			}, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })); err != nil {
-				return err
 			}
-			opened[stream] = true
-			return nil
 		}
-		if err := consume("EVENTS", account.Node+"_"+account.Module); err != nil {
+		if err := hub.Start(ctx, js, account); err != nil {
 			return err
 		}
 		hears, err := grants.OwnHears(ownStore{conn}, account.Node)
@@ -184,9 +183,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			return err
 		}
 		for _, h := range hears {
-			if err := consume(h.Stream, h.Consumer); err != nil {
+			if opened[h.Stream] {
+				continue
+			}
+			cons, err := js.Consumer(ctx, h.Stream, h.Consumer)
+			if err != nil {
+				return fmt.Errorf("the consumer %s on %s: %w", h.Consumer, h.Stream, err)
+			}
+			if _, err := cons.Consume(func(m jetstream.Msg) {
+				handle(m.Subject(), m.Data())
+				_ = m.Ack()
+			}, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })); err != nil {
 				return err
 			}
+			opened[h.Stream] = true
 		}
 		return nil
 	}
@@ -2214,6 +2214,16 @@ func TestGrants(t *testing.T) {
 				t.Errorf("the part consumes %s", c)
 			}
 		}
+		// The hub's seat handlers are the seats' events of Consumes on the seats' own subjects, the verdicts' and views' apart.
+		for _, subject := range grants.SeatEventSubjects() {
+			seat, event, _ := strings.Cut(strings.TrimPrefix(subject, "mesh.seat."), ".event.")
+			if !slices.Contains(grants.Consumes, seat+"."+event) || strings.Contains(subject, "mesh-catalog") {
+				t.Errorf("%s is no seat event of Consumes beside the verdicts and views", subject)
+			}
+		}
+		if n := len(grants.SeatEventSubjects()); n != 6 {
+			t.Errorf("the part hears %d seats' events, want the keeper's two, the vault's one and the broker's three", n)
+		}
 		for _, c := range b.calls {
 			if strings.Contains(c, "mesh-controller") {
 				t.Errorf("the part called %s", c)
@@ -42,8 +42,23 @@ const EventAccountIssued = "mesh-vault." + buscompose.AccountIssuedEvent
 // Consumes is every event the part hears, as its manifest's consumes name them.
 // The seat mesh-broker's created, recreated and dropped are heard as every seat's event, `<seat>.<event>` on EVENTS (the
 // lead's ruling, registry milestone 13:0x), never on the bus module's own stream.
-var Consumes = append([]string{EventKeeperRaised, EventKeeperCleared, EventAccountIssued, meshbroker.Seat + "." + BrokerCreated,
-	meshbroker.Seat + "." + BrokerDropped, meshbroker.Seat + "." + BrokerRecreated}, viewsEvents()...)
+var Consumes = append(slices.Clone(seatEvents), viewsEvents()...)
+
+// seatEvents are the events of seats the part follows its passes by, as `<seat>.<event>`.
+var seatEvents = []string{EventKeeperRaised, EventKeeperCleared, EventAccountIssued, meshbroker.Seat + "." + BrokerCreated,
+	meshbroker.Seat + "." + BrokerDropped, meshbroker.Seat + "." + BrokerRecreated}
+
+// SeatEventSubjects are the subjects of the seats' events the part follows its passes by besides the verdicts' and the
+// views': the keeper's condition events, the vault's account-issued and the seat mesh-broker's, each `<seat>.<event>` of
+// Consumes on the seat's own subject.
+func SeatEventSubjects() []string {
+	var out []string
+	for _, c := range seatEvents {
+		seat, event, _ := strings.Cut(c, ".")
+		out = append(out, "mesh.seat."+seat+".event."+event)
+	}
+	return out
+}

 // viewsEvents are the owners' events the part follows its first views by: `<owner>.views.<state>`, but the verdicts by
 // the indexer's manifest-accepted and manifest-refused (VerdictEvents, ADR 0622 §2).