From 119219b601cb922b6b1ccc90c963e2c28f9a84cd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 15:29:33 -0700 Subject: [PATCH 1/2] chore(quest): claim shaper-virtual-time Co-Authored-By: Claude Opus 5.5 From 3fad44dac6a546e072936b360a36733463a8ea6e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 15:52:00 -0700 Subject: [PATCH 2/2] test(shaper): run the unit tests on paused time over in-memory sockets The socket tests ran on the wall clock against real UDP, so under load a datagram could sit in the kernel past a 4ms burst window, a 200ms read deadline, or a step boundary. Under test, the shaper's sockets are now an in-memory stand-in and every async test pauses the clock: delivery stays inside the runtime, so the clock only jumps once nothing is in flight. tests/cli.rs still covers the binary over real sockets. Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/README.md | 1 - quest/m1/test-flakes-2/shaper-virtual-time.md | 18 ---- rs/moq-shaper/src/lib.rs | 49 +++++----- rs/moq-shaper/src/mem.rs | 93 +++++++++++++++++++ 4 files changed, 120 insertions(+), 41 deletions(-) delete mode 100644 quest/m1/test-flakes-2/shaper-virtual-time.md create mode 100644 rs/moq-shaper/src/mem.rs diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index c48c60452b..3ce5b15530 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -34,7 +34,6 @@ Public API: none. Wire: none. - [Subscription cut by disconnect](/quest/m1/test-flakes-2/subscription-cut.md) - a publisher disconnect never ends a subscription clean - [Broadcast race](/quest/m1/test-flakes-2/broadcast-race.md) - the QUIC-wins race test binds no shared port and has a deterministic winner - [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - `just test media` late join stays within one GOP, or the regression is fixed -- [Shaper virtual time](/quest/m1/test-flakes-2/shaper-virtual-time.md) - `moq-shaper` tests judge seeded decisions on paused time, not on wall-clock delivery under load - [Scoped WARN capture](/quest/m1/test-flakes-2/warn-capture.md) - the drop-unfinished tests count only their own WARNs - [moq-mux debounce clock](/quest/m1/test-flakes-2/mux-debounce-clock.md) - the TS export debounce test advances on the paused clock - [js/publish audio clock](/quest/m1/test-flakes-2/publish-audio-clock.md) - the audio encoder delay test runs on mock time diff --git a/quest/m1/test-flakes-2/shaper-virtual-time.md b/quest/m1/test-flakes-2/shaper-virtual-time.md deleted file mode 100644 index 3fea8a4fb4..0000000000 --- a/quest/m1/test-flakes-2/shaper-virtual-time.md +++ /dev/null @@ -1,18 +0,0 @@ -# [S] moq-shaper tests on virtual time - -## Goal - -The `moq-shaper` unit tests pass under any host load: their verdicts come -from the seeded decisions, not from how fast the kernel delivers datagrams. -`reorder_and_jitter_overtake` and `the_seed_reproduces_the_losses` saw every -datagram lost during a slowed full-workspace run, because `round_trip` stops -at the first 200 ms of silence on a real socket. - -## Plan - -- Drive the tests on paused tokio time (or an injected clock) so delay, - jitter, and the read deadline advance deterministically. -- Watch out: paused time auto-advances while the runtime idles, which can - fire a timer before a real UDP datagram lands. If the real sockets fight - the paused clock, separate the scheduling logic from the socket I/O and test - it directly, keeping one socket smoke test with a generous deadline. diff --git a/rs/moq-shaper/src/lib.rs b/rs/moq-shaper/src/lib.rs index 4e9ca01e5a..66e359f033 100644 --- a/rs/moq-shaper/src/lib.rs +++ b/rs/moq-shaper/src/lib.rs @@ -30,14 +30,20 @@ use std::{ time::Duration, }; +#[cfg(test)] +mod mem; mod preset; pub use preset::Preset; use anyhow::Context; +#[cfg(test)] +use mem::UdpSocket; use rand::{RngExt, SeedableRng, rngs::Xoshiro256PlusPlus}; use serde::{Deserialize, Serialize}; -use tokio::{net::UdpSocket, sync::mpsc, task::JoinSet, time::Instant}; +#[cfg(not(test))] +use tokio::net::UdpSocket; +use tokio::{sync::mpsc, task::JoinSet, time::Instant}; /// How one direction of the path treats each datagram. /// @@ -995,7 +1001,7 @@ mod tests { got } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn forwards_both_ways_untouched() { let (shaper, client) = setup(1, Profile::default(), Profile::default()).await; let got = round_trip(&client, 50).await; @@ -1022,7 +1028,7 @@ mod tests { assert_eq!(start.elapsed(), Duration::ZERO); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_seed_reproduces_the_losses() { let lossy = Profile { loss: 0.3, @@ -1042,7 +1048,7 @@ mod tests { assert_eq!(first_got.len() as u64, 200 - first.stats().up.lost); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn reorder_and_jitter_overtake() { let shuffled = Profile { delay: Duration::from_millis(20), @@ -1062,7 +1068,7 @@ mod tests { assert!(stats.up.reordered > 0, "{stats}"); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_rate_limit_queues_then_drops() { // 100 datagrams of 4 bytes is 3200 bits: at 8 kbit/s they need 400ms, // and the queue only holds 100ms of it. @@ -1085,7 +1091,7 @@ mod tests { assert!(got.windows(2).all(|pair| pair[0] < pair[1]), "{got:?}"); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_rate_limit_wider_than_the_traffic_passes_everything() { let wide = Profile { rate: Some(Rate { @@ -1102,7 +1108,7 @@ mod tests { assert_eq!(shaper.verify().unwrap().up.overflowed, 0); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_burst_counts_the_datagram_itself() { // A burst of one datagram lets the first out at once and holds the second. let one = Profile { @@ -1184,7 +1190,7 @@ mod tests { assert_eq!(unapplied(&phases, &stats), ["loss"]); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn an_invalid_profile_is_refused() { let refused = |bad: Profile, why: &'static str| async move { let err = Shaper::bind(Config { @@ -1268,7 +1274,7 @@ mod tests { steps: Vec::new(), }; - #[tokio::test] + #[tokio::test(start_paused = true)] async fn only_the_gaussian_model_keeps_the_order() { let jittery = Profile { delay: Duration::from_millis(20), @@ -1387,7 +1393,7 @@ mod tests { assert!(first_counters.lost > 0 && first_counters.reordered > 0); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_gaussian_sigma_may_exceed_the_delay() { // The uniform model refuses this; a gaussian clamps its draw at zero. let wide = Profile { @@ -1457,7 +1463,7 @@ mod tests { got } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_shared_path_keeps_the_order_across_clients() { let jittery = Profile { delay: Duration::from_millis(20), @@ -1530,7 +1536,7 @@ mod tests { } } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_batch_releases_on_count_and_on_the_window() { let socket = Arc::new(UdpSocket::bind(LOCALHOST).await.unwrap()); let ms = |ms| Duration::from_millis(ms); @@ -1609,7 +1615,7 @@ mod tests { assert_eq!(tally.batched.load(Ordering::Relaxed), 0); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_batch_releases_datagrams_together() { let (shaper, client) = shaped(|config| Setup { up: batched(7, Duration::from_millis(160)), @@ -1644,7 +1650,7 @@ mod tests { ); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_window_releases_a_batch_that_never_fills() { let (shaper, client) = shaped(|config| Setup { up: batched(100, Duration::from_millis(80)), @@ -1659,7 +1665,7 @@ mod tests { assert_eq!(shaper.verify().unwrap().up.delayed, 3); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_shared_batch_fills_from_every_client() { // Three datagrams from each of two clients, into batches of six: only a // shared path fills one, and separate links wait out the window. @@ -1684,7 +1690,7 @@ mod tests { } } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_batch_needs_a_count_and_a_window() { let refused = |options: Options, why: &'static str| async move { let config = Config { @@ -1812,7 +1818,7 @@ mod tests { assert_eq!(lost(start + secs(60)), 0, "the step back never cleared"); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_step_gets_worse_part_way_through() { let ms = Duration::from_millis; let (shaper, client) = shaped(|config| Setup { @@ -1849,7 +1855,7 @@ mod tests { shaper.verify().unwrap(); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_step_the_run_never_reached_is_unapplied() { let (shaper, client) = shaped(|config| Setup { down: stepped(vec![Step { @@ -1866,8 +1872,7 @@ mod tests { assert!(format!("{err:#}").contains("the down step at 60s"), "{err:#}"); } - // Real sockets, so the wall clock: a paused one could jump past a step while a datagram is still in the kernel. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_step_no_datagram_saw_is_unapplied() { let ms = Duration::from_millis; let (shaper, client) = shaped(|config| Setup { @@ -1909,7 +1914,7 @@ mod tests { assert!(!err.contains("the up step at 200ms"), "{err}"); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_profile_no_datagram_saw_before_its_step_is_unapplied() { let ms = Duration::from_millis; let (shaper, client) = shaped(|config| Setup { @@ -1954,7 +1959,7 @@ mod tests { assert_eq!(owed(&mut link, start + Duration::from_secs(1), &[16]), [Duration::ZERO]); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn steps_that_would_be_skipped_or_do_nothing_are_refused() { let refused = |steps: Vec, why: &'static str| async move { let config = Config { diff --git a/rs/moq-shaper/src/mem.rs b/rs/moq-shaper/src/mem.rs new file mode 100644 index 0000000000..bf7e896436 --- /dev/null +++ b/rs/moq-shaper/src/mem.rs @@ -0,0 +1,93 @@ +//! An in-memory stand-in for [`tokio::net::UdpSocket`], so the unit tests run +//! on a paused clock. A paused clock jumps to the next timer whenever the +//! runtime idles, which would fire timeouts and steps while a real datagram +//! sat in the kernel; a datagram here is delivered inside the runtime, so the +//! clock only moves once nothing is in flight. `tests/cli.rs` covers real sockets. + +use std::{ + collections::HashMap, + io, + net::SocketAddr, + sync::{ + LazyLock, Mutex, OnceLock, + atomic::{AtomicU16, Ordering}, + }, +}; + +use tokio::sync::mpsc; + +type Datagram = (SocketAddr, Vec); + +/// Every bound socket by port, which is the whole address on this network. +static PORTS: LazyLock>>> = LazyLock::new(Default::default); +static NEXT_PORT: AtomicU16 = AtomicU16::new(1); + +pub struct UdpSocket { + addr: SocketAddr, + peer: OnceLock, + queue: tokio::sync::Mutex>, +} + +impl UdpSocket { + pub async fn bind(mut addr: SocketAddr) -> io::Result { + if addr.port() == 0 { + addr.set_port(NEXT_PORT.fetch_add(1, Ordering::Relaxed)); + } + let (send, queue) = mpsc::unbounded_channel(); + let mut ports = PORTS.lock().unwrap(); + if ports.contains_key(&addr.port()) { + return Err(io::ErrorKind::AddrInUse.into()); + } + ports.insert(addr.port(), send); + Ok(Self { + addr, + peer: OnceLock::new(), + queue: tokio::sync::Mutex::new(queue), + }) + } + + pub fn local_addr(&self) -> io::Result { + Ok(self.addr) + } + + pub async fn connect(&self, peer: SocketAddr) -> io::Result<()> { + self.peer.set(peer).map_err(|_| io::ErrorKind::AlreadyExists.into()) + } + + /// Like UDP, a datagram to a port nobody bound is lost without an error. + pub async fn send_to(&self, buf: &[u8], dest: SocketAddr) -> io::Result { + if let Some(port) = PORTS.lock().unwrap().get(&dest.port()) { + let _ = port.send((self.addr, buf.to_vec())); + } + Ok(buf.len()) + } + + pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { + // The port's sender lives until this socket drops, so the queue never ends. + let (from, datagram) = self.queue.lock().await.recv().await.expect("bound port"); + let size = datagram.len().min(buf.len()); + buf[..size].copy_from_slice(&datagram[..size]); + Ok((size, from)) + } + + pub async fn send(&self, buf: &[u8]) -> io::Result { + self.send_to(buf, *self.peer.get().expect("not connected")).await + } + + /// Only what the connected peer sends, as a connected UDP socket filters. + pub async fn recv(&self, buf: &mut [u8]) -> io::Result { + let peer = *self.peer.get().expect("not connected"); + loop { + let (size, from) = self.recv_from(buf).await?; + if from == peer { + return Ok(size); + } + } + } +} + +impl Drop for UdpSocket { + fn drop(&mut self) { + PORTS.lock().unwrap().remove(&self.addr.port()); + } +}