From 4a0618ea117b50da0753d38d714a9fe5f1ce4765 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 15:29:41 +0100 Subject: [PATCH] Checker: its one pull of its durable on the shared events stream is the module's event hub: the registry seats' published and the views reader are two handlers of it (B7c finding 1; issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../mesh-delivery-checker/integration_test.go | 10 ++----- .../cmd/mesh-delivery-checker/main.go | 29 ++++++++++--------- 2 files changed, 17 insertions(+), 22 deletions(-) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index 1e30bbfe..e0fb6fad 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go @@ -27,7 +27,6 @@ import ( "git.novox.be/novox/mesh-contracts/go/seats/gopackageregistry" "git.novox.be/novox/mesh-contracts/go/seats/meshdeliverychecker" ne "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" - "git.novox.be/novox/mesh-contracts/go/seats/npmpackageregistry" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -425,8 +424,8 @@ func startBed(t *testing.T) *bed { meshdeliverychecker.HolderSubjectTagWanted(delivery.CheckerModule)}}); err != nil { t.Fatal(err) } - // The shared events stream with the package registry seats' published, and the checker's durable consumer of it. - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{gopackageregistry.SubjectPublished, npmpackageregistry.SubjectPublished}}); err != nil { + // The shared events stream with the seats' and modules' events (the package registry seats' published among them), and the checker's durable consumer of it. + if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil { t.Fatal(err) } if _, err := js.CreateOrUpdateConsumer(ctx, process.EventsStream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-checker", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { @@ -1057,11 +1056,6 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) { // the readable ones, which is what grants the forge module its read of owed (ADR 0495 §2-§3, ADR 0521; R4e finding R2-1). func TestTheCheckerSaysOwedReadableOnTheBus(t *testing.T) { // R2-1 b := startBed(t) - // The bed's shared events stream carries the registry seats' published; the module events go on it too. - if _, err := b.js.UpdateStream(b.ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>", - gopackageregistry.SubjectPublished, npmpackageregistry.SubjectPublished}}); err != nil { - t.Fatal(err) - } b.start(t, nil) events, err := b.js.Stream(b.ctx, "EVENTS") if err != nil { diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 3449b79f..55cc4c02 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -142,7 +142,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // first view of their read tools and then their views families, never their buckets (ADR 0615 §1, ADR 0596, DLV-M2). // An owner that is down never holds the start: the checks start once every first view is answered, and until then the // unanswered first view is the checker's finding. - reader := &views.ProcessReader{Module: delivery.CheckerModule, Call: caller.Call, Logf: logger.Printf} + // The module has one durable consumer of EVENTS: one hub pulls it for the views reader and for the registry seats' + // published, started once both are registered. + hub := process.NewEventHub(logger.Printf) + reader := &views.ProcessReader{Module: delivery.CheckerModule, Call: caller.Call, Logf: logger.Printf, Hub: hub} findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf) reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) } // Each by the owner's read tool and views family as the owner's manifest names them. @@ -260,20 +263,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // not there yet a group's check asks and waits in vain until its bound, said in its verdict. tags := tagAsks{js: js, node: cred.Node, heard: &check.PublishedHeard{}, logf: logger.Printf, now: time.Now} ch.Says = tags // a pull request's verdict event, on the same stream asks (design 26 *Out of step*) - switch cons, err := cred.EventConsumer(ctx, js, process.EventsStream); { - case err != nil: - logger.Printf("the consumer %s_%s on %s does not exist yet, so no package registry's published is heard and a group's check fails at its publish bound until the checker starts again: %v", - cred.Node, cred.Module, process.EventsStream, err) - default: - cc, err := process.HearEvents(cons, func(m jetstream.Msg) { - if !tags.heardPublished(m) { - _ = m.Ack() // nothing else the checker consumes rides the shared stream - } - }, logger.Printf) - if err != nil { - return err + if err := hub.Handle("mesh.seat.*.event.>", func(m jetstream.Msg) { + if !tags.heardPublished(m) { + _ = m.Ack() // nothing else the checker consumes rides the shared stream } - defer cc.Stop() + }); err != nil { + return err } ch.Judge = newJudge(judgeDeps{ asks: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound}, tools: tools, trunks: trunks, @@ -461,6 +456,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil } reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "checker_first_views reads nothing and checker_first_view_again is not heard", logger.Printf)) + // Every part has registered. Where the consumer is not there yet no package registry's published is heard and a group's + // check fails at its publish bound until the checker starts again. + if err := hub.Start(ctx, js, cred); err != nil { + logger.Printf("%v, so no package registry's published is heard and a group's check fails at its publish bound until the checker starts again", err) + } + defer hub.Stop() // The checks, the tries and the hand asks, once every first view is answered (the lead's ruling for DLV-M2): a check // judged without the candidates, the modules, the manifests, the published packages, the predefined keys or the // placement would be another check.