diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go index 7dace553..cbd06fde 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -59,7 +59,15 @@ func consumed(w *watch.Watchman, subject string, body []byte, logf func(string, // hearConsumed opens the watchman's durable consumers: on EVENTS now, and on each module stream its membership names as // that membership arrives. A consumer the bus does not hold yet is said and tried again at the membership's next change. -func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Account, w *watch.Watchman, logf func(string, ...any)) { +func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Account, hub *process.EventHub, w *watch.Watchman, logf func(string, ...any)) { + // The seats' events (manifest-accepted, the bus module's created and refused) are one handler of the module's one + // pull of EVENTS, beside the take-in events and the views reader. + if err := hub.Handle("mesh.seat.*.event.>", func(m jetstream.Msg) { + consumed(w, m.Subject(), m.Data(), logf) + _ = m.Ack() + }); err != nil { + w.NotHeard("EVENTS", "the seats' events cannot be handled", err) + } var mu sync.Mutex opened := map[string]bool{} open := func(stream, name string) { @@ -70,6 +78,15 @@ func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Acco } // Not bound, the stream is the finding family-not-heard (ADR 0615 §5) until it is, at a later change of the // membership or a later start. + if stream == "EVENTS" { + if err := hub.Start(ctx, js, cred); err != nil { + w.NotHeard(stream, "what it carries is heard once the grants part has made the consumer "+name, err) + return + } + w.Heard(stream) + opened[stream] = true + return + } cons, err := js.Consumer(ctx, stream, name) if err != nil { w.NotHeard(stream, "what it carries is heard once the grants part has made the consumer "+name, err) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index 59b36ae6..f522d41e 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -234,7 +234,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return fmt.Errorf("the watchman's connection for its tool calls: %w", err) } defer calls.Close() - reader := &views.ProcessReader{Module: delivery.WatchmanModule, Call: calls.Call, Logf: logger.Printf, + // The module has one durable consumer of EVENTS: one hub pulls it for the views reader, the indexer's take-in events + // and the seats' events, started once every part is registered. + hub := process.NewEventHub(logger.Printf) + reader := &views.ProcessReader{Module: delivery.WatchmanModule, Call: calls.Call, Logf: logger.Printf, Hub: hub, Findings: func() views.FindingWriter { return sdkfindings.Writer(ctx, func() jetstream.KeyValue { return kvs.findings }) }} @@ -450,17 +453,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro w.TakeInCaughtUp() } } - hc, err := process.HearEvents(tc, func(m jetstream.Msg) { takeInHeard(w, m, logger.Printf) }, logger.Printf) - if err != nil { + if err := hub.Handle("mesh.mod."+delivery.IndexerModule+".event.take-in.*", func(m jetstream.Msg) { takeInHeard(w, m, logger.Printf) }); err != nil { return err } - defer hc.Stop() } w.Reconcile() if err := hear(conn, w, dead, logger.Printf); err != nil { return err } - hearConsumed(ctx, js, cred, w, logger.Printf) errs := make(chan error, 16) if healthWatch != nil { @@ -468,6 +468,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // The first views of the wants and the modules, asked now and read each on its own goroutine once answered (ADR 0615). reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "watchman_first_views reads nothing and watchman_first_view_again is not heard", logger.Printf)) + // Every part has registered: the hub pulls the module's durable, and the watchman's other consumers open. + hearConsumed(ctx, js, cred, hub, w, logger.Printf) + defer hub.Stop() // The indexer's modules (ADR 0578 §1): a module whose value carries a refusal of its manifest is a finding. verdictEach := func(e jetstream.KeyValueEntry) { w.ModuleVerdict(e.Key(), e.Value(), opOf(e)) } go func() { errs <- readLate(ctx, w, modules, delivery.ModulesBucket, verdictEach) }()