Forge bridge: its one pull of its durable on the shared events stream is the module's event hub, for consistency with the modules that have a second part (B7c finding 1; issue 1966)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
2026-11-05 15:34:08 +01:00
parent 8c80e1b097
commit 376119d61e
@@ -195,23 +195,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
go onReconnect(ctx, again, said, logger)
// What a merge read of each module directory, by its tree (issue 1126): a merge reads only the directories it changed.
inputs := newKept(process.OptionalBucket(ctx, js, inputsBucket, "a merge's module directories are kept in memory only, so the first merge after a start reads every one", logger.Printf), logger)
cons, err := cred.EventConsumer(ctx, js, eventsStream)
if err != nil {
// Essential: the forge's events are what the bridge carries.
return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the bridge hears no forge event without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, eventsStream, err)
}
cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
// The module's one pull of its durable consumer on the shared events stream is its event hub; the bridge has the one
// handler, for every event its consumes filter lets through. Essential: the forge's events are what the bridge carries.
hub := process.NewEventHub(logger.Printf)
if err := hub.Handle("mesh.mod.*.event.>", func(m jetstream.Msg) {
hear(&turn, func() {
// Bounded: the turn is held, and no sync or other event goes on without it.
hctx, cancel := context.WithTimeout(ctx, syncDeadline)
defer cancel()
handleGrouped(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), inputs, groups, logger)
})
}, logger.Printf)
if err != nil {
}); err != nil {
return err
}
defer cc.Stop()
if err := hub.Start(ctx, js, cred); err != nil {
return fmt.Errorf("%w: the bridge hears no forge event without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", err)
}
defer hub.Stop()
select {
case <-ctx.Done():
return nil