novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

A-274: the watchman, forge bridge, teller and view pass their finding-events binding to the finding writer

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-12 11:56:08 +01:00
parent 8f7fd39e21
commit af96eeba86
18 changed files with 132 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go
+28 -6
View File
@@ -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<E>, SEAT_<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)
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go
+5 -2
View File
@@ -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)
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/findingwriter.go
+3 -3
View File
@@ -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
},
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go
+9 -1
View File
@@ -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)
}
modules/mesh-delivery-teller/go.mod
+1
View File
@@ -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
modules/mesh-delivery-teller/manifest_test.go
+2 -2
View File
@@ -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) {
modules/mesh-delivery-teller/module.json
+4 -1
View File
@@ -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",
modules/mesh-delivery-view/cmd/mesh-delivery-view/findingwriter.go
+3 -3
View File
@@ -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
},
modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go
+9 -1
View File
@@ -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)
}
modules/mesh-delivery-view/go.mod
+1
View File
@@ -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
modules/mesh-delivery-view/manifest_test.go
+3 -1
View File
@@ -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.
modules/mesh-delivery-view/module.json
+4 -1
View File
@@ -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",
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go
+13 -2
View File
@@ -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 {
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/findingwriter.go
+3 -2
View File
@@ -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
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go
+31 -6
View File
@@ -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)
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go
+7 -2
View File
@@ -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) {
modules/mesh-delivery-watchman/manifest_test.go
+2 -2
View File
@@ -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).
modules/mesh-delivery-watchman/module.json
+4 -1
View File
@@ -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": [