fix(net): prove native futures Send at the default recursion limit

.github/workflows/nightly.yml

@@ -163,6 +163,9 @@ jobs:
         # js/net, or the wire between them.
         # `gh test` exercises workflow and release tooling, which no pull request runs.
         recipe: ["rs doctest --workspace", "rs loom", "test drill-sensitivity", "rs uring", "test drain", "gh test", "test json-properties"]
+        include:
+          - recipe: rs send-depth
+            nightly: true
     steps:
       - name: Free disk space
         uses: jlumbroso/free-disk-space@ceedf095f4ec1a097402bc6bd80831f2e1a6fde6 # main
@@ -182,6 +185,11 @@ jobs:
             extra-trusted-public-keys = kixelated.cachix.org-1:CmFcV0lyM6KuVM2m9mih0q4SrAa0XyCsiM7GHrz3KKk=
       - name: Rust cache
         uses: ./.github/actions/rust-cache
+      - name: Install nightly Rust
+        if: matrix.nightly
+        uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+        with:
+          toolchain: nightly
       - name: Test
         run: nix develop --command just ${{ matrix.recipe }}
         env:

quest/m2/README.md

@@ -109,7 +109,6 @@ ranked just below deterministic TS legs, which it requires.
   with a delivery-delay histogram for jitter
 - [#3199](/quest/m2/3199-moq-uring-remove-sq-indirection-and-per-enter-ring-fd.md) - moq-uring: remove SQ indirection and per-enter ring fd lookup, if a profile shows the win
 - [#3200](/quest/m2/3200-moq-uring-batch-completion-wakeups-with-min-timeout.md) - moq-uring: batch completion wakeups with MIN_TIMEOUT
-- [Send depth](/quest/m2/send-depth.md) - moq-net futures prove Send at the default recursion limit, so downstream crates see no nightly lint
 - [Browser benchmarks](/quest/m2/browser-benchmarks.md) - measure JS transport, container, decode, and render costs in an identified browser
 - [Browser media report](/quest/m2/rs2ts-browser-report.md) - compare Rust-backed and handwritten @moq/net through full media workloads, supplementing the pre-adoption comparisons
 - [mTLS on tls://](/quest/m2/tls-listener-mtls.md) - a `tls://` listener can identify a cluster peer by its client certificate

rs/justfile

@@ -294,6 +294,10 @@ loom *args:
     RUSTFLAGS="--cfg loom" cargo test --locked --release -p kio --lib loom:: "$@"
     RUSTFLAGS="--cfg loom" cargo test --locked --release -p moq-net --test loom "$@"
 
+# Prove downstream futures are Send at the default recursion limit on nightly.
+send-depth:
+    bash sh/rs/send-depth.sh
+
 # Needs nightly and cargo-fuzz; CI replays the corpus on stable through
 # moq-net's own tests instead. See rs/moq-net/fuzz/README.md.
 

rs/kio/src/lock.rs

@@ -17,11 +17,31 @@ use crate::sync::{Mutex, MutexGuard};
 /// Poisoning is ignored: a panic while the lock is held must not cascade into every other
 /// handle sharing this state, or into a double panic (and an abort) from a `Drop` that locks it.
 ///
+/// State that cannot cross threads keeps the lock on its thread too.
+///
+/// ```compile_fail,E0277
+/// fn require_send(_: impl Send) {}
+/// require_send(kio::Lock::new(std::rc::Rc::new(())));
+/// ```
+///
+/// ```compile_fail,E0277
+/// fn require_sync(_: impl Sync) {}
+/// require_sync(kio::Lock::new(std::rc::Rc::new(())));
+/// ```
+///
 /// [`WaiterList`]: crate::WaiterList
 pub struct Lock<T> {
 	inner: Arc<Mutex<T>>,
 }
 
+// SAFETY: std and loom's Mutex<T> are Send + Sync when T: Send, so this is
+// exactly Arc<Mutex<T>>'s derived bound. Stopping here keeps recursive channel
+// graphs from expanding the same Arc/Mutex proof at every level.
+unsafe impl<T: Send> Send for Lock<T> {}
+
+// SAFETY: Arc<Mutex<T>> has the same T: Send bound for Sync, with either mutex.
+unsafe impl<T: Send> Sync for Lock<T> {}
+
 impl<T> Lock<T> {
 	/// Wrap a value.
 	pub fn new(value: T) -> Self {
@@ -142,6 +162,12 @@ impl<T> Default for WeakLock<T> {
 mod tests {
 	use super::*;
 
+	#[test]
+	fn send_state_does_not_have_to_be_sync() {
+		fn require_send_sync(_: impl Send + Sync) {}
+		require_send_sync(Lock::new(std::cell::Cell::new(1)));
+	}
+
 	#[test]
 	fn a_panic_under_the_lock_leaves_it_usable() {
 		let lock = Lock::new(1);

rs/moq-net/src/auth.rs

@@ -1400,15 +1400,14 @@ impl RequestGrant {
 
 	/// Run `body`, everything a request does once admitted, until it finishes or `grant`
 	/// (when its token admitted it) ends. See [`until`].
-	pub(crate) async fn guard<F: std::future::Future>(
+	pub(crate) fn guard<F: std::future::Future + kio::MaybeSend>(
 		grant: Option<&Self>,
 		body: F,
-	) -> std::result::Result<F::Output, Refusal> {
-		until(body, |waiter| match grant {
+	) -> impl std::future::Future<Output = std::result::Result<F::Output, Refusal>> + kio::MaybeSend {
+		crate::util::sendable(until(body, move |waiter| match grant {
 			Some(grant) => grant.poll_ended(waiter),
 			None => Poll::Pending,
-		})
-		.await
+		}))
 	}
 }
 

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

@@ -3531,7 +3531,14 @@ where
 		let passthrough = !fill.read().outstanding() && resume.is_none();
 		let opened = {
 			let mut opening = track.clone();
-			let mut open = std::pin::pin!(self.open_group(stream, &mut opening, &fill, resume, &group, &mut reading));
+			let mut open = std::pin::pin!(crate::util::sendable(self.open_group(
+				stream,
+				&mut opening,
+				&fill,
+				resume,
+				&group,
+				&mut reading
+			)));
 			kio::wait(|waiter| {
 				if let Poll::Ready(err) = track.poll_closed(waiter) {
 					return Poll::Ready(Err(err));
@@ -4086,7 +4093,14 @@ where
 		// track does not close a group producer, since those lifecycles are independent.
 		let res = {
 			let mut serving = track.clone();
-			let mut serve = std::pin::pin!(self.run_fill(stream, &mut serving, timescale, joining, largest, resume));
+			let mut serve = std::pin::pin!(crate::util::sendable(self.run_fill(
+				stream,
+				&mut serving,
+				timescale,
+				joining,
+				largest,
+				resume
+			)));
 			kio::wait(|waiter| {
 				if let Poll::Ready(err) = track.poll_closed(waiter) {
 					return Poll::Ready(Err(err));
@@ -4717,8 +4731,13 @@ where
 		let closed = producer.clone();
 		let mut producer = crate::recv::Group::new(producer);
 		let res = {
-			let mut receive =
-				std::pin::pin!(self.recv_group_fetch_objects(stream, &mut producer, start, end, timescale));
+			let mut receive = std::pin::pin!(crate::util::sendable(self.recv_group_fetch_objects(
+				stream,
+				&mut producer,
+				start,
+				end,
+				timescale
+			)));
 			kio::wait(|waiter| {
 				if let Poll::Ready(res) = waiter.poll_future(receive.as_mut()) {
 					return Poll::Ready(res);

rs/moq-net/src/lib.rs

@@ -74,8 +74,6 @@
 //! instant.
 
 #![warn(missing_docs)]
-// Proving receive futures are Send traverses the shared origin/broadcast/track types.
-#![recursion_limit = "256"]
 // The browser transport is `!Send`, so on wasm the shared state behind these `Arc`s is
 // too and clippy suggests `Rc`. The same code is genuinely cross-thread on native, so
 // `Arc` stays and the lint is unactionable here.

rs/moq-net/src/model/origin.rs

@@ -21,7 +21,7 @@ use crate::{
 	coding::{BoundsExceeded, Decode, DecodeError, Decoder, Encode, EncodeError, Encoder},
 	path::Segment,
 	time::{Clock, Instant},
-	util::{Keepalive, TaskSet, Tasks, TasksWeak},
+	util::{Keepalive, SendCell, TaskSet, Tasks, TasksWeak},
 };
 
 /// One relay's identity in a broadcast's hop chain: a 62-bit varint on the wire.
@@ -1608,7 +1608,7 @@ pub struct Producer {
 
 	// The origin's shared state: the route table, announce cursors, and the
 	// remotely-served fronts. Shared with every derived consumer.
-	shared: kio::Shared<OriginState>,
+	shared: SendCell<kio::Shared<OriginState>>,
 
 	// The cache pool inherited by broadcasts created under this origin (sessions
 	// mint their remote broadcasts with it). Unbounded by default.
@@ -1650,7 +1650,7 @@ impl Producer {
 		let (tasks, set) = TaskSet::new();
 		let scope = OriginScope::default();
 		let timers = Clock::default();
-		let shared = kio::Shared::new(OriginState::new(config.update_hold, timers.clone()));
+		let shared = SendCell::new(kio::Shared::new(OriginState::new(config.update_hold, timers.clone())));
 		let pool = config.pool.clone();
 		let producer = Self {
 			hop: config.hop,
@@ -1760,7 +1760,10 @@ impl Producer {
 			hop,
 			scope: OriginScope::empty(),
 			root: PathOwned::default(),
-			shared: kio::Shared::new(OriginState::new(DEFAULT_UPDATE_HOLD, Clock::default())),
+			shared: SendCell::new(kio::Shared::new(OriginState::new(
+				DEFAULT_UPDATE_HOLD,
+				Clock::default(),
+			))),
 			pool: cache::Pool::default(),
 			cache_duration: Duration::MAX,
 			stats: stats::Session::default(),
@@ -2066,7 +2069,7 @@ impl Producer {
 /// and [`Announcer`], which is the same advertisement re-issued from a broadcast.
 struct Announcing {
 	hop: Hop,
-	shared: kio::Shared<OriginState>,
+	shared: SendCell<kio::Shared<OriginState>>,
 	/// The absolute advertised prefix, which also keys the ingress announce counters.
 	requested: PathOwned,
 	/// The prefix inserted into the table, with its [`prefix_claim`]. Pattern
@@ -2215,7 +2218,7 @@ impl Announcer {
 /// from their peers.
 #[must_use = "dropping an announcement retracts the route"]
 pub(crate) struct AnnounceProducer {
-	shared: kio::Shared<OriginState>,
+	shared: SendCell<kio::Shared<OriginState>>,
 	/// The table entries this advertisement created, by prefix and id. A prefix
 	/// remains unchanged; pattern scopes only filter its visibility and requests.
 	entries: Vec<(PathOwned, u64)>,
@@ -2389,7 +2392,7 @@ struct DriverState {
 	set: TaskSet,
 	/// The route table, announce cursors, and the remotely-served fronts, for
 	/// ending everything on drop.
-	shared: kio::Shared<OriginState>,
+	shared: SendCell<kio::Shared<OriginState>>,
 	/// Cached completion so a poll after `Ready` doesn't re-poll the drained set.
 	done: bool,
 }
@@ -2498,7 +2501,7 @@ impl Drop for DriverState {
 /// Everything [`run_front`] owns, queued by [`Consumer::request_broadcast`].
 struct FrontTask {
 	/// The route table the front selects from.
-	shared: kio::Shared<OriginState>,
+	shared: SendCell<kio::Shared<OriginState>>,
 	/// The broadcast the front serves.
 	broadcast: broadcast::Producer,
 	/// Absolute path of the front.
@@ -3506,7 +3509,7 @@ impl Selected {
 /// The lifetime of path requests while this peer still selects the held instance.
 #[cfg_attr(not(feature = "async"), allow(dead_code))]
 pub(crate) struct Selection {
-	shared: kio::Shared<OriginState>,
+	shared: SendCell<kio::Shared<OriginState>>,
 	path: PathOwned,
 	id: u64,
 	signal: kio::Consumer<Selected>,
@@ -3549,7 +3552,7 @@ impl Drop for Selection {
 /// Dropping it unregisters, which takes the table lock: never drop one while
 /// holding it.
 struct Watch {
-	shared: kio::Shared<OriginState>,
+	shared: SendCell<kio::Shared<OriginState>>,
 	path: PathOwned,
 	id: u64,
 	signal: kio::Consumer<Watched>,
@@ -4069,7 +4072,7 @@ impl OriginState {
 	}
 
 	/// Register a [`Watch`] on the routes covering `path`.
-	fn watch(&mut self, shared: &kio::Shared<OriginState>, path: &Path) -> Watch {
+	fn watch(&mut self, shared: &SendCell<kio::Shared<OriginState>>, path: &Path) -> Watch {
 		let id = self.next_watch;
 		self.next_watch += 1;
 		let signal = self.routes.add_watch(path, id);
@@ -4798,7 +4801,7 @@ pub struct Consumer {
 
 	// The origin's shared state: the route table, announce cursors, and the
 	// remotely-served fronts.
-	shared: kio::Shared<OriginState>,
+	shared: SendCell<kio::Shared<OriginState>>,
 
 	// Egress stats context. Broadcasts handed out through this consumer (and any
 	// handle derived from them) are attributed to it (reads counted on the
@@ -5373,7 +5376,7 @@ pub struct AnnounceConsumer {
 	// One registration per table cursor feeding `state`: more than one when the
 	// consumer reads through a mount.
 	ids: Vec<ConsumerId>,
-	shared: kio::Shared<OriginState>,
+	shared: SendCell<kio::Shared<OriginState>>,
 	root: PathOwned,
 
 	// Pending updates queued for this cursor. Coalesced so a slow consumer
@@ -5413,7 +5416,7 @@ impl AnnounceConsumer {
 		cursors: Vec<TableCursor>,
 		state: kio::Producer<OriginConsumerState>,
 		stats: stats::Session,
-		shared: &kio::Shared<OriginState>,
+		shared: &SendCell<kio::Shared<OriginState>>,
 		timers: Clock,
 	) -> Self {
 		let mut ids = Vec::with_capacity(cursors.len());

rs/moq-net/src/util.rs

@@ -7,6 +7,14 @@
 
 use std::{collections::VecDeque, future::Future, pin::Pin, task::Poll};
 
+mod cell;
+pub(crate) use cell::SendCell;
+
+/// Prove a future's native Send bound here rather than inside every caller's future.
+pub(crate) fn sendable<F: Future + kio::MaybeSend>(future: F) -> impl Future<Output = F::Output> + kio::MaybeSend {
+	future
+}
+
 #[cfg(not(target_family = "wasm"))]
 pub(crate) type MaybeSendBox<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
 #[cfg(target_family = "wasm")]
@@ -46,11 +54,15 @@ fn future_task(mut task: MaybeSendBox<'static, ()>) -> MaybeSendTask {
 /// The building block for racing "watchdog" futures whose clean completion should
 /// not end the race (only their failure should).
 #[cfg(feature = "async")]
-pub(crate) async fn err_only<E>(fut: impl Future<Output = Result<(), E>>) -> E {
-	match fut.await {
-		Err(err) => err,
-		Ok(()) => std::future::pending().await,
-	}
+pub(crate) fn err_only<E>(
+	fut: impl Future<Output = Result<(), E>> + kio::MaybeSend,
+) -> impl Future<Output = E> + kio::MaybeSend {
+	sendable(async move {
+		if let Err(err) = fut.await {
+			return err;
+		}
+		std::future::pending().await
+	})
 }
 
 /// Cloneable handle for submitting futures to a driver-owned [`TaskSet`].