refactor(net)!: return the next deadline from driver polls

doc/lib/rs/moq-net.md

@@ -42,26 +42,27 @@ grammar lives on the concept page.
 `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.
+`moq_net::time::run` does that on tokio or in the browser.
 
 ```rust
 let now = tokio::time::Instant::now().into_std();
 let (session, driver) = client.connect(now, transport).await?;
-tokio::spawn(moq_tokio::runtime::run(driver));
+tokio::spawn(moq_net::time::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.
+`moq_net::time::Instant`. `Ok(Some(at))` asks to be polled again by `at` or
+on external activity, `Ok(None)` only on external activity, and `Err` is the
+terminal error (`Error::Closed` for a clean finish): stop polling. 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)`
+cleanup time into its returned deadline. A standalone pool needs `gc(now)`
 called by its owner, at least by the returned deadline; `None` means expiry is
 disabled.
 

rs/kio/src/lock.rs

@@ -95,6 +95,12 @@ impl<T> Clone for Lock<T> {
 	}
 }
 
+impl<T: Default> Default for Lock<T> {
+	fn default() -> Self {
+		Self::new(T::default())
+	}
+}
+
 impl<T: fmt::Debug> fmt::Debug for Lock<T> {
 	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
 		match self.inner.try_lock() {

rs/moq-e2ee/Cargo.toml

@@ -25,7 +25,6 @@ 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

@@ -11,7 +11,7 @@ use moq_net::{Client, Server, Session, Version, origin};
 
 use super::mock::create_mock_session_pair;
 
-pub use moq_net::time::test::run;
+pub use moq_net::time::run;
 
 pub fn now() -> moq_net::time::Instant {
 	tokio::time::Instant::now().into_std()

rs/moq-ffi/src/lib.rs

@@ -26,8 +26,6 @@ mod log;
 pub mod media;
 pub mod origin;
 pub mod producer;
-#[cfg(target_arch = "wasm32")]
-mod runtime;
 #[cfg(not(target_arch = "wasm32"))]
 pub mod server;
 pub mod session;

rs/moq-ffi/src/origin.rs

@@ -206,10 +206,9 @@ impl MoqOriginProducer {
 /// Build an origin producer, spawning its driver on the FFI runtime.
 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(moq_tokio::runtime::run(driver));
-	#[cfg(target_arch = "wasm32")]
-	crate::ffi::spawn(crate::runtime::run(driver));
+	crate::ffi::spawn(async move {
+		moq_net::time::run(driver).await;
+	});
 	producer
 }
 

rs/moq-ffi/src/session.rs

@@ -301,7 +301,7 @@ impl Client {
 			.await?;
 
 		crate::ffi::spawn(async move {
-			let _ = crate::runtime::run(driver).await;
+			moq_net::time::run(driver).await;
 		});
 
 		Ok(Arc::new(MoqSession::accepted(session, publish, subscribe)))

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

@@ -362,7 +362,7 @@ mod tests {
 	fn produce_origin() -> moq_net::origin::Producer {
 		let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default());
 		if tokio::runtime::Handle::try_current().is_ok() {
-			tokio::spawn(moq_tokio::runtime::run(driver));
+			tokio::spawn(moq_net::time::run(driver));
 		} 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::origin::Config::default());
-		let driver = tokio::spawn(moq_tokio::runtime::run(driver));
+		let driver = tokio::spawn(moq_net::time::run(driver));
 		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::origin::Config::default());
 		if tokio::runtime::Handle::try_current().is_ok() {
-			tokio::spawn(moq_tokio::runtime::run(driver));
+			tokio::spawn(moq_net::time::run(driver));
 		} 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/Cargo.toml

@@ -42,5 +42,4 @@ webm-iterable = "0.7"
 
 [dev-dependencies]
 futures = { workspace = true }
-moq-tokio = { workspace = true }
 tokio = { workspace = true, features = ["test-util"] }

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::origin::Config::default());
 	if tokio::runtime::Handle::try_current().is_ok() {
-		tokio::spawn(moq_tokio::runtime::run(driver));
+		tokio::spawn(moq_net::time::run(driver));
 	} 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-net/Cargo.toml

@@ -16,10 +16,6 @@ categories = ["multimedia", "network-programming", "web-programming"]
 ignored = ["getrandom"]
 
 [features]
-# The deterministic runtime::Test, for tests in this crate and downstream ones.
-# A feature rather than cfg(test) because integration tests and other crates'
-# tests compile against the normal library build.
-test-runtime = []
 # Exposes the wire codecs to the `fuzz/` harness through the hidden `fuzz` module.
 # Off by default and hidden from the docs; nothing depends on it but that harness.
 fuzz = []
@@ -41,9 +37,6 @@ web-transport-trait = { workspace = true }
 
 [dev-dependencies]
 criterion = { workspace = true }
-# Enable the test runtime for our own unit and integration tests. A crate may
-# dev-depend on itself to turn features on for its test builds only.
-moq-net = { path = ".", features = ["test-runtime"] }
 serde_json = { workspace = true }
 # test-util (tokio::time::pause/advance) is test-only and is NOT supported on
 # wasm, so it must not leak into the normal dependency feature set.

rs/moq-net/src/client.rs

@@ -682,7 +682,7 @@ mod tests {
 			.connect(tokio::time::Instant::now().into_std(), fake.clone())
 			.await
 			.unwrap();
-		tokio::spawn(crate::time::test::run(driver));
+		tokio::spawn(crate::time::run(driver));
 
 		// Verify the client setup was encoded using Draft14 framing (ALPN_LITE fallback path).
 		let mut setup_bytes = Bytes::from(fake.control_writes());
@@ -759,7 +759,7 @@ mod tests {
 		assert_eq!(session.version(), Version::Lite(lite::Version::Lite04));
 
 		// Construction leaves the driver idle until the caller polls it.
-		assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_pending());
+		assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_ok());
 
 		// The caller drops their only session clone; the machine observes the
 		// last handle going away and closes the transport.
@@ -982,7 +982,7 @@ mod tests {
 
 		let runtime = crate::runtime::Test::new();
 		let (session, mut driver) = futures::executor::block_on(client.connect_lite(runtime.now(), local)).unwrap();
-		assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_pending());
+		assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_ok());
 
 		fn assert_send_sync<T: Send + Sync>(_: &T) {}
 		assert_send_sync(&session);
@@ -1009,7 +1009,7 @@ mod tests {
 		let runtime = crate::runtime::Test::new();
 		let (session, mut driver) = futures::executor::block_on(server.accept_lite(runtime.now(), local)).unwrap();
 		assert_eq!(session.version(), Version::Lite(lite::Version::Lite04));
-		assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_pending());
+		assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_ok());
 
 		drop(session);
 		assert!(fake.state.close_events.lock().unwrap().is_empty());
@@ -1048,7 +1048,7 @@ mod tests {
 			.connect(tokio::time::Instant::now().into_std(), fake.clone())
 			.await
 			.unwrap();
-		tokio::spawn(crate::time::test::run(driver));
+		tokio::spawn(crate::time::run(driver));
 
 		// The construction-time snapshot, before the machine sampled anything.
 		assert_eq!(
@@ -1081,7 +1081,7 @@ mod tests {
 			.connect(tokio::time::Instant::now().into_std(), fake.clone())
 			.await
 			.unwrap();
-		tokio::spawn(crate::time::test::run(driver));
+		tokio::spawn(crate::time::run(driver));
 		assert!(
 			session.send_bandwidth().is_none(),
 			"no send-rate estimate, so nothing samples on its own"
@@ -1112,7 +1112,7 @@ mod tests {
 			.connect(tokio::time::Instant::now().into_std(), fake.clone())
 			.await
 			.unwrap();
-		tokio::spawn(crate::time::test::run(driver));
+		tokio::spawn(crate::time::run(driver));
 
 		let mut bandwidth = session.send_bandwidth().expect("backend reports an estimate");
 		assert_eq!(

rs/moq-net/src/driver.rs

@@ -8,9 +8,10 @@ use crate::time::{Clock, Instant};
 /// Drives a session with caller-supplied time.
 ///
 /// Returned by [`crate::Client::connect`] and [`crate::Server::accept`]. Call
-/// [`poll`](Self::poll) when external activity wakes the waiter or when
-/// [`timeout`](Self::timeout) is reached. Supply nondecreasing instants.
-/// Completion is cached, so subsequent polls return the same result.
+/// [`poll`](Self::poll) when external activity wakes the waiter or when the
+/// returned deadline is reached, supplying nondecreasing instants; see
+/// [`crate::time::Driver`] for the contract. Completion is cached, so
+/// subsequent polls return the same error.
 ///
 /// It holds no session handle: dropping the last [`crate::Session`] requests
 /// closure on the next poll. Dropping the driver cancels the session, and
@@ -59,9 +60,13 @@ impl<S: crate::transport::poll::Session> Driver<S> {
 	/// Process ready work at `now`, registering for external activity.
 	///
 	/// Panics if `now` is earlier than the previous poll or construction time.
-	pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Poll<Result<(), Error>> {
+	pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result<Option<Instant>, Error> {
 		self.clock.advance(now);
-		self.state.poll(waiter)
+		match self.state.poll(waiter) {
+			Poll::Ready(Ok(())) => Err(Error::Closed),
+			Poll::Ready(Err(err)) => Err(err),
+			Poll::Pending => Ok(self.clock.timeout()),
+		}
 	}
 }
 
@@ -103,20 +108,9 @@ impl<S: crate::transport::poll::Session> State<S> {
 }
 
 impl<S: crate::transport::poll::Session> crate::time::Driver for Driver<S> {
-	type Output = Result<(), Error>;
-	fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Poll<Self::Output> {
+	fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result<Option<Instant>, Error> {
 		self.poll(now, waiter)
 	}
-	fn timeout(&self) -> Option<Instant> {
-		self.timeout()
-	}
-}
-
-impl<S: crate::transport::poll::Session> Driver<S> {
-	/// The next instant the caller must poll at, or none while no timer is armed.
-	pub fn timeout(&self) -> Option<Instant> {
-		self.clock.timeout()
-	}
 }
 
 impl<S: crate::transport::poll::Session> std::fmt::Debug for Driver<S> {

rs/moq-net/src/ietf/subscriber.rs

@@ -16,7 +16,7 @@ use crate::{
 
 use super::{Message, Version, cluster, error::request, peer};
 
-use web_async::Lock;
+use kio::Lock;
 
 const TRACK_ALIAS_TIMEOUT: Duration = Duration::from_secs(1);
 

rs/moq-net/src/lib.rs

@@ -47,21 +47,22 @@
 //! last producer signals consumers that no more updates are coming.
 //!
 //! ## Driving and time
-//! This library never spawns tasks or schedules runtime timers. [`Client::connect`]
-//! and [`Server::accept`] take an initial [`time::Instant`] and return
+//! This library never spawns tasks or reads the clock. [`Client::connect`] and
+//! [`Server::accept`] take an initial [`time::Instant`] and return
 //! `(Session, Driver)`. Poll the [`Driver`] with the current instant and a
-//! [`kio::Waiter`], then wake on external activity or at [`Driver::timeout`].
-//! The last [`Session`] drop requests closure; dropping the driver cancels it.
+//! [`kio::Waiter`], then wake on external activity or at the deadline it
+//! returns; [`time::run`] does exactly that on tokio or the browser. The last
+//! [`Session`] drop requests closure; dropping the driver cancels it.
 //!
 //! [`origin::Producer::new`] also returns a producer and driver. Its driver runs
 //! route changes, serving, linger, teardown, and the origin's cache expiration.
 //! Standalone caches expose [`cache::Pool::gc`]. Frame read/write methods
 //! clear their expiration timestamp for the next cleanup pass. Datagrams use a bounded
 //! FIFO; model read/write APIs take no wall-clock time.
 //!
-//! Both drivers implement [`time::Driver`]. `moq-tokio` and `moq-wasm`
-//! supply runtime adapters, and `moq-uring` can drive thread-local transports.
-//! Tests can advance time simply by supplying a later instant.
+//! Both drivers implement [`time::Driver`]. `moq-uring` drives thread-local
+//! transports on its own timer heap. Tests advance time by supplying a later
+//! instant.
 
 #![warn(missing_docs)]
 // The browser transport is `!Send`, so on wasm the shared state behind these `Arc`s is

rs/moq-net/src/lite/subscriber.rs

@@ -16,7 +16,7 @@ use crate::{
 
 use super::Version;
 
-use web_async::Lock;
+use kio::Lock;
 
 pub(super) struct SubscriberConfig<S: crate::transport::poll::Session> {
 	pub runtime: crate::time::Clock,