gateway tests: timing assertions hold under heavy host load

crates/gateway/Cargo.toml

@@ -90,6 +90,10 @@ hyper-util = { version = "0.1", features = ["tokio", "server-auto"] }
 proptest = { version = "1.10", default-features = false, features = ["std"] }
 rcgen = "0.13"
 reqwest = { version = "0.13", default-features = false, features = ["json", "rustls", "http2"] }
+# Thread CPU time (`clock_gettime(CLOCK_THREAD_CPUTIME_ID)`) behind a safe API, for perf assertions
+# that must hold on a loaded host (`unsafe_code` is forbidden, so not raw `libc`). Already in the
+# build graph via `tempfile` and `clap`: this only adds its `time` feature.
+rustix = { version = "1", default-features = false, features = ["std", "time"] }
 tokio-rustls = "0.26"
 
 [[bench]]

crates/gateway/src/cache.rs

@@ -484,17 +484,29 @@ mod tests {
 
     #[test]
     fn insert_drops_an_expired_prefix_without_evicting_a_newer_live_entry() {
-        let c = ResponseCache::new(Duration::from_millis(40), 2, 1024);
+        const TTL: Duration = Duration::from_millis(40);
         let k1 = key_of(1, "/v1", b"a", &[]);
         let k2 = key_of(1, "/v1", b"b", &[]);
-        c.insert(k1, entry(b"old"));
-        std::thread::sleep(Duration::from_millis(50));
-        c.insert(k2, entry(b"new"));
         let k3 = key_of(1, "/v1", b"c", &[]);
-        c.insert(k3, entry(b"newer"));
-        assert!(c.get(&k1).is_none());
-        assert_eq!(c.get(&k2).unwrap().body.as_ref(), b"new");
-        assert_eq!(c.get(&k3).unwrap().body.as_ref(), b"newer");
+        // k2 and k3 must still be live when read back. A loaded host can stall this thread past
+        // the TTL between the inserts and the reads; such a run proves nothing, so it is re-run.
+        for _ in 0..50 {
+            let c = ResponseCache::new(TTL, 2, 1024);
+            c.insert(k1, entry(b"old"));
+            std::thread::sleep(TTL + Duration::from_millis(10));
+            let live_from = Instant::now();
+            c.insert(k2, entry(b"new"));
+            c.insert(k3, entry(b"newer"));
+            let (got1, got2, got3) = (c.get(&k1), c.get(&k2), c.get(&k3));
+            if live_from.elapsed() >= TTL {
+                continue;
+            }
+            assert!(got1.is_none());
+            assert_eq!(got2.unwrap().body.as_ref(), b"new");
+            assert_eq!(got3.unwrap().body.as_ref(), b"newer");
+            return;
+        }
+        panic!("every run stalled past the {TTL:?} TTL");
     }
 
     #[test]

crates/gateway/src/circuit_breaker.rs

@@ -917,15 +917,23 @@ mod tests {
 
     #[test]
     fn test_windowed_resets_after_window() {
-        // Note: window uses second-level precision, so use 1 second window
-        let cb = CircuitBreaker::new(CircuitBreakerConfig::windowed(3, Duration::from_secs(1)));
+        // A hand-stepped clock: on the real one, a second boundary falling between two of these
+        // calls (a preempted thread, a loaded host) would split one window into two.
+        static NOW: AtomicU64 = AtomicU64::new(100);
+        fn clock() -> u64 {
+            NOW.load(Ordering::Relaxed)
+        }
+        let cb = CircuitBreaker::with_clock(
+            CircuitBreakerConfig::windowed(3, Duration::from_secs(1)),
+            clock,
+        );
 
         cb.record_failure();
         cb.record_failure();
         assert_eq!(cb.state(), CircuitState::Closed { failure_count: 2 });
 
-        // Wait for window to expire (1 second + buffer)
-        thread::sleep(Duration::from_millis(1100));
+        // The window (second-level precision) expires.
+        NOW.store(101, Ordering::Relaxed);
 
         // This failure starts a new window
         cb.record_failure();

crates/gateway/src/deadline.rs

@@ -99,21 +99,54 @@ mod tests {
         let start = Instant::now();
         let d = at(start, 2);
         let left = remaining(d).unwrap();
+        // Measured after the read, so it covers however long this thread was descheduled: what is
+        // left plus what has passed is the whole two seconds, less the millisecond truncation.
+        let passed = start.elapsed();
         assert!(left <= Duration::from_secs(2), "{left:?}");
-        assert!(left > Duration::from_millis(1500), "{left:?}");
+        assert!(
+            left + passed + Duration::from_millis(1) >= Duration::from_secs(2),
+            "{left:?} left after {passed:?}"
+        );
         assert_eq!(
             remaining(0),
             Some(Duration::ZERO),
             "a deadline already passed"
         );
     }
 
+    /// The coarse clock advances on its own and expires a deadline, never early: it lags real time
+    /// (by up to a tick, more on a loaded host), so a deadline it calls expired has truly passed.
+    /// Polled rather than slept on, so a ticker thread a loaded host runs late cannot fail it.
     #[test]
     fn the_coarse_clock_ticks_and_expires_deadlines() {
         start();
-        let d = at(Instant::now(), 1);
-        assert!(!expired(d));
-        std::thread::sleep(Duration::from_millis(2300));
-        assert!(expired(d), "now {} deadline {d}", now_ms());
+        let begun = Instant::now();
+        let d = at(begun, 1);
+        let mut seen = now_ms();
+        loop {
+            let now = now_ms();
+            if now != seen {
+                // A tick. The ticker sleeps a whole TICK between stores, and load only adds to it.
+                assert!(
+                    now + 1 >= seen + TICK.as_millis() as u64,
+                    "ticked {seen} -> {now}"
+                );
+                seen = now;
+            }
+            if now >= d {
+                break;
+            }
+            assert!(
+                begun.elapsed() < Duration::from_secs(30),
+                "the coarse clock never reached the deadline: now {now} deadline {d}"
+            );
+            std::thread::sleep(Duration::from_millis(5));
+        }
+        let passed = begun.elapsed();
+        assert!(expired(d));
+        assert!(
+            passed + Duration::from_millis(1) >= Duration::from_secs(1),
+            "expired after only {passed:?}"
+        );
     }
 }

crates/gateway/src/route.rs

@@ -1237,11 +1237,16 @@ mod tests {
             ProviderMetrics::disconnected(),
             None,
         );
-        // A cooldown that ends 200 ms from now, rather than `KEY_COOLDOWN`'s minute.
-        p.pool_auth[0]
-            .bad_until_ms
-            .store(clock_ms().saturating_add(200), Ordering::Relaxed);
-        assert_eq!(p.first_key(), 1);
+        // A cooldown that ends 200 ms from now, rather than `KEY_COOLDOWN`'s minute. It only shows
+        // the key cooling if it is read before those 200 ms are up; a run a loaded host stalled
+        // past them proves nothing, so it is re-run.
+        let cooling_read = (0..50).find_map(|_| {
+            let until = clock_ms().saturating_add(200);
+            p.pool_auth[0].bad_until_ms.store(until, Ordering::Relaxed);
+            let first = p.first_key();
+            (clock_ms() < until).then_some(first)
+        });
+        assert_eq!(cooling_read, Some(1), "the cooling key is skipped");
         std::thread::sleep(Duration::from_millis(250));
         assert_eq!(p.first_key(), 0, "the cooldown has passed");
     }

crates/gateway/src/translate_request_tests.rs

@@ -1696,24 +1696,30 @@ fn consecutive_user_messages_keep_their_text_apart() {
 /// client split across messages) merges in time linear in its length, and every message keeps its
 /// block. Each merge used to copy every block gathered so far: 256 KiB of user turns took 4.4 s of
 /// CPU in a release build, 1 MiB over a minute, all on one request.
+///
+/// Asserted as scaling, in this thread's CPU time, so a loaded host cannot fail it: quadrupling the
+/// run must cost under 8x (linear is ~4x; the quadratic merge was ~16x). Each size keeps the
+/// cheapest of three runs, which drops a run that a cache-cold start or a preemption inflated.
 /// claim: TRN-3
 /// defect: D217
 #[test]
 fn a_long_run_of_same_role_messages_merges_in_linear_time() {
-    let users = vec![json!({"role": "user", "content": "hi"}); 6000];
-    let calls = vec![
-        json!({"role": "assistant", "content": null, "tool_calls": [
-            {"id": "c", "type": "function", "function": {"name": "f", "arguments": "{}"}}
-        ]});
-        3000
-    ];
-    let start = std::time::Instant::now();
-    let v = c2m(
-        &chat(json!({"messages": users.into_iter().chain(calls).collect::<Vec<_>>()})),
-        "claude-haiku-4-5",
-    );
-    let took = start.elapsed();
-    let blocks = |role: &str| -> usize {
+    fn thread_cpu() -> std::time::Duration {
+        let t = rustix::time::clock_gettime(rustix::time::ClockId::ThreadCPUTime);
+        std::time::Duration::new(t.tv_sec as u64, t.tv_nsec as u32)
+    }
+    // `users` user turns, then half as many assistant messages carrying one tool call each.
+    let request = |users: usize| {
+        let calls = vec![
+            json!({"role": "assistant", "content": null, "tool_calls": [
+                {"id": "c", "type": "function", "function": {"name": "f", "arguments": "{}"}}
+            ]});
+            users / 2
+        ];
+        let user = vec![json!({"role": "user", "content": "hi"}); users];
+        chat(json!({"messages": user.into_iter().chain(calls).collect::<Vec<_>>()}))
+    };
+    let blocks = |v: &Value, role: &str| -> usize {
         v["messages"]
             .as_array()
             .unwrap()
@@ -1722,8 +1728,31 @@ fn a_long_run_of_same_role_messages_merges_in_linear_time() {
             .map(|m| m["content"].as_array().map_or(1, Vec::len))
             .sum()
     };
-    assert_eq!((blocks("user"), blocks("assistant")), (6000, 3000));
-    assert!(took < std::time::Duration::from_secs(2), "took {took:?}");
+    let cheapest = |users: usize| {
+        let body = request(users);
+        (0..3)
+            .map(|_| {
+                let start = thread_cpu();
+                let v = c2m(&body, "claude-haiku-4-5");
+                let took = thread_cpu() - start;
+                assert_eq!(
+                    (blocks(&v, "user"), blocks(&v, "assistant")),
+                    (users, users / 2)
+                );
+                took
+            })
+            .min()
+            .unwrap()
+    };
+    // 1,000 user turns cost ~10 ms of CPU in a debug build (far above the clock's resolution);
+    // the quadratic merge took seconds at 4x that.
+    let small = cheapest(1_000);
+    let large = cheapest(4_000);
+    let ratio = large.as_secs_f64() / small.as_secs_f64();
+    assert!(
+        ratio < 8.0,
+        "4x the messages cost {ratio:.1}x the CPU ({small:?} -> {large:?}): not linear"
+    );
 }
 
 /// A Chat client's `cache_control` on a whole assistant or tool message (the same message-level

crates/gateway/tests/claims_security.rs

@@ -775,30 +775,60 @@ async fn the_rate_limit_counts_one_identity_across_credential_locations() {
         .await;
     let vk = vkey(&sk, 12);
     let url = format!("{}/openai/v1/chat/completions", gw.url());
-    // Two requests per location, six in all. Separate buckets would see two each and never trip a
-    // limit of two; one bucket sees six inside at most two windows, so at least one window has 3.
-    let mut statuses = Vec::new();
-    for round in 0..2 {
-        for loc in 0..3 {
-            let mut req = test_client()
-                .post(if loc == 2 {
-                    format!("{url}?key={vk}")
-                } else {
-                    url.clone()
-                })
-                .header("content-type", "application/json")
-                .body(CHAT);
-            req = match loc {
-                0 => req.header("authorization", format!("Bearer {vk}")),
-                1 => req.header("x-api-key", vk.clone()),
-                _ => req,
-            };
-            statuses.push((round, loc, req.send().await.unwrap().status().as_u16()));
+    // Two requests per location, six in all, sent at once. Separate buckets would see two each and
+    // never trip a limit of two; one bucket that sees six inside a second (at most two of the
+    // limiter's whole-second windows) has 3 in at least one. A loaded host can spread the six
+    // over more windows than that, which proves nothing either way: such a burst is re-sent, as a
+    // fresh identity so no earlier burst's count carries into it.
+    let send_six = |vk: String| {
+        let url = url.clone();
+        async move {
+            let started = Instant::now();
+            let mut set = tokio::task::JoinSet::new();
+            for round in 0..2 {
+                for loc in 0..3 {
+                    let mut req = test_client()
+                        .post(if loc == 2 {
+                            format!("{url}?key={vk}")
+                        } else {
+                            url.clone()
+                        })
+                        .header("content-type", "application/json")
+                        .body(CHAT);
+                    req = match loc {
+                        0 => req.header("authorization", format!("Bearer {vk}")),
+                        1 => req.header("x-api-key", vk.clone()),
+                        _ => req,
+                    };
+                    set.spawn(
+                        async move { (round, loc, req.send().await.unwrap().status().as_u16()) },
+                    );
+                }
+            }
+            let mut statuses = set.join_all().await;
+            statuses.sort();
+            (statuses, started.elapsed())
+        }
+    };
+    let mut bursts = Vec::new();
+    for tenant in 0..20 {
+        let (statuses, spread) = send_six(if tenant == 0 {
+            vk.clone()
+        } else {
+            vkey(&sk, 1200 + tenant)
+        })
+        .await;
+        let tripped = statuses.iter().any(|(_, _, s)| *s == 429);
+        bursts.push((statuses, spread));
+        if tripped || spread < Duration::from_secs(1) {
+            break;
         }
     }
     assert!(
-        statuses.iter().any(|(_, _, s)| *s == 429),
-        "one identity in three locations must share one bucket: {statuses:?}"
+        bursts
+            .last()
+            .is_some_and(|(statuses, _)| statuses.iter().any(|(_, _, s)| *s == 429)),
+        "one identity in three locations must share one bucket: {bursts:?}"
     );
 
     let respelled = vk.replacen("bai_v1.1.", "bai_v1.01.", 1);
@@ -1134,17 +1164,24 @@ async fn a_deny_lands_within_the_bound_and_in_flight_streams_finish() {
     }
     assert_eq!(up.hits(), 1, "the stream is in flight before the deny");
 
+    // The two seconds, unless this host is loaded enough to make every request slow (see
+    // `stretched`): measured now, on the same requests the poll below makes.
+    let round_trip = median_round_trip(|| async { post().await.status() }).await;
+    let bound = stretched(Duration::from_secs(2), round_trip);
     put_kv(nats.port, "blackhole.1717", b"spend").await;
     let written = Instant::now();
     let mut status = 0;
-    while written.elapsed() < Duration::from_secs(2) {
+    while written.elapsed() < bound {
         status = post().await.status().as_u16();
         if status == 402 {
             break;
         }
         tokio::time::sleep(Duration::from_millis(50)).await;
     }
-    assert_eq!(status, 402, "the deny did not land within 2s");
+    assert_eq!(
+        status, 402,
+        "the deny did not land within {bound:?} (requests take {round_trip:?})"
+    );
 
     let (status, body) = in_flight.await.unwrap();
     assert_eq!(