Merge pull request 'Tests no longer race the wall clock; the checker invokes compose-candidate again' (#143) from fix/timing-tests into main

modules/mesh-delivery-builder/buildrun/asks_test.go

@@ -53,38 +53,23 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
 	if len(g.Asks()) != 0 {
 		t.Fatal("a wait put an ask")
 	}
-	// Issue 1930: the bound is on a step that stands still, not on its age. A step that enters a new phase inside the bound is
+	// Issue 1930: the bound is on a step that stands still, not on its age. A step past one bound from its start that entered a
-	// waited for until it ends; one that stays in a phase the bound fails at once, naming the phase; one that never stops moving
+	// new phase inside the bound is waited for until it ends; one that stood a whole bound in a phase fails at once, naming
-	// ends at three bounds, said as such.
+	// the phase; one three bounds from its start ends, said as kept moving. The steps' own times stand for the clock, so the
-	k = Asker{Publisher: g, Waiter: g, Bound: 60 * time.Millisecond}
+	// bound is an hour no loaded machine reaches and no case waits on a pause (the check of 2026-11-02 22:35 tripped one
-	started := time.Now()
+	// of 60 ms).
-	go func() {
+	k = Asker{Publisher: g, Waiter: g, Bound: time.Hour}
-		for i := range 8 {
+	now := time.Now()
-			g.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "ace", State: delivery.StepRunning, Started: started, Phase: "p" + string(rune('a'+i)), PhaseAt: time.Now()})
+	g.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "ace", State: delivery.StepRunning, Started: now.Add(-90 * time.Minute), Phase: "pb", PhaseAt: now})
-			time.Sleep(15 * time.Millisecond)
+	g.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "ace", State: delivery.StepPassed})
-		}
-		g.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "ace", State: delivery.StepPassed})
-	}()
 	if step, err := k.Wait(context.Background(), goAsk("mvl0frv18d1980")); err != nil || step.State != delivery.StepPassed {
 		t.Fatalf("a step that kept moving past its bound: %+v, %v", step, err)
 	}
-	g.Write(delivery.Step{ID: "mvl0frv18d1981", Node: "ace", State: delivery.StepRunning, Started: time.Now(), Phase: "integration: tests", PhaseAt: time.Now()})
+	g.Write(delivery.Step{ID: "mvl0frv18d1981", Node: "ace", State: delivery.StepRunning, Started: now.Add(-time.Hour), Phase: "integration: tests", PhaseAt: now.Add(-time.Hour)})
 	if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1981")); err == nil || !strings.Contains(err.Error(), `stood still in "integration: tests"`) {
 		t.Fatalf("a step that stood still in a phase: %v", err)
 	}
-	loop := make(chan struct{})
+	g.Write(delivery.Step{ID: "mvl0frv18d1982", Node: "ace", State: delivery.StepRunning, Started: now.Add(-3 * time.Hour), Phase: "qz", PhaseAt: now})
-	defer close(loop)
-	started = time.Now()
-	go func() {
-		for i := 0; ; i++ {
-			select {
-			case <-loop:
-				return
-			case <-time.After(20 * time.Millisecond):
-				g.Write(delivery.Step{ID: "mvl0frv18d1982", Node: "ace", State: delivery.StepRunning, Started: started, Phase: "q" + string(rune('a'+i%26)), PhaseAt: time.Now()})
-			}
-		}
-	}()
 	if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1982")); err == nil || !strings.Contains(err.Error(), "kept moving") {
 		t.Fatalf("a step that never stops moving: %v", err)
 	}
@@ -184,22 +169,21 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
 }
 
 // Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the
-// new run's start, so its pass is answered after the first run's bound would have ended the wait.
+// new run's start, not the first run's. The steps' own times stand for the clock: the first run started now, the second a
+// whole bound ago, so the wait ends at once on the second run's bound where the first run's would wait an hour; no case
+// waits on a pause a loaded machine can outlast (the check of 2026-11-02 22:35 tripped one of 60 ms). The context's
+// bound is reached only by a wait that kept the first run's start.
 func TestARedeliveryToTheSameNodeRestartsTheBound(t *testing.T) {
 	f := &FakeAsks{}
-	k := Asker{Publisher: f, Waiter: f, Bound: 60 * time.Millisecond}
+	k := Asker{Publisher: f, Waiter: f, Bound: time.Hour}
 	const id = "mvl0frv18d1962"
-	first := time.Now()
+	now := time.Now()
-	f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: first})
+	f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: now})
-	go func() {
+	f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: now.Add(-time.Hour)})
-		time.Sleep(15 * time.Millisecond)
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-		f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: time.Now()})
+	defer cancel()
-		time.Sleep(50 * time.Millisecond) // past the first run's bound, inside the second's
+	if _, err := k.Wait(ctx, goAsk(id)); err == nil || !strings.Contains(err.Error(), "on ace did not end within 1h0m0s of its start") {
-		f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepPassed, Started: time.Now()})
+		t.Fatalf("a redelivered run a bound old: %v, want its own bound to end the wait", err)
-	}()
-	step, err := k.Wait(context.Background(), goAsk(id))
-	if err != nil || step.State != delivery.StepPassed {
-		t.Fatalf("step %+v, %v, want the redelivered run's pass", step, err)
 	}
 }
 

modules/mesh-delivery-builder/internal/build/askagain.go

@@ -318,7 +318,7 @@ func (b *Builder) answer(ctx context.Context, l *lane, ask delivery.StepAsk, ask
 			// Not an asking: a node that sits out would use them all while it idles and pulls the ask again at once.
 			// Bounded by time from the first decline, then failed loudly naming the nodes and their reasons (as the
 			// checker's decline loop is, issue 890).
-			now := time.Now()
+			now := b.now()
 			if firstDecline.IsZero() {
 				firstDecline = now
 			}

modules/mesh-delivery-builder/internal/build/run.go

@@ -75,7 +75,8 @@ type Builder struct {
 	Said func(AskedAgain)
 	// Log is told each build's start and end, the entries of the builder's log (novox/hq ADR 0571 §1). Nil tells nobody.
 	Log func(delivery.BuildLogEntry)
-	// Now is the clock of the log; nil is time.Now. Node is the node the builder runs on, which asked the build.
+	// Now is the builder's clock: of the log, and of the decline bound (DeclineBound runs from the first decline); nil is
+	// time.Now. Node is the node the builder runs on, which asked the build.
 	Now  func() time.Time
 	Node string
 

modules/mesh-delivery-builder/internal/build/run_test.go

@@ -318,7 +318,7 @@ func TestATrunkFingerprintIsOneAskOnTheDeliveryStream(t *testing.T) {
 	}
 	r.b.Module(context.Background(), "m", data)
 	var asks []delivery.StepAsk
-	for deadline := time.Now().Add(2 * time.Second); len(asks) == 0; time.Sleep(time.Millisecond) {
+	for deadline := time.Now().Add(10 * time.Second); len(asks) == 0; time.Sleep(time.Millisecond) {
 		if time.Now().After(deadline) {
 			t.Fatal("no ask was put")
 		}
@@ -705,7 +705,7 @@ func (g *goneAsks) asked() int {
 // waitForAsks waits until n asks were put.
 func waitForAsks(t *testing.T, r *rig, n int) []delivery.StepAsk {
 	t.Helper()
-	for deadline := time.Now().Add(2 * time.Second); len(r.asks.Asks()) < n; time.Sleep(time.Millisecond) {
+	for deadline := time.Now().Add(10 * time.Second); len(r.asks.Asks()) < n; time.Sleep(time.Millisecond) {
 		if time.Now().After(deadline) {
 			t.Fatalf("%d asks put, want %d", len(r.asks.Asks()), n)
 		}
@@ -725,7 +725,13 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
 		r.b.Gone, r.b.CheckAfter = gone, 5*time.Millisecond
 		r.b.Module(context.Background(), "m", moduleValue(t, "modules/m"))
 		waitForAsks(t, r, 1)
-		time.Sleep(100 * time.Millisecond)
+		// Five check bounds passed (five reads of whether it can be answered), so the ask waited past the step bound: an event
+		// to wait on, where a pause of 100 ms said nothing on a loaded machine.
+		for deadline := time.Now().Add(10 * time.Second); gone.asked() < 5; time.Sleep(time.Millisecond) {
+			if time.Now().After(deadline) {
+				t.Fatalf("whether the ask can still be answered was read %d times within 10 s, want 5", gone.asked())
+			}
+		}
 		if v := valueOf(t, r.st, r.key); v.State != delivery.BuildBuilding {
 			t.Fatalf("the key = %+v while its ask waits, want building", v)
 		}
@@ -741,11 +747,16 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
 	t.Run("declined by a node off the agreed toolchain: put again, no asking used, the bound ends it loudly", func(t *testing.T) { // ADR 0584 clause 2
 		r := newRig(t, goBundle)
 		r.asks.Answer = nil
-		r.b.DeclinePause, r.b.DeclineBound = -1, 150*time.Millisecond
+		// The builder's clock steps a second at each read, so the bound of a minute ends the declines after many asks however
+		// slow the machine (a bound of 150 ms of wall clock let a loaded one put fewer than DefaultAskings).
+		var clock atomic.Int64
+		start := time.Now()
+		r.b.Now = func() time.Time { return start.Add(time.Duration(clock.Add(1)) * time.Second) }
+		r.b.DeclinePause, r.b.DeclineBound = -1, time.Minute
 		r.b.Module(context.Background(), "m", moduleValue(t, "modules/m"))
 		declined, _ := json.Marshal(delivery.StepDeclined{Declined: true, Lacks: []string{"the agreed toolchain go1.27.1 (recipe 1)"}})
 		waitForAsks(t, r, 1)
-		for ended, deadline := 0, time.Now().Add(5*time.Second); valueOf(t, r.st, r.key).State == delivery.BuildBuilding && time.Now().Before(deadline); time.Sleep(5 * time.Millisecond) {
+		for ended, deadline := 0, time.Now().Add(20*time.Second); valueOf(t, r.st, r.key).State == delivery.BuildBuilding && time.Now().Before(deadline); time.Sleep(5 * time.Millisecond) {
 			if asks := r.asks.Asks(); len(asks) > ended { // each ask put again is declined in its turn
 				r.asks.End(delivery.Step{ID: asks[ended].ID, Node: "ace", State: delivery.StepRefused, Reason: "this node builds with go1.26.0", Result: declined})
 				ended++
@@ -926,7 +937,7 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
 			}
 			if c.byHand {
 				waitForAsks(t, r, 1)
-				for deadline := time.Now().Add(2 * time.Second); len(r.b.AskAgain(r.key, "stuck since 19:06")) == 0; time.Sleep(time.Millisecond) {
+				for deadline := time.Now().Add(10 * time.Second); len(r.b.AskAgain(r.key, "stuck since 19:06")) == 0; time.Sleep(time.Millisecond) {
 					if time.Now().After(deadline) {
 						t.Fatal("no lane of the key waits")
 					}
@@ -934,7 +945,7 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
 			}
 			if c.deaf {
 				r.asks.End(r.answer(waitForAsks(t, r, 1)[0]))
-				for deadline := time.Now().Add(2 * time.Second); heard.Waiting() == 0; time.Sleep(time.Millisecond) {
+				for deadline := time.Now().Add(10 * time.Second); heard.Waiting() == 0; time.Sleep(time.Millisecond) {
 					if time.Now().After(deadline) {
 						t.Fatal("no ask is waited on")
 					}

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go

@@ -7,7 +7,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"os"
 	"path/filepath"
 	"slices"
@@ -261,6 +260,26 @@ type bed struct {
 	srvURL   string
 	announce []string
 	mu       sync.Mutex
+	logs     saidLog
+}
+
+// saidLog is what the program said, safe for its goroutines and the test's.
+type saidLog struct {
+	mu sync.Mutex
+	b  strings.Builder
+}
+
+func (l *saidLog) Write(p []byte) (int, error) { l.mu.Lock(); defer l.mu.Unlock(); return l.b.Write(p) }
+func (l *saidLog) String() string              { l.mu.Lock(); defer l.mu.Unlock(); return l.b.String() }
+
+// said waits until the program said words, within 20 s: an event to wait on, not a pause a loaded machine can outlast.
+func (b *bed) said(t *testing.T, words string) {
+	t.Helper()
+	for deadline := time.Now().Add(20 * time.Second); !strings.Contains(b.logs.String(), words); time.Sleep(10 * time.Millisecond) {
+		if time.Now().After(deadline) {
+			t.Fatalf("the checker did not say %q within 20 s:\n%s", words, b.logs.String())
+		}
+	}
 }
 
 // engine places each node and serves compose-candidate for the first one, as its node-engine does on the bus (answered on
@@ -512,7 +531,7 @@ func (b *bed) start(t *testing.T, env map[string]string) {
 	}
 	runCtx, stop := context.WithCancel(b.ctx)
 	done := make(chan error, 1)
-	go func() { done <- run(runCtx, func(k string) string { return e[k] }, io.Discard) }()
+	go func() { done <- run(runCtx, func(k string) string { return e[k] }, &b.logs) }()
 	t.Cleanup(func() { stop(); <-done })
 }
 
@@ -639,12 +658,13 @@ func TestACheckOrACarryOfEqualInputsRunsNothing(t *testing.T) { // 1287/AC5, 128
 		t.Fatal(err)
 	}
 	tk, _ := delivery.TurnsKey(repo)
-	if _, err := b.buckets[delivery.TurnsBucket].Put(b.ctx, tk, turn); err != nil {
+	turnRev, err := b.buckets[delivery.TurnsBucket].Put(b.ctx, tk, turn)
+	if err != nil {
 		t.Fatal(err)
 	}
-	time.Sleep(300 * time.Millisecond) // the checker hears the turn
+	b.said(t, fmt.Sprintf("merge turn %s changed (revision %d)", tk, turnRev)) // the checker heard the turn
 	third := b.candidate(t, "feat/z", pull(repo, 9, "3333333333333333333333333333333333333333"))
-	time.Sleep(time.Second)
+	b.said(t, third+": not claimed: the approved merge of fix/other") // it held the check off
 	if _, err := b.buckets[delivery.CheckerVerdictsBucket].Get(b.ctx, third); !errors.Is(err, jetstream.ErrKeyNotFound) {
 		t.Fatalf("verdicts.%s exists (%v) while a merge turn holds its repository", third, err)
 	}

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go

@@ -303,6 +303,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		go func() {
 			errs <- follow(ctx, tw, func(e jetstream.KeyValueEntry) {
 				if turns.Put(e.Key(), e.Value(), !opOf(e)) {
+					// Said, so a wait on a held check can be read (and a test waits on it rather than on a pause).
+					logger.Printf("the gate's merge turn %s changed (revision %d): the checks it holds are held or claimed from now on", e.Key(), e.Revision())
 					ch.Reconsider(ctx) // a turn went or changed: a check it held is claimed now
 				}
 			})

modules/mesh-delivery-checker/internal/check/adr548_test.go

@@ -33,10 +33,11 @@ func turnKey(t testing.TB, repo string) string {
 	return k
 }
 
-// waitFor polls a condition the checker's goroutines make true, failing after two seconds.
+// waitFor polls a condition the checker's goroutines make true, failing after ten seconds: a bound a loaded machine does
+// not reach, waited out only by a failing test.
 func waitFor(t testing.TB, what string, ok func() bool) {
 	t.Helper()
-	for deadline := time.Now().Add(2 * time.Second); !ok(); time.Sleep(time.Millisecond) {
+	for deadline := time.Now().Add(10 * time.Second); !ok(); time.Sleep(time.Millisecond) {
 		if time.Now().After(deadline) {
 			t.Fatalf("%s did not happen", what)
 		}

modules/mesh-delivery-checker/internal/check/examiner_test.go

@@ -957,7 +957,7 @@ func TestACheckAskWaitingInItsStreamIsNotBounded(t *testing.T) {
 	b.ex.Asks = buildrun.Asker{Publisher: b.asks, Waiter: b.asks, Bound: 20 * time.Millisecond}
 	done := make(chan Outcome, 1)
 	go func() { done <- b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) }()
-	for deadline := time.Now().Add(2 * time.Second); len(b.asks.Asks()) == 0; time.Sleep(time.Millisecond) {
+	for deadline := time.Now().Add(10 * time.Second); len(b.asks.Asks()) == 0; time.Sleep(time.Millisecond) {
 		if time.Now().After(deadline) {
 			t.Fatal("no ask was put")
 		}

modules/mesh-delivery-checker/internal/check/lostask_test.go

@@ -82,10 +82,13 @@ func TestAnAskGivenUpOrNeverTakenEndsItsTryAsErrorAndItsHeadIsAskedAnew(t *testi
 		name  string
 		lose  int
 		never bool
-		want  string
+		// notTaken is the try's bound on an ask no node takes: short only where it is what ends the try, an hour where the
+		// dead letter must (a loaded machine let 200 ms pass before the dead letter was heard).
+		notTaken time.Duration
+		want     string
 	}{
-		{"given up by the build seat", 1, false, buildrun.LostWords + " after 5 deliveries at 2026-10-26T12:26:40Z"},
+		{"given up by the build seat", 1, false, time.Hour, buildrun.LostWords + " after 5 deliveries at 2026-10-26T12:26:40Z"},
-		{"taken by no build node within the bound", 0, true, buildrun.NotTakenWords + " within 200ms"},
+		{"taken by no build node within the bound", 0, true, 200 * time.Millisecond, buildrun.NotTakenWords + " within 200ms"},
 	} {
 		t.Run(row.name, func(t *testing.T) {
 			kv := newMemKV()
@@ -97,7 +100,7 @@ func TestAnAskGivenUpOrNeverTakenEndsItsTryAsErrorAndItsHeadIsAskedAnew(t *testi
 			b := newBench(t)
 			l := losing(b, row.lose)
 			l.never = row.never
-			tr := &Tryer{Tries: kv, Judge: b.ex, NotTaken: 200 * time.Millisecond}
+			tr := &Tryer{Tries: kv, Judge: b.ex, NotTaken: row.notTaken}
 			raw, _, _, _ := kv.Get(ctx, id)
 			tr.Put(ctx, id, raw)
 			tr.Wait()
@@ -119,7 +122,7 @@ func TestAnAskGivenUpOrNeverTakenEndsItsTryAsErrorAndItsHeadIsAskedAnew(t *testi
 		tr := &Tryer{Tries: kv, Judge: b.ex} // no bound: only the hand ends it
 		raw, _, _, _ := kv.Get(ctx, id)
 		tr.Put(ctx, id, raw)
-		deadline := time.Now().Add(2 * time.Second)
+		deadline := time.Now().Add(10 * time.Second)
 		for readTry(t, kv, id).State != TryRunning && time.Now().Before(deadline) {
 			time.Sleep(5 * time.Millisecond)
 		}
@@ -289,7 +292,7 @@ func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) {
 			key := keyOf(t, "feat/x", cand)
 			r.ch.Candidate(ctx, "feat%2Fx", cand)
 			if row.deaf > 0 {
-				for deadline := time.Now().Add(2 * time.Second); heard.Waiting() == 0 && time.Now().Before(deadline); {
+				for deadline := time.Now().Add(10 * time.Second); heard.Waiting() == 0 && time.Now().Before(deadline); {
 					time.Sleep(2 * time.Millisecond)
 				}
 				if v := r.verdict(t, key); v.State != delivery.VerdictRunning {

modules/mesh-delivery-checker/manifest_test.go

@@ -111,13 +111,11 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
 	}
 	// 1887/A5b: the checker reads nothing of the controller's plans-building (ADR 0483 superseded by ADR 0525 §7), calls no
 	// build module's tool and holds no capacity: it puts check asks on the Go and Node build seats (ADR 0525 §1, §4).
-	// The forge's reads for a mesh-sdk pin (ADR 0484 clause 9). Not yet the node seat node-engine's compose-candidate
+	// The forge's reads for a mesh-sdk pin (ADR 0484 clause 9), and of the node seat node-engine its one verb
-	// (ADR 0593 §7): the indexer that runs while this delivers was built before the verb existed and refuses a manifest
+	// compose-candidate, asked of every placed node (ADR 0593 §7), never the seat's other verbs.
-	// invoking it, which held the grants part's whole list (issue 1126, 2026-11-02); a follow-up adds it back once the
+	if want := []string{"gitea.gitea_commits", "gitea.gitea_compare", "gitea.gitea_get_file", "gitea.gitea_list_branches",
-	// new indexer runs.
+		"seat:node-engine.compose-candidate"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
-	if want := []string{"gitea.gitea_commits", "gitea.gitea_compare", "gitea.gitea_get_file",
+		t.Errorf("invokes = %v, want the forge's reads and compose-candidate", m.Invokes)
-		"gitea.gitea_list_branches"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
-		t.Errorf("invokes = %v, want the forge's reads", m.Invokes)
 	}
 	if want := []string{"node-build-go", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
 		t.Errorf("uses = %v, want the Go and Node build seats", m.Uses)

modules/mesh-delivery-checker/module.json

@@ -65,7 +65,8 @@
     "gitea.gitea_commits",
     "gitea.gitea_compare",
     "gitea.gitea_get_file",
-    "gitea.gitea_list_branches"
+    "gitea.gitea_list_branches",
+    "seat:node-engine.compose-candidate"
   ],
 
   "own-secrets": {

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go

@@ -308,14 +308,14 @@ func TestBridgeWritesPullMessages(t *testing.T) { // 1235/AC2
 		t.Fatal(err)
 	}
 	var info *jetstream.StreamInfo
-	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 		if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 2 {
 			break
 		}
 	}
 	// The open message the bridge writes at its start (ADR 0443 §5), and the event's message.
 	if info.State.Msgs != 2 {
-		t.Fatalf("FORGE holds %d messages within 5 s, want 2", info.State.Msgs)
+		t.Fatalf("FORGE holds %d messages within 20 s, want 2", info.State.Msgs)
 	}
 	raw, err := forge.GetLastMsgForSubject(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull")
 	if err != nil {
@@ -336,7 +336,7 @@ func TestBridgeWritesPullMessages(t *testing.T) { // 1235/AC2
 	if err != nil {
 		t.Fatal(err)
 	}
-	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 		ci, err := cons.Info(ctx)
 		if err == nil && ci.NumAckPending == 0 && ci.NumPending == 0 && ci.Delivered.Consumer >= 1 {
 			return
@@ -403,14 +403,14 @@ func TestBridgeWritesMergedMessage(t *testing.T) { // 1236/AC4, 1448/AC5
 		t.Fatal(err)
 	}
 	var info *jetstream.StreamInfo
-	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 		if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 2 {
 			break
 		}
 	}
 	// The open message the bridge writes at its start (ADR 0443 §5), and the event's message.
 	if info.State.Msgs != 2 {
-		t.Fatalf("FORGE holds %d messages within 5 s, want 2", info.State.Msgs)
+		t.Fatalf("FORGE holds %d messages within 20 s, want 2", info.State.Msgs)
 	}
 	raw, err := forge.GetLastMsgForSubject(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.merged")
 	if err != nil {
@@ -443,7 +443,7 @@ func TestBridgeWritesMergedMessage(t *testing.T) { // 1236/AC4, 1448/AC5
 	if err != nil {
 		t.Fatal(err)
 	}
-	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 		ci, err := cons.Info(ctx)
 		if err == nil && ci.NumAckPending == 0 && ci.NumPending == 0 && ci.Delivered.Consumer >= 1 {
 			return
@@ -465,13 +465,13 @@ func TestBridgeSaysTheOpenPullsAtStart(t *testing.T) { // 1237/AC1
 		t.Fatal(err)
 	}
 	var info *jetstream.StreamInfo
-	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 		if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 3 {
 			break
 		}
 	}
 	if info.State.Msgs != 3 {
-		t.Fatalf("FORGE holds %d messages within 5 s, want 3", info.State.Msgs)
+		t.Fatalf("FORGE holds %d messages within 20 s, want 3", info.State.Msgs)
 	}
 	for _, repo := range []string{"mesh-delivery", "mesh-catalog"} {
 		raw, err := forge.GetLastMsgForSubject(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox."+repo+".pull")

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go

@@ -65,7 +65,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10
 
 	hash, _ := delivery.HeadsHash([]delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: h1}, {Repository: "novox/mesh-catalog", Number: 300, Head: h2}})
 	var c delivery.Candidate
-	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 		e, err := candidates.Get(ctx, "feat/x")
 		if err != nil {
 			continue
@@ -75,7 +75,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10
 		}
 	}
 	if c.Group != "feat/x" || c.Base != 0 || c.HeadsHash != hash || len(c.Pulls) != 2 {
-		t.Fatalf("candidate = %+v within 5 s", c)
+		t.Fatalf("candidate = %+v within 20 s", c)
 	}
 	for _, p := range c.Pulls {
 		want := map[string]string{"novox/mesh-delivery": "a", "novox/mesh-catalog": "c"}[p.Repository]
@@ -195,7 +195,7 @@ func (l buildLog) end(t *testing.T, repo, module, path string) {
 }
 
 // TestIndexerWritesModulesAtAMerge: a module's record and its newest pointer are in mesh-controller_builds; a merged
-// message on FORGE makes modules.<m> within 5 s, the fingerprint of the merge's tree and the recorded inputs.
+// message on FORGE makes modules.<m> within 20 s, the fingerprint of the merge's tree and the recorded inputs.
 func TestIndexerWritesModulesAtAMerge(t *testing.T) { // 1239/AC9; I1, I3: the modules are the merged message's, no build record names them (issue 1851)
 	// The bucket holds only build-agent's record, which no fingerprint reads since R4g (ADR 0530 §1).
 	js, ctx, _, modules := startIndexer(t, func(ctx context.Context, builds buildLog) {
@@ -212,7 +212,7 @@ func TestIndexerWritesModulesAtAMerge(t *testing.T) { // 1239/AC9; I1, I3: the m
 		t.Fatal(err)
 	}
 	var m delivery.Module
-	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 		if e, err := modules.Get(ctx, "a"); err == nil {
 			if m, err = delivery.DecodeModule(e.Value()); err == nil {
 				break
@@ -220,12 +220,12 @@ func TestIndexerWritesModulesAtAMerge(t *testing.T) { // 1239/AC9; I1, I3: the m
 		}
 	}
 	if m.Repository != "novox/mesh-delivery" || m.Commit != merge || m.Path != "modules/a" || m.Manifest == "" || m.FP != "" || m.None != "" { // 1606/AC2
-		t.Errorf("module = %+v within 5 s, want path modules/a, its manifest, and no value of today's form", m)
+		t.Errorf("module = %+v within 20 s, want path modules/a, its manifest, and no value of today's form", m)
 	}
 }
 
 // TestIndexerKeepsManifestsAtAMerge: a merged message of mesh-catalog carrying the real manifest of the module avahi and the
-// hash of its go.sum (the bridge's Manifests and Locks, issue 1448) makes, within 5 s, modules.avahi with its manifest digest,
+// hash of its go.sum (the bridge's Manifests and Locks, issue 1448) makes, within 20 s, modules.avahi with its manifest digest,
 // its inputs and the reason it has no fingerprint of the new form, and manifests.<digest> with the manifest byte for byte.
 //
 // And it judges them (novox/hq ADR 0578, issue 1585): with three manifests more, one valid, one refused and one with a
@@ -272,7 +272,7 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4, 15
 		t.Fatal(err)
 	}
 	var m delivery.Module
-	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 		if e, err := modules.Get(ctx, "avahi"); err == nil {
 			if m, err = delivery.DecodeModule(e.Value()); err == nil {
 				break
@@ -282,7 +282,7 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4, 15
 	digest := delivery.ManifestDigest(manifest)
 	if m.Manifest != digest || m.None2 != "toolchain not yet reported" || m.FP2 != "" || m.Inputs2 == nil || m.Inputs2.Tree != "newtree" ||
 		m.Inputs2.Locks["tools"] != goSum || m.Path != "modules/avahi" { // 1606/AC2
-		t.Errorf("module = %+v within 5 s, want its path modules/avahi, its manifest digest %s, its inputs and none2 'toolchain not yet reported'", m, digest)
+		t.Errorf("module = %+v within 20 s, want its path modules/avahi, its manifest digest %s, its inputs and none2 'toolchain not yet reported'", m, digest)
 	}
 	kv, err := js.KeyValue(ctx, delivery.ManifestsBucket)
 	if err != nil {
@@ -345,7 +345,7 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4, 15
 // TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain: the bucket mesh-build-go_toolchains exists (the state
 // toolchains of the build module, delivery.ToolchainsBucket) and holds no key; a merged message of mesh-catalog carrying
 // avahi's real manifest makes modules.avahi with none2 'toolchain not yet reported'; the bucket made after the indexer started, a put of the key n1.go and the build module's event toolchain.reported (the build
-// module's own layout, delivery.ToolchainKey and EncodeToolchain) makes, within 5 s and with no merge, its fingerprint:
+// module's own layout, delivery.ToolchainKey and EncodeToolchain) makes, within 20 s and with no merge, its fingerprint:
 // no builder input is waited for (ADR 0530 §1; issue 1887 I1).
 func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *testing.T) { // 1444/AC10
 	manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json")
@@ -369,7 +369,7 @@ func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *tes
 	}
 	none2 := func(want string) string {
 		var got string
-		for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 			if e, err := modules.Get(ctx, "avahi"); err == nil {
 				if m, err := delivery.DecodeModule(e.Value()); err == nil {
 					if got = m.None2; got == want {
@@ -381,7 +381,7 @@ func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *tes
 		return got
 	}
 	if got := none2(delivery.ToolchainNoReport); got != delivery.ToolchainNoReport {
-		t.Fatalf("none2 = %q within 5 s, want %q", got, delivery.ToolchainNoReport)
+		t.Fatalf("none2 = %q within 20 s, want %q", got, delivery.ToolchainNoReport)
 	}
 	// The bucket is made only now, as when the build module is assigned after the indexer started (review finding 9): the
 	// indexer finds it on its own.
@@ -405,7 +405,7 @@ func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *tes
 		t.Fatal(err)
 	}
 	if got := none2(""); got != "" {
-		t.Errorf("none2 = %q within 5 s of the toolchain key, want a fingerprint and no reason", got)
+		t.Errorf("none2 = %q within 20 s of the toolchain key, want a fingerprint and no reason", got)
 	}
 }
 
@@ -413,7 +413,7 @@ func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *tes
 // and its durable consumer (made by the controller from the checker's declaration and the indexer's consumes) and no bucket
 // mesh-delivery-checker_previews; a pull request of mesh-catalog touching avahi makes the candidate with the reason
 // 'no merge preview is computed'; the checker then makes its bucket, writes the preview of that head onto the trunk head
-// the source names, and says so on its stream; the candidate's avahi gets, within 5 s and with no restart, the reason
+// the source names, and says so on its stream; the candidate's avahi gets, within 20 s and with no restart, the reason
 // computed from the preview (the toolchain is not reported here, so 'toolchain not yet reported').
 func TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBucket(t *testing.T) { // 1445/AC4
 	manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json")
@@ -453,7 +453,7 @@ func TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBuc
 	}
 	none2 := func(want string) string {
 		var got string
-		for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 			if e, err := candidates.Get(ctx, "feat/x"); err == nil {
 				if c, err := delivery.DecodeCandidate(e.Value()); err == nil && len(c.Pulls) == 1 {
 					got = c.Pulls[0].Fps["avahi"].None2
@@ -466,7 +466,7 @@ func TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBuc
 		return got
 	}
 	if got := none2("no merge preview is computed"); got != "no merge preview is computed" {
-		t.Fatalf("none2 = %q within 5 s of the pull request, want 'no merge preview is computed'", got)
+		t.Fatalf("none2 = %q within 20 s of the pull request, want 'no merge preview is computed'", got)
 	}
 	// The checker's bucket is made only now, as when the checker is assigned after the indexer started.
 	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.PreviewsBucket, History: delivery.PreviewsHistory})
@@ -487,13 +487,13 @@ func TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBuc
 		t.Fatal(err)
 	}
 	if got := none2(delivery.ToolchainNoReport); got != delivery.ToolchainNoReport {
-		t.Errorf("none2 = %q within 5 s of the preview, want %q (computed from the preview)", got, delivery.ToolchainNoReport)
+		t.Errorf("none2 = %q within 20 s of the preview, want %q (computed from the preview)", got, delivery.ToolchainNoReport)
 	}
 }
 
 // TestIndexerJudgesAnUnreadableVerdictAgainAtStart: the modules bucket holds avahi as an older SDK wrote it, with the reason
 // 'the manifest cannot be read', the inputs the merge carried (the tree, the lock hash by file name) and the digest of its
-// manifest, which manifests.<digest> holds and the running SDK reads; the go toolchain is reported. Within 5 s of the start
+// manifest, which manifests.<digest> holds and the running SDK reads; the go toolchain is reported. Within 20 s of the start
 // and with no repository merge the value carries a fingerprint and no reason (issue 1923).
 func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC3
 	manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json")
@@ -548,7 +548,7 @@ func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC
 		}
 	})
 	var got delivery.Module
-	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 		if e, err := modules.Get(ctx, "avahi"); err == nil {
 			if m, err := delivery.DecodeModule(e.Value()); err == nil {
 				if got = m; got.FP2 != "" {
@@ -558,7 +558,7 @@ func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC
 		}
 	}
 	if got.FP2 == "" || got.None2 != "" {
-		t.Errorf("avahi = fp2 %q none2 %q within 5 s of the start, want a fingerprint and no reason", got.FP2, got.None2)
+		t.Errorf("avahi = fp2 %q none2 %q within 20 s of the start, want a fingerprint and no reason", got.FP2, got.None2)
 	}
 }
 
@@ -626,7 +626,7 @@ func TestIndexerCountsOnlyThePlacedBuildNodesToolchains(t *testing.T) {
 	}
 	read := func(ok func(delivery.Module) bool) delivery.Module {
 		var got delivery.Module
-		for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 			if e, err := modules.Get(ctx, "avahi"); err == nil {
 				if m, err := delivery.DecodeModule(e.Value()); err == nil {
 					if got = m; ok(m) {
@@ -640,7 +640,7 @@ func TestIndexerCountsOnlyThePlacedBuildNodesToolchains(t *testing.T) {
 	agreed := func(m delivery.Module) bool { return m.FP2 != "" && m.Toolchains["go"].Version == "go1.27.3" }
 	held := read(agreed)
 	if !agreed(held) {
-		t.Fatalf("within 5 s avahi has fp2 %q, none2 %q, toolchains %v: want a fingerprint of n1's go1.27.3 alone", held.FP2, held.None2, held.Toolchains)
+		t.Fatalf("within 20 s avahi has fp2 %q, none2 %q, toolchains %v: want a fingerprint of n1's go1.27.3 alone", held.FP2, held.None2, held.Toolchains)
 	}
 	place(ctx, t, "n2", "mesh-build-go")
 	key, _ := delivery.ToolchainKey("n2", "go")
@@ -670,10 +670,10 @@ func TestIndexerCountsOnlyThePlacedBuildNodesToolchains(t *testing.T) {
 		return m.FP2 == "" && strings.HasPrefix(m.None2, "builders disagree: go1.27.2 recipe 1 on n2")
 	}
 	if m := read(disagree); !disagree(m) {
-		t.Fatalf("within 5 s of a merge after n2's placement avahi has fp2 %q, none2 %q: want the builders to disagree", m.FP2, m.None2)
+		t.Fatalf("within 20 s of a merge after n2's placement avahi has fp2 %q, none2 %q: want the builders to disagree", m.FP2, m.None2)
 	}
 	place(ctx, t, "n2")
 	if m := read(agreed); !agreed(m) {
-		t.Fatalf("within 5 s of n2's placement taken away avahi has fp2 %q, none2 %q", m.FP2, m.None2)
+		t.Fatalf("within 20 s of n2's placement taken away avahi has fp2 %q, none2 %q", m.FP2, m.None2)
 	}
 }

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go

@@ -20,7 +20,7 @@ import (
 // TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage (ADR 0593 §2-§4; D11): the indexer whole against the
 // bus. Two Go modules of the catalogue whose go.sum names mesh-sdk; their builds end reporting their imports (importer:
 // buildstep; plain: announce); then mesh-sdk merges a change of buildstep alone, which moves mesh-sdk's version. Within
-// 5 s the packages key of mesh-sdk/go names the merge with a digest per package, importer gets a new fingerprint, and
+// 20 s the packages key of mesh-sdk/go names the merge with a digest per package, importer gets a new fingerprint, and
 // plain keeps the one it had, though the version line would have moved it. The forge's tools answer on the bus as the
 // gitea module's do. A merge of go/ alone before it writes no value of mesh-sdk, whose tree is its named paths'.
 func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T) {
@@ -85,14 +85,14 @@ func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T)
 	module := func(name string, ok func(delivery.Module) bool) delivery.Module {
 		t.Helper()
 		var m delivery.Module
-		for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 			if e, err := modules.Get(ctx, name); err == nil {
 				if m, err = delivery.DecodeModule(e.Value()); err == nil && ok(m) {
 					return m
 				}
 			}
 		}
-		t.Fatalf("modules.%s = %+v within 5 s", name, m)
+		t.Fatalf("modules.%s = %+v within 20 s", name, m)
 		return m
 	}
 	sdkObjects := func(src string) map[string]map[string]string {
@@ -161,7 +161,7 @@ func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T)
 	}
 	key, _ := delivery.PackagesKey(sdkGo)
 	var pub delivery.PublishedPackages
-	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
 		if e, err := packages.Get(ctx, key); err == nil {
 			if pub, err = delivery.DecodePublishedPackages(e.Value()); err == nil && pub.Commit == m2 {
 				break
@@ -169,7 +169,7 @@ func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T)
 		}
 	}
 	if pub.Commit != m2 || len(pub.Packages) != 3 {
-		t.Fatalf("packages.%s = %+v within 5 s, want the merge %s with a digest of go, announce and buildstep", key, pub, m2)
+		t.Fatalf("packages.%s = %+v within 20 s, want the merge %s with a digest of go, announce and buildstep", key, pub, m2)
 	}
 	moved := module("importer", func(m delivery.Module) bool { return m.FP2 != importer.FP2 })
 	if moved.Packages[sdkGo] != m2 {

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go

@@ -128,7 +128,7 @@ func TestABuildModulesBucketThatAppearsLaterIsWatchedWhenItsEventArrives(t *test
 	}
 	// It is followed from now on: a key put later is applied, and a second report only recomputes.
 	bucket.w.ch <- fakeEntry{key: "ace.go", value: value}
-	for deadline := time.Now().Add(3 * time.Second); time.Now().Before(deadline); time.Sleep(5 * time.Millisecond) {
+	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(5 * time.Millisecond) {
 		mu.Lock()
 		n := len(applies)
 		mu.Unlock()

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go

@@ -20,6 +20,8 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/asks"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"mesh-delivery-teller/internal/ends"
 )
 
 // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
@@ -175,7 +177,9 @@ const (
 )
 
 func fullConf(dir string) string {
-	teller := append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), readOf(gateAsksKV)...),
+	// ends is the teller's own state (manifest state entry ends), which the controller makes before it starts: a bed
+	// without it held the start ten seconds on a read the bus never answered, and the tests raced that stall.
+	teller := append(append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), keepsOf(ends.Bucket)...), readOf(gateAsksKV)...),
 		seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_"+messengerAsks+".$KV."+messengerAsks+".mesh-delivery-teller.>")
 	teller = append(append(teller, readOf(assignerAsks)...), readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads
 	view := append(append(append(readOf(tellerAnswers), readOf(gateAsksKV)...), readOf(assignerAsks)...), keepsOf(viewAnswers)...)
@@ -244,7 +248,7 @@ func startFull(t *testing.T) *fullBed {
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
 	t.Cleanup(cancel)
-	for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1, messengerAsks: 1, assignerAsks: delivery.AssignerAsksHistory, viewAnswers: delivery.ViewAnswersHistory} {
+	for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1, ends.Bucket: 1, messengerAsks: 1, assignerAsks: delivery.AssignerAsksHistory, viewAnswers: delivery.ViewAnswersHistory} {
 		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
 			t.Fatal(err)
 		}
@@ -256,6 +260,21 @@ func startFull(t *testing.T) *fullBed {
 	return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx}
 }
 
+// listening waits until the program hears the router's warrants (its subscription on the decided subject is on the bus),
+// which it does before it watches any bucket: an event, not a pause a loaded machine can outlast.
+func (b *fullBed) listening(t *testing.T) {
+	t.Helper()
+	acc, err := b.srv.LookupAccount("MESH")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for deadline := time.Now().Add(20 * time.Second); !acc.SubscriptionInterest(seatDecided); time.Sleep(10 * time.Millisecond) {
+		if time.Now().After(deadline) {
+			t.Fatal("the program did not subscribe to the router's warrants within 20 s")
+		}
+	}
+}
+
 func (b *fullBed) as(t *testing.T, name string) jetstream.JetStream {
 	t.Helper()
 	c, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo(name, "pw"), nats.CustomInboxPrefix("_INBOX."+name))
@@ -318,7 +337,7 @@ func TestCredentialsWriteOnlyTheirOwnBuckets(t *testing.T) { // 1343/AC4
 // warrant on the teller's decided subject, and the answer is under the ask's key in the teller's bucket; withdrawing a
 // second ask cancels it (1343/AC1, AC2).
 func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
-	now = time.Now().UTC()
+	t.Cleanup(nowIs(time.Now().UTC()))
 	b := startFull(t)
 	tools := b.as(t, "n1.node-tools")
 	broker := filepath.Join(t.TempDir(), "broker")
@@ -351,8 +370,8 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
 	if err != nil {
 		t.Fatal(err)
 	}
-	// The program hears warrants before it reads the asks; give it its subscription before the ask is written.
+	// The program hears warrants before it reads the asks; it has its subscription before the ask is written.
-	time.Sleep(500 * time.Millisecond)
+	b.listening(t)
 	if _, err := kv.Put(b.ctx, key, value); err != nil {
 		t.Fatal(err)
 	}
@@ -423,7 +442,7 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
 // lapsed; the refused ask is on the seat's work queue again, which the bus's duplicate window would have dropped under its
 // first message id (1610/AC1 on the real bus).
 func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
-	now = time.Now().UTC()
+	t.Cleanup(nowIs(time.Now().UTC()))
 	b := startFull(t)
 	tools := b.as(t, "n1.node-tools")
 	broker := filepath.Join(t.TempDir(), "broker")
@@ -443,7 +462,7 @@ func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
 	if err != nil {
 		t.Fatal(err)
 	}
-	time.Sleep(500 * time.Millisecond)
+	b.listening(t)
 	// lastAsk waits for the seat's queue to hold an ask newer than seq under id, and gives its sequence.
 	lastAsk := func(id string, after uint64) uint64 {
 		t.Helper()
@@ -495,7 +514,7 @@ func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
 // (rewriting the view's answer does not write a second), and a view answer under the merge's key fills nothing
 // (1297/B10 AC: a view answer fills the ask once; an ask needing proof is not answered from a view key).
 func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
-	now = time.Now().UTC()
+	t.Cleanup(nowIs(time.Now().UTC()))
 	b := startFull(t)
 	assigner, view := b.as(t, "n1.mesh-delivery-assigner"), b.as(t, "n1.mesh-delivery-view")
 	broker := filepath.Join(t.TempDir(), "broker")
@@ -521,7 +540,7 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
 	if err != nil {
 		t.Fatal(err)
 	}
-	time.Sleep(500 * time.Millisecond) // the program is watching
+	b.listening(t) // the program hears warrants and watches
 	if _, err := askKV.Put(b.ctx, key, value); err != nil {
 		t.Fatal(err)
 	}
@@ -568,7 +587,7 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
 // ask's key, where the assigner reads it; a second ask the assigner takes away (a hold it lifted) is cancelled at the router
 // (1906/H1, H2 on the real bus).
 func TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer(t *testing.T) {
-	now = time.Now().UTC()
+	t.Cleanup(nowIs(time.Now().UTC()))
 	b := startFull(t)
 	assigner := b.as(t, "n1.mesh-delivery-assigner")
 	broker := filepath.Join(t.TempDir(), "broker")
@@ -586,7 +605,7 @@ func TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	time.Sleep(500 * time.Millisecond) // the program hears warrants and watches
+	b.listening(t) // the program hears warrants and watches
 	if _, err := askKV.Put(b.ctx, key, value); err != nil {
 		t.Fatal(err)
 	}
@@ -662,7 +681,7 @@ func TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer(t *testing.T) {
 // TestTellerTakesAChangedHoldAnswerFromTheView starts the program against the bus: a hold ask answered "keep" earlier, and the
 // view's answer "release" written now, replace the answer of that asking with "release" (1906/N23 on the real bus).
 func TestTellerTakesAChangedHoldAnswerFromTheView(t *testing.T) {
-	now = time.Now().UTC()
+	t.Cleanup(nowIs(time.Now().UTC()))
 	b := startFull(t)
 	assigner, view := b.as(t, "n1.mesh-delivery-assigner"), b.as(t, "n1.mesh-delivery-view")
 	broker := filepath.Join(t.TempDir(), "broker")
@@ -694,7 +713,7 @@ func TestTellerTakesAChangedHoldAnswerFromTheView(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	time.Sleep(500 * time.Millisecond) // the program is watching
+	b.listening(t) // the program hears warrants and watches
 	if _, err := viewKV.Put(b.ctx, key+".view", viewAnswer(t, "release", "it was a flake")); err != nil {
 		t.Fatal(err)
 	}

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go

@@ -25,6 +25,14 @@ const (
 
 var now = time.Date(2026, 10, 20, 9, 0, 0, 0, time.UTC)
 
+// nowIs sets the tests' clock now to at and returns what puts it back; a test sets it only as t.Cleanup(nowIs(at)), so no
+// test leaves its clock to the next one (-shuffle runs them in any order).
+func nowIs(at time.Time) (restore func()) {
+	was := now
+	now = at
+	return func() { now = was }
+}
+
 // memStore is a state bucket in memory with the revisions and the compare-and-set of the runtime's state.
 type memStore struct {
 	entries map[string]*stdio.StateEntry

modules/mesh-delivery-watchman/internal/watch/findings_test.go

@@ -2,6 +2,7 @@ package watch
 
 import (
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -372,8 +373,8 @@ func TestTheWatchmansFindings(t *testing.T) {
 			expire(w, c, signalKey("S9", "bus.ace.web.slow-consumer"), AdvisoryQuiet)
 		}, map[string]string{}},
 		{"WM5: while letters are held DEAD_LETTERS is counted again at its own pulse's expiry, and at a letter's age", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
-			n := 0
+			var n atomic.Int32
-			w.SetRecount(func() { n++ })
+			w.SetRecount(func() { n.Add(1) })
 			w.DeadLettersHeld(map[string]int{"EVENTS.ace_web": 1}, func(s, c string) string { return c })
 			if o.ttls[signalKey("dead", "recount")] != DeadRecountEvery || !strings.Contains(o.findings["bus.EVENTS.ace_web.max-deliveries"].Said, "every 5m0s") {
 				t.Fatalf("pulses %v findings %v", o.ttls, o.findings)
@@ -381,9 +382,11 @@ func TestTheWatchmansFindings(t *testing.T) {
 			w.DeadAging("EVENTS.ace_web", c.t.Add(time.Minute))
 			expire(w, c, signalKey("dead", "EVENTS.ace_web"), time.Minute)
 			expire(w, c, signalKey("dead", "recount"), DeadRecountEvery)
-			time.Sleep(10 * time.Millisecond)
+			// The recount runs in a goroutine of its own: waited for, within seconds no loaded machine reaches.
-			if n != 2 {
+			for deadline := time.Now().Add(10 * time.Second); n.Load() < 2 && time.Now().Before(deadline); time.Sleep(time.Millisecond) {
-				t.Fatalf("counted again %d times", n)
+			}
+			if n.Load() != 2 {
+				t.Fatalf("counted again %d times", n.Load())
 			}
 			w.DeadLettersHeld(map[string]int{"EVENTS.ace_web": 1}, func(s, c string) string { return c }) // the count renews its pulse
 			w.DeadLettersHeld(map[string]int{}, nil)