From 62bb25675de22436366f228f95ffdc7801a80d85 Mon Sep 17 00:00:00 2001
From: jochen <jschoubben@gmail.com>
Date: Thu, 5 Nov 2026 15:14:12 +0100
Subject: [PATCH] distribution: its wake events are heard through the events
 hub, the module's one durable pull (B5d; B9d hub; issue 1966)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
---
 .../distribution/cmd/distribution/windows.go  | 25 ++++++++++++-------
 1 file changed, 16 insertions(+), 9 deletions(-)

diff --git a/modules/distribution/cmd/distribution/windows.go b/modules/distribution/cmd/distribution/windows.go
index 3d40e32b0..15b166776 100644
--- a/modules/distribution/cmd/distribution/windows.go
+++ b/modules/distribution/cmd/distribution/windows.go
@@ -212,6 +212,9 @@ func sayClosed(js jetstream.JetStream, cred process.Account) func(holder.Closed)
 	}
 }
 
+// wakePatterns are the subject patterns of the events that wake the driver, as wakes judges them.
+var wakePatterns = []string{"mesh.*.mesh-delivery-view.event.delivered", "mesh.*.node-engine.event.manifest-reads-changed.*"}
+
 // wakes says which events wake the driver, by subject: the status view's delivered, a node-engine's
 // manifest-reads-changed (its node the subject's last token).
 func wakes(subject string) (string, bool) {
@@ -238,12 +241,10 @@ func drive(ctx context.Context, js jetstream.JetStream, cred process.Account, d
 		return err
 	}
 	_ = wake("distribution started")
-	cons, err := cred.EventConsumer(ctx, js, process.EventsStream)
-	if err != nil {
-		logger.Printf("the events stream cannot be heard (%v): the windows move only at the next start", err)
-		return
-	}
-	cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
+	// The module's one durable on the events stream is pulled by the events hub (mesh-sdk process), each waking event
+	// registered by its subject pattern: the seat of the view says delivered, the node-engine says a manifest read changed.
+	hub := process.NewEventHub(logger.Printf)
+	handle := func(m jetstream.Msg) {
 		why, ok := wakes(m.Subject())
 		if !ok {
 			_ = m.Ack()
@@ -254,11 +255,17 @@ func drive(ctx context.Context, js jetstream.JetStream, cred process.Account, d
 			return
 		}
 		_ = m.Ack()
-	}, logger.Printf)
-	if err != nil {
+	}
+	for _, pattern := range wakePatterns {
+		if err := hub.Handle(pattern, handle); err != nil {
+			logger.Printf("the events cannot be heard (%v): the windows move only at the next start", err)
+			return
+		}
+	}
+	if err := hub.Start(ctx, js, cred); err != nil {
 		logger.Printf("the events stream cannot be heard (%v): the windows move only at the next start", err)
 		return
 	}
 	<-ctx.Done()
-	cc.Stop()
+	hub.Stop()
 }