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) <noreply@anthropic.com>
This commit is contained in:
1 parent
37a80a9b44
commit
b661bfead8
4 files changed
+57
-17
No files matched your search
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user