From f73b0bf6f63b6f5e971cb9dce2d47918d271994c Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 23:04:19 +0100 Subject: [PATCH] Tests wait on events or the clock they are given, never on milliseconds a loaded machine outlasts The check of 2026-11-02 22:35 on a loaded shanks failed two buildrun tests on wall-clock bounds of 60 ms, the second failure of this kind tonight (the first: mesh-sdk's busretry tests, f02f142). Quick fix: give the two buildrun asker tests bounds of seconds. Thorough fix (chosen, because the kind recurs across every module of the repository): every test that raced a wall clock now waits on an event, steps a clock it is given, or has a bound only a failing test reaches. - buildrun asker tests: the steps' own times stand for the clock (bound an hour, starts and phases written in the past), no goroutine paces phases with sleeps. - builder: the decline bound reads Builder.Now; the decline test steps it; the stream-wait test waits for five check reads instead of 100 ms. - checker: the lost-ask try bounds an untaken ask at 200 ms only where that is what ends it; the program says when a merge turn changes, and the integration test waits on that line instead of 300 ms. - teller: the bed makes the teller's own ends bucket (a 10 s stall at start the tests raced by 0.5 s), waits for its warrant subscription instead of 500 ms pauses, and the package-level now is restored with t.Cleanup. - watchman: the recount, run in its own goroutine, is waited for instead of 10 ms. - polls of 2-5 s become 10 s (unit) and 20 s (integration). Under a throttled cgroup (half a core shared with two busy loops) beside the node-engine's suite, the buildrun tests failed 5 of 20 runs before and passed 30 of 30 after; no test added. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- .../buildrun/asks_test.go | 62 +++++++------------ .../internal/build/askagain.go | 2 +- .../internal/build/run.go | 3 +- .../internal/build/run_test.go | 25 +++++--- .../mesh-delivery-checker/integration_test.go | 30 +++++++-- .../cmd/mesh-delivery-checker/main.go | 2 + .../internal/check/adr548_test.go | 5 +- .../internal/check/examiner_test.go | 2 +- .../internal/check/lostask_test.go | 15 +++-- .../integration_test.go | 16 ++--- .../mesh-delivery-indexer/integration_test.go | 46 +++++++------- .../packages_integration_test.go | 10 +-- .../mesh-delivery-indexer/toolchains_test.go | 2 +- .../mesh-delivery-teller/integration_test.go | 45 ++++++++++---- .../cmd/mesh-delivery-teller/teller_test.go | 8 +++ .../internal/watch/findings_test.go | 13 ++-- 16 files changed, 169 insertions(+), 117 deletions(-) diff --git a/modules/mesh-delivery-builder/buildrun/asks_test.go b/modules/mesh-delivery-builder/buildrun/asks_test.go index 5403a54..ec6b94f 100644 --- a/modules/mesh-delivery-builder/buildrun/asks_test.go +++ b/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 - // waited for until it ends; one that stays in a phase the bound fails at once, naming the phase; one that never stops moving - // ends at three bounds, said as such. - k = Asker{Publisher: g, Waiter: g, Bound: 60 * time.Millisecond} - started := time.Now() - go func() { - for i := range 8 { - g.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "ace", State: delivery.StepRunning, Started: started, Phase: "p" + string(rune('a'+i)), PhaseAt: time.Now()}) - time.Sleep(15 * time.Millisecond) - } - g.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "ace", State: delivery.StepPassed}) - }() + // 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 + // new phase inside the bound is waited for until it ends; one that stood a whole bound in a phase fails at once, naming + // the phase; one three bounds from its start ends, said as kept moving. The steps' own times stand for the clock, so the + // 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 + // of 60 ms). + k = Asker{Publisher: g, Waiter: g, Bound: time.Hour} + now := time.Now() + g.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "ace", State: delivery.StepRunning, Started: now.Add(-90 * time.Minute), Phase: "pb", PhaseAt: now}) + 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{}) - 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()}) - } - } - }() + g.Write(delivery.Step{ID: "mvl0frv18d1982", Node: "ace", State: delivery.StepRunning, Started: now.Add(-3 * time.Hour), Phase: "qz", PhaseAt: 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() - f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: first}) - go func() { - time.Sleep(15 * time.Millisecond) - f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: time.Now()}) - time.Sleep(50 * time.Millisecond) // past the first run's bound, inside the second's - f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepPassed, Started: time.Now()}) - }() - 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) + now := time.Now() + f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: now}) + f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: now.Add(-time.Hour)}) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if _, err := k.Wait(ctx, goAsk(id)); err == nil || !strings.Contains(err.Error(), "on ace did not end within 1h0m0s of its start") { + t.Fatalf("a redelivered run a bound old: %v, want its own bound to end the wait", err) } } diff --git a/modules/mesh-delivery-builder/internal/build/askagain.go b/modules/mesh-delivery-builder/internal/build/askagain.go index 5a4b4d1..e91e6ba 100644 --- a/modules/mesh-delivery-builder/internal/build/askagain.go +++ b/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 } diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index e01bdce..6cae694 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/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 diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index e2ecd62..03d64ce 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/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") } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index f42d8ec..4a03d30 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/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) } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index a5fbef9..42e99b8 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/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 } }) diff --git a/modules/mesh-delivery-checker/internal/check/adr548_test.go b/modules/mesh-delivery-checker/internal/check/adr548_test.go index 87d7d88..6b72ac1 100644 --- a/modules/mesh-delivery-checker/internal/check/adr548_test.go +++ b/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) } diff --git a/modules/mesh-delivery-checker/internal/check/examiner_test.go b/modules/mesh-delivery-checker/internal/check/examiner_test.go index 5cdf4cc..6e5b473 100644 --- a/modules/mesh-delivery-checker/internal/check/examiner_test.go +++ b/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") } diff --git a/modules/mesh-delivery-checker/internal/check/lostask_test.go b/modules/mesh-delivery-checker/internal/check/lostask_test.go index 9a1581c..3c547e4 100644 --- a/modules/mesh-delivery-checker/internal/check/lostask_test.go +++ b/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"}, - {"taken by no build node within the bound", 0, true, buildrun.NotTakenWords + " within 200ms"}, + {"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, 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 { diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go index 5f1592f..84644d6 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go +++ b/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") diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 8f7275a..5685dc0 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/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. within 5 s, the fingerprint of the merge's tree and the recorded inputs. +// message on FORGE makes modules. 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. 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. holds and the running SDK reads; the go toolchain is reported. Within 5 s of the start +// manifest, which manifests. 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) } } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go index 11192c9..069be8b 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go +++ b/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 { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go index 7427efb..00f72f5 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go +++ b/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() diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go index 9975628..5754987 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go +++ b/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. - time.Sleep(500 * time.Millisecond) + // The program hears warrants before it reads the asks; it has its subscription before the ask is written. + 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) } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go index 3f6aee0..fc118d8 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go +++ b/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 diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 9c291e5..573abe7 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/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 - w.SetRecount(func() { n++ }) + var n atomic.Int32 + 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) - if n != 2 { - t.Fatalf("counted again %d times", n) + // The recount runs in a goroutine of its own: waited for, within seconds no loaded machine reaches. + for deadline := time.Now().Add(10 * time.Second); n.Load() < 2 && time.Now().Before(deadline); time.Sleep(time.Millisecond) { + } + 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)