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

Tests wait on events or the clock they are given, never on milliseconds a loaded machine outlasts

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-02 23:04:44 +01:00
parent 15d28ddf5b
commit f73b0bf6f6
16 changed files with 169 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-builder/buildrun/asks_test.go
+23 -39
View File
@@ -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)
}
}
modules/mesh-delivery-builder/internal/build/askagain.go
+1 -1
View File
@@ -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
+2 -1
View File
@@ -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
+18 -7
View File
@@ -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
+25 -5
View File
@@ -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
+2
View File
@@ -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
+3 -2
View File
@@ -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
+1 -1
View File
@@ -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
+9 -6
View File
@@ -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 {
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go
+8 -8
View File
@@ -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
+23 -23
View File
@@ -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
+5 -5
View File
@@ -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
+1 -1
View File
@@ -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
+32 -13
View File
@@ -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)
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go
+8
View File
@@ -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
+8 -5
View File
@@ -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)