From 92b430792210f6c310c212db3af9808f1be2ddee Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 17:18:36 +0100 Subject: [PATCH] =?UTF-8?q?The=20watchman=20mirrors=20a=20want=20held=20by?= =?UTF-8?q?=20a=20rule,=20a=20provision=20asked=20and=20a=20build=20queued?= =?UTF-8?q?=20behind=20a=20package=20as=20waits=20with=20a=20bound,=20and?= =?UTF-8?q?=20raises=20a=20finding=20past=20it=20(ADR=200603=20=C2=A78)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cmd/mesh-delivery-watchman/main.go | 22 +++++++- .../internal/watch/findings_test.go | 32 +++++++++++ .../internal/watch/mirror.go | 56 +++++++++++++++++++ .../internal/watch/signals.go | 31 ++++++++++ .../internal/watch/watch.go | 3 +- .../mesh-delivery-watchman/manifest_test.go | 2 +- modules/mesh-delivery-watchman/module.json | 1 + 7 files changed, 144 insertions(+), 3 deletions(-) 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 1ded4bd7..8ec5142d 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -246,6 +246,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro w.ReadWhole(delivery.AssignerWantsBucket) } } + // 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, delivery.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, delivery.BuilderBuildsBucket, err, "a build waiting for a package is not watched until the watchman starts again") + buildSeen = nil + } else { + w.ReadWhole(delivery.BuilderBuildsBucket) + } + } // The build seats' steps: a step key `.` written is the ask taken (issue 1887 A8b). A seat whose build module // keeps no steps bucket (yet) is said and left; the asks of its stream are mirrored all the same. taken := map[string]jetstream.KeyWatcher{} @@ -304,7 +315,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The seat mesh-provisioner's provisions (ADR 0570 §7): a provision in waits is needs-operator. Read before the findings are // reconciled when the bucket is there; bound once it appears when it is not (mesh-provisioner comes in its own delivery). provisionsWatch := func(kv jetstream.KeyValue) (jetstream.KeyWatcher, error) { - return settle(ctx, kv, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) }) + pw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) }) + if err == nil { + w.ReadWhole(provisions.ProvisionsBucket) + } + return pw, err } var provWatch jetstream.KeyWatcher if kv, err := js.KeyValue(ctx, provisions.ProvisionsBucket); err == nil { @@ -395,6 +410,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro errs <- follow(ctx, wantSeen, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), 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()) }) + }() + } for lang, sw := range taken { lang, sw := lang, sw go func() { errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) }) }() diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 57f26142..b0a04e78 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -457,6 +457,38 @@ 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) { + w.SetStepBound(30 * time.Minute) + held, _ := delivery.EncodeWant(delivery.Want{FP: fpOld, Self: true, Newest: fpNew, Rule: delivery.RuleCanary, Words: "the canary ace does not run it yet"}) + released, _ := delivery.EncodeWant(delivery.Want{FP: fpNew, Self: true}) + asked, _ := provisions.Encode(provisions.Provision{Provision: "oidc", Consumer: provisions.End{Module: "photos", Node: "ace"}, + Provider: provisions.End{Module: "keycloak", Node: "novox"}, State: provisions.Asked, At: t0}) + provided, _ := provisions.Encode(provisions.Provision{Provision: "oidc", Consumer: provisions.End{Module: "photos", Node: "ace"}, + Provider: provisions.End{Module: "keycloak", Node: "novox"}, State: provisions.Provided, At: t0}) + bkey, _ := delivery.BuilderBuildKey("web", fpNew) + queued, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildQueued, Waits: "mesh-sdk@v0.1.14"}) + building, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.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{delivery.AssignerWantsBucket: "want.ace.web", provisions.ProvisionsBucket: "ace.photos.oidc", delivery.BuilderBuildsBucket: bkey} + if len(o.pulses) != 3 || len(o.findings) != 0 { + t.Fatalf("pulses %v findings %v", o.pulses, o.findings) + } + for bucket, key := range waits { + expire(w, c, pulseKeyOf(t, bucket, key), 31*time.Minute) + } + for bucket, key := range waits { + f := o.findings[FindingKey(ScopeDelivery, stuckKeyOf(t, bucket, key), KindStuck)] + if f.Kind != KindStuck || !f.Since.Equal(t0) || !strings.Contains(f.Summary, "bound 30m0s") { + t.Fatalf("%s: findings = %v", bucket, o.findings) + } + } + 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 { t.Run(tc.name, func(t *testing.T) { diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go index 5bf05e7f..b2f0e180 100644 --- a/modules/mesh-delivery-watchman/internal/watch/mirror.go +++ b/modules/mesh-delivery-watchman/internal/watch/mirror.go @@ -78,6 +78,7 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) { return case Delete: delete(w.wants, key) + delete(w.held, key) w.moved(delivery.AssignerWantsBucket, key) return } @@ -89,10 +90,27 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) { } if want.Removed { delete(w.wants, key) + delete(w.held, key) w.moved(delivery.AssignerWantsBucket, key) return } w.wants[key] = want.FP + if want.Newest != "" || want.Rule != 0 { + // A want held by a rule (ADR 0603 §8) waits on the rule, not on the node: its wait is the hold's, from the time the + // assigner wrote it, and the node running what the want names does not end it. + if at.IsZero() { + at = w.now() + } + w.held[key] = true + w.mirrorWithin(w.stepBound, delivery.AssignerWantsBucket, key, + describe("the hold on %s on %s at %s (rule %d) to end, bound %s: %s", module, node, shortFP(want.FP), want.Rule, w.stepBound, want.Words), + "held|"+want.FP+"|"+want.Newest, at, 0) + return + } + if w.held[key] { + delete(w.held, key) + w.stuckMoved(delivery.AssignerWantsBucket, key) // the hold ended: its finding is cleared + } if w.have(node, module) == want.FP { w.moved(delivery.AssignerWantsBucket, key) return @@ -104,6 +122,44 @@ 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 := delivery.ParseBuilderBuildKey(key) + if err != nil { + return + } + w.mu.Lock() + defer w.mu.Unlock() + if op == Expired { + return + } + if op == Delete { + w.moved(delivery.BuilderBuildsBucket, key) + return + } + w.stands(delivery.BuilderBuildsBucket, key) + b, err := delivery.DecodeBuilderBuild(value) + if err != nil { + w.say("the build %s cannot be read, and is left: %v", key, err) + return + } + if b.State != delivery.BuildQueued || b.Waits == "" { + w.moved(delivery.BuilderBuildsBucket, key) + return + } + if at.IsZero() { + at = w.now() + } + w.mirrorWithin(w.stepBound, delivery.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, 0) +} + // 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) { diff --git a/modules/mesh-delivery-watchman/internal/watch/signals.go b/modules/mesh-delivery-watchman/internal/watch/signals.go index ee44118a..9c2e8501 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -589,6 +589,7 @@ func (w *Watchman) Provision(key string, value []byte, op Op) { w.waiting[id][key] = fmt.Sprintf("%s from %s on %s: %s", p.Provision, p.Provider.Module, p.Provider.Node, p.Why) } } + w.askedWait(key, value, op) fkey := FindingKey(ScopeModule, id, KindNeedsOperator) if len(w.waiting[id]) == 0 { delete(w.waiting, id) @@ -604,6 +605,36 @@ func (w *Watchman) Provision(key string, value []byte, op Op) { Said: strings.Join(said, "; ")}) } +// askedWait is the wait of a provision its provider has not provided (ADR 0603 §8): a key asked is mirrored from the time the +// holder wrote it, bounded as a step is, and provided, failed, waiting for the operator (its own finding), retiring or deleted +// ends it. Called with the lock held. +func (w *Watchman) askedWait(key string, value []byte, op Op) { + bucket := provisions.ProvisionsBucket + if op == Expired { + return + } + w.stands(bucket, key) + if op == Delete { + w.moved(bucket, key) + return + } + p, err := provisions.Decode(value) + if err != nil || p.State != provisions.Asked { + w.moved(bucket, key) + return + } + since := p.At + if since.IsZero() { + since = w.now() + } + words := fmt.Sprintf("%s from %s on %s to be provided for %s on %s, bound %s", p.Provision, p.Provider.Module, p.Provider.Node, + p.Consumer.Module, p.Consumer.Node, w.stepBound) + if p.Why != "" { + words += ": " + p.Why + } + w.mirrorWithin(w.stepBound, bucket, key, words, "asked", since, 0) +} + func sortedKeys(m map[string]string) []string { out := make([]string, 0, len(m)) for k := range m { diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 3c769493..ef16ebbd 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -64,6 +64,7 @@ type Watchman struct { keptSince map[string]time.Time wants map[string]string // the fingerprint of each want that stands, by want key haves map[string]string // the fingerprint each node runs of a module, by have key + held map[string]bool // each want key held by a rule (ADR 0603 §8), mirrored as the hold's wait bound time.Duration // findings is every finding written, by key, as written; seededFindings those read at the start and not yet observed // again. @@ -103,7 +104,7 @@ type Watchman struct { // New is a watchman writing to out, reading the time from now. func New(out Out, now func() time.Time) *Watchman { return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{}, - pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, bound: delivery.PulseMoveBound, stepBound: StepBound, + pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, held: map[string]bool{}, bound: delivery.PulseMoveBound, stepBound: StepBound, findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, away: map[string]bool{}, haveState: map[string]delivery.Have{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{}, control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{}, diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 7f146ca1..a577361e 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -91,7 +91,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 reads := slices.Clone(m.Reads) slices.Sort(reads) if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", - "mesh-delivery-assigner.wants", "mesh-delivery-indexer.modules", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) { + "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-indexer.modules", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) { t.Errorf("reads = %v", reads) } // 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 f405cf75..4c85f5d7 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -20,6 +20,7 @@ "mesh-build-image.steps", "mesh-build-node.steps", "mesh-delivery-assigner.wants", + "mesh-delivery-builder.builds", "mesh-delivery-indexer.modules", "mesh-host.health", "mesh-provisioner.provisions",