diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 384b382e..12895cd8 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -401,9 +401,13 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, return err } } + // The assigner's one pull of its durable on the shared events stream: the indexer's views and the engines' reads changed + // are handlers of it, started once both are registered (process.EventHub). + hub := process.NewEventHub(logger.Printf) // The indexer's manifests, modules and module graph, through its first views and views family, never its buckets // (DLV-F2r, ADR 0631): the first pass waits for the three answers, as it waits for every watch to hand what stands. if iv != nil { + iv.hub = hub started += iv.start(ctx, w, wake, settled) } // The seat's states arrive with the same delivery as this assigner (ADR 0567 §1): one not on the bus at this start is @@ -451,7 +455,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, // Heard on the assigner's durable consumer on EVENTS, which the grants part makes from the manifest's consumes; where it // is not there yet, a node is asked again only when it comes back online. if js != nil { - hearReadsChanged(ctx, js, reads, findings, logger.Printf) + hearReadsChanged(ctx, js, hub, reads, findings, logger.Printf) } if bus != nil && kvs[delivery.AssignerPutsBucket] != nil { p.acts = newPutActs(bus, wake, logger.Printf) @@ -518,18 +522,13 @@ func sayVia(ctx context.Context, js jetstream.JetStream) func(oc.Condition) erro } } -// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's durable consumer on EVENTS, -// and asks the node it names again. Any other event the consumer hands is acknowledged and left. -func hearReadsChanged(ctx context.Context, js jetstream.JetStream, reads *enginereads.Keeper, findings assign.Findings, logf func(string, ...any)) { +// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's one hub of its durable +// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of +// indexerViews, and starts once both are registered. +func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, logf func(string, ...any)) { account := process.Account{Node: thisNode, Module: delivery.AssignerModule} w := notHeardWriter(findings, logf) - cons, err := account.EventConsumer(ctx, js, process.EventsStream) - if err != nil { - w.NotHeard(process.EventsStream, enginereads.Event+" is not heard until the assigner starts again (a node is asked again when it comes back online)", - fmt.Errorf("the consumer %s on %s does not exist yet: %w", account.EventConsumerName(), process.EventsStream, err)) - return - } - cc, err := process.HearEvents(cons, func(m jetstream.Msg) { + handle := func(m jetstream.Msg) { if err := m.Ack(); err != nil { logf("acknowledging %s: %v", m.Subject(), err) } @@ -542,14 +541,18 @@ func hearReadsChanged(ctx context.Context, js jetstream.JetStream, reads *engine default: reads.AskNode(node, enginereads.Event) } - }, logf) + } + err := hub.Handle(enginereads.SubjectPrefix+"*", handle) + if err == nil { + err = hub.Start(ctx, js, account) + } if err != nil { - w.NotHeard(process.EventsStream, enginereads.Event+" is not heard until the assigner starts again", err) + w.NotHeard(process.EventsStream, enginereads.Event+" is not heard until the assigner starts again (a node is asked again when it comes back online)", err) return } w.Heard(process.EventsStream) go func() { <-ctx.Done() - cc.Stop() + hub.Stop() }() } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go index 9de44e35..9906624e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -3,7 +3,6 @@ package main import ( "context" "encoding/json" - "fmt" "slices" "time" @@ -132,6 +131,7 @@ type indexerViews struct { findings assign.Findings status *process.AwaitedBucket logf func(string, ...any) + hub *process.EventHub } // indexerFirstViews are the indexer's states the assigner takes by first view and the views family, each by the name of @@ -195,15 +195,10 @@ func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settle } }) } - // Not heard, the indexer's views family is the finding family-not-heard until it binds (ADR 0615 §5). - stream := process.EventsStream - _ = notHeardWriter(iv.findings, iv.logf).Bind(stream, "the indexer's manifests, modules and graph are the first views' alone until the assigner starts again", func() error { - cons, err := iv.account.EventConsumer(ctx, iv.js, stream) - if err != nil { - return fmt.Errorf("the consumer %s on %s does not exist yet: %w", iv.account.EventConsumerName(), stream, err) - } - _, err = process.HearEvents(cons, views.Hear(ctx, followers, iv.logf), iv.logf) - return err + // Not heard, the indexer's views family is the finding family-not-heard until it binds (ADR 0615 §5): the handler is + // registered on the module's hub, which the assigner starts once every handler is. + _ = notHeardWriter(iv.findings, iv.logf).Bind(process.EventsStream, "the indexer's manifests, modules and graph are the first views' alone until the assigner starts again", func() error { + return iv.hub.Handle("mesh.mod."+delivery.IndexerModule+".event."+views.EventPrefix+"*", views.Hear(ctx, followers, iv.logf)) }) return len(all) }