diff --git a/quest/m1/README.md b/quest/m1/README.md index 641ec4a11a4..6945afa42d1 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -66,7 +66,6 @@ bench all require its crate. - [More tests under load](/quest/m1/test-flakes-2/README.md) - the second round of load-only failures, one quest per flake, fixed at the cause - [Test TypeScript check](/quest/m1/test-ts-check.md) - `just check` type-checks the TypeScript harnesses under test/ - [Release backports, second batch](/quest/m1/release-backports/README.md) - the 0.17.x line picks up the smaller release-only fixes left from the 2026-10-09 triage -- [IETF request headers](/quest/m1/ietf-dispatch-headers.md) - each bidi request reads its header in its own task, so a slow one never blocks the next - [END_OF_TRACK placement](/quest/m1/ietf-end-of-track-placement.md) - END_OF_TRACK rides the upstream's Location, and Rust's header stops claiming END_OF_GROUP - [FFI runtime](/quest/m1/ffi-runtime.md) - moq-ffi drives moq on a multi-thread runtime instead of one thread - [Pipelined requests](/quest/m1/pipeline-requests/README.md) - SUBSCRIBE and the first FETCH go out with the track-info request at every hop, so first data arrives a round trip sooner per hop diff --git a/quest/m1/ietf-dispatch-headers.md b/quest/m1/ietf-dispatch-headers.md deleted file mode 100644 index 4e4468e8546..00000000000 --- a/quest/m1/ietf-dispatch-headers.md +++ /dev/null @@ -1,38 +0,0 @@ -# [S] A slow IETF request header never blocks the next request - -## Goal - -`rs/moq-net`'s moq-transport session reads each incoming bidi request -stream's header in its own task, so a request stream whose header stalls -behind flow control never holds up accepting the next one. - -## Plan - -[#5086](https://github.com/moq-dev/moq/pull/5086) fixed the same pattern for -uni streams: `run_unis` waited for each stream's type before accepting the -next, and stalled under connection flow control. `run_dispatch` -(`rs/moq-net/src/ietf/session.rs`) still reads `id` and `ietf::Body` inside -the accept loop. Move the header read into the per-stream task, keeping the -current error handling: a stream that dies before its header is dropped, and -any other header error still fails the session. JS already handles each bidi -stream in its own task (`#runBidis` in `js/net/src/ietf/connection.ts`), so -this is Rust only (decided 2026-10-08). - -Test, beside #5086's `a_silent_stream_does_not_hold_up_the_next`: a silent -bidi request stream does not block a SUBSCRIBE on the next one. - -Review of [#5208](https://github.com/moq-dev/moq/pull/5208) found that -concurrent partial headers return transport credit before parsing completes, -allowing retention far beyond the receive window. Keep that PR held until -the shared native receive-memory work below bounds headers and frame data; -do not substitute a separate header quota or speculative-allocation fallback. - -`release` has the same loop. Its backport needs a separate compatibility -assessment because the new receive-credit prerequisite targets the in-tree -stack; this plan does not authorize moving that stack onto `release`. - -Public API: none. Wire: none. - -## Required - -- [Bound message assembly](/quest/m1/quic/receive-memory.md) - concurrent headers retain connection credit and pressure cancellation restores progress diff --git a/quest/m1/quic/receive-memory.md b/quest/m1/quic/receive-memory.md index ad6267897e3..f027c782966 100644 --- a/quest/m1/quic/receive-memory.md +++ b/quest/m1/quic/receive-memory.md @@ -92,7 +92,6 @@ separate guide quest is needed. ## Related -- [IETF request headers](/quest/m1/ietf-dispatch-headers.md) - concurrent request parsing must not amplify unbounded partial-header retention - [Scope track priority](/quest/m1/track-priority-scope.md) - broadcast fairness domains and subscription priority semantics - [Hierarchical stream scheduling](/quest/m1/quic/scheduler.md) - share priority semantics without requiring sender-side recovery - [QUIC bitrate caps](/quest/m2/rate-quic.md) - pace credit grants inside the independently bounded memory allowance diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index e5f9dfb6a97..0c28bbad06a 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -1091,11 +1091,11 @@ fn stop_on_error(reader: &mut Reader( - session: S, + mut session: S, publisher: Publisher, - mut subscriber: Subscriber, + subscriber: Subscriber, peer_setup: peer::PeerSetup, // Answers the peer's Auth requests, on the versions that can negotiate them. serve: Option, @@ -1104,14 +1104,14 @@ async fn run_dispatch( where S: crate::transport::poll::Boxable, { + use std::task::Poll; + // PUBLISH_NAMESPACE decodes differently once the MoQ Cluster extension is - // negotiated, so the whole dispatch loop waits for the peer's SETUP first. The peer - // must send it before anything else, and `run_unis` reads it independently, so this - // costs a handshake round rather than blocking. + // negotiated, so dispatch waits for SETUP, which `run_unis` reads independently. let peer = subscriber.peer().await; // An AUTH from a peer that did not negotiate MoQ Auth is an unknown request, which - // falls through to the protocol violation below. + // falls through to the protocol violation in `Bidi::serve`. let serve = match peer_setup.get().await.auth { true => serve, false => None, @@ -1121,7 +1121,6 @@ where // advertisement is the peer ignoring our own SETUP (MoQ Solicit). let declared = subscriber.solicit().await; - let mut tasks = TaskSet::owned(); // Each AUTH serve task settles `Issued::closed` from whatever the auth handle // holds when it drops, and the first reason wins. So the error that ends the // session has to reach the handle while `tasks` is still alive, or every grant @@ -1129,101 +1128,148 @@ where // which leaves no serve task to settle. let handle = serve.as_ref().map(|serve| serve.handle.clone()); - // Scoped so `tasks` outlives the close below: the loop borrows it, so it only - // drops once this block's future is done. - let res: Result<(), Error> = async { - let mut accept = session.clone(); - loop { - let mut stream = tasks - .drive(|waiter| { - let mut cx = waiter.context(); - Stream::poll_accept(&mut accept, version, &mut cx) - }) - .await?; - - // The intermediate results live outside the poll closure, so a Pending - // mid-header resumes where it left off. - let mut hdr_id: Option = None; - let header = tasks - .drive(|waiter| { - let mut cx = waiter.context(); - let id = match hdr_id { - Some(id) => id, - None => *hdr_id.insert(std::task::ready!(stream.reader.poll_varint(&mut cx))?), - }; - let body = std::task::ready!(stream.reader.poll_decode::(&mut cx))?; - std::task::Poll::Ready(Ok::<_, Error>((id, body))) - }) - .await; - // Same tolerance as `run_unis`: a request stream that dies before its header - // is the peer abandoning that request, not the session. Anything else, a - // header that does not parse included, still fails the session. - let (id, data) = match header { - Ok(header) => header, - Err(err) if died_before_header(&err) => { - tracing::debug!(%err, "dropping bidi stream that died before its header"); - continue; - } - Err(err) => return Err(err), - }; + let bidi = Bidi { + publisher, + subscriber, + peer, + declared, + serve, + version, + }; + let mut tasks = TaskSet::owned(); + let fatal = kio::Shared::>::default(); - match id { - // Draft-16 moved SUBSCRIBE_NAMESPACE to its own stream, past the control stream - // that admits every other request, but it still takes a request ID from the - // MAX_REQUEST_ID window. Held until the request ends, like the rest. - ietf::SubscribeNamespaceLegacy::ID if version == Version::Draft16 => { - let request_id = - RequestId::decode(&mut crate::coding::Decoder::new(&data.0, version.into()), version)?; - let permit = publisher.control.accept(request_id)?; - let task = publisher.handle_stream(id, data, stream)?; - tasks.push( - async move { - let _permit = permit; - task.await - } - .maybe_boxed(), - ); - } - // Publisher handles: Subscribe, Fetch, SubscribeNamespace (0x50 modern / - // 0x11 legacy), SubscribeTracks, TrackStatus - ietf::Subscribe::ID - | ietf::Fetch::ID - | ietf::SubscribeNamespace::ID - | ietf::SubscribeNamespaceLegacy::ID - | ietf::SUBSCRIBE_TRACKS_ID - | ietf::TrackStatus::ID => { - tasks.push(publisher.handle_stream(id, data, stream)?); - } - // Subscriber handles: Publish, PublishNamespace - ietf::Publish::ID | ietf::PublishNamespace::ID => { - tasks.push(subscriber.handle_stream(id, data, stream, peer, declared)?); - } - auth::Auth::ID if let Some(serve) = &serve => { - let mut data = data.decoder(version); - let msg = auth::Auth::decode_msg(&mut data, version)?; - if !data.is_empty() { - return Err(Error::WrongSize); - } - tasks.push(serve.clone().run(stream, msg, version)); - } - _ => { - tracing::warn!(id, "unexpected bidi stream type"); - return Err(Error::UnexpectedStream); + let err = loop { + let accepted = tasks + .drive(|waiter| { + let broken = fatal.poll(waiter, |fatal| match fatal.is_some() { + true => Poll::Ready(()), + false => Poll::Pending, + }); + if let Poll::Ready(mut fatal) = broken + && let Some(err) = fatal.take() + { + return Poll::Ready(Err(err)); } + let mut cx = waiter.context(); + Stream::poll_accept(&mut session, version, &mut cx) + }) + .await; + let stream = match accepted { + Ok(stream) => stream, + Err(err) => break err, + }; + + // QUIC may expose an earlier stream before its bytes arrive. Reading the + // header here could block later streams whose unread data holds flow control. + let bidi = bidi.clone(); + let fatal = fatal.clone(); + tasks.push(async move { + if let Err(err) = bidi.serve(stream).await { + fatal.lock().get_or_insert(err); } + }); + }; + + // `tasks` still holds the AUTH serve tasks, so the session's error lands on the + // handle before they drop. The driver closes it again with the same error, which + // then does nothing. + if let Some(handle) = &handle { + handle.close(err.clone()); + } + Err(err) +} + +/// What serving one incoming bidi request needs from its session. +struct Bidi { + publisher: Publisher, + subscriber: Subscriber, + peer: cluster::Peer, + declared: Option, + serve: Option, + version: Version, +} + +impl Clone for Bidi { + fn clone(&self) -> Self { + Self { + publisher: self.publisher.clone(), + subscriber: self.subscriber.clone(), + peer: self.peer, + declared: self.declared, + serve: self.serve.clone(), + version: self.version, } } - .await; +} - // Every error exit above drops the AUTH serve tasks in `tasks`, so the session's - // error has to land on the handle first. The driver closes it again with the same - // error, which then does nothing. - if let Some(handle) = &handle - && let Err(err) = &res - { - handle.close(err.clone()); +impl Bidi { + /// Read and serve a request, failing only when it breaks the session. + async fn serve(self, mut stream: Stream) -> Result<(), Error> { + let Self { + publisher, + mut subscriber, + peer, + declared, + serve, + version, + } = self; + let header = async { + let id = stream.reader.varint().await?; + let body = stream.reader.decode::().await?; + Ok::<_, Error>((id, body)) + } + .await; + // Abandoning a request before its header is not a session failure. Bytes + // that arrive and do not parse, and dispatch errors, remain session-fatal. + let (id, data) = match header { + Ok(header) => header, + Err(err) if died_before_header(&err) => { + tracing::debug!(%err, "dropping bidi stream that died before its header"); + return Ok(()); + } + Err(err) => return Err(err), + }; + match id { + // Draft-16 moved SUBSCRIBE_NAMESPACE to its own stream, past the control stream + // that admits every other request, but it still takes a request ID from the + // MAX_REQUEST_ID window. Held until the request ends, like the rest. + ietf::SubscribeNamespaceLegacy::ID if version == Version::Draft16 => { + let request_id = RequestId::decode(&mut crate::coding::Decoder::new(&data.0, version.into()), version)?; + let permit = publisher.control.accept(request_id)?; + let task = publisher.handle_stream(id, data, stream)?; + let _permit = permit; + task.await; + } + // Publisher handles: Subscribe, Fetch, SubscribeNamespace (0x50 modern / + // 0x11 legacy), SubscribeTracks, TrackStatus + ietf::Subscribe::ID + | ietf::Fetch::ID + | ietf::SubscribeNamespace::ID + | ietf::SubscribeNamespaceLegacy::ID + | ietf::SUBSCRIBE_TRACKS_ID + | ietf::TrackStatus::ID => { + publisher.handle_stream(id, data, stream)?.await; + } + // Subscriber handles: Publish, PublishNamespace + ietf::Publish::ID | ietf::PublishNamespace::ID => { + subscriber.handle_stream(id, data, stream, peer, declared)?.await; + } + auth::Auth::ID if let Some(serve) = serve => { + let mut data = data.decoder(version); + let msg = auth::Auth::decode_msg(&mut data, version)?; + if !data.is_empty() { + return Err(Error::WrongSize); + } + serve.run(stream, msg, version).await; + } + _ => { + tracing::warn!(id, "unexpected bidi stream type"); + return Err(Error::UnexpectedStream); + } + } + Ok(()) } - res } /// Monitor the peer's SETUP stream for a GOAWAY, surfacing it through @@ -2046,6 +2092,97 @@ mod tests { assert!(result.is_none(), "the silent stream ended the session: {result:?}"); } + /// A stalled bidi header must not block a SUBSCRIBE on the next stream. + #[moq_net_sim::test] + async fn a_silent_bidi_does_not_hold_up_the_next() { + const VERSION: Version = Version::Draft19; + let mut subscribe = Vec::new(); + let mut encoder = crate::coding::Encoder::new(&mut subscribe, VERSION.into()); + encoder.varint(ietf::Subscribe::ID).unwrap(); + ietf::Subscribe { + request_id: RequestId(1), + track_namespace: crate::Path::new("room"), + track_name: "video".into(), + subscriber_priority: 128, + group_order: ietf::GroupOrder::Descending, + filter: ietf::Filter::NextObject, + fill: None, + properties_wanted: true, + forward: true, + range_filters: false, + } + .encode(&mut encoder, VERSION) + .unwrap(); + + // Stall before the type, then after it while the body has not arrived. + for stalled in [Vec::new(), vec![ietf::Subscribe::ID as u8]] { + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let broadcast = origin.publish("room", crate::origin::Route::default()).unwrap(); + let mut track = broadcast.create_track("video", None).unwrap(); + let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()) + .with_incoming_bidis(vec![stalled, subscribe.clone()]); + let driver = bidi_driver(session, Some(origin.consume())); + let subscribed = Box::pin(async { while track.subscription_changed().await.unwrap().is_none() {} }); + let result = moq_net_sim::timeout( + std::time::Duration::from_secs(10), + futures::future::select(driver, subscribed), + ) + .await + .expect("SUBSCRIBE behind the stalled header was never dispatched"); + assert!( + matches!(result, futures::future::Either::Right(_)), + "session ended before SUBSCRIBE" + ); + } + } + + /// Child tasks must report malformed and unknown requests to the session driver. + #[moq_net_sim::test] + async fn invalid_bidi_requests_still_end_the_session() { + for id in [ietf::Subscribe::ID, 0x3f] { + let mut header = Vec::new(); + let mut encoder = crate::coding::Encoder::new(&mut header, Version::Draft19.into()); + encoder.varint(id).unwrap(); + encoder.slice(&[0, 0]); // Complete, empty body: malformed for SUBSCRIBE. + let session = + crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_bidis(vec![header]); + let err = moq_net_sim::timeout(std::time::Duration::from_secs(10), bidi_driver(session, None)) + .await + .expect("invalid request did not end the session") + .expect_err("invalid request was accepted"); + if id == ietf::Subscribe::ID { + assert!(matches!(err, Error::Decode(_)), "{err:?}"); + } else { + assert!(matches!(err, Error::UnexpectedStream), "{err:?}"); + } + } + } + + /// Start a draft-19 session with SETUP already settled, so bidi requests run immediately. + fn bidi_driver(session: crate::lite::test_transport::ScriptedSession, publish: Option) -> Driver { + start(Config { + runtime: crate::time::Clock::sim(), + session, + setup: None, + request_id_max: None, + limits: Default::default(), + client: true, + publish, + subscribe: None, + peer_hop: None, + cost: None, + version: Version::Draft19, + path: None, + authority: None, + peer_setup_stream: None, + peer_declared: Some(peer::Peer::default()), + auth: crate::auth::Handle::new(false), + early_unis: Vec::new(), + }) + .unwrap() + .0 + } + /// A non-zero subgroup is refused on its own stream, never by closing the session. #[moq_net_sim::test] async fn a_non_zero_subgroup_is_stopped_without_closing_the_session() {