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