From e482ebc667b82abff96a0e32367dddf98d2fe9ca Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 15:14:17 +0100 Subject: [PATCH] Indexer: its one pull of its durable on the shared events stream is the module's event hub: the node-engine's reads changed, the registries' published, the builder's numbered and the assigner's views are handlers of it (B7c finding 1; issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-indexer/main.go | 32 ++++++++++++++++--- .../cmd/mesh-delivery-indexer/views.go | 17 +++------- 2 files changed, 33 insertions(+), 16 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index b114ded7..f2a48b7f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -30,6 +30,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-delivery/sdk/enginereads" "git.novox.be/novox/mesh-delivery/sdk/graph" "git.novox.be/novox/mesh-sdk/go/process" @@ -501,7 +502,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro }) // The placement is the assigner's: read through its first view and its views family, never its bucket (DLV-F1r). placementErrs := make(chan error) - followPlacement(ctx, js, cred, calls.Call, gate, func(ctx context.Context) { + // The indexer's one pull of its durable on the shared events stream (process.EventHub): every handler is registered + // before it starts. + hub := process.NewEventHub(logger.Printf) + followPlacement(ctx, hub, cred, calls.Call, gate, func(ctx context.Context) { if err := ix.Recompute(ctx); err != nil { logger.Printf("writing the modules again after the placement of a build module changed: %v", err) } @@ -609,13 +613,33 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The versions each package's series holds (DLV-I5): the registry seats' published and the builder's numbered, on the // indexer's durable consumer of the shared events stream. Where it is not there yet, no series moves until the indexer // starts again, said. - hear(process.EventsStream, "no package's series moves and no node's manifest-reads-changed is heard until the indexer starts again", func(m jetstream.Msg) { - if reads.heard(m, logger.Printf) { // the node-engine seat's manifest-reads-changed (ADR 0618 §4): its node asked again - return + if err := hub.Handle(enginereads.SubjectPrefix+"*", func(m jetstream.Msg) { // the node-engine seat's manifest-reads-changed (ADR 0618 §4): its node asked again + if !reads.heard(m, logger.Printf) { + _ = m.Ack() } + }); err != nil { + return err + } + versions := func(m jetstream.Msg) { if !heardVersion(ctx, ix, m, logger.Printf) { _ = m.Ack() // nothing else the indexer consumes rides the shared stream } + } + for subject := range publishedSeats { + if err := hub.Handle(subject, versions); err != nil { + return err + } + } + if err := hub.Handle(numberedSubject, versions); err != nil { + return err + } + _ = notHeard.Bind(process.EventsStream, "no package's series moves, no node's manifest-reads-changed and no placement view is heard until the indexer starts again", func() error { + if err := hub.Start(ctx, js, cred); err != nil { + stopped = cmp.Or(stopped, err) + return err + } + go func() { <-ctx.Done(); hub.Stop() }() + return nil }) // The checker's verdicts carry a pull request's out-of-step files (DLV-I20): on the indexer's durable consumer of the // checker's stream asks. Where it is not there yet, no pull request's out-of-step files are kept until the indexer diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go index c80d92e1..22166b5d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go @@ -3,7 +3,6 @@ package main import ( "context" "encoding/json" - "fmt" "sync/atomic" "time" @@ -117,7 +116,7 @@ const placementTool = delivery.AssignerModule + "." + delivery.PlacementState // changed told whenever what the gate passes on changed. It waits for the first answer at most views.Bound, so a // normal start, the two owners coming up together, computes the fingerprints once; past it the indexer goes on, the gate // passing every report, and raises its condition naming the assigner and placement until the answer clears it. -func followPlacement(ctx context.Context, js jetstream.JetStream, cred process.Account, call func(string, any) (json.RawMessage, error), +func followPlacement(ctx context.Context, hub *process.EventHub, cred process.Account, call func(string, any) (json.RawMessage, error), gate *buildrun.ToolchainGate, changed func(context.Context), findings, status *process.AwaitedBucket, logf func(string, ...any)) *views.Follower { finding := func(s views.Status, raised bool) { w := sdkfindings.Writer(ctx, findings.KV) @@ -164,16 +163,10 @@ func followPlacement(ctx context.Context, js jetstream.JetStream, cred process.A } close(answered) }() - // The assigner's views family, on the indexer's durable consumer, once the first view is asked; not heard, it is the - // finding family-not-heard until it binds (ADR 0615 §5). - stream := process.EventsStream - _ = notHeardWriter(ctx, findings, logf).Bind(stream, "the placement is the first view's alone, asked again by hand, until the indexer starts again", func() error { - cons, err := cred.EventConsumer(ctx, js, stream) - if err != nil { - return fmt.Errorf("the consumer %s on %s does not exist yet: %w", cred.EventConsumerName(), stream, err) - } - _, err = process.HearEvents(cons, views.Hear(ctx, map[string]*views.Follower{delivery.PlacementState: f}, logf), logf) - return err + // The assigner's views family, on the module's one hub (the host starts it once every handler is registered); a family the + // hub cannot register is the finding family-not-heard until the indexer starts again (ADR 0615 §5). + _ = notHeardWriter(ctx, findings, logf).Bind(process.EventsStream, "the placement is the first view's alone, asked again by hand, until the indexer starts again", func() error { + return hub.Handle("mesh.mod."+delivery.AssignerModule+".event."+views.EventPrefix+"*", views.Hear(ctx, map[string]*views.Follower{delivery.PlacementState: f}, logf)) }) wait, cancel := context.WithTimeout(ctx, views.Bound) // a context's bound, no timer (ADR 0437 §8) defer cancel()