From 678f47f626583818d0c483b69920d9669c9ce030 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 08:10:02 -0700 Subject: [PATCH 1/2] quest: claim IETF dispatch headers Co-Authored-By: GPT-6 From dae30aaecf28c841f49bc3f859213caad281f17c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 08:54:28 -0700 Subject: [PATCH 2/2] fix(net): read each IETF request header in its own task Co-Authored-By: GPT-6 --- quest/m1/README.md | 1 - quest/m1/ietf-dispatch-headers.md | 27 ---- rs/moq-net/src/ietf/session.rs | 214 ++++++++++++++++++++++++------ 3 files changed, 173 insertions(+), 69 deletions(-) delete mode 100644 quest/m1/ietf-dispatch-headers.md diff --git a/quest/m1/README.md b/quest/m1/README.md index e06cb20693f..d109544e760 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -63,7 +63,6 @@ ladders. - [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 - [moq-uring tests under load](/quest/m1/uring-tests-under-load.md) - uring tests pass while parallel checks share locked memory - [FFI runtime](/quest/m1/ffi-runtime.md) - moq-ffi drives moq on a multi-thread runtime instead of one thread diff --git a/quest/m1/ietf-dispatch-headers.md b/quest/m1/ietf-dispatch-headers.md deleted file mode 100644 index 9a019dbf744..00000000000 --- a/quest/m1/ietf-dispatch-headers.md +++ /dev/null @@ -1,27 +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. - -`release` has the same loop; once this lands on `main`, backport it the way -#5125 carried the uni half. - -Public API: none. Wire: none. diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index b8c2cded2e6..29515f2906f 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -1006,62 +1006,108 @@ fn stop_on_error(reader: &mut Reader( - session: S, + mut session: S, publisher: Publisher, - mut subscriber: Subscriber, + subscriber: Subscriber, version: Version, ) -> Result<(), Error> 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; - - // From the same slot, so this costs nothing extra: it decides whether an unsolicited - // advertisement is the peer ignoring our own SETUP (MoQ Solicit). let declared = subscriber.solicit().await; - + let bidi = Bidi { + publisher, + subscriber, + peer, + declared, + version, + }; let mut tasks = TaskSet::owned(); - let mut accept = session.clone(); + let fatal = kio::Shared::>::default(); + loop { - let mut stream = tasks + let stream = 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 accept, version, &mut cx) + Stream::poll_accept(&mut session, 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. + // 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); + } + }); + } +} + +/// What serving one incoming bidi request needs from its session. +struct Bidi { + publisher: Publisher, + subscriber: Subscriber, + peer: cluster::Peer, + declared: 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, + version: self.version, + } + } +} + +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, + 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"); - continue; + 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 @@ -1070,13 +1116,8 @@ where 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(), - ); + let _permit = permit; + task.await; } // Publisher handles: Subscribe, Fetch, SubscribeNamespace (0x50 modern / // 0x11 legacy), SubscribeTracks, TrackStatus @@ -1086,17 +1127,18 @@ where | ietf::SubscribeNamespaceLegacy::ID | ietf::SUBSCRIBE_TRACKS_ID | ietf::TrackStatus::ID => { - tasks.push(publisher.handle_stream(id, data, stream)?); + publisher.handle_stream(id, data, stream)?.await; } // Subscriber handles: Publish, PublishNamespace ietf::Publish::ID | ietf::PublishNamespace::ID => { - tasks.push(subscriber.handle_stream(id, data, stream, peer, declared)?); + subscriber.handle_stream(id, data, stream, peer, declared)?.await; } _ => { tracing::warn!(id, "unexpected bidi stream type"); return Err(Error::UnexpectedStream); } } + Ok(()) } } @@ -1775,6 +1817,96 @@ 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()), + 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() {