From b661bfead87d7a3489d12727e94bfbea36bffe97 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Wed, 30 Sep 2026 01:36:55 +0200 Subject: [PATCH] Wait for what tests check instead of sleeping and hoping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI's test job has failed intermittently — two of three runs on some days — while every run here passed, including six repetitions on one CPU under the race detector with eight packages at once. The difference is a shared, loaded Linux runner, and the tests most exposed to it are the ones that sleep a fixed time and then assert: - admission: idle admission bounded at 50ms of wall clock against a 1ms tick, and a queued request checked for after a 50ms sleep; - scrape: a 10ms scrape timeout, which on a slow runner reports the engine unreachable and then the violation a second time, in a test that counts one; - supervisor: log lines, three lines of output and a removed state file each checked after a fixed sleep, and an exited worker bounded at 1.5s; - agent: a heartbeat checked for after four intervals. Each now waits for its condition up to a deadline of seconds, and the two wall-clock bounds are set against what they exist to distinguish — the admission queue's tick raised to ten seconds, the exited worker bounded below the five-second grace it must not wait out — rather than against how fast this laptop is. Without the CI logs this cannot be proven to be the failure; these are the tests that fail when the machine is slow, and they no longer do. Co-Authored-By: Claude Opus 5.5 (1M context) --- internal/agent/agent_test.go | 13 ++++--- internal/agent/supervisor/supervisor_test.go | 37 +++++++++++++++----- internal/router/admission/admission_test.go | 19 +++++++--- internal/router/scrape/llamacpp_test.go | 5 ++- 4 files changed, 57 insertions(+), 17 deletions(-) diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index a3dc052..8bf4929 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -400,11 +400,16 @@ func TestHeartbeatKeepsNodeAlive(t *testing.T) { n, _ := h.inv.Get("n1") first := n.LastSeen - time.Sleep(200 * time.Millisecond) - n, _ = h.inv.Get("n1") - if !n.LastSeen.After(first) { - t.Error("heartbeats are not refreshing LastSeen; the node would be declared lost") + // Waited for rather than slept through: four heartbeat intervals was not + // always enough on a loaded runner for one to have landed. + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if n, _ = h.inv.Get("n1"); n.LastSeen.After(first) { + return + } + time.Sleep(20 * time.Millisecond) } + t.Error("heartbeats are not refreshing LastSeen; the node would be declared lost") } func TestStartAndStopShardEndToEnd(t *testing.T) { diff --git a/internal/agent/supervisor/supervisor_test.go b/internal/agent/supervisor/supervisor_test.go index 1fb998d..3039a7a 100644 --- a/internal/agent/supervisor/supervisor_test.go +++ b/internal/agent/supervisor/supervisor_test.go @@ -243,7 +243,9 @@ func TestWaitReadyTimeoutIncludesLogs(t *testing.T) { if err != nil { t.Fatal(err) } - time.Sleep(100 * time.Millisecond) // let the line be captured + eventually(t, 3*time.Second, "the line was never captured", func() bool { + return strings.Contains(strings.Join(sh.Logs(10), "\n"), "CUDA out of memory") + }) err = s.WaitReady(t.Context(), sh, "http://x", "m", 200*time.Millisecond) if err == nil { @@ -295,7 +297,10 @@ func TestLogsAreCaptured(t *testing.T) { if err != nil { t.Fatal(err) } - time.Sleep(400 * time.Millisecond) + eventually(t, 3*time.Second, "not every line arrived", func() bool { + l := strings.Join(sh.Logs(10), "\n") + return strings.Contains(l, "line1") && strings.Contains(l, "line2") && strings.Contains(l, "line3") + }) logs := strings.Join(sh.Logs(10), "\n") for _, want := range []string{"line1", "line2", "line3"} { @@ -372,11 +377,11 @@ func TestStateFileRemovedOnExit(t *testing.T) { sh, _ := s.Start(t.Context(), "a", engines.ShardSpec{}, shellCommand("exit 0")) waitState(t, sh, Failed, 3*time.Second) - time.Sleep(50 * time.Millisecond) - - if _, err := os.Stat(s.statePath("a")); !os.IsNotExist(err) { - t.Error("the state file outlived the process; a later scan would find a dead orphan") - } + eventually(t, 3*time.Second, + "the state file outlived the process; a later scan would find a dead orphan", func() bool { + _, err := os.Stat(s.statePath("a")) + return os.IsNotExist(err) + }) } // --- state machine and ring buffer --- @@ -743,7 +748,9 @@ func TestWaitAliveFailsOnEarlyExit(t *testing.T) { if err := s.WaitAlive(t.Context(), sh); err == nil { t.Fatal("a worker that died immediately was reported ready") } - if d := time.Since(start); d > 1500*time.Millisecond { + // Below AliveGrace (5s), which is what waiting it out would take; the + // headroom is for a loaded runner, where 1.5s was sometimes not enough. + if d := time.Since(start); d > 4*time.Second { t.Errorf("waited %v for a process that had already exited", d) } } @@ -827,3 +834,17 @@ func TestAWorkersOnlyProcessWaitsForTheLeader(t *testing.T) { t.Errorf("the worker started after %v, before the leader was listening", d) } } + +// eventually waits for cond, failing with msg if it does not hold within +// limit. Sleeping a fixed time and then checking measured the machine the +// test ran on, and a loaded CI runner under the race detector lost that race. +func eventually(t *testing.T, limit time.Duration, msg string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(limit) + for !cond() { + if time.Now().After(deadline) { + t.Fatal(msg) + } + time.Sleep(10 * time.Millisecond) + } +} diff --git a/internal/router/admission/admission_test.go b/internal/router/admission/admission_test.go index 7d9dd02..be4987f 100644 --- a/internal/router/admission/admission_test.go +++ b/internal/router/admission/admission_test.go @@ -42,14 +42,20 @@ func newController(t *testing.T, cfg Config, sat SaturationFunc) *Controller { func TestAdmitsImmediatelyWhenIdle(t *testing.T) { s := &fixedSat{} - c := newController(t, testConfig(), s.fn) + // A queue that is looked at every ten seconds, so a request that went + // through it rather than the fast path would take that long. A bound of + // 50ms against a 1ms tick measured the machine the test ran on, and a + // loaded CI runner under the race detector was slower than that. + cfg := testConfig() + cfg.Interval = 10 * time.Second + c := newController(t, cfg, s.fn) start := time.Now() lease, err := c.Acquire(t.Context(), Request{Model: "m", Band: "bronze"}) if err != nil { t.Fatal(err) } - if d := time.Since(start); d > 50*time.Millisecond { + if d := time.Since(start); d > 2*time.Second { t.Errorf("idle admission took %v; the fast path is not being used", d) } lease.Release() @@ -131,8 +137,13 @@ func TestQueuedRequestAdmittedWhenSaturationDrops(t *testing.T) { done <- err }() - // Give it time to actually queue, then open the gate. - time.Sleep(50 * time.Millisecond) + // Wait for it to actually queue, then open the gate. Waited for, not + // slept through: on a loaded runner 50ms was sometimes not long enough + // for the goroutine to have been scheduled at all. + deadline := time.Now().Add(2 * time.Second) + for c.Depths()[3].Depth != 1 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } if d := c.Depths(); d[3].Depth != 1 { t.Fatalf("expected 1 queued bronze request, got %+v", d) } diff --git a/internal/router/scrape/llamacpp_test.go b/internal/router/scrape/llamacpp_test.go index a03e23b..93df598 100644 --- a/internal/router/scrape/llamacpp_test.go +++ b/internal/router/scrape/llamacpp_test.go @@ -175,7 +175,10 @@ func TestABrokenContractIsReportedOnceAndLoudly(t *testing.T) { in := endpoint.Info{ID: "e1", URL: srv.URL, Engine: "vllm", Models: []string{"m"}} reg.Replace([]endpoint.Info{in}) - p := NewPoller(reg, Config{Interval: 20 * time.Millisecond, Timeout: 10 * time.Millisecond}, + // A generous timeout: a scrape that timed out would be reported as + // unreachable, and the violation reported again once it answered — two + // warnings where the test counts one, on a slow runner only. + p := NewPoller(reg, Config{Interval: 20 * time.Millisecond, Timeout: 2 * time.Second}, slog.New(cap)) ctx, cancel := context.WithCancel(context.Background()) defer cancel()