fix(moq-net): hold IETF uni streams that arrive before SETUP

js/net/src/connection/accept.ts

@@ -115,14 +115,15 @@ async function acceptAlpn(
 	version: Ietf.IetfVersion,
 	wiring: SessionProps,
 ): Promise<Established> {
-	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,
 		client: false,
 		url,
 		quic: transport,
 		control,
+		early,
 		solicit,
 		hidden,
 		cluster,

js/net/src/connection/connect.ts

@@ -362,14 +362,15 @@ async function handshakeAlpn(
 	version: Ietf.IetfVersion,
 	wiring: SessionProps,
 ): Promise<Established> {
-	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,
 		client: true,
 		url,
 		quic: session,
 		control,
+		early,
 		solicit,
 		hidden,
 		cluster,

js/net/src/connection/handshake.ts

@@ -1,14 +1,16 @@
+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
  * message, and reads the peer's Setup off an incoming uni stream. The two
  * 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<Uint8Array>
 	>;
-	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),

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);
+});

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<void> {
+	async #run(early: Reader[]): Promise<void> {
 		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) {