diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 982ac279cd..c592098537 100644 --- a/.github/workflows/nightly.yml +++ b/.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: diff --git a/quest/m2/README.md b/quest/m2/README.md index 2db979d6b1..0b62ae1d8a 100644 --- a/quest/m2/README.md +++ b/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 diff --git a/quest/m2/send-depth.md b/quest/m2/send-depth.md deleted file mode 100644 index 582d4e2662..0000000000 --- a/quest/m2/send-depth.md +++ /dev/null @@ -1,33 +0,0 @@ -# [M] moq-net futures prove Send without a raised recursion limit - -## Goal - -A downstream crate that requires `Send` on a future awaiting -`announce::Consumer::next()`, or a track or group read, compiles without -nightly's `recursion_depth_exceeding_limit` future-incompat warning and -without raising its own `recursion_limit`. - -## Plan - -moq-net itself needs `#![recursion_limit = "256"]` (#4746), which downstream -crates do not inherit. Each `kio::Shared` adds about five levels to the -auto-trait chain (Shared, Lock, Arc, Mutex, State), nested from origin to -broadcast to track to group. The recursive `RouteNode` is a cycle the -compiler closes, so flattening the route tree likely does not help. - -Decided (2026-10-04), in order: - -1. Measure: an integration test crate in `rs/moq-net/tests`, compiled at the - default limit, proves these futures are `Send`, and runs on nightly in CI - so the lint shows up. Settle whether it runs in `nightly.yml`. -2. Shorten every chain in kio with `unsafe impl Send for Lock`, - the same condition the compiler derives. -3. If that is not enough, a zero-cost `SendCell` whose only constructor - requires the bound, gated for wasm. - -Done when the `recursion_limit` line is deleted. Revisit the milestone if the -lint becomes a hard error on stable. - -## Closes - -- [#4648](https://github.com/moq-dev/moq/issues/4648) - close this issue when the quest finishes diff --git a/rs/justfile b/rs/justfile index 5b9a2d7b35..30ca1efda4 100644 --- a/rs/justfile +++ b/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. diff --git a/rs/kio/src/lock.rs b/rs/kio/src/lock.rs index 23662d2bcb..a814a47f8b 100644 --- a/rs/kio/src/lock.rs +++ b/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 { inner: Arc>, } +// SAFETY: std and loom's Mutex are Send + Sync when T: Send, so this is +// exactly Arc>'s derived bound. Stopping here keeps recursive channel +// graphs from expanding the same Arc/Mutex proof at every level. +unsafe impl Send for Lock {} + +// SAFETY: Arc> has the same T: Send bound for Sync, with either mutex. +unsafe impl Sync for Lock {} + impl Lock { /// Wrap a value. pub fn new(value: T) -> Self { @@ -142,6 +162,12 @@ impl Default for WeakLock { 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); diff --git a/rs/moq-net/src/auth.rs b/rs/moq-net/src/auth.rs index 207632e0b8..b68ab281d2 100644 --- a/rs/moq-net/src/auth.rs +++ b/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( + pub(crate) fn guard( grant: Option<&Self>, body: F, - ) -> std::result::Result { - until(body, |waiter| match grant { + ) -> impl std::future::Future> + kio::MaybeSend { + crate::util::sendable(until(body, move |waiter| match grant { Some(grant) => grant.poll_ended(waiter), None => Poll::Pending, - }) - .await + })) } } diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 98e00ea042..01a7409937 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/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); diff --git a/rs/moq-net/src/lib.rs b/rs/moq-net/src/lib.rs index 63ff8b119c..d010e75180 100644 --- a/rs/moq-net/src/lib.rs +++ b/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. diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 58043030d2..cce5e7bd30 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/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, + shared: SendCell>, // 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, + shared: SendCell>, /// 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, + shared: SendCell>, /// 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, + shared: SendCell>, /// 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, + shared: SendCell>, /// 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, + shared: SendCell>, path: PathOwned, id: u64, signal: kio::Consumer, @@ -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, + shared: SendCell>, path: PathOwned, id: u64, signal: kio::Consumer, @@ -4069,7 +4072,7 @@ impl OriginState { } /// Register a [`Watch`] on the routes covering `path`. - fn watch(&mut self, shared: &kio::Shared, path: &Path) -> Watch { + fn watch(&mut self, shared: &SendCell>, 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, + shared: SendCell>, // 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, - shared: kio::Shared, + shared: SendCell>, root: PathOwned, // Pending updates queued for this cursor. Coalesced so a slow consumer @@ -5413,7 +5416,7 @@ impl AnnounceConsumer { cursors: Vec, state: kio::Producer, stats: stats::Session, - shared: &kio::Shared, + shared: &SendCell>, timers: Clock, ) -> Self { let mut ids = Vec::with_capacity(cursors.len()); diff --git a/rs/moq-net/src/util.rs b/rs/moq-net/src/util.rs index 12e8b81921..03f38068ab 100644 --- a/rs/moq-net/src/util.rs +++ b/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(future: F) -> impl Future + kio::MaybeSend { + future +} + #[cfg(not(target_family = "wasm"))] pub(crate) type MaybeSendBox<'a, T> = Pin + 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(fut: impl Future>) -> E { - match fut.await { - Err(err) => err, - Ok(()) => std::future::pending().await, - } +pub(crate) fn err_only( + fut: impl Future> + kio::MaybeSend, +) -> impl Future + 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`]. diff --git a/rs/moq-net/src/util/cell.rs b/rs/moq-net/src/util/cell.rs new file mode 100644 index 0000000000..a51afe523b --- /dev/null +++ b/rs/moq-net/src/util/cell.rs @@ -0,0 +1,55 @@ +/// A value whose native thread-safety proof is checked once at construction. +/// +/// Recursive channel graphs otherwise repeat the entire auto-trait proof at every +/// caller. The private field keeps construction behind these bounds; wasm keeps +/// its single-threaded, non-Send transport values. +/// +/// The include stays at item level: in statement position it fails to parse, +/// which would pass these tests for the wrong reason. +/// +/// ```compile_fail,E0277 +/// include!(concat!(env!("CARGO_MANIFEST_DIR"), "/src/util/cell.rs")); +/// fn main() { +/// let _ = SendCell::new(std::rc::Rc::new(())); +/// } +/// ``` +/// +/// ```compile_fail,E0277 +/// include!(concat!(env!("CARGO_MANIFEST_DIR"), "/src/util/cell.rs")); +/// fn main() { +/// let _ = SendCell::new(std::cell::Cell::new(0)); +/// } +/// ``` +#[derive(Clone, Debug)] +pub(crate) struct SendCell(T); + +#[cfg(not(target_family = "wasm"))] +impl SendCell { + pub fn new(value: T) -> Self { + Self(value) + } +} + +#[cfg(target_family = "wasm")] +impl SendCell { + pub fn new(value: T) -> Self { + Self(value) + } +} + +// SAFETY: the only native constructor requires T: Send + Sync. The private +// field cannot be replaced with another type; cloning preserves T's bounds. +#[cfg(not(target_family = "wasm"))] +unsafe impl Send for SendCell {} + +// SAFETY: construction proves T: Sync before any shared reference can escape. +#[cfg(not(target_family = "wasm"))] +unsafe impl Sync for SendCell {} + +impl std::ops::Deref for SendCell { + type Target = T; + + fn deref(&self) -> &T { + &self.0 + } +} diff --git a/rs/moq-net/tests/send.rs b/rs/moq-net/tests/send.rs new file mode 100644 index 0000000000..909a76cd32 --- /dev/null +++ b/rs/moq-net/tests/send.rs @@ -0,0 +1,43 @@ +//! Compile these downstream futures at Rust's default recursion limit. +#![cfg(all(feature = "async", not(target_family = "wasm")))] + +fn require_send(_: impl Send) {} + +async fn watch_announcements(mut consumer: moq_net::announce::Consumer) { + while let Some(event) = consumer.next().await { + drop(event); + } +} + +async fn receive_groups(mut consumer: moq_net::track::Subscriber) { + while let Ok(Some(group)) = consumer.recv_group().await { + drop(group); + } +} + +async fn read_frames(mut consumer: moq_net::group::Consumer) { + while let Ok(Some(frame)) = consumer.read_frame().await { + drop(frame); + } +} + +#[test] +fn announcement_next_is_send() { + let _: fn(moq_net::announce::Consumer) = |consumer| { + require_send(watch_announcements(consumer)); + }; +} + +#[test] +fn track_receive_is_send() { + let _: fn(moq_net::track::Subscriber) = |consumer| { + require_send(receive_groups(consumer)); + }; +} + +#[test] +fn group_read_is_send() { + let _: fn(moq_net::group::Consumer) = |consumer| { + require_send(read_frames(consumer)); + }; +} diff --git a/sh/rs/send-depth.sh b/sh/rs/send-depth.sh new file mode 100644 index 0000000000..77d842b78a --- /dev/null +++ b/sh/rs/send-depth.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +# Prove native downstream futures at the default recursion limit on nightly. +set -euo pipefail + +send_rust_version=$(rustc --version) +if [[ "$send_rust_version" == *-nightly* ]]; then + send_sysroot=$(rustc --print sysroot) +elif command -v rustup >/dev/null 2>&1; then + send_sysroot=$(rustup run nightly rustc --print sysroot) +else + echo "rs: nightly Rust is missing; select it or run rustup toolchain install nightly" >&2 + exit 1 +fi + +# Nix puts its pinned stable compiler first; cargo and rustc must both be nightly. +export PATH="$send_sysroot/bin:$PATH" +export RUSTC="$send_sysroot/bin/rustc" +export RUSTDOC="$send_sysroot/bin/rustdoc" +export RUSTFLAGS="${RUSTFLAGS:+$RUSTFLAGS }-D recursion_depth_exceeding_limit" +rustc --version +cargo check --locked -p moq-net --test send