From 89aa46255b10cff9a7db3ac85d6d600f73a10228 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 16:36:36 -0700 Subject: [PATCH 1/4] quest: claim ietf-early-streams From 2f46f83e92f44b001f55028d137912e07e022140 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 16:59:40 -0700 Subject: [PATCH 2/4] fix(moq-net): hold IETF uni streams that arrive before SETUP A draft-17+ server aborted every uni stream ahead of the peer's SETUP, and the JS handshake failed outright. The drafts say to buffer early data, so both now queue those streams and hand them to the normal uni classifier once the session starts. Co-Authored-By: Claude Opus 5.5 --- js/net/src/connection/accept.ts | 3 +- js/net/src/connection/connect.ts | 3 +- js/net/src/connection/handshake.ts | 55 ++++++++-- js/net/src/ietf/connection.test.ts | 72 ++++++++++++ js/net/src/ietf/connection.ts | 41 ++++--- js/net/src/stream.ts | 18 ++- quest/m0/README.md | 1 - quest/m0/ietf-early-streams.md | 34 ------ rs/moq-net/src/client.rs | 2 + rs/moq-net/src/ietf/session.rs | 171 +++++++++++++++++++++-------- rs/moq-net/src/server.rs | 3 + 11 files changed, 292 insertions(+), 111 deletions(-) delete mode 100644 quest/m0/ietf-early-streams.md 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..a7f9bdc573 100644 --- a/js/net/src/connection/handshake.ts +++ b/js/net/src/connection/handshake.ts @@ -8,7 +8,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 +20,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 +47,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 +63,54 @@ 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); + // Full width, so an unknown type past 2^53 is held for the classifier to refuse. + const streamType = await stream.peekU62().catch(() => undefined); + // A stream that died before its type is that stream's failure, not the session's. + 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..4bbc424bfd 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,71 @@ 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(); + } +}); 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..066007d56e 100644 --- a/js/net/src/stream.ts +++ b/js/net/src/stream.ts @@ -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); } 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(), }, })), }), From 3ba46e69079a6c68c8c3324dff19f4ed045d51a2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 17:22:55 -0700 Subject: [PATCH 3/4] fix(net): keep a malformed early uni stream fatal to the handshake Only a stream that ends or resets before its type is skipped; a bad varint still fails SETUP, matching Rust's died_before_header. Co-Authored-By: Claude Opus 5.5 --- js/net/src/connection/handshake.ts | 10 ++++++++-- js/net/src/ietf/connection.test.ts | 21 +++++++++++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/js/net/src/connection/handshake.ts b/js/net/src/connection/handshake.ts index a7f9bdc573..de58709125 100644 --- a/js/net/src/connection/handshake.ts +++ b/js/net/src/connection/handshake.ts @@ -1,3 +1,4 @@ +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"; @@ -91,9 +92,14 @@ async function receiveSetup( if (next.done) throw new Error("no incoming uni stream for SETUP"); const stream = new Reader(next.value, undefined, version); - // Full width, so an unknown type past 2^53 is held for the classifier to refuse. - const streamType = await stream.peekU62().catch(() => undefined); // A stream that died before its 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) return undefined; + throw err; + }); if (streamType === undefined) continue; if (streamType === BigInt(Ietf.Setup.id)) { reader = stream; diff --git a/js/net/src/ietf/connection.test.ts b/js/net/src/ietf/connection.test.ts index 4bbc424bfd..4feb7911fd 100644 --- a/js/net/src/ietf/connection.test.ts +++ b/js/net/src/ietf/connection.test.ts @@ -148,3 +148,24 @@ test("uni streams before SETUP are held until it lands", async () => { 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"); +}); From dacfc819262bf53dd4c03d82240fcda4cdcfadf1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 17:36:05 -0700 Subject: [PATCH 4/4] fix(net): skip an early uni stream that ends partway through its type A clean FIN mid-varint is the stream dying, like Rust's DecodeError::Short, so it no longer fails the handshake. Reader now throws a typed UnexpectedEnd. Co-Authored-By: Claude Opus 5.5 --- js/net/src/connection/handshake.ts | 6 +++--- js/net/src/ietf/connection.test.ts | 20 ++++++++++++++++++++ js/net/src/stream.ts | 9 ++++++++- 3 files changed, 31 insertions(+), 4 deletions(-) diff --git a/js/net/src/connection/handshake.ts b/js/net/src/connection/handshake.ts index de58709125..81bb2052a9 100644 --- a/js/net/src/connection/handshake.ts +++ b/js/net/src/connection/handshake.ts @@ -1,7 +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 @@ -92,12 +92,12 @@ async function receiveSetup( if (next.done) throw new Error("no incoming uni stream for SETUP"); const stream = new Reader(next.value, undefined, version); - // A stream that died before its type is that stream's failure, not the session's. + // 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) return undefined; + if (err instanceof StreamError || err instanceof UnexpectedEnd) return undefined; throw err; }); if (streamType === undefined) continue; diff --git a/js/net/src/ietf/connection.test.ts b/js/net/src/ietf/connection.test.ts index 4feb7911fd..a9c171321d 100644 --- a/js/net/src/ietf/connection.test.ts +++ b/js/net/src/ietf/connection.test.ts @@ -169,3 +169,23 @@ test("malformed uni stream before SETUP fails the handshake", async () => { 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/stream.ts b/js/net/src/stream.ts index 066007d56e..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(); } } @@ -480,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.