refactor(net)!: drive time and cache cleanup explicitly

doc/lib/rs/moq-net.md

@@ -24,7 +24,7 @@ above ([hang](/lib/rs/hang)); relays and CDNs implement only this.
 - **Routes** record the relay hops and a cost, which is what the relay [cluster](/bin/relay/cluster) routes on. A hop of 0 marks the chain anonymous: `Route::is_anonymous()` is true, and that route ranks below every fully identified one.
 - **Stats** counters per broadcast and session, drained by [`moq-stats`](https://docs.rs/moq-stats).
 
-It runs over anything implementing `web_transport_trait::Session`: noq, the
+It runs over anything implementing `web_transport_trait::poll::Session`: noq, the
 browser, iroh, or qmux over TCP, Unix sockets, and
 WebSockets. [`moq-tokio`](https://docs.rs/moq-tokio) wires those up.
 
@@ -37,6 +37,38 @@ See the [Rust quick start](/lib/rs/#quick-start) and
 [`@moq/net`](/lib/js/net). The [path pattern](/concept/moq-lite#path-patterns)
 grammar lives on the concept page.
 
+## Driving sessions
+
+`Client::connect(now, transport)`, `Server::accept(now, transport)`, and
+`server::Handshake::ok()` return `(Session, Driver)`. `moq-net` never spawns
+tasks or reads the clock: the caller polls the driver and supplies the time.
+
+```rust
+let now = tokio::time::Instant::now().into_std();
+let (session, driver) = client.connect(now, transport).await?;
+tokio::spawn(moq_tokio::runtime::run(driver));
+```
+
+A custom event loop calls `driver.poll(now, waiter)` with a nondecreasing
+`moq_net::time::Instant`. After `Pending`, wait for external activity or the
+instant returned by `driver.timeout()` (`None` means no timer is armed), then
+poll again with fresh time. Tests drive the same interface with explicitly
+advanced instants.
+
+Dropping the last session handle requests closure on the next poll. Dropping
+the driver cancels the session. `moq-tokio` and `moq-wasm` drive sessions for
+their callers.
+
+`origin::Producer::new` returns a driver with the same `time::Driver`
+interface. It calls `cache::Pool::gc(now)` after each poll and folds the next
+cleanup time into its own `timeout()`. A standalone pool needs `gc(now)`
+called by its owner, at least by the returned deadline; `None` means expiry is
+disabled.
+
+Cache activity is dated lazily: reads and writes mark a group active without
+reading a clock, and the next `gc` pass stamps it with the supplied instant.
+Expiry is therefore approximate; a late `gc` extends retention.
+
 ## Patterns
 
 `Pattern` describes a set of paths; `Patterns` is a union reduced by

js/net/src/track.test.ts

@@ -120,6 +120,22 @@ test("appendDatagram delivers to a subscriber", async () => {
 	expect(got && dec.decode(got.payload)).toBe("hello");
 });
 
+test("datagram buffers drop oldest at capacity without delaying active subscribers", async () => {
+	const producer = new TrackProducer("test");
+	const slow = producer.subscribe();
+	const fast = producer.subscribe();
+	const count = 192;
+	for (let sequence = 0; sequence < count; sequence++) {
+		producer.appendDatagram(Timestamp.fromMillis(0), enc.encode("x"));
+		expect((await fast.recvDatagram())?.sequence).toBe(sequence);
+	}
+	producer.close();
+	for (let sequence = count - 64; sequence < count; sequence++) {
+		expect((await slow.recvDatagram())?.sequence).toBe(sequence);
+	}
+	expect(await slow.recvDatagram()).toBeUndefined();
+});
+
 test("insertDatagram preserves an explicit sequence", async () => {
 	const producer = new TrackProducer("test");
 	const track = producer.subscribe();

js/net/src/track.ts

@@ -27,17 +27,8 @@ const MAX_TIMEOUT_MS = 2 ** 31 - 1;
 /** Default {@link Info.maxAge} window (milliseconds) when the publisher does not set one. */
 export const DEFAULT_MAX_AGE_MS = Milli(5000);
 
-/**
- * How long (milliseconds) a datagram stays in the per-subscriber buffer before it is dropped.
- *
- * Datagrams are a best-effort send buffer, not a replay cache (unlike groups): only the last few
- * tens of milliseconds are kept, so a consumer that stalls loses stale datagrams instead of
- * replaying them. Mirrors the Rust `MAX_DATAGRAM_AGE`.
- */
-const MAX_DATAGRAM_AGE_MS = 50;
-
-/** A datagram buffered with its arrival time, so the send buffer can evict by age. */
-type BufferedDatagram = { datagram: Datagram; time: number };
+/** Maximum buffered datagrams per subscriber; mirrors Rust's bounded send buffer. */
+const MAX_DATAGRAMS = 64;
 
 /**
  * Sanity cap on a datagram payload: the QUIC DATAGRAM frame ceiling. The real limit is
@@ -306,8 +297,8 @@ class TrackState {
 	// First timestamps mutate group state rather than track state, so held groups
 	// watch this revision as well as arrivals when enforcing latency after handoff.
 	timelineChanged = new Signal(0);
-	/** Best-effort datagram channel, parallel to {@link groups}; an age-evicted send buffer per subscriber. */
-	datagrams = new Signal<BufferedDatagram[]>([]);
+	/** Best-effort datagram channel, parallel to {@link groups}; a bounded send buffer per subscriber. */
+	datagrams = new Signal<Datagram[]>([]);
 	latest?: number;
 	/**
 	 * The exclusive final boundary, stamped when the producer closes cleanly: one past the
@@ -722,12 +713,10 @@ export class Producer {
 	// Fan a datagram out to every live subscriber, dropping the oldest once the ring is full.
 	// Late subscribers do NOT replay old datagrams (best-effort, unlike the group cache).
 	#publishDatagram(datagram: Datagram): void {
-		const now = performance.now();
 		for (const sink of this.#sinks) {
 			sink.datagrams.mutate((list) => {
-				list.push({ datagram, time: now });
-				// Drop anything older than the send-buffer window.
-				while (list.length > 0 && now - list[0].time > MAX_DATAGRAM_AGE_MS) list.shift();
+				if (list.length === MAX_DATAGRAMS) list.shift();
+				list.push(datagram);
 			});
 		}
 	}
@@ -1214,13 +1203,8 @@ export class Subscriber {
 		for (;;) {
 			const datagrams = this.#state.datagrams.peek();
 
-			// Evict datagrams older than the send-buffer window (also enforced on write), so a
-			// reader that stalled skips stale datagrams instead of replaying them.
-			const cutoff = performance.now() - MAX_DATAGRAM_AGE_MS;
-			while (datagrams.length > 0 && datagrams[0].time < cutoff) datagrams.shift();
-
 			if (datagrams.length > 0) {
-				return datagrams.shift()?.datagram;
+				return datagrams.shift();
 			}
 
 			const closed = this.#state.closed.peek();

rs/moq-e2ee/Cargo.toml

@@ -25,6 +25,7 @@ zeroize = { workspace = true }
 [dev-dependencies]
 criterion = { workspace = true }
 hex = { workspace = true }
+moq-net = { workspace = true, features = ["test-runtime"] }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] }
 web-transport-trait = { workspace = true }

rs/moq-e2ee/tests/support/harness.rs

@@ -2,90 +2,19 @@
 //! [`moq_net::Session`]s over the in-memory mock transport.
 //!
 //! The harness runs the full MoQ handshake (Client::connect + Server::accept)
-//! over a [`MockSession`] pair and spawns both protocol drivers, giving tests
+//! over a [`super::mock::MockSession`] pair and spawns both protocol drivers, giving tests
 //! two live sessions ready for pub/sub without any real QUIC or network I/O.
 
 #![allow(dead_code)]
 
-use std::{marker::PhantomData, pin::Pin, task::Poll};
-
 use moq_net::{Client, Server, Session, Version, origin};
 
-use super::mock::{MockSession, create_mock_session_pair};
-
-/// A tokio-backed [`moq_net::runtime::Runtime`] for these tests: machines are
-/// spawned onto tokio and timers are tokio sleeps, so `tokio::time::pause` keeps
-/// working. A copy of the crate's own `runtime::tokio_test` module, which is
-/// `cfg(test)` and therefore invisible to integration tests.
-pub struct TokioRuntime<S = MockSession>(PhantomData<fn(S)>);
-
-impl<S> TokioRuntime<S> {
-	pub fn new() -> Self {
-		Self(PhantomData)
-	}
-}
-
-impl<S> Clone for TokioRuntime<S> {
-	fn clone(&self) -> Self {
-		Self(PhantomData)
-	}
-}
-
-impl<S> Default for TokioRuntime<S> {
-	fn default() -> Self {
-		Self::new()
-	}
-}
+use super::mock::create_mock_session_pair;
 
-// Unbounded: timers don't involve the transport, so origin drivers can borrow
-// a transportless handle (`TokioRuntime::<()>::new()`).
-impl<S> moq_net::runtime::Timers for TokioRuntime<S> {
-	type Timer = TokioTimer;
+pub use moq_net::time::test::run;
 
-	fn timer(&self) -> Self::Timer {
-		TokioTimer { at: None, sleep: None }
-	}
-
-	fn now(&self) -> moq_net::runtime::Instant {
-		tokio::time::Instant::now().into_std()
-	}
-}
-
-// Boxable: tokio work-steals, so the machine must be `Send`.
-impl<S: moq_net::transport::poll::Boxable> moq_net::runtime::Runtime for TokioRuntime<S> {
-	type Transport = S;
-
-	fn spawn(&self, machine: moq_net::runtime::Machine<Self>) {
-		tokio::spawn(machine);
-	}
-}
-
-/// The [`moq_net::runtime::Timer`] handed out by [`TokioRuntime`].
-pub struct TokioTimer {
-	at: Option<moq_net::runtime::Instant>,
-	// Allocated on the first poll after arming, then re-armed in place via
-	// `Sleep::reset`; construction panics without a live tokio time driver.
-	sleep: Option<Pin<Box<tokio::time::Sleep>>>,
-}
-
-impl moq_net::runtime::Timer for TokioTimer {
-	fn set(&mut self, at: Option<moq_net::runtime::Instant>) {
-		self.at = at;
-		if let (Some(at), Some(sleep)) = (at, &mut self.sleep) {
-			sleep.as_mut().reset(tokio::time::Instant::from_std(at));
-		}
-	}
-
-	fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> {
-		let Some(at) = self.at else { return Poll::Pending };
-		let sleep = self
-			.sleep
-			.get_or_insert_with(|| Box::pin(tokio::time::sleep_until(tokio::time::Instant::from_std(at))));
-		if sleep.is_elapsed() {
-			return Poll::Ready(());
-		}
-		waiter.poll_future(sleep.as_mut())
-	}
+pub fn now() -> moq_net::time::Instant {
+	tokio::time::Instant::now().into_std()
 }
 
 /// Options for [`connect_mock`].
@@ -151,21 +80,25 @@ pub async fn connect_mock(opts: MockConnectOptions) -> MockPair {
 		server = server.with_subscriber(subscribe);
 	}
 
-	// Run both handshakes concurrently; the runtime spawns each side's machine
+	// Run both handshakes concurrently and spawn each side's driver
 	// the moment its handshake resolves: on draft-17+ the server's accept blocks
 	// on the client's SETUP, which only reaches the wire once the client's
 	// machine is polled (and vice versa for the server's own SETUP).
 	let client_fut = async {
-		client
-			.connect(TokioRuntime::new(), client_transport)
+		let (session, driver) = client
+			.connect(now(), client_transport)
 			.await
-			.expect("client handshake failed")
+			.expect("client handshake failed");
+		tokio::spawn(run(driver));
+		session
 	};
 	let server_fut = async {
-		server
-			.accept(TokioRuntime::new(), server_transport)
+		let (session, driver) = server
+			.accept(now(), server_transport)
 			.await
-			.expect("server handshake failed")
+			.expect("server handshake failed");
+		tokio::spawn(run(driver));
+		session
 	};
 	let (client_session, server_session) = tokio::join!(client_fut, server_fut);
 

rs/moq-e2ee/tests/transport.rs

@@ -6,13 +6,13 @@ use std::time::Duration;
 
 use moq_e2ee::{Credential, PROFILE};
 use moq_net::{Hop, Timestamp, Version};
-use support::harness::{MockConnectOptions, MockPair, TokioRuntime, connect_mock};
+use support::harness::{MockConnectOptions, MockPair, connect_mock};
 
 const TEST_TIMEOUT: Duration = Duration::from_secs(10);
 
 fn produce_origin(hop: u64) -> moq_net::origin::Producer {
 	let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap()));
-	tokio::spawn(driver.run(TokioRuntime::<()>::new()));
+	tokio::spawn(support::harness::run(driver));
 	producer
 }
 

rs/moq-ffi/src/origin.rs

@@ -207,9 +207,9 @@ impl MoqOriginProducer {
 pub(crate) fn spawn(config: moq_net::origin::Config) -> moq_net::origin::Producer {
 	let (producer, driver) = moq_net::origin::Producer::new(config);
 	#[cfg(not(target_arch = "wasm32"))]
-	crate::ffi::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new()));
+	crate::ffi::spawn(moq_tokio::runtime::run(driver));
 	#[cfg(target_arch = "wasm32")]
-	crate::ffi::spawn(driver.run(crate::runtime::Runtime));
+	crate::ffi::spawn(crate::runtime::run(driver));
 	producer
 }
 

rs/moq-ffi/src/runtime.rs

@@ -1,58 +1,28 @@
-//! The browser runtime: microtask spawning and `wasmtimer`-backed timers.
-
-use std::{pin::Pin, task::Poll};
-
-/// The [`moq_net::Runtime`] for the browser: machines run on the microtask
-/// queue via `web_async::spawn`, timers are `setTimeout`-backed sleeps.
-#[derive(Clone, Copy, Debug, Default)]
-pub(crate) struct Runtime;
-
-impl moq_net::Timers for Runtime {
-	type Timer = Timer;
-
-	fn timer(&self) -> Self::Timer {
-		Timer { at: None, sleep: None }
-	}
-}
-
-impl moq_net::Runtime for Runtime {
-	type Transport = web_transport_wasm::Session;
-
-	fn spawn(&self, machine: moq_net::runtime::Machine<Self>) {
-		web_async::spawn(async move {
-			// The session surfaces the result through `closed()`; nothing to do
-			// with it here.
-			let _ = machine.await;
-		});
-	}
-}
-
-/// A `web_async::time::Sleep` driven through the [`moq_net::runtime::Timer`]
-/// contract.
-pub(crate) struct Timer {
-	at: Option<moq_net::runtime::Instant>,
-	// Allocated on the first poll after arming, then re-armed in place.
-	sleep: Option<Pin<Box<web_async::time::Sleep>>>,
-}
-
-impl moq_net::runtime::Timer for Timer {
-	fn set(&mut self, at: Option<moq_net::runtime::Instant>) {
-		self.at = at;
-		// Reuse the allocation when there is one; `reset` also clears the
-		// elapsed state.
-		if let (Some(at), Some(sleep)) = (at, &mut self.sleep) {
-			sleep.as_mut().reset(at);
-		}
-	}
-
-	fn poll(&mut self, waiter: &moq_net::kio::Waiter) -> Poll<()> {
-		let Some(at) = self.at else { return Poll::Pending };
-		let sleep = self
-			.sleep
-			.get_or_insert_with(|| Box::pin(web_async::time::sleep_until(at)));
-		if sleep.is_elapsed() {
-			return Poll::Ready(());
+//! Run MoQ drivers with the runtime clock and timer.
+
+use std::task::Poll;
+
+/// Run an explicit-time driver on the runtime's clock and timer.
+pub async fn run<D: moq_net::time::Driver>(mut driver: D) -> D::Output {
+	let mut timer = None;
+	moq_net::kio::wait(|waiter| {
+		loop {
+			let now = web_async::time::Instant::now();
+			if let Poll::Ready(result) = driver.poll(now, waiter) {
+				return Poll::Ready(result);
+			}
+			let Some(at) = driver.timeout() else {
+				timer = None;
+				return Poll::Pending;
+			};
+			let sleep = timer.get_or_insert_with(|| Box::pin(web_async::time::sleep_until(at)));
+			if sleep.deadline() != at {
+				sleep.as_mut().reset(at);
+			}
+			if waiter.poll_future(sleep.as_mut()).is_pending() {
+				return Poll::Pending;
+			}
 		}
-		waiter.poll_future(sleep.as_mut())
-	}
+	})
+	.await
 }