From 62de8d0bb81ddd54f5578b0bb02ad7116788034c Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:49:48 +0100 Subject: [PATCH] WM4, WM5: the watchman hears the node-engine's wait-held and wait-ended MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A hold of the composition standing past 30 minutes is delivery-waits (node, module, reason), a wait for a kept build past 60 minutes is kept-build-waits; each clears on wait-ended. The node-engine's version is not yet named in them (it needs the catalog entry of the fingerprint running.. names). Builds: design 47 G §499, §541, §271, §376 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-watchman/consumes.go | 20 ++++++ .../cmd/mesh-delivery-watchman/main_test.go | 14 ++++ .../internal/watch/waits.go | 71 +++++++++++++++++++ .../internal/watch/waits_test.go | 36 ++++++++++ .../mesh-delivery-watchman/manifest_test.go | 2 +- modules/mesh-delivery-watchman/module.json | 2 + 6 files changed, 144 insertions(+), 1 deletion(-) 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 912f1fe4..864cab05 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" @@ -46,6 +47,7 @@ func consumed(w *watch.Watchman, subject string, body []byte, logf func(string, } w.Accepted(a) case pipelineSubject(w, subject, body): + case engineWaitSubject(w, subject, body): case recordSubject(w, subject, body): case buildSeatSubject(w, subject, body): case brokerSubject(subject) && strings.HasSuffix(subject, ".event."+eventCreated): @@ -190,6 +192,24 @@ func recordSubject(w *watch.Watchman, subject string, body []byte) bool { return true } +// engineWaitSubject hands the node-engine seat's `wait-held` and `wait-ended`, on +// mesh.seat.node-engine.event..., to the watchman (design 47 G §499, §541). It says whether the subject was one. +func engineWaitSubject(w *watch.Watchman, subject string, body []byte) bool { + parts := strings.Split(subject, ".") + if len(parts) != 7 || parts[0] != "mesh" || parts[1] != "seat" || parts[2] != nodeengine.Seat || parts[3] != "event" { + return false + } + switch parts[4] { + case nodeengine.EventWaitHeld: + w.EngineWaitHeld(parts[5], parts[6], body) + case nodeengine.EventWaitEnded: + w.EngineWaitEnded(parts[5], parts[6]) + default: + return false + } + return true +} + // buildSeatSubject hands the build seats' `ask-taken`, `step-progress` and `step-ended`, on // mesh.seat..event.., to the watchman (design 47 G §483, §496, §602, A-136): it times every ask of a // build seat from them. It says whether the subject was one of them. diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go index a0f47f2d..aecea500 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go @@ -75,6 +75,20 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 t.Fatalf("created left %v", out.findings) } }) + t.Run("design 47 G §499, §541: a node-engine's wait-held and wait-ended reach the watchman by their subjects", func(t *testing.T) { + out := &findingsOut{findings: map[string]bool{}} + w := watch.New(out, time.Now) + say := func(string, ...any) {} + body := fmt.Sprintf(`{"step":"compose","reason":"entry-missing","since":%q}`, time.Now().Add(-2*time.Hour).UTC().Format(time.RFC3339)) + consumed(w, "mesh.seat.node-engine.event.wait-held.ace.web", []byte(body), say) + if !out.findings[watch.FindingKey(watch.ScopeDelivery, "ace.web", watch.KindDeliveryWaits)] { + t.Fatalf("a hold two hours old reached %v", out.findings) + } + consumed(w, "mesh.seat.node-engine.event.wait-ended.ace.web", []byte(body), say) + if len(out.findings) != 0 { + t.Fatalf("wait-ended left %v", out.findings) + } + }) err := run(context.Background(), func(string) string { return "" }, io.Discard) if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") { t.Fatalf("run = %v, want a refusal naming MESH_BROKER_FILE", err) diff --git a/modules/mesh-delivery-watchman/internal/watch/waits.go b/modules/mesh-delivery-watchman/internal/watch/waits.go index b9d4d0c4..c9931cd1 100644 --- a/modules/mesh-delivery-watchman/internal/watch/waits.go +++ b/modules/mesh-delivery-watchman/internal/watch/waits.go @@ -6,6 +6,7 @@ import ( "strings" "time" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" ) @@ -110,3 +111,73 @@ func (w *Watchman) putStands(id string, a meshwants.Wait, bound time.Duration, f } w.find(f) } + +// What a node-engine says of a hold it keeps (design 47 G §499, §541), heard on the seat node-engine's `wait-held` and `wait-ended` +// (`mesh.seat.node-engine.event...`, the node the one that says it): a hold of the composition (`step: +// compose`, with its reason) standing past composing's 30 minutes (§266) is the finding `delivery-waits`, naming the node, the +// module and the reason, and a wait for a kept build (`step: kept-build`, naming the build node) standing past building's default +// bound of 60 minutes is `kept-build-waits`. Each clears on `wait-ended`. + +// The findings of a node-engine's waits, and the bounds that stand when its event names none. +const ( + KindDeliveryWaits = "delivery-waits" + KindKeptBuildWaits = "kept-build-waits" + ComposeBound = 30 * time.Minute + KeptBuildBound = 60 * time.Minute +) + +// engineRow is the row of the signals table of a node-engine's wait. +const engineRow = "engine-wait" + +func engineWaitID(node, module string) string { return node + "." + module } + +// EngineWaitHeld hears the node-engine of node saying `wait-held` for module. +func (w *Watchman) EngineWaitHeld(node, module string, body []byte) { + var a nodeengine.Wait + if err := json.Unmarshal(body, &a); err != nil || a.Since.IsZero() { + w.say("a wait-held of the node-engine of %s for %s does not read, left: %v", node, module, err) + return + } + kind, bound := KindDeliveryWaits, ComposeBound + if a.Step == nodeengine.WaitStepKeptBuild { + kind, bound = KindKeptBuildWaits, KeptBuildBound + } + if d, err := time.ParseDuration(a.Bound); err == nil && d > 0 { + bound = d + } + from := a.Since + if a.ClockFrom != nil { + from = *a.ClockFrom + } + id := engineWaitID(node, module) + what := "holds the composition of " + module + if a.Step == nodeengine.WaitStepKeptBuild { + what = "waits for the kept build of " + module + " on " + orWords(a.Node, "no node") + } + f := Finding{Scope: ScopeDelivery, ID: id, Kind: kind, Severity: Warning, Machine: node, Source: engineRow, Since: a.Since, + Summary: fmt.Sprintf("the node-engine of %s %s: %s, since %s (bound %s)", node, what, orWords(a.Reason, "no reason said"), stamp(a.Since), bound), + Said: "held for " + bound.String(), + Headline: "A module's delivery is held on a machine", + Explanation: "The node-engine of " + node + " " + what + " (" + orWords(a.Reason, "no reason said") + ") and its bound of " + + bound.String() + " has run, so the module is not at the build wanted there.", + Needs: "what it waits on to come: " + orWords(a.Words, orWords(a.Reason, "the node-engine's wait to end"))} + w.mu.Lock() + defer w.mu.Unlock() + w.unsignal(engineRow, id) + w.unfind(f.Key()) + if left := bound - w.now().Sub(from); left > 0 { + w.signal(engineRow, id, left, signalDue{Do: dueFind, Finding: f}) + return + } + w.find(f) +} + +// EngineWaitEnded hears the node-engine of node saying `wait-ended` for module: the hold is over, whichever kind it was. +func (w *Watchman) EngineWaitEnded(node, module string) { + id := engineWaitID(node, module) + w.mu.Lock() + defer w.mu.Unlock() + w.unsignal(engineRow, id) + w.unfind(FindingKey(ScopeDelivery, id, KindDeliveryWaits)) + w.unfind(FindingKey(ScopeDelivery, id, KindKeptBuildWaits)) +} diff --git a/modules/mesh-delivery-watchman/internal/watch/waits_test.go b/modules/mesh-delivery-watchman/internal/watch/waits_test.go index 481bd363..9e587b8c 100644 --- a/modules/mesh-delivery-watchman/internal/watch/waits_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/waits_test.go @@ -2,9 +2,11 @@ package watch import ( "encoding/json" + "strings" "testing" "time" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" ) @@ -93,3 +95,37 @@ func TestAPutWithoutHoldThatStandsIsPutStands(t *testing.T) { t.Fatalf("wait-ended left %v", o.findings) } } + +// design 47 G §499, §541 (WM4, WM5): a node-engine's hold of the composition standing 30 minutes is `delivery-waits` naming the node, +// the module and the reason, a wait for a kept build standing 60 minutes is `kept-build-waits`, and `wait-ended` clears either. +func TestANodeEnginesWaitIsAFindingPastItsBound(t *testing.T) { + held := func(step, reason string, since time.Time) []byte { + b, _ := json.Marshal(nodeengine.Wait{Step: step, Reason: reason, Node: "build1", Since: since, Owner: "node-engine"}) + return b + } + for _, tc := range []struct { + name, step, reason, kind string + bound time.Duration + }{ + {"a hold of the composition", nodeengine.WaitStepCompose, "restore point missing", KindDeliveryWaits, 30 * time.Minute}, + {"a wait for a kept build", nodeengine.WaitStepKeptBuild, "node away", KindKeptBuildWaits, 60 * time.Minute}, + } { + t.Run(tc.name, func(t *testing.T) { + w, o, c := setup() + id := "ace.web" + w.EngineWaitHeld("ace", "web", held(tc.step, tc.reason, c.t)) + if o.ttls[signalKey(engineRow, id)] != tc.bound || len(o.findings) != 0 { + t.Fatalf("within the bound: ttl %v, findings %v", o.ttls[signalKey(engineRow, id)], o.findings) + } + expire(w, c, signalKey(engineRow, id), tc.bound) + f, ok := o.findings[FindingKey(ScopeDelivery, id, tc.kind)] + if !ok || len(o.findings) != 1 || !strings.Contains(f.Summary, tc.reason) { + t.Fatalf("past its bound: %v", o.findings) + } + w.EngineWaitEnded("ace", "web") + if len(o.findings) != 0 { + t.Fatalf("wait-ended left %v", o.findings) + } + }) + } +} diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 8fb2ae9a..dea3bc92 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -130,7 +130,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 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-delivery-indexer.views.modules", "mesh-catalog.manifest-accepted", "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-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-delivery-indexer.views.modules", "mesh-catalog.manifest-accepted", "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", "node-engine.wait-held", "node-engine.wait-ended", "mesh-broker.created", "mesh-broker.refused", "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 52e7103b..5e6fe470 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -67,6 +67,8 @@ "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", + "node-engine.wait-held", + "node-engine.wait-ended", "mesh-broker.created", "mesh-broker.refused", "mesh-conditions.findings-asked"