test(shaper): run the unit tests on paused time over in-memory sockets

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

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<Step>, why: &'static str| async move {
 			let config = Config {

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<u8>);
+
+/// Every bound socket by port, which is the whole address on this network.
+static PORTS: LazyLock<Mutex<HashMap<u16, mpsc::UnboundedSender<Datagram>>>> = LazyLock::new(Default::default);
+static NEXT_PORT: AtomicU16 = AtomicU16::new(1);
+
+pub struct UdpSocket {
+	addr: SocketAddr,
+	peer: OnceLock<SocketAddr>,
+	queue: tokio::sync::Mutex<mpsc::UnboundedReceiver<Datagram>>,
+}
+
+impl UdpSocket {
+	pub async fn bind(mut addr: SocketAddr) -> io::Result<Self> {
+		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<SocketAddr> {
+		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<usize> {
+		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<usize> {
+		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<usize> {
+		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());
+	}
+}