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

WM13 (watchman half): the builder's builds are no longer read; a package wait is heard as wait-held

Browse Source 
The watchman's read of mesh-delivery-builder.builds (Build in mirror.go, the bucket, its manifest reads) goes; it hears the
builder's wait-held and wait-ended on the builder's stream, naming a package as their part, and times them as an acting
part's wait. A toolchain wait (no part) is left to the builder's own finding. The node-engine's have and mesh-host.health
reads stay: no event carries a have's phase, kept data, node record or a node's health report (asked of the lead).

Builds: design 47 G A-192, §484, §496, §602, §548

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 05:53:23 +01:00
parent 524fbb02ef
commit faec6b5333
10 changed files with 95 additions and 72 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
@@ -48,6 +48,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 moduleWaitSubject(w, subject, body):
case recordSubject(w, subject, body):
case buildSeatSubject(w, subject, body):
case brokerSubject(subject) && strings.HasSuffix(subject, ".event."+eventCreated):
@@ -210,6 +211,25 @@ func engineWaitSubject(w *watch.Watchman, subject string, body []byte) bool {
return true
}
// moduleWaitSubject hands a module's `wait-held` and `wait-ended` on its own stream, on
// mesh.stream.<module>.event.<event>.<module>.<fingerprint>, to the watchman: the form the builder says a key queued behind a package
// in (design 47 G A-192). It says whether the subject was one.
func moduleWaitSubject(w *watch.Watchman, subject string, body []byte) bool {
parts := strings.Split(subject, ".")
if len(parts) != 7 || parts[0] != "mesh" || parts[1] != "stream" || parts[3] != "event" {
return false
}
switch parts[4] {
case nodeengine.EventWaitHeld:
w.ModuleWaitHeld(parts[2], body)
case nodeengine.EventWaitEnded:
w.ModuleWaitEnded(parts[2], body)
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.go
-16
View File
@@ -323,17 +323,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
haveSeen = nil
}
}
// The builder's builds (ADR 0603 §8): a key queued behind a package is a wait, mirrored from the time it was written.
var buildSeen jetstream.KeyWatcher
if kv := process.OptionalBucket(ctx, js, watch.BuilderBuildsBucket, "a build waiting for a package is not watched until the watchman starts again", logger.Printf); kv != nil {
buildEach := func(e jetstream.KeyValueEntry) { w.Build(e.Key(), e.Value(), opOf(e), e.Created()) }
if buildSeen, err = settle(ctx, kv, buildEach); err != nil {
process.NotWatched(logger.Printf, watch.BuilderBuildsBucket, err, "a build waiting for a package is not watched until the watchman starts again")
buildSeen = nil
} else {
w.ReadWhole(watch.BuilderBuildsBucket)
}
}
// The events first, then the one request of the open connections, so no connection falls between them.
onEvent := func(open bool) nats.MsgHandler {
return func(m *nats.Msg) {
@@ -491,11 +480,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
go func() {
errs <- follow(ctx, pulses, func(e jetstream.KeyValueEntry) { w.Pulse(e.Key(), opOf(e), e.Created()) })
}()
if buildSeen != nil {
go func() {
errs <- follow(ctx, buildSeen, func(e jetstream.KeyValueEntry) { w.Build(e.Key(), e.Value(), opOf(e), e.Created()) })
}()
}
go func() {
for {
select {
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go
+13
View File
@@ -89,6 +89,19 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
t.Fatalf("wait-ended left %v", out.findings)
}
})
t.Run("design 47 G A-192: a module's wait-held on its stream, naming a part, reaches the watchman", func(t *testing.T) {
out := &findingsOut{findings: map[string]bool{}}
w := watch.New(out, time.Now)
body := fmt.Sprintf(`{"step":"building","part":"mesh-sdk@v0.1.14","module":"web","fingerprint":"src2:f","bound":"1h0m0s","since":%q}`, time.Now().Add(-2*time.Hour).UTC().Format(time.RFC3339))
consumed(w, "mesh.stream.mesh-delivery-builder.event.wait-held.web.src2:f", []byte(body), func(string, ...any) {})
if len(out.findings) != 1 {
t.Fatalf("a package wait two hours old reached %v", out.findings)
}
consumed(w, "mesh.stream.mesh-delivery-builder.event.wait-ended.web.src2:f", []byte(body), func(string, ...any) {})
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/findings_test.go
+3 -10
View File
@@ -12,7 +12,6 @@ import (
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
@@ -737,7 +736,7 @@ func TestTheWatchmansFindings(t *testing.T) {
w.DeadLettersHeld(map[string]int{"SEAT_X.y_worker": 1}, who)
w.NoticesWaiting(true)
}, map[string]string{"bus.SEAT_X.y_worker.max-deliveries": Warning, "bus.DEAD_LETTER_NOTICES.controller.notices-waiting": Warning}},
{"ADR 0603 §8: a want held by a rule, a provision asked and a build queued behind a package each raise a finding naming the wait past their bound, and clear when it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
{"ADR 0603 §8: a want held by a rule and a provision asked each raise a finding naming the wait past their bound, and clear when it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.SetStepBound(30 * time.Minute)
held, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fpOld, Self: true, Newest: fpNew, Rule: meshwantsclient.RuleCanary, Words: "the canary ace does not run it yet"})
released, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fpNew, Self: true})
@@ -745,15 +744,11 @@ func TestTheWatchmansFindings(t *testing.T) {
Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: meshprovisionerclient.Asked, At: t0})
provided, _ := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "oidc", Consumer: meshprovisionerclient.End{Module: "photos", Node: "ace"},
Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: meshprovisionerclient.Provided, At: t0})
bkey, _ := builds.BuilderBuildKey("web", fpNew)
queued, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildQueued, Waits: "mesh-sdk@v0.1.14"})
building, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildBuilding, By: "n1"})
w.Have("ace.web", haveValue(t, fpOld), Put) // the node runs what the held want names: the hold is still the wait
w.Want("want.ace.web", held, Put, t0)
w.Provision("ace.photos.oidc", asked, Put)
w.Build(bkey, queued, Put, t0)
waits := map[string]string{AssignerWantsBucket: "want.ace.web", ProvisionsBucket: "ace.photos.oidc", BuilderBuildsBucket: bkey}
if len(o.pulses) != 3 || len(o.findings) != 0 {
waits := map[string]string{AssignerWantsBucket: "want.ace.web", ProvisionsBucket: "ace.photos.oidc"}
if len(o.pulses) != 2 || len(o.findings) != 0 {
t.Fatalf("pulses %v findings %v", o.pulses, o.findings)
}
// The keys are written again as their parts move on, with new words and a new time: the wait still began at t0.
@@ -763,7 +758,6 @@ func TestTheWatchmansFindings(t *testing.T) {
Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: meshprovisionerclient.Asked, Why: "the provider is applying", At: c.t})
w.Want("want.ace.web", held2, Put, c.t)
w.Provision("ace.photos.oidc", asked2, Put)
w.Build(bkey, queued, Put, c.t)
for bucket, key := range waits {
expire(w, c, pulseKeyOf(t, bucket, key), 31*time.Minute)
}
@@ -775,7 +769,6 @@ func TestTheWatchmansFindings(t *testing.T) {
}
w.Want("want.ace.web", released, Put, c.t)
w.Provision("ace.photos.oidc", provided, Put)
w.Build(bkey, building, Put, c.t)
}, map[string]string{}},
}
for _, tc := range cases {
modules/mesh-delivery-watchman/internal/watch/mirror.go
-39
View File
@@ -8,7 +8,6 @@ import (
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo"
@@ -165,44 +164,6 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) {
describe("the node agent of %s to run %s at %s", node, module, shortFP(want.Fp)), want.Fp, at, 0)
}
// A build key queued behind a package is a wait of the builder's (ADR 0603 §8, ADR 0500 clause 6): its Waits names the package
// version the registry does not yet hold, and the key is not claimed until it is published. Its pulse is the pulse of the
// builder's builds bucket under the key, from the time the key was written, bounded as a step is (StepBound); the key
// claimed, done, failed or deleted ends it.
// Build is a change of a key of the builder's builds bucket, read from the watch (at the start the keys that stand, then
// each change), at the time the entry was written.
func (w *Watchman) Build(key string, value []byte, op Op, at time.Time) {
module, fp, err := builds.ParseBuilderBuildKey(key)
if err != nil {
return
}
w.mu.Lock()
defer w.mu.Unlock()
if op == Expired {
return
}
if op == Delete {
w.moved(BuilderBuildsBucket, key)
return
}
w.stands(BuilderBuildsBucket, key)
b, err := builds.DecodeBuilderBuild(value)
if err != nil {
w.say("the build %s cannot be read, and is left: %v", key, err)
return
}
if b.State != builds.BuildQueued || b.Waits == "" {
w.moved(BuilderBuildsBucket, key)
return
}
if at.IsZero() {
at = w.now()
}
w.mirrorKept(BuilderBuildsBucket, key,
describe("the package %s to be published before the build of %s at %s, bound %s", b.Waits, module, shortFP(fp), w.stepBound), b.Waits, at)
}
// Have is a change of a key of the node-engine's have bucket: the fingerprint a node runs of a module. One that names the
// fingerprint of the node's want for the module ends the want's pulse. A node's own key and a key deleted are left.
func (w *Watchman) Have(key string, value []byte, op Op) {
modules/mesh-delivery-watchman/internal/watch/names.go
-3
View File
@@ -14,7 +14,6 @@ const (
ProvisionsBucket = "mesh-provisioner_provisions"
// IndexerModule is the holder of the seat mesh-catalog, named from this module's manifest (consumes and invokes of mesh-delivery-indexer).
IndexerModule = "mesh-delivery-indexer"
builderModule = "mesh-delivery-builder"
// The watchman's own states of the seat mesh-watchdogs.
PresenceBucket = WatchmanModule + "_" + meshwatchdogs.StatePresence
@@ -24,8 +23,6 @@ const (
// HaveBucket and NodeHealthBucket are the node-engine's states have and health (reads: node-engine.have, mesh-host.health).
HaveBucket = "node-engine_have"
NodeHealthBucket = "mesh-host_health"
// BuilderBuildsBucket is the builder's state builds (reads: mesh-delivery-builder.builds).
BuilderBuildsBucket = builderModule + "_builds"
// AssignerWantsBucket is the assigner's state wants, followed by its views family (consumes: mesh-delivery-assigner.views.wants).
AssignerWantsBucket = AssignerModule + "_wants"
// ModulesBucket is the indexer's state modules, followed by its views family (consumes: mesh-delivery-indexer.views.modules).
modules/mesh-delivery-watchman/internal/watch/waits.go
+26
View File
@@ -181,3 +181,29 @@ func (w *Watchman) EngineWaitEnded(node, module string) {
w.unfind(FindingKey(ScopeDelivery, id, KindDeliveryWaits))
w.unfind(FindingKey(ScopeDelivery, id, KindKeptBuildWaits))
}
// ModuleWaitHeld hears a module's `wait-held` on its own stream (`mesh.stream.<module>.event.wait-held.<module>.<fingerprint>`), the form
// the builder says a key queued behind a package in, with the package as its `part` (design 47 G A-192, ADR 0603 §8): it is timed as any
// acting part's wait. A wait naming no part is a toolchain wait, which the builder times itself (§672), and is left.
func (w *Watchman) ModuleWaitHeld(holder string, body []byte) {
var a meshwants.Wait
if json.Unmarshal(body, &a) == nil && a.Part == "" {
return
}
a2, ok := decodeWait(w, holder, "wait-held", body)
if !ok {
return
}
w.mu.Lock()
defer w.mu.Unlock()
w.waitHeld(holder, a2)
}
// ModuleWaitEnded hears the same module's `wait-ended`.
func (w *Watchman) ModuleWaitEnded(holder string, body []byte) {
var a meshwants.Wait
if json.Unmarshal(body, &a) == nil && a.Part == "" {
return
}
w.WaitEnded(holder, body)
}
modules/mesh-delivery-watchman/internal/watch/waits_test.go
+29
View File
@@ -129,3 +129,32 @@ func TestANodeEnginesWaitIsAFindingPastItsBound(t *testing.T) {
})
}
}
// design 47 G A-192 (WM13): a key queued behind a package is a wait the builder says on its own stream with the package as its part;
// past its bound it is `acting-part-late`, `wait-ended` clears it, and a toolchain wait (no part) is left to the builder.
func TestAKeyQueuedBehindAPackageIsAnActingPartsWait(t *testing.T) {
body := func(part string, since time.Time) []byte {
b, _ := json.Marshal(meshwants.Wait{Step: "building", Part: part, On: "package:" + part, Module: "web", Fingerprint: "src2:f", Since: since, Bound: "1h0m0s"})
return b
}
const holder = "builder"
w, o, c := setup()
id := waitID(holder, meshwants.Wait{Step: "building", Part: "mesh-sdk@v0.1.14", Module: "web"})
w.ModuleWaitHeld(holder, body("mesh-sdk@v0.1.14", c.t))
if o.ttls[signalKey(actingRow, id)] != time.Hour || len(o.findings) != 0 {
t.Fatalf("within the bound: ttl %v, findings %v", o.ttls[signalKey(actingRow, id)], o.findings)
}
expire(w, c, signalKey(actingRow, id), time.Hour)
if _, ok := o.findings[FindingKey(ScopeDelivery, id, KindActingPartLate)]; !ok || len(o.findings) != 1 {
t.Fatalf("past its bound: %v", o.findings)
}
w.ModuleWaitEnded(holder, body("mesh-sdk@v0.1.14", c.t))
if len(o.findings) != 0 {
t.Fatalf("wait-ended left %v", o.findings)
}
w, o, c = setup()
w.ModuleWaitHeld(holder, body("", c.t.Add(-2*time.Hour)))
if len(o.findings) != 0 || len(o.ttls) != 0 {
t.Fatalf("a toolchain wait was timed by the watchman: %v %v", o.ttls, o.findings)
}
}
modules/mesh-delivery-watchman/manifest_test.go
+2 -3
View File
@@ -103,7 +103,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
reads := slices.Clone(m.Reads)
slices.Sort(reads)
// The build seats' steps are no read (design 47 G §483, A-136): the watchman times an ask from the seats' step-progress.
if !reflect.DeepEqual(reads, []string{"mesh-delivery-builder.builds", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) {
if !reflect.DeepEqual(reads, []string{"mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) {
t.Errorf("reads = %v", reads)
}
// Design 47 G §368: the version is MAJOR.MINOR and nothing else.
@@ -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", "node-engine.wait-held", "node-engine.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", "mesh-delivery-builder.wait-held", "mesh-delivery-builder.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).
@@ -174,7 +174,6 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
"stuck": {watch.StuckBucket, m.Module + "_" + m.State[2].Name},
"have": {watch.HaveBucket, bucket(m.Reads, "node-engine.have")},
"health": {watch.NodeHealthBucket, bucket(m.Reads, "mesh-host.health")},
"builds": {watch.BuilderBuildsBucket, bucket(m.Reads, "mesh-delivery-builder.builds")},
"wants": {watch.AssignerWantsBucket, bucket(m.Consumes, "mesh-delivery-assigner.views.wants")},
"provisions": {watch.ProvisionsBucket, bucket(m.Reads, "mesh-provisioner.provisions")},
"modules": {watch.ModulesBucket, bucket(m.Consumes, "mesh-delivery-indexer.views.modules")},
modules/mesh-delivery-watchman/module.json
+2 -1
View File
@@ -16,7 +16,6 @@
}
],
"reads": [
"mesh-delivery-builder.builds",
"mesh-host.health",
"mesh-provisioner.provisions",
"node-engine.have"
@@ -67,6 +66,8 @@
"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",
"mesh-broker.created",