diff --git a/js/net/src/connection/accept.ts b/js/net/src/connection/accept.ts index fb8adb089c..4a6ea050ef 100644 --- a/js/net/src/connection/accept.ts +++ b/js/net/src/connection/accept.ts @@ -115,7 +115,7 @@ async function acceptAlpn( version: Ietf.IetfVersion, wiring: SessionProps, ): Promise { - const { control, solicit, hidden, cluster } = await exchangeSetup(transport, version, "moq-lite-js"); + const { control, early, solicit, hidden, cluster } = await exchangeSetup(transport, version, "moq-lite-js"); return new Ietf.Connection({ ...wiring, @@ -123,6 +123,7 @@ async function acceptAlpn( url, quic: transport, control, + early, solicit, hidden, cluster, diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index a31dcc6dee..28fce7e376 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -362,7 +362,7 @@ async function handshakeAlpn( version: Ietf.IetfVersion, wiring: SessionProps, ): Promise { - const { control, solicit, hidden, cluster } = await exchangeSetup(session, version, "moq-lite-js"); + const { control, early, solicit, hidden, cluster } = await exchangeSetup(session, version, "moq-lite-js"); return new Ietf.Connection({ ...wiring, @@ -370,6 +370,7 @@ async function handshakeAlpn( url, quic: session, control, + early, solicit, hidden, cluster, diff --git a/js/net/src/connection/handshake.ts b/js/net/src/connection/handshake.ts index 05cd214c8b..81bb2052a9 100644 --- a/js/net/src/connection/handshake.ts +++ b/js/net/src/connection/handshake.ts @@ -1,6 +1,7 @@ +import { StreamError } from "../error.ts"; import { type Hop, randomHop } from "../hop.ts"; import * as Ietf from "../ietf/index.ts"; -import { Reader, Stream, Writer } from "../stream.ts"; +import { Reader, Stream, UnexpectedEnd, Writer } from "../stream.ts"; /** * Draft-17+ SETUP exchange. Each side opens a uni stream, writes its Setup @@ -8,7 +9,8 @@ import { Reader, Stream, Writer } from "../stream.ts"; * halves run in parallel and the protocol is symmetric, so both `connect` * (client) and `accept` (server) use this same function. * - * Returns the control stream plus what the peer's SETUP declared: whether it requires + * Returns the control stream, the uni streams that arrived before it (see + * {@link receiveSetup}), plus what the peer's SETUP declared: whether it requires * solicitation, which decides whether we announce namespaces unprompted (see the MoQ Solicit * extension), and its Hop ID (see the MoQ Cluster extension). We declare both ourselves on * every session: we send SUBSCRIBE_NAMESPACE for each prefix we want, so an unsolicited @@ -19,7 +21,13 @@ export async function exchangeSetup( transport: WebTransport, version: Ietf.IetfVersion, implementation: string, -): Promise<{ control: Stream; solicit: boolean | undefined; hidden: boolean; cluster: Ietf.Cluster.Hops }> { +): Promise<{ + control: Stream; + early: Reader[]; + solicit: boolean | undefined; + hidden: boolean; + cluster: Ietf.Cluster.Hops; +}> { const encoder = new TextEncoder(); const params = new Ietf.SetupOptions(); params.setBytes(Ietf.SetupOption.Implementation, encoder.encode(implementation)); @@ -40,6 +48,7 @@ export async function exchangeSetup( return { control: new Stream({ writer, reader: received.reader }), + early: received.early, solicit: received.solicit, hidden: received.hidden, cluster: { self, peer: received.cluster }, @@ -55,27 +64,59 @@ async function sendSetup(transport: WebTransport, version: Ietf.IetfVersion, set return writer; } +/** + * Read the peer's SETUP off its uni stream. Any other uni stream that beats it (padding, + * or group data for a subscription the peer already holds) is held, type unread, for the + * session to classify once it starts: the drafts say to buffer early data rather than + * reject it. QUIC stream credit bounds how many can pile up. + */ async function receiveSetup( transport: WebTransport, version: Ietf.IetfVersion, -): Promise<{ reader: Reader; solicit: boolean | undefined; hidden: boolean; cluster: Hop | undefined }> { +): Promise<{ + reader: Reader; + early: Reader[]; + solicit: boolean | undefined; + hidden: boolean; + cluster: Hop | undefined; +}> { const uniReader = transport.incomingUnidirectionalStreams.getReader() as ReadableStreamDefaultReader< ReadableStream >; - const next = await uniReader.read(); - uniReader.releaseLock(); - if (next.done) throw new Error("no incoming uni stream for SETUP"); + const early: Reader[] = []; - const reader = new Reader(next.value, undefined, version); + let reader: Reader; + try { + for (;;) { + const next = await uniReader.read(); + if (next.done) throw new Error("no incoming uni stream for SETUP"); - const streamType = await reader.u53(); - if (streamType !== Ietf.Setup.id) { - throw new Error(`unexpected stream type on setup uni: 0x${streamType.toString(16)}`); + const stream = new Reader(next.value, undefined, version); + // A stream that ended or reset before its full type is that stream's failure, not the session's. + // Malformed bytes are still the peer's fault, so a bad type fails the handshake. + if (await stream.done().catch(() => true)) continue; + // Full width, so an unknown type past 2^53 is held for the classifier to refuse. + const streamType = await stream.peekU62().catch((err: unknown) => { + if (err instanceof StreamError || err instanceof UnexpectedEnd) return undefined; + throw err; + }); + if (streamType === undefined) continue; + if (streamType === BigInt(Ietf.Setup.id)) { + reader = stream; + break; + } + early.push(stream); + } + } finally { + uniReader.releaseLock(); } + + await reader.u53(); // the SETUP type, peeked above const setup = await Ietf.Setup.decode(reader, version); return { reader, + early, solicit: Ietf.solicitFromSetup(setup.parameters), hidden: Ietf.hiddenFromSetup(setup.parameters), cluster: Ietf.Cluster.fromSetup(setup.parameters, version), diff --git a/js/net/src/ietf/connection.test.ts b/js/net/src/ietf/connection.test.ts index 8b65bc6db1..a9c171321d 100644 --- a/js/net/src/ietf/connection.test.ts +++ b/js/net/src/ietf/connection.test.ts @@ -1,8 +1,12 @@ import { expect, spyOn, test } from "bun:test"; +import { exchangeSetup } from "../connection/handshake.ts"; import { SessionCode } from "../error.ts"; import { createMockTransportPair } from "../mock.ts"; import { Stream, Writer } from "../stream.ts"; import { Connection } from "./connection.ts"; +import { Group } from "./object.ts"; +import { SetupOption, SetupOptions } from "./parameters.ts"; +import { Setup } from "./setup.ts"; import { ALPN, type IetfVersion, Version } from "./version.ts"; const PADDING = 0x132b3e28n; @@ -76,3 +80,112 @@ test("an unknown uni stream type closes the session", async () => { } } }); + +/** + * Padding and group streams that beat the peer's SETUP are held and classified once it + * lands. The drafts say to buffer early data; failing the handshake over it broke a peer + * that already held a subscription or probed for bandwidth straight away. + */ +test("uni streams before SETUP are held until it lands", async () => { + const version = Version.DRAFT_19; + const pair = createMockTransportPair(ALPN.DRAFT_19); + + const padding = new Writer(await pair.client.createUnidirectionalStream(), version); + await padding.u62(PADDING); + await padding.write(new Uint8Array(4096)); + padding.close(); + + const group = new Writer(await pair.client.createUnidirectionalStream(), version); + await new Group({ + trackAlias: 7n, + groupId: 0, + subGroupId: 0, + publisherPriority: 128, + flags: { + hasExtensions: false, + hasSubgroup: false, + hasSubgroupObject: false, + hasEnd: false, + hasPriority: true, + firstObject: true, + }, + }).encode(group, version); + + const setup = new Writer(await pair.client.createUnidirectionalStream(), version); + await setup.u53(Setup.id); + const parameters = new SetupOptions(); + parameters.setBytes(SetupOption.Implementation, new TextEncoder().encode("test")); + await new Setup({ parameters }).encode(setup, version); + + const { control, early, solicit, hidden, cluster } = await exchangeSetup(pair.server, version, "test"); + expect(early.length).toBe(2); + + let closed = false; + void pair.server.closed.then(() => { + closed = true; + }); + + const connection = new Connection({ + url: new URL("https://example.com"), + quic: pair.server, + control, + early, + solicit, + hidden, + cluster, + maxRequestId: 0n, + version, + client: false, + }); + + try { + // A STOP_SENDING would reject this; a clean close means the padding was read to its end. + await padding.closed; + + await Bun.sleep(0); + expect(closed).toBe(false); + } finally { + connection.close(); + } +}); + +/** An early stream that dies before its type is skipped, but a malformed type is still fatal. */ +test("malformed uni stream before SETUP fails the handshake", async () => { + const version = Version.DRAFT_17; + const pair = createMockTransportPair(ALPN.DRAFT_17); + + const empty = new Writer(await pair.client.createUnidirectionalStream(), version); + empty.close(); + + // 0b1111110x is a reserved varint prefix on draft-17. + const malformed = new Writer(await pair.client.createUnidirectionalStream(), version); + await malformed.write(new Uint8Array([0xfc, 0, 0, 0, 0, 0, 0, 0, 0])); + + const setup = new Writer(await pair.client.createUnidirectionalStream(), version); + await setup.u53(Setup.id); + const parameters = new SetupOptions(); + parameters.setBytes(SetupOption.Implementation, new TextEncoder().encode("test")); + await new Setup({ parameters }).encode(setup, version); + + await expect(exchangeSetup(pair.server, version, "test")).rejects.toThrow("leading-ones varint"); +}); + +/** An early stream that ends partway through its type died; it is skipped like an empty one. */ +test("truncated uni stream before SETUP is skipped", async () => { + const version = Version.DRAFT_17; + const pair = createMockTransportPair(ALPN.DRAFT_17); + + // A two-byte varint prefix, then FIN. + const truncated = new Writer(await pair.client.createUnidirectionalStream(), version); + await truncated.write(new Uint8Array([0x80])); + truncated.close(); + + const setup = new Writer(await pair.client.createUnidirectionalStream(), version); + await setup.u53(Setup.id); + const parameters = new SetupOptions(); + parameters.setBytes(SetupOption.Implementation, new TextEncoder().encode("test")); + await new Setup({ parameters }).encode(setup, version); + + const { early } = await exchangeSetup(pair.server, version, "test"); + expect(early.length).toBe(0); +}); diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 4dde38159b..579744cf9c 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -83,6 +83,7 @@ export class Connection implements Established { * @param version - The negotiated protocol version * @param solicit - What the peer's SETUP declared (undefined when it declared nothing) * @param cluster - The Hop IDs the SETUP exchange settled, on the versions that negotiate them + * @param early - Uni streams that arrived before the peer's SETUP * * @internal */ @@ -98,6 +99,7 @@ export class Connection implements Established { solicit, hidden = false, cluster, + early = [], }: { url: URL; quic: WebTransport; @@ -121,6 +123,8 @@ export class Connection implements Established { * cannot negotiate the extension, as is a `peer` the peer never declared. */ cluster?: Cluster.Hops; + /** Uni streams that arrived before the peer's SETUP, type unread (v17+). */ + early?: Reader[]; }) { this.url = url; this.discovery = discovery; @@ -159,7 +163,7 @@ export class Connection implements Established { this.#subscriber = new Subscriber({ session: this.#session, quic, cluster, hidden }); registerWire(this, { consume: (path) => this.#subscriber.consume(path), goaway: this.#goaway }); - void this.#run(); + void this.#run(early); } /** Snapshot the transport's counters; see {@link Established.stats}. */ @@ -198,9 +202,9 @@ export class Connection implements Established { this.#close({ closeCode: SessionCode.ProtocolViolation, reason: err.message }); } - async #run(): Promise { + async #run(early: Reader[]): Promise { try { - await Promise.all([this.#runBidis(), this.#runUnis(), this.#publisher.runPublishNamespaces()]); + await Promise.all([this.#runBidis(), this.#runUnis(early), this.#publisher.runPublishNamespaces()]); } catch (err) { if (!this.#closed) { console.error("fatal error running connection", err); @@ -323,25 +327,30 @@ export class Connection implements Established { /** * Handles unidirectional streams for media delivery (groups). */ - async #runUnis() { - const readers = new Readers(this.#quic, this.#session.version); + async #runUnis(early: Reader[]) { + // Streams that beat the SETUP go first, in arrival order. + for (const stream of early) this.#spawnUni(stream); + const readers = new Readers(this.#quic, this.#session.version); for (;;) { const stream = await readers.next(); if (!stream) break; + this.#spawnUni(stream); + } + } - this.#runUni(stream) - .then(() => { - stream.stop(new StreamError(StreamCode.Cancel, { message: "cancel" })); - }) - .catch((err: unknown) => { - console.error("error processing object stream", err); - stream.stop(err); + #spawnUni(stream: Reader) { + this.#runUni(stream) + .then(() => { + stream.stop(new StreamError(StreamCode.Cancel, { message: "cancel" })); + }) + .catch((err: unknown) => { + console.error("error processing object stream", err); + stream.stop(err); - // An unknown or invalid stream type MUST close the session, not just the stream. - if (err instanceof ProtocolViolation) this.#violated(err); - }); - } + // An unknown or invalid stream type MUST close the session, not just the stream. + if (err instanceof ProtocolViolation) this.#violated(err); + }); } async #runUni(stream: Reader) { diff --git a/js/net/src/stream.ts b/js/net/src/stream.ts index e4c8401c5a..44b0aaea61 100644 --- a/js/net/src/stream.ts +++ b/js/net/src/stream.ts @@ -304,7 +304,7 @@ export class Reader { while (this.#buffer.byteLength + this.#chunked < size) { if (!(await this.#fill())) { - throw new Error("unexpected end of stream"); + throw new UnexpectedEnd(); } } @@ -364,13 +364,22 @@ export class Reader { } } + // Like decode, but leaves the bytes buffered for the next read. + async #peek(decode: (c: Cursor) => T): Promise { + for (;;) { + const result = this.#try(decode, false); + if (!(result instanceof Short)) return result; + await this.#fillTo(result.need); + } + } + /** Like {@link decode}, but returns undefined if the stream ends cleanly first. */ async decodeMaybe(decode: (c: Cursor) => T): Promise { if (await this.done()) return undefined; return this.decode(decode); } - #try(decode: (c: Cursor) => T): T | Short { + #try(decode: (c: Cursor) => T, consume = true): T | Short { // A retry of the decode that last ran short, before the bytes it needs have arrived, // would only throw again. Every decode reads at least a byte, so none can succeed on // an empty buffer either. @@ -382,7 +391,7 @@ export class Reader { const cursor = new Cursor(this.#buffer, this.version); try { const result = decode(cursor); - this.#slice(cursor.offset); + if (consume) this.#slice(cursor.offset); this.#short = undefined; return result; } catch (err: unknown) { @@ -442,6 +451,11 @@ export class Reader { return this.decode(U62); } + /** Like {@link u62}, but leaves the varint buffered, so a stream's type can be read twice. */ + async peekU62(): Promise { + return this.#peek(U62); + } + async varint(): Promise { return this.decode(VARINT); } @@ -466,6 +480,13 @@ export class Reader { } } +/** The stream ended cleanly partway through a read. */ +export class UnexpectedEnd extends Error { + constructor() { + super("unexpected end of stream"); + } +} + // Thrown by a Cursor read that runs past the buffered bytes, carrying how many bytes from the // start of the buffer the decode needs. Not an Error: it ends every chunk, so it must not // capture a stack. diff --git a/quest/m0/README.md b/quest/m0/README.md index 2781ce6732..d0eab40500 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -40,7 +40,6 @@ Published API or wire breaks still land on dev; each quest's Plan says so. ## Required - [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 -- [IETF early streams](/quest/m0/ietf-early-streams.md) - a moq-transport stream that arrives before SETUP is held until SETUP lands, never aborted - [SUBSCRIBE_TRACKS refusal](/quest/m0/ietf-subscribe-tracks.md) - a draft-18+ SUBSCRIBE_TRACKS gets NOT_SUPPORTED on its stream, not a session close - [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 diff --git a/quest/m0/ietf-early-streams.md b/quest/m0/ietf-early-streams.md deleted file mode 100644 index 8c2af0a245..0000000000 --- a/quest/m0/ietf-early-streams.md +++ /dev/null @@ -1,34 +0,0 @@ -# [S] IETF streams that arrive before SETUP are held - -## Goal - -A moq-transport stream that arrives before SETUP completes is held and -handled once SETUP lands, never aborted and never fatal, in Rust and JS, on -every draft the tree negotiates. - -## Plan - -Facts (`origin/main`, 2026-09-30): - -- Rust server, drafts 17+: `accept_setup` aborts every uni stream that is not - SETUP (0x2F00) with `UnexpectedStream`, which reaches the wire as - INTERNAL_ERROR. Bidi streams wait in the transport queue. -- JS: `receiveSetup` reads the first incoming uni stream and fails the whole - handshake if it is not SETUP. -- Drafts 16 to 21 say early data SHOULD be buffered until the control - streams arrive, and permit resetting only bidi streams. Drafts 18+ add that - parameters needing negotiation SHOULD NOT be used before the peer's SETUP. -- Group and fetch decoding needs only the draft version, which ALPN fixes; - what needs SETUP state is PUBLISH_NAMESPACE (cluster), solicit, and hidden. - -Decided (2026-09-30): queue non-SETUP uni streams during the handshake and -hand them to the normal classifier (padding, group, fetch, unknown, per -the uni-stream classifier from moq-dev/moq#4603) once SETUP lands. -QUIC stream credit already bounds the queue, so no extra cap. Do the same in -the JS handshake. Bidi streams keep waiting as today. Check draft 22's text -when it is reachable. - -Tests in both languages: a padding stream and a group stream that arrive -before SETUP are handled after it, and the session opens. - -Public API: none. Wire: none; fixes conformance. diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 8531ea788c..7574b768ca 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -276,6 +276,7 @@ impl Client { authority: self.setup_authority.clone(), peer_setup_stream: None, peer_declared: None, + early_unis: Vec::new(), })?; tracing::debug!(version = ?v, "connected"); @@ -424,6 +425,7 @@ impl Client { authority: None, peer_setup_stream: None, peer_declared: Some(peer_declared), + early_unis: Vec::new(), })?; (None, crate::driver::Protocol::Ietf(protocol), goaway) } diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 66cfe5683f..e52497310e 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -65,6 +65,10 @@ pub struct Config { /// What that pre-read SETUP declared, so the session does not have to parse it /// twice. `None` when [`Self::peer_setup_stream`] is. pub peer_declared: Option, + + /// Uni streams that arrived before that pre-read SETUP (see [`PeerSetup::early`]), + /// classified by the session before it accepts any more. + pub early_unis: Vec>, } pub(crate) struct Driver { @@ -98,6 +102,7 @@ where authority, peer_setup_stream, peer_declared, + early_unis, } = config; // GOAWAY wiring: the public Session holds one half (drain trigger, received @@ -210,6 +215,7 @@ where UniSetup { peer: None, read: false, + early: Vec::new(), version, goaway, local_close: closing.clone() @@ -361,6 +367,7 @@ where UniSetup { peer: Some(peer_setup.clone()), read: setup_read, + early: early_unis, version, goaway, local_close: closing.clone() @@ -484,6 +491,11 @@ pub struct PeerSetup { /// The Setup Options it declared (see [`cluster`] and [`solicit`]). pub declared: peer::Peer, + + /// Uni streams that arrived before the SETUP, type peeked but unread. The drafts say + /// to buffer early data until the control streams arrive, so the session classifies + /// these once it starts. QUIC stream credit bounds how many can pile up. + pub early: Vec>, } /// The Hop ID this session declares and detects loops against. @@ -501,23 +513,32 @@ fn self_origin(publish: Option<&origin::Consumer>, subscribe: Option<&origin::Pr /// Server (draft-17+): read the peer's SETUP off its uni stream before starting the /// session, returning that stream plus what it declared. /// -/// Blocks on the peer's Setup Stream; any other uni stream racing ahead of it is -/// `STOP_SENDING`-ed and skipped (group data needs a prior subscribe, so nothing -/// legitimate precedes the SETUP at connect). Pass the returned reader to [`start`] -/// as its `peer_setup_stream` so GOAWAY monitoring continues without re-reading it. +/// Blocks on the peer's Setup Stream. Any other uni stream racing ahead of it (padding, +/// or group data for a subscription the peer already holds) is held in +/// [`PeerSetup::early`] for the session to classify. Pass the returned reader to +/// [`start`] as its `peer_setup_stream` so GOAWAY monitoring continues without +/// re-reading it. pub async fn accept_setup( session: &mut S, version: Version, ) -> Result, Error> { let outer_version = crate::Version::Ietf(version); + let mut early = Vec::new(); loop { let recv = session.accept_uni().await.map_err(Error::from_transport)?; let mut reader: Reader = Reader::new(recv, outer_version); - if reader.decode_peek::().await? != setup::SETUP_V17 { - // Not the SETUP (group data this early is unexpected). Reject and keep waiting. - reader.abort(&Error::UnexpectedStream); + let kind = match reader.decode_peek::().await { + Ok(kind) => kind, + Err(err) if died_before_header(&err) => { + tracing::debug!(%err, "dropping uni stream that died before its type"); + continue; + } + Err(err) => return Err(err), + }; + if kind != setup::SETUP_V17 { + early.push(reader); continue; } @@ -540,6 +561,7 @@ pub async fn accept_setup( path, token, declared, + early, }); } } @@ -688,22 +710,25 @@ impl UniType { } /// Setup and lifecycle state for the incoming uni stream dispatcher. -struct UniSetup { +struct UniSetup { peer: Option, read: bool, + /// Streams accepted before the session started, handled before accepting more. + early: Vec>, version: Version, goaway: crate::goaway::Protocol, local_close: std::sync::Arc, } /// Accept incoming uni streams, including SETUP and GOAWAY on draft-17+. -async fn run_unis(mut session: S, subscriber: Subscriber, setup: UniSetup) -> Result<(), Error> +async fn run_unis(mut session: S, subscriber: Subscriber, setup: UniSetup) -> Result<(), Error> where S: crate::transport::poll::Boxable, { let UniSetup { peer: peer_setup, read: setup_read, + early, version, goaway, local_close, @@ -713,29 +738,35 @@ where // A gated server accept already read the peer's one SETUP off its own uni stream, // so anything arriving here is a second one. let mut seen_setup = setup_read; + let mut early = early.into_iter(); loop { - let recv = tasks - .drive(|waiter| { - let mut cx = std::task::Context::from_waker(waiter.waker()); - session.poll_accept_uni(&mut cx) - }) - .await; - let recv = match recv { - Ok(recv) => recv, - Err(err) => { - let err = Error::from_transport(err); - // Settle tracks and open groups before the owned receive tasks drop - // their cancellation guards, which would otherwise record Cancel. - if local_close.load(std::sync::atomic::Ordering::Relaxed) { - subscriber.close(); - } else { - subscriber.abort(&err); - } - return Err(err); + let mut reader = match early.next() { + Some(reader) => reader, + None => { + let recv = tasks + .drive(|waiter| { + let mut cx = std::task::Context::from_waker(waiter.waker()); + session.poll_accept_uni(&mut cx) + }) + .await; + let recv = match recv { + Ok(recv) => recv, + Err(err) => { + let err = Error::from_transport(err); + // Settle tracks and open groups before the owned receive tasks drop + // their cancellation guards, which would otherwise record Cancel. + if local_close.load(std::sync::atomic::Ordering::Relaxed) { + subscriber.close(); + } else { + subscriber.abort(&err); + } + return Err(err); + } + }; + Reader::new(recv, outer_version) } }; - let mut reader: Reader = Reader::new(recv, outer_version); // A stream that dies before its type varint is that stream's failure, not the // session's. RESET_STREAM is how a peer drops a group, and QUIC does not order // the reset behind the data, so one can beat the first byte even of a stream @@ -756,13 +787,7 @@ where .await { Ok(kind) => kind, - Err( - err @ (Error::Cancel - | Error::Stream(_) - | Error::Remote(_) - | Error::Transport(_) - | Error::Decode(DecodeError::Short)), - ) => { + Err(err) if died_before_header(&err) => { tracing::debug!(%err, "dropping uni stream that died before its type"); continue; } @@ -850,6 +875,15 @@ where } } +/// Whether reading an incoming stream's header failed because the stream died, which is +/// that stream's failure and not the session's. Bytes that arrive and do not parse are not. +fn died_before_header(err: &Error) -> bool { + matches!( + err, + Error::Cancel | Error::Stream(_) | Error::Remote(_) | Error::Transport(_) | Error::Decode(DecodeError::Short) + ) +} + /// Receive QUIC datagrams, each an OBJECT_DATAGRAM for one of our subscriptions. /// /// A transport without datagrams never delivers one, so this parks. A transport failure @@ -936,13 +970,7 @@ where // header that does not parse included, still fails the session. let (id, data) = match header { Ok(header) => header, - Err( - err @ (Error::Cancel - | Error::Stream(_) - | Error::Remote(_) - | Error::Transport(_) - | Error::Decode(DecodeError::Short)), - ) => { + Err(err) if died_before_header(&err) => { tracing::debug!(%err, "dropping bidi stream that died before its header"); continue; } @@ -1105,6 +1133,7 @@ mod tests { }, ..Default::default() }), + early_unis: Vec::new(), }) .expect("start the session"); @@ -1157,6 +1186,7 @@ mod tests { peer_setup_stream: None, // The requests wait on the peer's SETUP (MoQ Hidden). peer_declared: Some(peer::Peer::default()), + early_unis: Vec::new(), }) .expect("start the session"); let _driver = tokio::spawn(driver); @@ -1208,6 +1238,7 @@ mod tests { authority: None, peer_setup_stream: None, peer_declared, + early_unis: Vec::new(), }) .expect("start the session"); let _driver = tokio::spawn(driver); @@ -1318,6 +1349,7 @@ mod tests { // Pre-settled, so nothing waits on a SETUP the dead stream will never // carry and the dispatch loop actually runs. peer_declared: Some(peer::Peer::default()), + early_unis: Vec::new(), }) .expect("start the session"); @@ -1356,6 +1388,7 @@ mod tests { authority: Some(String::from_utf8(AUTHORITY.to_vec()).unwrap()), peer_setup_stream: None, peer_declared: Some(peer::Peer::default()), + early_unis: Vec::new(), }) .expect("start the session"); @@ -1426,10 +1459,21 @@ mod tests { payload: Vec, retired_alias: Option, ) -> (crate::lite::test_transport::Log, Option>) { - let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); // The peer opens one uni stream and then goes quiet, so // the loop is still running when the assertion is taken. let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_unis(vec![payload]); + drive_unis(version, session, None, retired_alias).await + } + + /// Run the uni dispatch loop until it stops a stream or ends. `accepted` is a SETUP + /// already read by [`accept_setup`], with the streams that raced ahead of it. + async fn drive_unis( + version: Version, + session: crate::lite::test_transport::ScriptedSession, + accepted: Option>, + retired_alias: Option, + ) -> (crate::lite::test_transport::Log, Option>) { + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); let log = session.log.clone(); let (tasks, _task_set) = TaskSet::new(); @@ -1453,14 +1497,17 @@ mod tests { // Held so the trigger side stays alive for as long as the loop runs. let (_goaway, goaway) = crate::goaway::Handle::new(false); - // The peer's SETUP has not arrived yet, which is the state a group stream racing - // ahead of it lands in. + // Without `accepted`, the peer's SETUP has not arrived yet, which is the state a + // group stream racing ahead of it lands in. + let read = accepted.is_some(); + let early = accepted.map(|accepted| accepted.early).unwrap_or_default(); let mut unis = std::pin::pin!(run_unis( session, subscriber, UniSetup { peer: Some(peer_setup), - read: false, + read, + early, version, goaway, local_close: Default::default() @@ -1535,6 +1582,37 @@ mod tests { } } + /// A server waiting on the client's SETUP holds the padding and group streams that beat + /// it, then classifies them once the session starts. The drafts say to buffer early + /// data; rejecting it reached the wire as INTERNAL_ERROR. + #[tokio::test(start_paused = true)] + async fn uni_streams_before_setup_are_held_until_it_lands() { + const VERSION: Version = Version::Draft19; + + let mut setup = Vec::new(); + setup::Setup { + parameters: ietf::Parameters::default().encode_bytes(VERSION).unwrap(), + } + .encode(&mut setup, crate::Version::Ietf(VERSION)) + .unwrap(); + + let mut session = crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_unis(vec![ + uni_stream(VERSION, PADDING), + subgroup_header(VERSION, 7, 0).await, + setup, + ]); + let accepted = accept_setup(&mut session, VERSION).await.expect("accept the SETUP"); + assert_eq!(accepted.early.len(), 2, "both early streams are held"); + assert!(session.log.stops().is_empty(), "stopped {:?}", session.log.stops()); + + // The group answers a retired subscription, so the classifier reaching it shows + // as CANCELLED; the padding is read to its end and stops nothing. + let (log, result) = drive_unis(VERSION, session, Some(accepted), Some(7)).await; + assert_eq!(log.stops(), vec![crate::ietf::error::CANCELLED]); + assert!(result.is_none(), "an early stream ended the session: {result:?}"); + assert_eq!(log.closes(), vec![]); + } + /// An unknown or invalid stream type MUST close the session, so it stops nothing on its own: /// the session close takes the stream with it. #[tokio::test(start_paused = true)] @@ -1642,6 +1720,7 @@ mod tests { authority: None, peer_setup_stream: None, peer_declared: None, + early_unis: Vec::new(), }) .expect("start the session"); let driver = tokio::spawn(driver); diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 346b8a200d..0ba506db7c 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -534,6 +534,7 @@ where authority: None, peer_setup_stream: Some(peer_setup.stream), peer_declared: Some(peer_setup.declared), + early_unis: peer_setup.early, })?; tracing::debug!(?version, "connected"); Ok(Session::new( @@ -641,6 +642,7 @@ where authority: None, peer_setup_stream: None, peer_declared: Some(peer_declared), + early_unis: Vec::new(), })?; (None, crate::driver::Protocol::Ietf(protocol), goaway) } @@ -1243,6 +1245,7 @@ mod tests { path: None, token: None, declared: ietf::peer::Peer::default(), + early: Vec::new(), }, })), }),