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

WM4, WM5: the watchman hears the node-engine's wait-held and wait-ended

Browse Source 
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.<node>.<node-engine's module> names).

Builds: design 47 G §499, §541, §271, §376

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 05:49:48 +01:00
parent 464206d2de
commit 62de8d0bb8
6 changed files with 144 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go
+20
View File
@@ -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.<event>.<node>.<module>, 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.<seat>.event.<event>.<node>, 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.
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go
+14
View File
@@ -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)
modules/mesh-delivery-watchman/internal/watch/waits.go
+71
View File
@@ -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.<event>.<node>.<module>`, 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))
}
modules/mesh-delivery-watchman/internal/watch/waits_test.go
+36
View File
@@ -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)
}
})
}
}
modules/mesh-delivery-watchman/manifest_test.go
+1 -1
View File
@@ -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).
modules/mesh-delivery-watchman/module.json
+2
View File
@@ -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"