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
This commit is contained in:
2026-11-05 15:14:12 +01:00
parent 4104143d16
commit 62bb25675d
@@ -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()
}