refactor(net)!: return session drivers to callers

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`: quinn,
+It runs over anything implementing `web_transport_trait::poll::Session`: quinn,
 quiche, 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,29 @@ 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(timers, transport)`, `Server::accept(timers, transport)`, and
+`server::Handshake::ok()` return `(Session, Driver)`. The lite-only entry points
+return the same pair. `moq-net` never spawns tasks: poll or spawn the driver to
+run the protocol, process close requests, and update session statistics.
+
+```rust
+let (session, driver) = client
+    .connect(moq_tokio::runtime::Runtime::new(), transport)
+    .await?;
+tokio::spawn(driver);
+```
+
+Dropping the last session handle requests closure when the driver next runs.
+Dropping the driver cancels the session. Keep both alive while using the
+connection. `moq-tokio` and `moq-wasm` spawn the drivers for their callers.
+
+`Timers` supplies a clock and re-armable timers independently of the transport
+and executor. `runtime::Test` (feature `test-runtime`) provides virtual time;
+tests advance its clock and poll their drivers explicitly. The lite-only path
+also supports native `!Send` transports when driven on their owning thread.
+
 ## Patterns
 
 `Pattern` describes a set of paths; `Patterns` is a union reduced by

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

@@ -2,44 +2,28 @@
 //! [`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 std::{pin::Pin, task::Poll};
 
 use moq_net::{Client, Server, Session, Version, origin};
 
-use super::mock::{MockSession, create_mock_session_pair};
+use super::mock::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)>);
+/// Tokio's clock and timers for these tests.
+#[derive(Clone, Default)]
+pub struct TokioRuntime;
 
-impl<S> TokioRuntime<S> {
+impl TokioRuntime {
 	pub fn new() -> Self {
-		Self(PhantomData)
+		Self
 	}
 }
 
-impl<S> Clone for TokioRuntime<S> {
-	fn clone(&self) -> Self {
-		Self(PhantomData)
-	}
-}
-
-impl<S> Default for TokioRuntime<S> {
-	fn default() -> Self {
-		Self::new()
-	}
-}
-
-// 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> {
+impl moq_net::runtime::Timers for TokioRuntime {
 	type Timer = TokioTimer;
 
 	fn timer(&self) -> Self::Timer {
@@ -51,15 +35,6 @@ impl<S> moq_net::runtime::Timers for TokioRuntime<S> {
 	}
 }
 
-// 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>,
@@ -151,21 +126,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
+		let (session, driver) = client
 			.connect(TokioRuntime::new(), client_transport)
 			.await
-			.expect("client handshake failed")
+			.expect("client handshake failed");
+		tokio::spawn(driver);
+		session
 	};
 	let server_fut = async {
-		server
+		let (session, driver) = server
 			.accept(TokioRuntime::new(), server_transport)
 			.await
-			.expect("server handshake failed")
+			.expect("server handshake failed");
+		tokio::spawn(driver);
+		session
 	};
 	let (client_session, server_session) = tokio::join!(client_fut, server_fut);
 

rs/moq-e2ee/tests/transport.rs

@@ -12,7 +12,7 @@ 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(driver.run(TokioRuntime::new()));
 	producer
 }
 

rs/moq-ffi/src/origin.rs

@@ -206,7 +206,7 @@ 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(driver.run(moq_tokio::runtime::Runtime::new()));
 	#[cfg(target_arch = "wasm32")]
 	crate::ffi::spawn(driver.run(crate::runtime::Runtime));
 	producer

rs/moq-ffi/src/runtime.rs

@@ -1,9 +1,8 @@
-//! The browser runtime: microtask spawning and `wasmtimer`-backed timers.
+//! Browser-backed timers for MoQ drivers.
 
 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.
+/// Supplies the browser clock and timers to MoQ drivers.
 #[derive(Clone, Copy, Debug, Default)]
 pub(crate) struct Runtime;
 
@@ -15,18 +14,6 @@ impl moq_net::Timers for Runtime {
 	}
 }
 
-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 {

rs/moq-ffi/src/session.rs

@@ -291,15 +291,19 @@ impl Client {
 		}
 		.map_err(|err| MoqError::Connect(format!("{err}")))?;
 
-		// The runtime spawns the protocol machine on the microtask queue. The machine
+		// Run the driver on the microtask queue. The driver
 		// holds no session clone, so dropping the last handle still closes the
 		// transport and ends that task.
-		let session = moq_net::Client::new()
+		let (session, driver) = moq_net::Client::new()
 			.with_publisher(&publish)
 			.with_subscriber(subscribe.clone())
 			.connect(crate::runtime::Runtime, transport)
 			.await?;
 
+		crate::ffi::spawn(async move {
+			let _ = driver.await;
+		});
+
 		Ok(Arc::new(MoqSession::accepted(session, publish, subscribe)))
 	}
 }

rs/moq-hls/src/export/mod.rs

@@ -360,7 +360,7 @@ mod tests {
 	fn produce_origin() -> moq_net::origin::Producer {
 		let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into());
 		if tokio::runtime::Handle::try_current().is_ok() {
-			tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new()));
+			tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new()));
 		} else {
 			// A sync test: nothing polls the driver, and dropping it would tear
 			// the origin down, so leak it and rely on the synchronous half.

rs/moq-hls/src/export/upstream.rs

@@ -40,7 +40,7 @@ mod tests {
 	#[tokio::test]
 	async fn self_references_keep_the_catalog_broadcast_after_replacement() {
 		let (origin, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into());
-		let driver = tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new()));
+		let driver = tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new()));
 		let old = origin.create_broadcast("a/live").unwrap();
 		let source = moq_mux::Source::new(origin.consume(), "a/live");
 		let upstream = Upstream {

rs/moq-hls/src/server/mod.rs

@@ -146,7 +146,7 @@ mod tests {
 	fn produce_origin() -> moq_net::origin::Producer {
 		let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into());
 		if tokio::runtime::Handle::try_current().is_ok() {
-			tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new()));
+			tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new()));
 		} else {
 			// A sync test: nothing polls the driver, and dropping it would tear
 			// the origin down, so leak it and rely on the synchronous half.

rs/moq-mux/src/source.rs

@@ -288,7 +288,7 @@ impl BroadcastConfig for hang::catalog::BinaryConfig {
 pub(crate) fn produce_origin() -> moq_net::origin::Producer {
 	let (producer, driver) = moq_net::origin::Producer::new(moq_net::Hop::random().into());
 	if tokio::runtime::Handle::try_current().is_ok() {
-		tokio::spawn(driver.run(moq_tokio::runtime::Runtime::<()>::new()));
+		tokio::spawn(driver.run(moq_tokio::runtime::Runtime::new()));
 	} else {
 		// A sync test: nothing polls the driver, and dropping it would tear
 		// the origin down, so leak it and rely on the synchronous half.