From d4287eb1199449ad5bea932a392dde599af4422b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 21:18:00 -0700 Subject: [PATCH 1/7] chore: claim lite early streams quest Co-Authored-By: GPT-6 From 7ddfa101ba3123151653f0159a80e7bfa8507c52 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 21:30:41 -0700 Subject: [PATCH 2/7] fix(net): retain early lite streams and reject duplicate setup Co-Authored-By: GPT-6 --- doc/concept/moq-lite.md | 4 ++ js/net/src/lite/connection.test.ts | 35 ++++++++-- js/net/src/lite/connection.ts | 20 +++++- quest/m0/README.md | 1 - quest/m0/lite-early-streams.md | 52 -------------- rs/moq-net/src/lite/session.rs | 101 +++++++++++++++++++++------ rs/moq-net/src/lite/setup.rs | 26 +++++-- rs/moq-net/src/lite/subscriber.rs | 106 +++++++++++++++++++++++++++-- rs/moq-net/src/server.rs | 51 ++++++++++---- 9 files changed, 297 insertions(+), 99 deletions(-) delete mode 100644 quest/m0/lite-early-streams.md diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 79e8378167..2784004134 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -29,6 +29,10 @@ implement in an afternoon. The wire spec is A dedicated ALPN selects the wire version for moq-lite 03 and newer. The legacy `moql` ALPN negotiates moq-lite 01 or 02 via `SETUP`. In moq-lite 05 and newer, each side also sends a `SETUP` message with its capabilities. +Rust holds unidirectional streams arriving before the peer's `SETUP` until the +peer's `SETUP` arrives. Streams that depend on negotiated capabilities wait for +`SETUP` in both Rust and JavaScript. A second Setup Stream closes the session +with `PROTOCOL_VIOLATION`. Rust and TypeScript speak moq-lite 01 through 06 and moq-transport drafts 14 through 22. Clients offer `moq-lite-06` first by default. moq-lite 07 is still in progress: it negotiates as `moq-lite-07-wip`, and only when both diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index fc11239b1f..562b875a82 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -1,12 +1,13 @@ import { expect, test } from "bun:test"; +import { SessionCode } from "../error.ts"; import { createMockTransportPair } from "../mock.ts"; -import { Stream } from "../stream.ts"; +import { Stream, Writer } from "../stream.ts"; import { wireOf } from "../wire.ts"; import { Connection, probeLevel } from "./connection.ts"; import { Goaway } from "./goaway.ts"; -import { ProbeLevel } from "./setup.ts"; -import { StreamId } from "./stream.ts"; -import { ALPN_04, Version } from "./version.ts"; +import { ProbeLevel, Setup } from "./setup.ts"; +import { DataType, StreamId } from "./stream.ts"; +import { ALPN_04, ALPN_05, ALPN_06, ALPN_07_WIP, Version } from "./version.ts"; /** A transport whose `getStats` behaves as described, or is absent entirely. */ function transport(getStats?: () => Promise): WebTransport { @@ -86,3 +87,29 @@ test("a lite GOAWAY keeps the session open, and a second one closes it", async ( connection.close(); } }); + +for (const [alpn, version] of [ + [ALPN_05, Version.DRAFT_05], + [ALPN_06, Version.DRAFT_06], + [ALPN_07_WIP, Version.DRAFT_07], +] as const) { + for (const complete of [false, true]) { + test(`duplicate SETUP closes ${alpn} with PROTOCOL_VIOLATION (complete=${complete})`, async () => { + const pair = createMockTransportPair(alpn); + const connection = new Connection({ url: new URL("https://relay.example/"), quic: pair.client, version }); + try { + for (let i = 0; i < 2; i++) { + const writer = await Writer.open(pair.server, { version }); + await writer.u53(DataType.Setup); + if (complete && i === 0) { + await new Setup({}).encode(writer, version); + writer.close(); + } + } + expect((await pair.client.closed).closeCode).toBe(SessionCode.ProtocolViolation); + } finally { + connection.close(); + } + }); + } +} diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index 31ae5c92c9..79cc10e0a1 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -4,7 +4,16 @@ import type { Established } from "../connection/established.ts"; import type { Drain } from "../connection/goaway.ts"; import { type Probe, type Stats, transportStats } from "../connection/stats.ts"; import { type Transport, transportOf } from "../connection/transport.ts"; -import { closeError, error, fromClose, ProtocolViolation, StreamCode, StreamError, sessionCause } from "../error.ts"; +import { + closeError, + error, + fromClose, + ProtocolViolation, + SessionCode, + StreamCode, + StreamError, + sessionCause, +} from "../error.ts"; import { type Hop, randomHop } from "../hop.ts"; import type { Consumer as OriginConsumer } from "../origin.ts"; import type * as Path from "../path.ts"; @@ -87,6 +96,7 @@ export class Connection implements Established { // encoding depends on a negotiated capability (e.g. PROBE) wait on this. undefined // until the peer's SETUP arrives; stays undefined forever on older drafts. #peerSetup = new Signal(undefined); + #setupSeen = false; // Mirrors the role out of #peerSetup, so the public surface exposes the peer's declared // direction without handing out the whole SETUP (whose probe level gates our own streams). @@ -286,6 +296,9 @@ export class Connection implements Established { }) .catch((err: unknown) => { stream.stop(err); + if (err instanceof ProtocolViolation) { + this.#quic.close({ closeCode: SessionCode.ProtocolViolation, reason: err.message }); + } }); } } @@ -296,8 +309,11 @@ export class Connection implements Established { const msg = await Group.decode(stream, this.#version); await this.#subscriber.runGroup(msg, stream); } else if (typ === DataType.Setup) { + // Claim the stream before decoding, so two incomplete SETUPs are duplicates too. + if (this.#setupSeen) throw new ProtocolViolation("duplicate SETUP"); + this.#setupSeen = true; // The peer sends exactly one SETUP, then FINs. Record it so capability-gated - // streams (e.g. PROBE) can react, then drain to the FIN. + // streams (e.g. PROBE) can react. const setup = await Setup.decode(stream, this.#version); this.#peerSetup.set(setup); this.#peerRole.set(setup.role); diff --git a/quest/m0/README.md b/quest/m0/README.md index a07461f573..a3b2e00056 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -41,7 +41,6 @@ Published API or wire breaks still land on dev; each quest's Plan says so. - [IETF FIN semantics](/quest/m0/ietf-fin-not-cancel.md) - a request stream FIN stops updates without cancelling, and REQUEST_UPDATE on a subscribe is parsed - [JS unknown bidi](/quest/m0/ietf-unknown-bidi.md) - an unknown moq-transport bidi stream type closes the session in JS, as in Rust -- [moq-lite early streams](/quest/m0/lite-early-streams.md) - a lite uni stream that arrives before SETUP is held until SETUP lands, and a second SETUP is fatal - [Request caps](/quest/m0/request-caps.md) - lite message sizes, IETF request IDs, and per-session announces and subscriptions are bounded - [noq reassembly cap](/quest/m0/noq-reassembly-cap.md) - noq carries quinn's stream reassembly cap and the connection receive window is finite by default - [qmux credit](/quest/m0/qmux-credit.md) - qmux returns connection credit for dropped and stopped streams and delivers its close frame, on both lines diff --git a/quest/m0/lite-early-streams.md b/quest/m0/lite-early-streams.md deleted file mode 100644 index b9d005c299..0000000000 --- a/quest/m0/lite-early-streams.md +++ /dev/null @@ -1,52 +0,0 @@ -# [S] moq-lite streams that arrive before SETUP are held - -## Goal - -On lite-05 and later (the versions with a Setup Stream), a uni stream that -arrives before the peer's SETUP is held and handled once SETUP lands, never -aborted, in Rust and JS. A second Setup Stream closes the session with -PROTOCOL_VIOLATION in both languages, as the lite draft requires. - -## Plan - -Facts (`origin/main`, 2026-10-01): - -- Rust `lite::accept_setup` (`rs/moq-net/src/lite/session.rs`) aborts every - non-SETUP uni stream with `UnexpectedStream`. - `accept_request_skips_uni_stream_before_setup` in `rs/moq-net/src/server.rs` - pins that. An unknown `DataType` fails the handshake. -- JS lite has no pre-SETUP gate: `#runUni` in `js/net/src/lite/connection.ts` - handles SETUP like any other uni stream, so early groups already work. It - also accepts a second SETUP silently, and the `#runUnis` catch only stops the - stream, so throwing `ProtocolViolation` there would not close the session. - The `#runBidis` catch closes on `ProtocolViolation`, but with a bare - `close()` rather than the PROTOCOL_VIOLATION code. -- `drafts/draft-lcurley-moq-lite.md`: neither endpoint waits for the peer's - SETUP. A receiver MUST buffer only streams whose encoding depends on a - negotiated extension, and a second Setup Stream MUST close with - PROTOCOL_VIOLATION. Rust's abort is stricter than the draft. -- Lite-01 to lite-04 have no Setup Stream and are out of scope. - -Decided (2026-10-01): - -- ✅ Rust holds early uni streams until SETUP, then hands them to the normal - classifier, the same shape as - IETF early streams (#4686). Rust can't - route before SETUP anyway, since path and role come from it. Rejected: - processing immediately unless extension-dependent, which needs a session - before SETUP. -- ✅ The duplicate-SETUP check rides along, in both languages. In JS the uni - catch must close the session with the PROTOCOL_VIOLATION code, not copy the - bidi path's bare `close()`. - -The remaining JS gap is the duplicate SETUP and its close path; the hold is -Rust work unless JS turns out to need an extension-dependent hold. - -Crib from #4686: its `died_before_header` helper and peek-then-hold pattern. -QUIC stream credit bounds the held queue. Check whether JS lite must buffer any -extension-dependent stream type today, and hold those until SETUP. - -Tests: flip `accept_request_skips_uni_stream_before_setup` to expect the group -to be delivered. In both languages, a duplicate SETUP closes the session. - -Public API: none. Wire: none; matches the lite draft. diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index 3a9120a13c..af32fb7465 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -24,31 +24,50 @@ pub(crate) struct SessionStart { /// the session, so the caller can inspect the advertised path (and gate on it) before /// serving. lite-05+ only. /// -/// Blocks on the peer's Setup Stream, which every lite-05+ endpoint opens at startup. -/// Almost always the first unidirectional stream; any other uni stream that races -/// ahead of it is `STOP_SENDING`-ed and skipped (we don't support proactive uni -/// PUBLISH, so nothing legitimate precedes the SETUP today). The eventual home for -/// out-of-order tolerance is the full session loop with deferred origin binding. +/// Other uni streams racing ahead of SETUP are held for the normal session +/// classifier. QUIC stream credit bounds the held queue. /// -/// Pass the returned [`Setup`] to [`start`] as its `peer_setup` so PROBE gating still -/// resolves without re-reading the (consumed) stream. +/// Pass the result to [`start`] as its `peer_setup` so capability gating resolves +/// without re-reading SETUP, and every held stream is handed back to the driver. pub async fn accept_setup( session: &mut S, version: Version, -) -> Result { +) -> Result, Error> { + let mut early = Vec::new(); loop { let stream = session.accept_uni().await.map_err(Error::from_transport)?; let mut reader = Reader::new(stream, version); - - match reader.decode::().await? { - DataType::Setup => return reader.decode::().await, - // A non-SETUP uni stream this early is unexpected (GROUP needs a prior - // subscribe). Reject it and keep waiting rather than failing the session. - _ => reader.abort(&Error::UnexpectedStream), + let kind = match reader.decode_peek::().await { + Ok(kind) => kind, + Err( + Error::Cancel + | Error::Stream(_) + | Error::Remote(_) + | Error::Transport(_) + | Error::Decode(crate::DecodeError::Short), + ) => continue, + Err(err) => return Err(err), + }; + if kind != DataType::Setup as u64 { + early.push(reader); + continue; } + reader.decode::().await?; + return Ok(AcceptedSetup { + setup: reader.decode().await?, + early, + }); } } +/// The peer's pre-read SETUP and the streams held until it arrived. +pub struct AcceptedSetup { + /// The SETUP decoded before origin binding. + pub setup: Setup, + /// Streams with their type still buffered for the session classifier. + pub early: Vec>, +} + /// Everything one moq-lite session needs to start. pub struct Config { /// The runtime that arms the session's timers. @@ -85,7 +104,7 @@ pub struct Config { /// The peer's SETUP, when it was already read before [`start`] (e.g. a server that /// gated on the client's path via [`accept_setup`]). Seeds the peer-setup slot so /// the Setup Stream isn't expected again. `None` reads it from the wire as usual. - pub peer_setup: Option, + pub peer_setup: Option>, } /// Start a lite session. @@ -151,9 +170,12 @@ where // When the caller already read it (a gated server accept), seed the slot so the // Setup stream isn't expected on the wire again. let peer_setup_slot = PeerSetup::default(); - if let Some(setup) = peer_setup { - peer_setup_slot.set(setup); - } + let early = if let Some(accepted) = peer_setup { + peer_setup_slot.set(accepted.setup); + accepted.early + } else { + Vec::new() + }; let peer_setup = peer_setup_slot; // GOAWAY wiring: the public Session holds one half (send trigger, received @@ -196,7 +218,7 @@ where goaway: Some(SendGoaway::new(runtime, session.clone(), goaway, version)), session_stream: setup_stream, publisher, - subscriber: SubscriberDriver::new(subscriber), + subscriber: SubscriberDriver::new(subscriber, early), session, }; @@ -524,3 +546,44 @@ impl SendGoaway { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::coding::Encode; + + #[tokio::test(start_paused = true)] + async fn duplicate_setup_closes_the_session() { + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + for complete in [false, true] { + let mut first = Vec::new(); + DataType::Setup.encode(&mut first, version).unwrap(); + if complete { + Setup::default().encode(&mut first, version).unwrap(); + } + let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()) + .with_incoming_unis(vec![first.clone(), first]); + let log = session.log.clone(); + let mut started = start(Config { + runtime: crate::time::Clock::tokio(), + session, + setup_stream: None, + publish: None, + subscribe: None, + peer_hop: None, + version, + our_setup: Setup::default(), + peer_setup: None, + }) + .unwrap(); + let _ = started.driver.poll(&kio::Waiter::noop()); + assert!( + log.closes() + .iter() + .any(|(code, _)| *code == SessionError::ProtocolViolation.to_code()), + "{version:?}, complete={complete}: duplicate SETUP must close the session" + ); + } + } + } +} diff --git a/rs/moq-net/src/lite/setup.rs b/rs/moq-net/src/lite/setup.rs index 53810df047..3bfe511e25 100644 --- a/rs/moq-net/src/lite/setup.rs +++ b/rs/moq-net/src/lite/setup.rs @@ -257,12 +257,30 @@ impl Message for Setup { /// stream) wait on this before deciding what to do. Cheap to clone: every handle /// shares the same slot. #[derive(Clone, Default)] -pub(crate) struct PeerSetup(kio::Shared>); +pub(crate) struct PeerSetup(kio::Shared); + +#[derive(Default)] +struct PeerSetupState { + seen: bool, + setup: Option, +} impl PeerSetup { /// Record the peer's SETUP. pub fn set(&self, setup: Setup) { - *self.0.lock() = Some(setup); + let mut state = self.0.lock(); + state.seen = true; + state.setup = Some(setup); + } + + /// Claim the Setup Stream before decoding its body, refusing any second stream. + pub fn claim(&self) -> Result<(), crate::Error> { + let mut state = self.0.lock(); + if state.seen { + return Err(crate::Error::ProtocolViolation); + } + state.seen = true; + Ok(()) } /// Poll for the peer's advertised probe level, waiting until its SETUP arrives. @@ -289,13 +307,13 @@ impl PeerSetup { /// driver. fn poll_get(&self, waiter: &kio::Waiter, f: impl FnOnce(&Setup) -> T) -> std::task::Poll { let slot = std::task::ready!(self.0.poll(waiter, |setup| { - if setup.is_some() { + if setup.setup.is_some() { std::task::Poll::Ready(()) } else { std::task::Poll::Pending } })); - std::task::Poll::Ready(f(slot.as_ref().expect("waited for Some"))) + std::task::Poll::Ready(f(slot.setup.as_ref().expect("waited for Some"))) } } diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 41bf3ce820..ce88a4e8f5 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -525,7 +525,7 @@ pub(super) struct SubscriberDriver { } impl SubscriberDriver { - pub fn new(subscriber: Subscriber) -> Self { + pub fn new(subscriber: Subscriber, early: Vec>) -> Self { // The wire speaks announce interest by prefix: ask for each granted // pattern's literal head and let the origin's scope filter what arrives. let prefixes = crate::model::interest_prefixes(&subscriber.origin.allowed()) @@ -536,7 +536,7 @@ impl SubscriberDriver { Self { prefixes, cleanup: SubscriptionCleanup(subscriber.subscribes.clone()), - uni: UniAccept::new(subscriber.clone()), + uni: UniAccept::new(subscriber.clone(), early), bandwidth: Some(RecvBandwidth::new(subscriber.clone())), datagrams: Some(DatagramRecv::new(subscriber.clone())), sources: kio::Tasks::new(), @@ -616,12 +616,19 @@ struct UniAccept { } impl UniAccept { - fn new(subscriber: Subscriber) -> Self { + fn new(subscriber: Subscriber, early: Vec>) -> Self { let accept = subscriber.session.clone(); + let mut children = kio::Tasks::new(); + for reader in early { + children.push(UniServe { + subscriber: subscriber.clone(), + state: UniState::Start { reader }, + }); + } Self { subscriber, accept, - children: kio::Tasks::new(), + children, } } @@ -677,6 +684,12 @@ impl kio::Task for UniServe { fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { if let Err(err) = ready!(self.poll_serve(waiter)) { tracing::debug!(%err, "error running uni stream"); + if matches!(err, Error::ProtocolViolation) { + self.subscriber + .session + .clone() + .close(crate::SessionError::ProtocolViolation.to_code(), &err.to_string()); + } } Poll::Ready(()) } @@ -699,7 +712,16 @@ impl UniServe { let mut cx = std::task::Context::from_waker(waiter.waker()); // A decode error here is only logged; the peer hung up or spoke garbage // before the stream had a type. - let kind = ready!(reader.poll_decode::(&mut cx))?; + let kind = ready!(reader.poll_decode_peek::(&mut cx))?; + if self.subscriber.version.has_setup_stream() { + match kind { + lite::DataType::Setup => self.subscriber.peer_setup.claim()?, + lite::DataType::Group => { + ready!(self.subscriber.peer_setup.poll_probe_level(waiter)); + } + } + } + ready!(reader.poll_decode::(&mut cx))?; let UniState::Start { reader } = std::mem::replace(&mut self.state, UniState::Done) else { unreachable!() }; @@ -1472,6 +1494,80 @@ mod tests { ); } + #[tokio::test(start_paused = true)] + async fn early_group_is_delivered_after_setup() { + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + for pre_read in [false, true] { + let mut bytes = Vec::new(); + lite::DataType::Group.encode(&mut bytes, version).unwrap(); + lite::Group { + subscribe: 7, + sequence: 3, + frame_start: 0, + } + .encode(&mut bytes, version) + .unwrap(); + 0u64.encode(&mut bytes, version).unwrap(); + 1u64.encode(&mut bytes, version).unwrap(); + bytes.push(42); + let mut setup = Vec::new(); + lite::DataType::Setup.encode(&mut setup, version).unwrap(); + lite::Setup::default().encode(&mut setup, version).unwrap(); + let mut session = + crate::lite::test_transport::ScriptedSession::eof(Vec::new()).with_incoming_unis(if pre_read { + vec![Vec::new(), vec![42], bytes, setup] + } else { + vec![bytes] + }); + let log = session.log.clone(); + let peer_setup = lite::PeerSetup::default(); + let early = if pre_read { + let accepted = lite::accept_setup(&mut session, version).await.unwrap(); + assert_eq!(accepted.early.len(), 2); + assert!(log.stops().is_empty()); + peer_setup.set(accepted.setup); + accepted.early + } else { + Vec::new() + }; + let subscriber = Subscriber::new(SubscriberConfig { + runtime: crate::time::Clock::tokio(), + session, + origin: origin::Config::new(crate::Hop::new(1).unwrap()).produce(), + recv_bandwidth: None, + version, + peer_setup: peer_setup.clone(), + peer_hop: None, + cost: None, + going_away: Default::default(), + }); + let track = track::Producer::new(std::sync::Arc::new(crate::broadcast::Info::default()), "video", None); + let mut consumer = track.consume().subscribe(None).await.unwrap(); + subscriber.subscribes.lock().insert( + 7, + TrackEntry { + producer: track, + timescale: Some(Timescale::default()), + tail: Default::default(), + }, + ); + let mut accept = UniAccept::new(subscriber, early); + assert!(accept.poll(&kio::Waiter::noop()).is_pending()); + if !pre_read { + assert!(consumer.recv_group().now_or_never().is_none(), "GROUP waits for SETUP"); + assert!(log.stops().is_empty()); + peer_setup.set(lite::Setup::default()); + assert!(accept.poll(&kio::Waiter::noop()).is_pending()); + } + let mut group = consumer.recv_group().await.unwrap().unwrap(); + assert_eq!(group.sequence, 3); + let frame = group.read_frame().await.unwrap().unwrap(); + assert_eq!(frame.payload.as_ref(), &[42]); + assert!(group.read_frame().await.unwrap().is_none()); + } + } + } + /// A group at or past the end the publisher declared contradicts that end, which no later /// stream can repair, so the whole track fails rather than ending clean without it. lite-05 /// specified an inclusive end, so there it costs only that group's stream. diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 0ba506db7c..5781e507d0 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -85,7 +85,7 @@ impl Server { runtime: Clock, session: S, version: lite::Version, - client_setup: Option, + client_setup: Option>, peer_hop: Option, ) -> Result<(Session, crate::Driver), Error> where @@ -180,9 +180,9 @@ impl Server { // re-reading the (consumed) Setup Stream. let client_setup = lite::accept_setup(&mut session, version).await?; ( - client_setup.path.clone(), - client_setup.role, - client_setup.hop, + client_setup.setup.path.clone(), + client_setup.setup.role, + client_setup.setup.hop, PausedHandshake::LiteSetup { session, version, @@ -464,7 +464,7 @@ enum PausedHandshake { LiteSetup { session: S, version: lite::Version, - client_setup: lite::Setup, + client_setup: lite::AcceptedSetup, }, /// An IETF (or legacy bidi-SETUP) handshake, boxed where its /// thread-affinity bounds held. The boxing is what keeps [`Handshake`] and @@ -844,6 +844,7 @@ mod tests { uni: Arc>>>, bi: Arc>>>, closed: Arc>>, + stops: Arc>>, } impl FakeSession { @@ -853,6 +854,7 @@ mod tests { uni: Arc::new(Mutex::new(uni.into_iter().collect())), bi: Default::default(), closed: Default::default(), + stops: Default::default(), } } @@ -876,7 +878,10 @@ mod tests { _cx: &mut std::task::Context<'_>, ) -> std::task::Poll> { match self.uni.lock().unwrap().pop_front() { - Some(data) => std::task::Poll::Ready(Ok(FakeRecv { data: data.into() })), + Some(data) => std::task::Poll::Ready(Ok(FakeRecv { + data: data.into(), + stops: self.stops.clone(), + })), None => std::task::Poll::Pending, } } @@ -885,7 +890,13 @@ mod tests { _cx: &mut std::task::Context<'_>, ) -> std::task::Poll> { match self.bi.lock().unwrap().pop_front() { - Some(data) => std::task::Poll::Ready(Ok((FakeSend, FakeRecv { data: data.into() }))), + Some(data) => std::task::Poll::Ready(Ok(( + FakeSend, + FakeRecv { + data: data.into(), + stops: self.stops.clone(), + }, + ))), None => std::task::Poll::Pending, } } @@ -954,6 +965,7 @@ mod tests { struct FakeRecv { data: VecDeque, + stops: Arc>>, } impl web_transport_trait::poll::RecvStream for FakeRecv { type Error = FakeError; @@ -971,7 +983,9 @@ mod tests { } std::task::Poll::Ready(Ok(Some(size))) } - fn stop(&mut self, _code: u32) {} + fn stop(&mut self, code: u32) { + self.stops.lock().unwrap().push(code); + } fn poll_closed(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll> { std::task::Poll::Ready(Ok(())) } @@ -1191,18 +1205,31 @@ mod tests { } #[tokio::test(start_paused = true)] - async fn accept_request_skips_uni_stream_before_setup() { - // A GROUP racing ahead of the SETUP is STOP_SENDING-ed and skipped; the gate - // keeps reading until it finds the SETUP. + async fn accept_request_holds_uni_stream_before_setup() { let session = FakeSession::new( ALPN_LITE_05, [lite05_group(), lite05_setup(Some("/team/room"), None, None)], ); + let stops = session.stops.clone(); let request = Server::new() - .accept_request(tokio::time::Instant::now().into_std(), session) + .accept_request_lite(tokio::time::Instant::now().into_std(), session) .await .unwrap(); assert_eq!(request.path(), "/team/room"); + assert!(stops.lock().unwrap().is_empty(), "the early stream must stay open"); + } + + #[tokio::test(start_paused = true)] + async fn accepted_lite_setup_refuses_a_second_setup_stream() { + let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None), vec![1]]); + let transport = session.clone(); + let request = Server::new() + .accept_request_lite(tokio::time::Instant::now().into_std(), session) + .await + .unwrap(); + let (_session, mut driver) = request.ok().await.unwrap(); + let _ = driver.poll(tokio::time::Instant::now().into_std(), &kio::Waiter::noop()); + assert_eq!(transport.closed(), Some(SessionError::ProtocolViolation.to_code())); } #[tokio::test(start_paused = true)] From 5309d20f6a9bc6311f2115410fd686bbf1ccdd8a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 21:39:56 -0700 Subject: [PATCH 3/7] fix(net): accept setup while earlier streams are incomplete Co-Authored-By: GPT-6 --- rs/moq-net/src/lite/session.rs | 117 ++++++++++++++++++++++++++------- 1 file changed, 94 insertions(+), 23 deletions(-) diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index af32fb7465..0fa6f0d05e 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -33,30 +33,78 @@ pub async fn accept_setup( session: &mut S, version: Version, ) -> Result, Error> { - let mut early = Vec::new(); - loop { - let stream = session.accept_uni().await.map_err(Error::from_transport)?; - let mut reader = Reader::new(stream, version); - let kind = match reader.decode_peek::().await { - Ok(kind) => kind, - Err( - Error::Cancel - | Error::Stream(_) - | Error::Remote(_) - | Error::Transport(_) - | Error::Decode(crate::DecodeError::Short), - ) => continue, - Err(err) => return Err(err), - }; - if kind != DataType::Setup as u64 { - early.push(reader); - continue; - } - reader.decode::().await?; - return Ok(AcceptedSetup { - setup: reader.decode().await?, + let mut accept = SetupAccept:: { + version, + pending: Vec::new(), + early: Vec::new(), + setup: None, + }; + kio::wait(|waiter| accept.poll(session, waiter)).await +} + +struct SetupAccept { + version: Version, + pending: Vec>, + early: Vec>, + setup: Option>, +} + +impl SetupAccept { + fn poll(&mut self, session: &mut S, waiter: &kio::Waiter) -> Poll, Error>> { + let Self { + version, + pending, early, - }); + setup, + } = self; + let version = *version; + + let mut cx = Context::from_waker(waiter.waker()); + while let Poll::Ready(stream) = session.poll_accept_uni(&mut cx) { + pending.push(Reader::new(stream.map_err(Error::from_transport)?, version)); + } + let mut index = 0; + while index < pending.len() { + let kind = match pending[index].poll_decode_peek::(&mut cx) { + Poll::Pending => { + index += 1; + continue; + } + Poll::Ready(Ok(kind)) => kind, + Poll::Ready(Err( + Error::Cancel + | Error::Stream(_) + | Error::Remote(_) + | Error::Transport(_) + | Error::Decode(crate::DecodeError::Short), + )) => { + pending.swap_remove(index); + continue; + } + Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), + }; + let mut reader = pending.swap_remove(index); + if kind != DataType::Setup as u64 { + early.push(reader); + continue; + } + if setup.is_some() { + return Poll::Ready(Err(Error::ProtocolViolation)); + } + ready!(reader.poll_decode::(&mut cx))?; + *setup = Some(reader); + } + let Some(reader) = setup else { + return Poll::Pending; + }; + let setup = ready!(reader.poll_decode::(&mut cx))?; + // A stream need not have delivered even its type yet. Keep it intact for the + // session instead of letting it block SETUP or dropping it at the handoff. + early.append(pending); + Poll::Ready(Ok(AcceptedSetup { + setup, + early: std::mem::take(early), + })) } } @@ -552,6 +600,29 @@ mod tests { use super::*; use crate::coding::Encode; + #[tokio::test(start_paused = true)] + async fn accept_setup_does_not_wait_on_an_early_stream_header() { + let version = Version::Lite05; + let mut setup = Vec::new(); + DataType::Setup.encode(&mut setup, version).unwrap(); + Setup::default().encode(&mut setup, version).unwrap(); + let mut session = + crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_unis(vec![Vec::new(), setup]); + let accepted = accept_setup(&mut session, version).await.unwrap(); + assert_eq!(accepted.early.len(), 1); + } + + #[tokio::test(start_paused = true)] + async fn accept_setup_refuses_two_incomplete_setup_streams() { + use futures::FutureExt; + let mut session = + crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_unis(vec![vec![1], vec![1]]); + assert!(matches!( + accept_setup(&mut session, Version::Lite05).now_or_never(), + Some(Err(Error::ProtocolViolation)) + )); + } + #[tokio::test(start_paused = true)] async fn duplicate_setup_closes_the_session() { for version in [Version::Lite05, Version::Lite06, Version::Lite07] { From e2ce4b5fc3f811eef898454375aa09a107cd65f2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 21:42:44 -0700 Subject: [PATCH 4/7] test(net): inject duplicate setup after the handshake Co-Authored-By: GPT-6 --- rs/moq-net/src/server.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 5781e507d0..0c4884219c 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -1221,13 +1221,14 @@ mod tests { #[tokio::test(start_paused = true)] async fn accepted_lite_setup_refuses_a_second_setup_stream() { - let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None), vec![1]]); + let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None)]); let transport = session.clone(); let request = Server::new() .accept_request_lite(tokio::time::Instant::now().into_std(), session) .await .unwrap(); let (_session, mut driver) = request.ok().await.unwrap(); + transport.uni.lock().unwrap().push_back(vec![1]); let _ = driver.poll(tokio::time::Instant::now().into_std(), &kio::Waiter::noop()); assert_eq!(transport.closed(), Some(SessionError::ProtocolViolation.to_code())); } From d7d5d57cd4bdb22640bb8d65900666f0c088d39e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 22:21:35 -0700 Subject: [PATCH 5/7] fix(lite): read buffered setup before transport close Co-Authored-By: GPT-6 --- rs/moq-net/src/lite/session.rs | 24 ++++++++++++++++++++---- rs/moq-net/src/server.rs | 12 ++++++++++++ 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index 0fa6f0d05e..38d57b8c47 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -60,8 +60,15 @@ impl SetupAccept { let version = *version; let mut cx = Context::from_waker(waiter.waker()); + let mut accept_error = None; while let Poll::Ready(stream) = session.poll_accept_uni(&mut cx) { - pending.push(Reader::new(stream.map_err(Error::from_transport)?, version)); + match stream { + Ok(stream) => pending.push(Reader::new(stream, version)), + Err(err) => { + accept_error = Some(Error::from_transport(err)); + break; + } + } } let mut index = 0; while index < pending.len() { @@ -94,10 +101,19 @@ impl SetupAccept { ready!(reader.poll_decode::(&mut cx))?; *setup = Some(reader); } - let Some(reader) = setup else { - return Poll::Pending; + // A closed transport can still yield a complete buffered SETUP. Decode it + // before reporting the terminal error from accepting the next stream. + let result = setup + .as_mut() + .map(|reader| reader.poll_decode::(&mut cx)) + .unwrap_or(Poll::Pending); + let setup = match result { + Poll::Ready(result) => result?, + Poll::Pending => match accept_error { + Some(err) => return Poll::Ready(Err(err)), + None => return Poll::Pending, + }, }; - let setup = ready!(reader.poll_decode::(&mut cx))?; // A stream need not have delivered even its type yet. Keep it intact for the // session instead of letting it block SETUP or dropping it at the handoff. early.append(pending); diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 0c4884219c..89852bf3d9 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -882,6 +882,7 @@ mod tests { data: data.into(), stops: self.stops.clone(), })), + None if self.closed().is_some() => std::task::Poll::Ready(Err(FakeError)), None => std::task::Poll::Pending, } } @@ -1219,6 +1220,17 @@ mod tests { assert!(stops.lock().unwrap().is_empty(), "the early stream must stay open"); } + #[tokio::test(start_paused = true)] + async fn accept_request_reads_buffered_setup_after_transport_close() { + let mut session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some("/closed"), None, None)]); + web_transport_trait::poll::Session::close(&mut session, SessionError::Cancel.to_code(), "closed"); + let request = Server::new() + .accept_request_lite(tokio::time::Instant::now().into_std(), session) + .await + .unwrap(); + assert_eq!(request.path(), "/closed"); + } + #[tokio::test(start_paused = true)] async fn accepted_lite_setup_refuses_a_second_setup_stream() { let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None)]); From 01593bda70a86c002d4b6f625c33bcb8ff5d3c10 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 09:23:29 -0700 Subject: [PATCH 6/7] fix(lite): let client GROUP streams proceed before SETUP The draft buffers only extension-dependent streams; GROUP needs none. Matches JS, which never gated it. Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 6 +++--- rs/moq-net/src/lite/subscriber.rs | 24 ++++++++---------------- 2 files changed, 11 insertions(+), 19 deletions(-) diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 2784004134..2f0dda79e5 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -29,9 +29,9 @@ implement in an afternoon. The wire spec is A dedicated ALPN selects the wire version for moq-lite 03 and newer. The legacy `moql` ALPN negotiates moq-lite 01 or 02 via `SETUP`. In moq-lite 05 and newer, each side also sends a `SETUP` message with its capabilities. -Rust holds unidirectional streams arriving before the peer's `SETUP` until the -peer's `SETUP` arrives. Streams that depend on negotiated capabilities wait for -`SETUP` in both Rust and JavaScript. A second Setup Stream closes the session +Streams that depend on negotiated capabilities wait for the peer's `SETUP`; +everything else proceeds immediately. A Rust server holds unidirectional streams +that arrive before the client's `SETUP`, since it routes by the path in `SETUP`. A second Setup Stream closes the session with `PROTOCOL_VIOLATION`. Rust and TypeScript speak moq-lite 01 through 06 and moq-transport drafts 14 through 22. Clients offer `moq-lite-06` first by default. moq-lite 07 is diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index ce88a4e8f5..bac8089d8a 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -712,16 +712,11 @@ impl UniServe { let mut cx = std::task::Context::from_waker(waiter.waker()); // A decode error here is only logged; the peer hung up or spoke garbage // before the stream had a type. - let kind = ready!(reader.poll_decode_peek::(&mut cx))?; - if self.subscriber.version.has_setup_stream() { - match kind { - lite::DataType::Setup => self.subscriber.peer_setup.claim()?, - lite::DataType::Group => { - ready!(self.subscriber.peer_setup.poll_probe_level(waiter)); - } - } + let kind = ready!(reader.poll_decode::(&mut cx))?; + // Claim before decoding the body, so two incomplete SETUPs are duplicates too. + if matches!(kind, lite::DataType::Setup) && self.subscriber.version.has_setup_stream() { + self.subscriber.peer_setup.claim()?; } - ready!(reader.poll_decode::(&mut cx))?; let UniState::Start { reader } = std::mem::replace(&mut self.state, UniState::Done) else { unreachable!() }; @@ -1494,8 +1489,10 @@ mod tests { ); } + /// A GROUP needs no negotiated extension, so it proceeds before SETUP, or after a + /// server's gated accept held it. #[tokio::test(start_paused = true)] - async fn early_group_is_delivered_after_setup() { + async fn early_group_is_delivered() { for version in [Version::Lite05, Version::Lite06, Version::Lite07] { for pre_read in [false, true] { let mut bytes = Vec::new(); @@ -1553,12 +1550,7 @@ mod tests { ); let mut accept = UniAccept::new(subscriber, early); assert!(accept.poll(&kio::Waiter::noop()).is_pending()); - if !pre_read { - assert!(consumer.recv_group().now_or_never().is_none(), "GROUP waits for SETUP"); - assert!(log.stops().is_empty()); - peer_setup.set(lite::Setup::default()); - assert!(accept.poll(&kio::Waiter::noop()).is_pending()); - } + assert!(log.stops().is_empty()); let mut group = consumer.recv_group().await.unwrap().unwrap(); assert_eq!(group.sequence, 3); let frame = group.read_frame().await.unwrap().unwrap(); From 0f6a8d1f9feff5ddc8c16421e4eb9a1ae94561b1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 23:01:42 -0700 Subject: [PATCH 7/7] fix(net): close the session when a claimed SETUP fails to decode Once the Setup Stream slot is claimed, no other SETUP can arrive, so a truncated or malformed body left SETUP-gated waiters pending forever. Close the session with PROTOCOL_VIOLATION in Rust and JS, like a duplicate SETUP. Main's raw QUIC stream-code test relied on early uni streams being refused; it now sends SETUP and reads the CANCEL code from a GROUP for an unknown subscription. Also fixes a floating promise from main's async Connection.close() in the duplicate-SETUP test. Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/connection.test.ts | 17 +++++++++++++++- js/net/src/lite/connection.ts | 6 +++++- rs/moq-net/src/lite/session.rs | 32 ++++++++++++++++++++++++++++++ rs/moq-net/src/lite/subscriber.rs | 5 ++++- rs/moq-tokio/tests/stream_code.rs | 25 +++++++++++++---------- 5 files changed, 72 insertions(+), 13 deletions(-) diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index 9bb23e92d0..4cc3748703 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -108,8 +108,23 @@ for (const [alpn, version] of [ } expect((await pair.client.closed).closeCode).toBe(SessionCode.ProtocolViolation); } finally { - connection.close(); + connection.abort(); } }); } + + // The Setup Stream is claimed before its body decodes, so a truncated one leaves + // nothing for SETUP-gated streams to wait on. + test(`truncated SETUP closes ${alpn} with PROTOCOL_VIOLATION`, async () => { + const pair = createMockTransportPair(alpn); + const connection = new Connection({ url: new URL("https://relay.example/"), quic: pair.client, version }); + try { + const writer = await Writer.open(pair.server, { version }); + await writer.u53(DataType.Setup); + writer.close(); + expect((await pair.client.closed).closeCode).toBe(SessionCode.ProtocolViolation); + } finally { + connection.abort(); + } + }); } diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index 0a18fcdf11..6e3faf45b2 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -322,7 +322,11 @@ export class Connection implements Established { this.#setupSeen = true; // The peer sends exactly one SETUP, then FINs. Record it so capability-gated // streams (e.g. PROBE) can react. - const setup = await Setup.decode(stream, this.#version); + // The slot is claimed, so no other SETUP can arrive and the streams waiting on it + // would hang. The session cannot continue. + const setup = await Setup.decode(stream, this.#version).catch((err: unknown) => { + throw new ProtocolViolation("invalid SETUP", { cause: err }); + }); this.#peerSetup.set(setup); this.#peerRole.set(setup.role); } else { diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index ba29b1cf1a..be643cbeae 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -678,4 +678,36 @@ mod tests { } } } + + /// A SETUP that ends before its body decodes leaves nothing to wait on, since the + /// Setup Stream is already claimed. + #[tokio::test(start_paused = true)] + async fn truncated_setup_closes_the_session() { + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + let mut truncated = Vec::new(); + DataType::Setup.encode(&mut truncated, version).unwrap(); + let session = + crate::lite::test_transport::ScriptedSession::eof(Vec::new()).with_incoming_unis(vec![truncated]); + let log = session.log.clone(); + let mut started = start(Config { + runtime: crate::time::Clock::tokio(), + session, + setup_stream: None, + publish: None, + subscribe: None, + peer_hop: None, + version, + our_setup: Setup::default(), + peer_setup: None, + }) + .unwrap(); + let _ = started.driver.poll(&kio::Waiter::noop()); + assert!( + log.closes() + .iter() + .any(|(code, _)| *code == SessionError::ProtocolViolation.to_code()), + "{version:?}: a truncated SETUP must close the session" + ); + } + } } diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 9fec980387..474af286bd 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -741,9 +741,12 @@ impl UniServe { self.subscriber.peer_setup.set(setup); return Poll::Ready(Ok(())); } + // The slot is claimed, so no other SETUP can arrive and the streams + // waiting on it would hang. The session cannot continue. Err(err) => { + tracing::debug!(%err, "failed to read peer setup"); self.abort(&err); - return Poll::Ready(Ok(())); + return Poll::Ready(Err(Error::ProtocolViolation)); } } } diff --git a/rs/moq-tokio/tests/stream_code.rs b/rs/moq-tokio/tests/stream_code.rs index 23775dc55f..922fe36488 100644 --- a/rs/moq-tokio/tests/stream_code.rs +++ b/rs/moq-tokio/tests/stream_code.rs @@ -2,9 +2,9 @@ //! mapped into the HTTP/3 WebTransport code space, so other raw QUIC MoQ stacks //! read it. //! -//! The peer here is a plain QUIC client. A moq-lite server refuses a uni stream opened -//! ahead of the SETUP with STOP_SENDING, and the client reads that code off the wire. -//! moq-transport holds such a stream instead, but its codes take the same raw session. +//! The peer here is a plain QUIC client. A moq-lite server cancels a GROUP stream for a +//! subscription it never made with STOP_SENDING, and the client reads that code off the +//! wire. moq-transport codes take the same raw session. #![cfg(feature = "noq")] @@ -12,8 +12,8 @@ use std::{sync::Arc, time::Duration}; use web_transport_moq::noq; -/// moq-lite sends INTERNAL_ERROR (0x0) for a refused stream. -const REFUSED: u32 = 0x0; +/// moq-lite sends CANCEL (0x1) for a group it has no subscription for. +const CANCEL: u32 = 0x1; struct Server { port: u16, @@ -66,8 +66,8 @@ async fn serve() -> Server { } } -/// Dial the server over raw QUIC with `alpn`, open a uni stream that is not a SETUP, -/// and return the code the server stopped it with. +/// Dial the server over raw QUIC with `alpn`, send an empty SETUP and a GROUP for an +/// unknown subscription, and return the code the server stopped the GROUP with. async fn refused_code(server: &Server, alpn: &str) -> Option { let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); @@ -90,9 +90,14 @@ async fn refused_code(server: &Server, alpn: &str) -> Option { .await .expect("handshake"); - // A group stream, which moq-lite refuses before the SETUP. + // SETUP type, then a one-byte message holding zero parameters. + let mut setup = conn.open_uni().await.expect("open"); + setup.write_all(&[0x01, 0x01, 0x00]).await.expect("write"); + setup.finish().expect("finish"); + + // GROUP type, then a three-byte header: subscribe 7, sequence 0, frame start 0. let mut send = conn.open_uni().await.expect("open"); - send.write_all(&[0x00]).await.expect("write"); + send.write_all(&[0x00, 0x03, 0x07, 0x00, 0x00]).await.expect("write"); tokio::time::timeout(Duration::from_secs(10), send.stopped()) .await @@ -103,5 +108,5 @@ async fn refused_code(server: &Server, alpn: &str) -> Option { #[tokio::test] async fn a_refused_stream_code_is_not_mapped() { let server = serve().await; - assert_eq!(refused_code(&server, "moq-lite-06").await, Some(REFUSED.into())); + assert_eq!(refused_code(&server, "moq-lite-06").await, Some(CANCEL.into())); }