fix(net): retain early lite streams and reject duplicate setup

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

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

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<Setup | undefined>(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 {

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