diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index a69c9f7595..2ac125e377 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. +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 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 501f461c27..4cc3748703 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,44 @@ test("a lite GOAWAY keeps the session open, and a second one closes it", async ( connection.abort(); } }); + +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.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 a613690487..6e3faf45b2 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"; @@ -89,6 +98,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). @@ -294,6 +304,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 }); + } }); } } @@ -304,9 +317,16 @@ 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. - const setup = await Setup.decode(stream, this.#version); + // streams (e.g. PROBE) can react. + // 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/quest/m0/README.md b/quest/m0/README.md index 4f2124588c..0239a19337 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -40,7 +40,6 @@ a published `@moq/watch` break. - [Branch flip](/quest/m0/branch-flip.md) - `dev` becomes the default `main` trunk and today's `main` becomes `release`, where publishing runs - [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 -- [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) - iroh's upstream noq carries quinn's stream reassembly cap, once n0 releases it - [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 f9552fbe84..be643cbeae 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -24,31 +24,114 @@ 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 { - 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), +) -> Result, Error> { + 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()); + let mut accept_error = None; + while let Poll::Ready(stream) = session.poll_accept_uni(&mut cx) { + 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() { + 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); + } + // 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, + }, + }; + // 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), + })) } } +/// 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 +168,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 +234,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 +282,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, }; @@ -529,3 +615,99 @@ impl SendGoaway { } } } + +#[cfg(test)] +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] { + 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" + ); + } + } + } + + /// 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/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 137aa17c48..474af286bd 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, } } @@ -679,6 +686,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(()) } @@ -702,6 +715,10 @@ impl UniServe { // 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))?; + // 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()?; + } let UniState::Start { reader } = std::mem::replace(&mut self.state, UniState::Done) else { unreachable!() }; @@ -724,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)); } } } @@ -1524,6 +1544,77 @@ 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() { + 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()); + 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(); + 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 777f9df25f..dea6c7b0eb 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, @@ -465,7 +465,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 @@ -846,6 +846,7 @@ mod tests { uni: Arc>>>, bi: Arc>>>, closed: Arc>>, + stops: Arc>>, } impl FakeSession { @@ -855,6 +856,7 @@ mod tests { uni: Arc::new(Mutex::new(uni.into_iter().collect())), bi: Default::default(), closed: Default::default(), + stops: Default::default(), } } @@ -878,7 +880,11 @@ 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 if self.closed().is_some() => std::task::Poll::Ready(Err(FakeError)), None => std::task::Poll::Pending, } } @@ -887,7 +893,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, } } @@ -956,6 +968,7 @@ mod tests { struct FakeRecv { data: VecDeque, + stops: Arc>>, } impl web_transport_trait::poll::RecvStream for FakeRecv { type Error = FakeError; @@ -973,7 +986,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(())) } @@ -1193,18 +1208,43 @@ 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 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)]); + 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())); } #[tokio::test(start_paused = true)] 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())); }