From 9d360e701a0fd237c9c5493ab59a817582f3b120 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 15:27:41 +0100 Subject: [PATCH] Builder: its one pull of its durable on the shared events stream is the module's event hub: the seats' events and the condition keeper's are one handler, the views reader another, the hub started last (B7c finding 1; issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../gone_integration_test.go | 2 +- .../mesh-delivery-builder/integration_test.go | 9 ++------ .../cmd/mesh-delivery-builder/main.go | 22 ++++++++++++++++--- 3 files changed, 22 insertions(+), 11 deletions(-) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go index 58395d16..bf59a418 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go @@ -65,7 +65,7 @@ func (h *heldVerb) asks() int { // from the manifest's consumes. func eventsConsumer(t *testing.T, b *bed, node string) { t.Helper() - if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: eventsStream, Subjects: []string{delivery.ArtifactsGoneSubject}}); err != nil { + if _, err := b.js.CreateOrUpdateStream(b.ctx, jetstream.StreamConfig{Name: eventsStream, Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil { t.Fatal(err) } if _, err := b.js.CreateOrUpdateConsumer(b.ctx, eventsStream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder", diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index fe813e7a..795fd0c0 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -1191,11 +1191,10 @@ func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *regis mk(versions.FloorWaitsBucket) // EVENTS with two seats' outcomes, and the forge bridge's stream (a branch's end): each with the builder's durable consumer. for stream, subjects := range map[string][]string{ - eventsStream: {delivery.ArtifactsGoneSubject, ociartifactregistry.SubjectPublished, ociartifactregistry.SubjectRefused, ociartifactregistry.SubjectRemoved, - gopackageregistry.SubjectPublished, gopackageregistry.SubjectRefused, gopackageregistry.SubjectRemoved}, + eventsStream: {"mesh.mod.*.event.>", "mesh.seat.*.event.>"}, forgeStream: {process.ModuleStreamSubject(delivery.ForgeBridgeModule, ">")}, } { - if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: stream, Subjects: subjects}); err != nil { + if _, err := b.js.CreateOrUpdateStream(b.ctx, jetstream.StreamConfig{Name: stream, Subjects: subjects}); err != nil { t.Fatal(err) } if _, err := b.js.CreateOrUpdateConsumer(b.ctx, stream, jetstream.ConsumerConfig{Durable: node + "_" + delivery.BuilderModule, @@ -1216,10 +1215,6 @@ func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *regis t.Fatal(err) } t.Cleanup(func() { _ = rm.Unsubscribe() }) - own := "mesh.mod." + delivery.BuilderModule + ".event.>" - if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "BUILDER_EVENTS", Subjects: []string{own}}); err != nil { - t.Fatal(err) - } nc := b.js.Conn() // As a build module hears it: on the contract's subject, read by the contract's decoder (BG2's late publish). sub, err := nc.Subscribe(delivery.VersionNumberedSubject, func(m *nats.Msg) { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index c9493ceb..927a8ad9 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -75,7 +75,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } defer calls.Close() - reader := &views.ProcessReader{Module: delivery.BuilderModule, Call: calls.Call, Logf: logger.Printf} + // The module has one durable consumer of EVENTS: one hub pulls it and hands each message to the part that hears it (the + // views reader and the builder's own events), started once every part is registered. + hub := process.NewEventHub(logger.Printf) + reader := &views.ProcessReader{Module: delivery.BuilderModule, Call: calls.Call, Logf: logger.Printf, Hub: hub} findingsKV := process.AwaitBucket(ctx, js, conn, findingsBucket, "the builder's findings, a first view unanswered past its bound among them, are said in its log only", logger.Printf) reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findingsKV.KV) } modules, err := reader.Follow(delivery.IndexerModule, modulesState, delivery.IndexerModule+".indexer_read_modules") @@ -330,16 +333,29 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro floor := floorWake{b: b, waits: floorWaits, logf: logger.Printf} // And the four registry seats' published, refused and removed (DLV-B4), which move a numbered version. seats := seatHearer{n: numberer, b: b, logf: logger.Printf, pub: pubs} - hear(eventsStream, delivery.ArtifactsGoneEvent+", what wakes a floor, the registry seats' words and manifest-reads-changed are not heard until the builder starts again", func(m jetstream.Msg) { + events := func(m jetstream.Msg) { if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4) return } heardGone(ctx, b, m, logger.Printf) - }) + } + // The seats' events (artifacts-gone, the registry seats' words, manifest-accepted and manifest-reads-changed) and the + // condition keeper's condition-cleared; the owners' views events are the views reader's. + for _, pattern := range []string{"mesh.seat.*.event.>", conditionClearedSubject} { + if err := hub.Handle(pattern, events); err != nil { + return err + } + } if stopped != nil { return stopped } reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "builder_first_views reads nothing and builder_first_view_again is not heard", logger.Printf)) + // Every part has registered: the hub pulls the module's durable. Where the consumer is not there yet the builder says so + // and binds it at a later change (heard.Writer; ADR 0615 §5). + _ = notHeard.Bind(eventsStream, delivery.ArtifactsGoneEvent+", what wakes a floor, the registry seats' words, the views and manifest-reads-changed are not heard until the builder starts again", func() error { + return hub.Start(ctx, js, cred) + }) + go func() { <-ctx.Done(); hub.Stop() }() // The modules, once their first views and the manifests' are answered: what a module builds is read from the manifests. go func() { errs <- buildFromModules(ctx, b, modules, manifests, candidates) }() // A package's name heard in the indexer's packages looks at its module again, so a done build is numbered then.