From af96eeba86e172184f2222ba02b2e9c560d9abcd Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 11:56:08 +0100 Subject: [PATCH] A-274: the watchman, forge bridge, teller and view pass their finding-events binding to the finding writer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each asks the seat mesh-broker's show for its finding-events entry (process.Binding, started without waiting), passes it as the writer's Bindings and hears the seat's recreated and dropped; the watchman's one handler of the seats' events hands those two to the binding, since a message has one taker. The teller, view and watchman manifests gain the invoke seat:mesh-broker.show and consume mesh-broker.recreated and dropped. The forge bridge's and the watchman's whole tests answer show and read the finding-written under the binding's subject. Builds: design 47 G A-274, §529, §577, A-13 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../integration_test.go | 34 ++++++++++++++--- .../cmd/mesh-delivery-forge-bridge/main.go | 7 +++- .../cmd/mesh-delivery-teller/findingwriter.go | 6 +-- .../cmd/mesh-delivery-teller/main.go | 10 ++++- modules/mesh-delivery-teller/go.mod | 1 + modules/mesh-delivery-teller/manifest_test.go | 4 +- modules/mesh-delivery-teller/module.json | 5 ++- .../cmd/mesh-delivery-view/findingwriter.go | 6 +-- .../cmd/mesh-delivery-view/main.go | 10 ++++- modules/mesh-delivery-view/go.mod | 1 + modules/mesh-delivery-view/manifest_test.go | 4 +- modules/mesh-delivery-view/module.json | 5 ++- .../cmd/mesh-delivery-watchman/consumes.go | 15 +++++++- .../mesh-delivery-watchman/findingwriter.go | 5 ++- .../integration_test.go | 37 ++++++++++++++++--- .../cmd/mesh-delivery-watchman/main.go | 9 ++++- .../mesh-delivery-watchman/manifest_test.go | 4 +- modules/mesh-delivery-watchman/module.json | 5 ++- 18 files changed, 132 insertions(+), 36 deletions(-) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go index ca8dd51e..ac275259 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go @@ -77,9 +77,9 @@ accounts { "$JS.API.CONSUMER.CREATE.EVENTS.>" "$JS.API.CONSUMER.DELETE.EVENTS.>" "$JS.API.CONSUMER.MSG.NEXT.EVENTS.>" - "mesh.stream.mesh-delivery-forge-bridge.event.finding-written.n1" - "mesh.stream.mesh-delivery-forge-bridge.event.finding-cleared.n1" - "mesh.stream.mesh-delivery-forge-bridge.event.findings-stated.n1" + "mesh.bus.mesh-delivery-forge-bridge.finding-events.finding-written.n1" + "mesh.bus.mesh-delivery-forge-bridge.finding-events.finding-cleared.n1" + "mesh.bus.mesh-delivery-forge-bridge.finding-events.findings-stated.n1" "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_inputs.>" "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_inputs.>" "$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_inputs.>" @@ -234,6 +234,13 @@ const boundPerformed = "mod_mesh-delivery-forge-bridge_performed_b1" // boundFindings is the bus name the stand-in seat mesh-broker's show answers for the state findings. const boundFindings = "mod_mesh-delivery-forge-bridge_findings_b1" +// boundFindingEvents and boundFindingEventsSubject are the stream and subject the stand-in seat mesh-broker's show answers +// for the bridge's finding events (design 47 G §529, A-274). +const ( + boundFindingEvents = "mod_mesh-delivery-forge-bridge_finding-events_b1" + boundFindingEventsSubject = "mesh.bus.mesh-delivery-forge-bridge.finding-events" +) + // startBed is startBridge with a step between the streams and the consumers: prepare writes what the streams hold before // the bridge's consumers are made, as the grants part makes them, from now (ADR 0443 §1), and reply sets the answer of the // seat git's open-pulls from then on. A stand-in holds the seats the bridge calls: git (every verb, trees r0 by default, @@ -257,9 +264,9 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>", "mesh.stream.mesh-delivery-forge-bridge.event.owed.ended"}}); err != nil { t.Fatal(err) } - // The bridge's own stream of its findings' events (design 47 G §383, §529), which its finding writer says into. - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "mesh-delivery-forge-bridge-findings", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.finding-written.*", - "mesh.stream.mesh-delivery-forge-bridge.event.finding-cleared.*", "mesh.stream.mesh-delivery-forge-bridge.event.findings-stated.*"}}); err != nil { + // The bridge's own stream of its findings' events, its requires entry finding-events under the names the seat + // mesh-broker's show answers (design 47 G §529, A-274), which its finding writer says into. + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: boundFindingEvents, Subjects: []string{boundFindingEventsSubject + ".>"}}); err != nil { t.Fatal(err) } // Each seat's own stream owed, as its contract keeps it (Stream, SEAT__owed), as the grants part makes it: no @@ -304,6 +311,8 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, "git.pull": `{"error":"no pull request merged at that commit","error-kind":"not-found"}`, "mesh-broker.show": `{"state":"created","binding":{"local":"performed","storage":"key-value","template":"state","bucket":"` + boundPerformed + `"}}`, "mesh-broker.show findings": `{"state":"created","binding":{"local":"findings","storage":"key-value","template":"state","bucket":"` + boundFindings + `"}}`, + "mesh-broker.show finding-events": `{"state":"created","binding":{"local":"finding-events","storage":"stream","template":"events","stream":"` + + boundFindingEvents + `","subject":"` + boundFindingEventsSubject + `"}}`, } for _, verb := range []string{"status", "comment", "view", "note", "merge-pull"} { answers["git."+verb] = string(gitfixture.Given(t, verb+" answers", "verb:"+verb+".answer")) @@ -926,6 +935,19 @@ func TestBridgeSaysAnUnprotectedBranchAsAFinding(t *testing.T) { if err := json.Unmarshal(e.Value(), &f); err != nil || f.Kind != "branch-unprotected" || f.Observer != "mesh-delivery-forge-bridge" { t.Errorf("the finding = %s %v", e.Value(), err) } + // A-274: the finding is said under the bridge's binding finding-events, never on the old stream. + events, err := b.js.Stream(b.ctx, boundFindingEvents) + if err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(20 * time.Second); ; time.Sleep(50 * time.Millisecond) { + if _, err = events.GetLastMsgForSubject(b.ctx, boundFindingEventsSubject+".finding-written.n1"); err == nil || time.Now().After(deadline) { + break + } + } + if err != nil { + t.Errorf("no finding-written under the binding finding-events: %v", err) + } forge, err := b.js.Stream(b.ctx, "FORGE") if err != nil { t.Fatal(err) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index bedda22d..6acb2c34 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -148,12 +148,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The state performed is a `requires` entry: opened by its binding, asked of the seat mesh-broker's show at this start // as a kept call and again on its recreated and dropped (design 47 G §577). Until show answers, nothing owed is // performed and the bridge says it waits, naming the entry. - binding := process.NewBinding(brokerSeat, forgeBridgeModule, keptCallOf(ctx, call), logger.Printf, performedLocal, findingsLocal) + binding := process.NewBinding(brokerSeat, forgeBridgeModule, keptCallOf(ctx, call), logger.Printf, performedLocal, findingsLocal, views.FindingEventsLocal) if err := binding.Register(hub); err != nil { return err } binding.Ask(performedLocal) binding.Ask(findingsLocal) + binding.Ask(views.FindingEventsLocal) if w := binding.Waiting(); w != "" { logger.Printf("the bridge %s: nothing owed to the forge is performed until it is bound", w) } @@ -163,7 +164,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // of its asks is not there yet) is said, and what it holds is stated at the next start. findingsWriter, err := views.NewHeldWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ Module: cred.Module, State: findingState{&boundBucket{binding: binding, local: findingsLocal, js: js}}, Node: cred.Node, - Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logger.Printf}) + Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logger.Printf, + // Its finding events under its binding finding-events (design 47 G A-274, §529). + Bindings: binding}) if err != nil { return fmt.Errorf("the bridge's finding writer cannot be made: %w", err) } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/findingwriter.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/findingwriter.go index be9d2cf1..d1c327fe 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/findingwriter.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/findingwriter.go @@ -13,10 +13,10 @@ import ( // findingWriter is the teller's mesh-sdk finding writer (design 47 G §461, §689): the teller observes only what its views // reader finds, so its own observe states none and the writer's statement covers the reader's open findings. The writer -// forms its subjects itself (design 47 G §383, A-13). -func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, logf func(string, ...any)) (*views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent], error) { +// forms its subjects itself (design 47 G §383, A-13), under the teller's binding finding-events (A-274, §529). +func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, bindings views.FindingBindings, logf func(string, ...any)) (*views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent], error) { w, err := views.NewWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ - Module: cred.Module, State: state, Node: cred.Node, Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logf, + Module: cred.Module, State: state, Node: cred.Node, Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logf, Bindings: bindings, Observe: func(context.Context) (views.Observed[condseat.Finding], error) { return views.Observed[condseat.Finding]{}, nil }, diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index 53ed4ec2..51cf7eac 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -10,6 +10,7 @@ package main import ( "context" "fmt" + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" "io" "log" @@ -107,7 +108,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return calls.Emit(answeredEvent, Answered{Key: key, Ask: askID, Answer: answer}, stdio.EmitOptions{MessageID: answeredID(key, askID)}) } reader := &views.ProcessReader{Module: tellerModule, Call: calls.Call, Logf: logger.Printf, Hub: hub} - writer, err := findingWriter(ctx, cred, conn, js, calls.State("findings"), logger.Printf) + // The teller's finding stream is its `finding-events` entry, asked of the seat mesh-broker's show (design 47 G A-274, + // §529, §577): the writer says its events under that binding's subjects. + binding := process.NewBinding(meshbroker.Seat, cred.Module, calls.CallKept, logger.Printf, views.FindingEventsLocal) + if err := binding.Register(hub); err != nil { + return err + } + binding.Start() + writer, err := findingWriter(ctx, cred, conn, js, calls.State("findings"), binding, logger.Printf) if err != nil { return fmt.Errorf("the teller's finding writer: %w", err) } diff --git a/modules/mesh-delivery-teller/go.mod b/modules/mesh-delivery-teller/go.mod index 96a10779..f946b858 100644 --- a/modules/mesh-delivery-teller/go.mod +++ b/modules/mesh-delivery-teller/go.mod @@ -3,6 +3,7 @@ module mesh-delivery-teller go 1.26.0 require ( + git.novox.be/novox/mesh-contracts/go/seats/meshbroker v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-teller/manifest_test.go b/modules/mesh-delivery-teller/manifest_test.go index 58a374e6..c043cadf 100644 --- a/modules/mesh-delivery-teller/manifest_test.go +++ b/modules/mesh-delivery-teller/manifest_test.go @@ -104,12 +104,12 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 } // Its own states answers and teller-marks are said as its views family views.answers, read by teller_read_answers (A-175, A-178). // The join: its first view asked again by hand is a hand act, recorded through the seat (views.StatusTools). - if want := []string{"mesh-delivery-assigner.assigner_read_asks", "mesh-delivery-view.view_read_answers", "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.registrations", "seat:operator-channel.notify"}; !reflect.DeepEqual(m.Invokes, want) { + if want := []string{"mesh-delivery-assigner.assigner_read_asks", "mesh-delivery-view.view_read_answers", "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.registrations", "seat:mesh-broker.show", "seat:operator-channel.notify"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } // design 47 G §335 (TL1): the widening of a pipeline end is told through the seat operator-channel and kept in widens-said. // design 47 G §387: the keeper's ask for the findings is heard, and the finding stream is declared (§383). - if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks", "mesh-delivery-view.views.answers", "mesh-catalog.registration-changed", "mesh-pipeline-runner.pipeline-ended", "mesh-delivery.recovery-merged", "mesh-delivery.record-opened"}; !reflect.DeepEqual(m.Consumes, want) { + if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks", "mesh-delivery-view.views.answers", "mesh-catalog.registration-changed", "mesh-pipeline-runner.pipeline-ended", "mesh-delivery.recovery-merged", "mesh-delivery.record-opened", "mesh-broker.recreated", "mesh-broker.dropped"}; !reflect.DeepEqual(m.Consumes, want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if want := []string{"operator-channel"}; !reflect.DeepEqual(sorted(m.Uses), want) { diff --git a/modules/mesh-delivery-teller/module.json b/modules/mesh-delivery-teller/module.json index a7b7e6ac..79b3fc66 100644 --- a/modules/mesh-delivery-teller/module.json +++ b/modules/mesh-delivery-teller/module.json @@ -15,6 +15,7 @@ "mesh-delivery-view.view_read_answers", "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.registrations", + "seat:mesh-broker.show", "seat:operator-channel.notify" ], "consumes": [ @@ -24,7 +25,9 @@ "mesh-catalog.registration-changed", "mesh-pipeline-runner.pipeline-ended", "mesh-delivery.recovery-merged", - "mesh-delivery.record-opened" + "mesh-delivery.record-opened", + "mesh-broker.recreated", + "mesh-broker.dropped" ], "emits": [ "answered", diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/findingwriter.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/findingwriter.go index 4cf3b9b2..777b9266 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/findingwriter.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/findingwriter.go @@ -13,10 +13,10 @@ import ( // findingWriter is the view's mesh-sdk finding writer (design 47 G §461, §689): the view observes only what its views // reader finds, so its own observe states none and the writer's statement covers the reader's open findings. The writer -// forms its subjects itself (design 47 G §383, A-13). -func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, logf func(string, ...any)) (*views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent], error) { +// forms its subjects itself (design 47 G §383, A-13), under the view's binding finding-events (A-274, §529). +func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, bindings views.FindingBindings, logf func(string, ...any)) (*views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent], error) { w, err := views.NewWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ - Module: cred.Module, State: state, Node: cred.Node, Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logf, + Module: cred.Module, State: state, Node: cred.Node, Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logf, Bindings: bindings, Observe: func(context.Context) (views.Observed[condseat.Finding], error) { return views.Observed[condseat.Finding]{}, nil }, diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index e1b98ecb..65a5b543 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -25,6 +25,7 @@ import ( "syscall" "time" + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/trunk" @@ -132,7 +133,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro hub := process.NewEventHub(logger.Printf) defer hub.Stop() reader := &views.ProcessReader{Module: viewstate.ViewModule, Call: calls.Call, Logf: logger.Printf, Hub: hub} - writer, err := findingWriter(ctx, cred, conn, js, calls.State("findings"), logger.Printf) + // The view's finding stream is its `finding-events` entry, asked of the seat mesh-broker's show (design 47 G A-274, + // §529, §577): the writer says its events under that binding's subjects. + binding := process.NewBinding(meshbroker.Seat, cred.Module, calls.CallKept, logger.Printf, views.FindingEventsLocal) + if err := binding.Register(hub); err != nil { + return err + } + binding.Start() + writer, err := findingWriter(ctx, cred, conn, js, calls.State("findings"), binding, logger.Printf) if err != nil { return fmt.Errorf("the view's finding writer: %w", err) } diff --git a/modules/mesh-delivery-view/go.mod b/modules/mesh-delivery-view/go.mod index 8cf7924c..6f8ad4f2 100644 --- a/modules/mesh-delivery-view/go.mod +++ b/modules/mesh-delivery-view/go.mod @@ -3,6 +3,7 @@ module mesh-delivery-view go 1.26.0 require ( + git.novox.be/novox/mesh-contracts/go/seats/meshbroker v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index acbf4d1d..06b1beed 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -155,6 +155,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 {Verb: "seat:mesh-catalog.entry", Fields: []string{"manifest-digest"}}, // §558: the module's value is the entry's manifest-digest (§506) {Verb: "seat:mesh-delivery.source", Fields: []string{"branches", "group", "repositories", "revision"}}, // §507, §614, A-173: the branches before the first registrations, and each watched branch's source (§558: fields G adds) // ADR 0663: the modules are the catalog half's, read through the seat's verb revision, never the indexer's {Verb: "seat:mesh-catalog.seat-definitions"}, // design 47 G A-82, §204: the build seats are read from the seat list + {Verb: "seat:mesh-broker.show"}, // A-274: the binding of its finding-events {Verb: "seat:mesh-hand-acts.hand-act"}, // the join: its first view asked again by hand is a hand act (views.StatusTools) {Verb: "seat:mesh-module-assigner.assignments"}, // A-82: the holder of each build seat {Verb: "seat:mesh-wants.revision"}, // §512, A-242: view_tiers compares kept assignment revisions @@ -175,7 +176,8 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks", "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.wants", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-delivery.source-moved", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.taken", "mesh-delivery-teller.views.answers", "mesh-wants.wait-held", "mesh-wants.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", - "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended"}; !reflect.DeepEqual(m.Consumes, want) { // design 47 G §512, A-242: every part's wait-held and wait-ended + "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended", + "mesh-broker.recreated", "mesh-broker.dropped"}; !reflect.DeepEqual(m.Consumes, want) { // design 47 G §512, A-242: every part's wait-held and wait-ended t.Errorf("consumes = %v, want %v", m.Consumes, want) } // The one event the view says: a source revision is delivered (1906/D9, ADR 0539 §17), derived by it from the wants and the haves. diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 3df085f5..90ee08be 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -84,7 +84,9 @@ "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-delivery-builder.wait-held", - "mesh-delivery-builder.wait-ended" + "mesh-delivery-builder.wait-ended", + "mesh-broker.recreated", + "mesh-broker.dropped" ], "invokes": [ "mesh-delivery-assigner.assigner_read_asks", @@ -105,6 +107,7 @@ {"verb": "seat:mesh-catalog.entry", "fields": ["manifest-digest"]}, {"verb": "seat:mesh-delivery.source", "fields": ["branches", "group", "repositories", "revision"]}, "seat:mesh-catalog.seat-definitions", + "seat:mesh-broker.show", "seat:mesh-hand-acts.hand-act", "seat:mesh-module-assigner.assignments", "seat:mesh-wants.revision", 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 8dac5cfb..624de9cc 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -10,13 +10,16 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" + "slices" "strings" "sync" + "time" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/membership" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatdef" "mesh-delivery-watchman/internal/watch" ) @@ -75,10 +78,18 @@ 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, hub *process.EventHub, w *watch.Watchman, logf func(string, ...any)) { +func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Account, hub *process.EventHub, w *watch.Watchman, binding *process.Binding, 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. + // pull of EVENTS, beside the take-in events and the views reader. The seat mesh-broker's recreated and dropped go to + // the binding of the watchman's finding-events (design 47 G A-274, §577), which this handler holds, since a message has + // one taker; and the binding asks show again at each bind of the hub's consumer, as its Register does. + brokerEvents := []string{seatdef.EventSubject(meshbroker.Seat, "recreated"), seatdef.EventSubject(meshbroker.Seat, "dropped")} + hub.OnBound(func(time.Time) { binding.Start() }) if err := hub.Handle("mesh.seat.*.event.>", func(m jetstream.Msg) { + if slices.Contains(brokerEvents, m.Subject()) { + binding.Hear(m) + return + } consumed(w, m.Subject(), m.Data(), logf) _ = m.Ack() }); err != nil { diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/findingwriter.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/findingwriter.go index cd0c5b19..5b300efb 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/findingwriter.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/findingwriter.go @@ -52,9 +52,10 @@ func seedFindings(state process.State) (*heldFindings, map[string][]byte, error) } // start makes and starts the held writer, which reads the state's findings itself before it states anything. -func (h *heldFindings) start(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, logf func(string, ...any)) error { +// Its events are said under the watchman's binding finding-events (design 47 G A-274, §529). +func (h *heldFindings) start(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, bindings views.FindingBindings, logf func(string, ...any)) error { w, err := views.NewHeldWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ - Module: cred.Module, State: state, Node: cred.Node, Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logf, + Module: cred.Module, State: state, Node: cred.Node, Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logf, Bindings: bindings, }) if err != nil { return err diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go index 0b08b916..39028cea 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go @@ -92,10 +92,26 @@ func sayAsk(t *testing.T, ctx context.Context, js jetstream.JetStream, event, id } } -// findingsStream is the watchman's own stream of its findings' events (design 47 G §383), which its finding writer says into. -func findingsStream(t *testing.T, ctx context.Context, js jetstream.JetStream) { +// findingEventsSubject is the subject the stand-in seat mesh-broker's show answers for the watchman's finding-events. +const findingEventsSubject = "mesh.bus." + watch.WatchmanModule + ".finding-events" + +// findingsStream is the watchman's own stream of its findings' events, its requires entry finding-events (design 47 G +// §383, §529), which its finding writer says into under the names a stand-in of the seat mesh-broker's show answers (A-274). +func findingsStream(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) { t.Helper() - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "WATCHMAN_FINDINGS", Subjects: []string{"mesh.stream." + watch.WatchmanModule + ".event.>"}}); err != nil { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "WATCHMAN_FINDINGS", Subjects: []string{findingEventsSubject + ".>"}}); err != nil { + t.Fatal(err) + } + answer := []byte(`{"result":{"state":"created","binding":{"local":"finding-events","storage":"stream","template":"events",` + + `"stream":"WATCHMAN_FINDINGS","subject":"` + findingEventsSubject + `"}},"node":"n0"}`) + for _, subject := range []string{"mesh.seat.mesh-broker.call.show.>", "mesh.seat.mesh-broker.tool.show"} { + sub, err := nc.Subscribe(subject, func(m *nats.Msg) { _ = m.Respond(answer) }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + } + if err := nc.Flush(); err != nil { t.Fatal(err) } } @@ -159,7 +175,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 kvs[c.Bucket] = kv } serveOwners(t, ctx, js, ctl) - findingsStream(t, ctx, js) + findingsStream(t, ctx, js, ctl) makeEvents(t, ctx, js) // The controller's consumer on DEAD_LETTER_NOTICES, the dead letters' streams and a source stream with one message a @@ -297,6 +313,15 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 return finding(watch.FindingKey(watch.ScopeDelivery, stuckOf(askStream, ask1), watch.KindStuck)) }) until("node n1's silence", func() bool { return finding("machine.n1.silent") }) + // A-274: the finding is said under the watchman's binding finding-events. + until("a finding-written under the binding finding-events", func() bool { + st, err := js.Stream(ctx, "WATCHMAN_FINDINGS") + if err != nil { + return false + } + _, err = st.GetLastMsgForSubject(ctx, findingEventsSubject+".finding-written.n1") + return err == nil + }) // 1952/WM5, 1126/A2/AC2: with no lease written, while the controller's consumer stands the watchman says the notices // wait; the consumer removed, it makes its own and keeps a notice with its headers. @@ -398,7 +423,7 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 kvs[c.Bucket] = kv } serveOwners(t, ctx, js, ctl) - findingsStream(t, ctx, js) + findingsStream(t, ctx, js, ctl) makeEvents(t, ctx, js) const id = "m0000009ffffff" sayAsk(t, ctx, js, "ask-taken", id) @@ -479,7 +504,7 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { kvs[c.Bucket] = kv } serveOwners(t, ctx, js, ctl) - findingsStream(t, ctx, js) + findingsStream(t, ctx, js, ctl) makeEvents(t, ctx, js) fpNew := meshcatalogclient.SourcePrefix2 + strings.Repeat("b", 64) fpOld := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64) 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 ed02decb..dbd28216 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -24,6 +24,7 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" @@ -249,7 +250,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return fmt.Errorf("the watchman's findings cannot be read: %w", err) } - if err := held.start(ctx, cred, conn, js, findingsState, logger.Printf); err != nil { + // The watchman's finding stream is its `finding-events` entry, asked of the seat mesh-broker's show (design 47 G A-274, + // §529, §577); the seats' events handler below hands it the seat's recreated and dropped. + binding := process.NewBinding(meshbroker.Seat, cred.Module, calls.CallKept, logger.Printf, views.FindingEventsLocal) + binding.Start() + if err := held.start(ctx, cred, conn, js, findingsState, binding, logger.Printf); err != nil { return fmt.Errorf("the watchman's finding writer: %w", err) } defer held.w.Close() @@ -413,7 +418,7 @@ 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) + hearConsumed(ctx, js, cred, hub, w, binding, logger.Printf) defer hub.Stop() go func() { errs <- readLate(ctx, w, wants, watch.AssignerWantsBucket, func(e jetstream.KeyValueEntry) { diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index d78bcf2e..c7056d48 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -126,11 +126,11 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 } // The join: its first view asked again by hand is a hand act, recorded through the seat (views.StatusTools). The waits the // indexer holds at the watchman's start are asked of its tool (ADR 0633 §5), a verb edge as the view's. - if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_verdicts", "seat:mesh-hand-acts.hand-act", "seat:mesh-conditions.conditions"}; !reflect.DeepEqual(m.Invokes, want) { + if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_verdicts", "seat:mesh-hand-acts.hand-act", "seat:mesh-conditions.conditions", "seat:mesh-broker.show"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } // DLV-M5 its first views, 1585/T20 the indexer's take-in events, and DLV-W1 (ADR 0595 §4): the accepted verdicts and the bus module's created and refused, durably. - if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-catalog.manifest-accepted", "mesh-catalog.manifest-refused", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "node-build-go.ask-taken", "node-build-go.step-progress", "node-build-go.step-ended", "node-build-image.ask-taken", "node-build-image.step-progress", "node-build-image.step-ended", "node-build-node.ask-taken", "node-build-node.step-progress", "node-build-node.step-ended", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", "node-engine.running", "mesh-broker.created", "mesh-broker.refused", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) { + if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-catalog.manifest-accepted", "mesh-catalog.manifest-refused", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "node-build-go.ask-taken", "node-build-go.step-progress", "node-build-go.step-ended", "node-build-image.ask-taken", "node-build-image.step-progress", "node-build-image.step-ended", "node-build-node.ask-taken", "node-build-node.step-progress", "node-build-node.step-ended", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", "node-engine.running", "mesh-broker.created", "mesh-broker.refused", "mesh-broker.recreated", "mesh-broker.dropped", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } // 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4). diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index e113c701..c474595e 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -23,7 +23,8 @@ "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_verdicts", "seat:mesh-hand-acts.hand-act", - "seat:mesh-conditions.conditions" + "seat:mesh-conditions.conditions", + "seat:mesh-broker.show" ], "consumes": [ "mesh-delivery-assigner.views.wants", @@ -71,6 +72,8 @@ "node-engine.running", "mesh-broker.created", "mesh-broker.refused", + "mesh-broker.recreated", + "mesh-broker.dropped", "mesh-conditions.findings-asked" ], "tools": [