test(interop): exercise real IETF control sessions

js/net/src/ietf/adapter.test.ts

@@ -14,33 +14,55 @@ import {
 } from "./publish_namespace.ts";
 import { MaxRequestId, RequestError, RequestOk } from "./request.ts";
 import { initialMaxRequestId, REQUEST_WINDOW } from "./setup_options.ts";
-import { Subscribe, SubscribeUpdate } from "./subscribe.ts";
+import { Subscribe, SubscribeError, SubscribeUpdate } from "./subscribe.ts";
 import { SubscribeNamespaceLegacy } from "./subscribe_namespace.ts";
-import { TrackStatusRequest } from "./track.ts";
+import { TrackStatusOk, TrackStatusRequest } from "./track.ts";
 import { ALPN, type IetfVersion, Version } from "./version.ts";
 
-test("draft-14 TRACK_STATUS_OK cannot be routed as NAMESPACE_DONE", async () => {
+test("draft-14 status answers route by request ID and a refusal leaves the session usable", async () => {
+	const version = Version.DRAFT_14;
 	const pair = createMockTransportPair(ALPN.DRAFT_14);
-	const control = await Stream.open(pair.server, { version: Version.DRAFT_14 });
-	const adapter = new ControlStreamAdapter(pair.server, control, Version.DRAFT_14, 100n, true);
-	const running = adapter.run();
-	// A writer may yield a task before the rejection assertion below.
-	void running.catch(() => {});
-	const peer = await Stream.accept(pair.client, Version.DRAFT_14);
+	const control = await Stream.open(pair.server, { version });
+	const adapter = new ControlStreamAdapter(pair.server, control, version, 100n, true);
+	void adapter.run().catch(() => {});
+	const peer = await Stream.accept(pair.client, version);
 	if (!peer) throw new Error("no control stream");
-	await peer.writer.u53(0x0e);
-	await peer.writer.u16(0);
-	// The adapter may reject while the writer yields a browser task.
-	await new Promise<void>((resolve) => {
-		const channel = new MessageChannel();
-		channel.port1.onmessage = () => {
-			channel.port1.close();
-			channel.port2.close();
-			resolve();
-		};
-		channel.port2.postMessage(null);
-	});
-	await expect(running).rejects.toThrow("unexpected message 0x0e");
+	try {
+		for (const refused of [false, true, false]) {
+			const requestId = await adapter.nextRequestId();
+			if (requestId === undefined) throw new Error("no request ID");
+			const stream = await adapter.openBi();
+			await stream.writer.u53(TrackStatusRequest.id);
+			await new TrackStatusRequest({ requestId, trackNamespace: Path.from("room"), trackName: "video" }).encode(
+				stream.writer,
+				version,
+			);
+			expect(await peer.reader.u53()).toBe(TrackStatusRequest.id);
+			expect((await TrackStatusRequest.decode(peer.reader, version)).requestId).toBe(requestId);
+			if (refused) {
+				await peer.writer.u53(0x0f);
+				await new SubscribeError({ requestId, errorCode: 4, reasonPhrase: "missing" }).encode(
+					peer.writer,
+					version,
+				);
+				expect(await stream.reader.u53()).toBe(0x0f);
+				expect((await SubscribeError.decode(stream.reader, version)).requestId).toBe(requestId);
+			} else {
+				await peer.writer.u53(TrackStatusOk.id(version));
+				await new TrackStatusOk({ requestId, largest: { groupId: 4n, objectId: 1n } }).encode(
+					peer.writer,
+					version,
+				);
+				expect(await stream.reader.u53()).toBe(TrackStatusOk.id(version));
+				const status = await TrackStatusOk.decode(stream.reader, version);
+				expect(status.requestId).toBe(requestId);
+				expect(status.largest).toEqual({ groupId: 4n, objectId: 1n });
+			}
+			stream.close();
+		}
+	} finally {
+		adapter.close();
+	}
 });
 
 // Draft-15 is the interesting one: it names its namespace withdrawals instead of

js/net/src/ietf/adapter.ts

@@ -1055,8 +1055,17 @@ export class ControlStreamAdapter implements Session {
 				if (subNs08 === undefined) throw new Error("unexpected message 0x08: no SubscribeNamespace stream");
 				return { route: Route.FollowUp, requestId: subNs08 };
 			}
+			case 0x0f: {
+				// TRACK_STATUS_ERROR has the SUBSCRIBE_ERROR body on draft 14.
+				if (this.version !== Version.DRAFT_14) throw new Error("unexpected TrackStatusError");
+				return { route: Route.ErrorResponse, requestId: await readRequestId() };
+			}
 			case 0x0e: {
-				if (this.version === Version.DRAFT_14 || this.version === Version.DRAFT_15) {
+				if (this.version === Version.DRAFT_14) {
+					// TRACK_STATUS_OK names its request, unlike draft 16's namespace entry.
+					return { route: Route.Response, requestId: await readRequestId() };
+				}
+				if (this.version === Version.DRAFT_15) {
 					throw new Error("unexpected message 0x0e");
 				}
 				// v16+: NamespaceDone entry (no requestId) — route to SubscribeNamespace stream

js/net/src/ietf/subscriber.ts

@@ -972,7 +972,8 @@ export class Subscriber {
 		} catch (err) {
 			stream.abort(err);
 			const e = await sessionCause(this.#quic, err);
-			producer.close(this.#localClose ? undefined : e);
+			if (this.#localClose) hooks.releaseTrack(producer);
+			else producer.close(e);
 			console.warn(
 				`subscribe error: id=${requestId} broadcast=${broadcast} track=${request.name} error=${reason(e)}`,
 			);

js/net/src/integration.test.ts

@@ -27,7 +27,7 @@ import { Producer as OriginProducer } from "./origin.ts";
 import * as Path from "./path.ts";
 import { Stream } from "./stream.ts";
 import { Timescale, Timestamp } from "./time.ts";
-import type { Producer as TrackProducer } from "./track.ts";
+import type { Location, Producer as TrackProducer } from "./track.ts";
 import { withTimeout } from "./util/timeout.ts";
 import { wireOf } from "./wire.ts";
 
@@ -2776,7 +2776,7 @@ async function runSessionDeath(
 	protocol: string,
 	version?: number,
 	local = false,
-): Promise<[Error | null, Error | null]> {
+): Promise<[Error | null, Error | null, Location | undefined]> {
 	const pair = createMockTransportPair(protocol);
 	const origin = new OriginProducer();
 
@@ -2815,7 +2815,7 @@ async function runSessionDeath(
 	client.abort();
 	server.abort();
 	origin.close();
-	return [closed, await withTimeout(groupEnd, Nano.fromMilli(Milli(2000)), "the group never ended")];
+	return [closed, await withTimeout(groupEnd, Nano.fromMilli(Milli(2000)), "the group never ended"), track.end()];
 }
 
 for (const [name, protocol, version] of [
@@ -2831,9 +2831,12 @@ for (const [name, protocol, version] of [
 		expect(groupEnd).toBeInstanceOf(SessionError);
 		expect((groupEnd as SessionError).code).toBe(DEATH);
 	});
-	test(`integration: ${name} ends a locally closed track cleanly`, async () => {
-		const [closed] = await runSessionDeath(protocol, version, true);
+	// A local close is a dead route, not finished content: it declares no end, so a resumed
+	// reader's replacement route continues past it.
+	test(`integration: ${name} ends a locally closed track cleanly without declaring an end`, async () => {
+		const [closed, , end] = await runSessionDeath(protocol, version, true);
 		expect(closed).toBeNull();
+		expect(end).toBeUndefined();
 	});
 }
 

js/net/src/internal.ts

@@ -405,6 +405,13 @@ export const hooks: {
 	 * a higher one. Mirrors the Rust `finish_at` on a `received()` copy; assigned by `track.ts`.
 	 */
 	endTrack: (producer: Producer, end: Location) => void;
+	// pending: /quest/m1/public-model/deferred.md
+	/**
+	 * End a wire subscription's copy cleanly without declaring an end, for a session closed
+	 * locally: a dead route, not finished content, so a resumed reader's replacement route
+	 * continues past it. Mirrors the Rust `close` on a `received()` copy; assigned by `track.ts`.
+	 */
+	releaseTrack: (producer: Producer) => void;
 	// private: capability gap
 	/** The producer's largest reported location and its subsequent changes. */
 	producerLargest: (producer: Producer) => Getter<Location | undefined>;
@@ -586,6 +593,9 @@ export const hooks: {
 	endTrack: () => {
 		throw new Error("track.ts not loaded");
 	},
+	releaseTrack: () => {
+		throw new Error("track.ts not loaded");
+	},
 	producerLargest: () => {
 		throw new Error("track.ts not loaded");
 	},

js/net/src/lite/subscriber.ts

@@ -1782,8 +1782,8 @@ export class Subscriber {
 	}
 
 	/**
-	 * Ends every subscribed track: cleanly for a deliberate close, or with `err` when the
-	 * session died, since those tracks were cut off rather than ended.
+	 * Ends every subscribed track: cleanly without declaring an end for a deliberate close, or
+	 * with `err` when the session died, since those tracks were cut off rather than ended.
 	 */
 	close(err?: Error) {
 		// A fetch or setup exchange cut off by the session is incomplete even on a deliberate
@@ -1792,7 +1792,8 @@ export class Subscriber {
 		this.#closed.abort(cut);
 
 		for (const { track } of this.#subscribes.values()) {
-			track.close(err);
+			if (err) track.close(err);
+			else hooks.releaseTrack(track);
 		}
 
 		this.#subscribes.clear();

js/net/src/resume.test.ts

@@ -2098,6 +2098,35 @@ test("an end a replaced copy declared holds once the copy is let go in the same
 	b.close();
 });
 
+test("a replaced copy whose session closed locally binds no end", async () => {
+	const front = new Front();
+	const a = new BroadcastProducer();
+	const b = new BroadcastProducer();
+	const ta = copy(a);
+	const tb = copy(b);
+	front.serve(a.consume());
+	const sub = front.consumer.track("seq").subscribe();
+	await settle();
+	ta.write(0);
+	await settle();
+	expect(drain(sub)).toEqual([0]);
+
+	front.serve(b.consume());
+	await settle();
+	// A clean close would declare an end at group 1, refusing b's next group.
+	hooks.releaseTrack(ta.producer);
+	tb.write(1);
+	await settle();
+	expect(drain(sub)).toEqual([1]);
+	expect(sub.end()).toBeUndefined();
+	expect(sub.closed.peek()).toBeUndefined();
+
+	sub.close();
+	front.consumer.close();
+	a.close();
+	b.close();
+});
+
 test("largest ends once the serving copy reaches its end, with the front still open", async () => {
 	const front = new Front();
 	const a = new BroadcastProducer();

js/net/src/track.ts

@@ -806,6 +806,7 @@ export class Producer {
 		hooks.producerLargest = (producer) => producer.#state.largest;
 		hooks.joinProducer = (producer) => producer.#join();
 		hooks.endTrack = (producer, end) => producer.#endPending(end);
+		hooks.releaseTrack = (producer) => producer.#close(undefined, false);
 		hooks.startTrack = (producer, served) => {
 			producer.#served = served;
 			producer.#owe();
@@ -1724,13 +1725,19 @@ export class Producer {
 	 * them.
 	 */
 	close(abort?: Error) {
+		this.#close(abort, true);
+	}
+
+	// `declare` is false for a received copy whose session closed locally: a dead route, not
+	// finished content, so a clean close there leaves the end undeclared.
+	#close(abort: Error | undefined, declare: boolean) {
 		if (this.#state.closed.peek() !== undefined) return;
 		// A gap past the grace was lost before the abort rather than cut off by it.
 		if (abort && this.#tailPending) this.#arrivals.expire(now(), this.#grace());
 		if (abort && this.#settled()) abort = undefined;
 		// A replacement continues its name's timeline: ending below an earlier producer's
 		// groups would contradict the ones subscribers already received.
-		if (abort === undefined && this.#state.end.peek() === undefined) {
+		if (declare && abort === undefined && this.#state.end.peek() === undefined) {
 			this.#declareEnd({ group: this.#liveEdge(), frame: 0 });
 		}
 		// Nobody will finish these, so a subscriber that has not taken one yet never sees it.

quest/m1/interop-coverage/README.md

@@ -44,7 +44,6 @@ coverage on `main` at `3ab6cbf81d`.
 - [JS namespace lifecycle](/quest/m1/interop-coverage/js-namespace-done.md) - moq-dev-js subscribe-namespace-lifecycle sees NAMESPACE_DONE with publisher and subscriber in one process
 - [JS subgroup data through imquic](/quest/m1/interop-coverage/js-imquic-objects.md) - moq-dev-js data-subgroup-basic receives objects from imquic over WebTransport
 - [Late join and FETCH](/quest/m1/interop-coverage/late-join.md) - a late subscriber gets the current group from object 0, and a standalone FETCH returns a past group exactly
-- [Control cases](/quest/m1/interop-coverage/control.md) - GOAWAY redirect, TRACK_STATUS, datagrams, and request-ID limits pass across implementations
 
 ## Related
 

quest/m1/public-model/deferred.md

@@ -5,8 +5,9 @@
 The hidden model items that other quests reshape are public, unnecessary,
 or marked private once those quests land: fetch ingest and answering, the
 IETF cache peeks, JS `Track.Subscriber.fork` and `Group.Producer.mirror`,
-and JS `hooks.endTrack`. A custom session then answers fetches and track
-info, and declares a received end in JS, through public calls only.
+and JS `hooks.endTrack` and `hooks.releaseTrack`. A custom session then
+answers fetches and track info, and declares or withholds a received end in
+JS, through public calls only.
 
 ## Plan
 
@@ -36,10 +37,12 @@ the prerequisites will have deleted or renamed most of these.
   answered through the public `Track.Request` and `Track.Dynamic` range
   requests that Model ranges defines, completing the answering quest's
   `Broadcast.requested` decision.
-- JS `hooks.endTrack` and `hooks.startTrack` (the JS `finish_at_pending` and
-  `start_at`): public under the received-end shape Rust settled in
+- JS `hooks.endTrack`, `hooks.releaseTrack`, and `hooks.startTrack` (the JS
+  `finish_at_pending`, `close`, and `start_at`): public under the received-end
+  shape Rust settled in
   [moq-dev/moq#6111](https://github.com/moq-dev/moq/pull/6111): `finish_at`
-  on a `received()` copy, and `start_at` after `track::Request::request_start`.
+  and `close` on a `received()` copy, and `start_at` after
+  `track::Request::request_start`.
 
 Tests: as in the other children, mock time in each language, with the
 built-in sessions moved onto the public calls.

rs/moq-cli/Cargo.toml

@@ -179,6 +179,18 @@ harness = false
 name = "interop-tail"
 path = "../../test/interop/clients/rust/tail.rs"
 
+[[example]]
+name = "interop-goaway"
+path = "../../test/interop/clients/rust/goaway.rs"
+
+[[example]]
+name = "interop-control"
+path = "../../test/interop/clients/rust/control.rs"
+
+[[example]]
+name = "interop-goaway-client"
+path = "../../test/interop/clients/rust/goaway-client.rs"
+
 [[example]]
 name = "interop-finish"
 path = "../../test/interop/clients/rust/finish.rs"

test/interop/README.md

@@ -180,6 +180,38 @@ opened landed on another one, such as a client silently falling back to
 moq-lite. The relay names no client, so a fallback fails the round rather than
 one cell. The tails check their own connections, so a fallback fails the lane.
 
+The in-tree control lanes also run with `--tail`:
+
+- Four untimed 900-byte datagrams cross the relay in both directions between Rust
+  and each native JS runtime. The subscriber acknowledges resolved metadata before
+  sending begins, then acknowledges each packet, so this checks framing and bytes
+  rather than best-effort burst capacity. The publisher stays open until the read is
+  acknowledged, keeping a reliable track end from overtaking the datagrams.
+- A JS wire requester gets the Rust publisher's exact largest location through
+  TRACK\_STATUS on every draft. From draft 18, Rust also queries the JS publisher's
+  non-default priority before subscribing, when the answer can describe a track.
+- On drafts 14 and 16, a JS publisher advertises one usable track slot beyond namespace discovery, holds the first
+  Rust subscription open through a data round trip while a second is queued, and
+  grants credit when the first is cancelled. Sending the second without waiting
+  violates the peer adapter's live-request limit. A JS subscriber runs the same
+  held-first/queued-second exchange over QMux against a controlled JS peer, whose
+  grant and replenishment codecs are also exercised by the Rust subscriber.
+- On drafts 18, 20, and 22, a Rust GOAWAY names a second WebSocket endpoint serving
+  the same publisher epoch. The JS reconnect loop retains its reader across the
+  move. A separate read through the replacement proves it serves data before the
+  original transport is cut; only the replacement can deliver the final group.
+  The reverse JS-server-to-Rust-client case uses a marker broadcast excluded from
+  the old endpoint's scope, then cuts that endpoint before publishing the final
+  group. Both directions use the original publisher stream.
+- Drafts 14 and 16 carry no epoch, so nothing proves the replacement is the same
+  publisher. The same GOAWAY lanes there check that the held reader ends with the
+  original transport instead, and that requesting the path again subscribes
+  through the new URI and reads the final group.
+
+The standalone control endpoints use OS-assigned ports and production QMux,
+SETUP codecs, and request adapters. They verify their negotiated version directly.
+The datagram and query lanes through moq-relay use its usual version-log check.
+
 Some cases change shape on moq-transport:
 
 - moq-net subscribes there from the largest group, not from a requested range,