The watchman mirrors a want held by a rule, a provision asked and a build queued behind a package as waits with a bound, and raises a finding past it (ADR 0603 §8)

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 `<node>.<ask id>` 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) }) }()

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) {

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) {

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 {

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{},

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).

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",