From 20ff147267575b93ad32e6a53eae9e7dbd3c3def Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 04:46:52 -0700 Subject: [PATCH 1/7] chore(quest): claim quest/m1/interop-coverage/control.md Quest-Claim: b758f0728197d910 From 2e468668f5c7a9b82ec8f30b4da551491f279097 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 05:21:44 -0700 Subject: [PATCH 2/7] Add real IETF control interop coverage Checkpoint draft14 status response routing and real datagram, metadata, request-credit, and GOAWAY fixtures. Draft16 credit and a GOAWAY Bun22 fixture interaction remain under investigation; the broad quest remains active. Co-Authored-By: GPT-6 --- js/net/src/ietf/adapter.test.ts | 66 ++++++++---- js/net/src/ietf/adapter.ts | 11 +- quest/m1/interop-coverage/control.md | 25 +++++ rs/moq-cli/Cargo.toml | 8 ++ test/interop/README.md | 25 +++++ test/interop/clients/js-native/control.ts | 114 +++++++++++++++++++++ test/interop/clients/js-native/goaway.ts | 86 ++++++++++++++++ test/interop/clients/js-native/tail.ts | 117 ++++++++++++++-------- test/interop/clients/rust/control.rs | 93 +++++++++++++++++ test/interop/clients/rust/goaway.rs | 87 ++++++++++++++++ test/interop/clients/rust/tail.rs | 73 +++++++++++--- test/interop/interop.sh | 98 ++++++++++++++++-- test/smoke/cloudflare.sh | 1 - 13 files changed, 720 insertions(+), 84 deletions(-) create mode 100644 test/interop/clients/js-native/control.ts create mode 100644 test/interop/clients/js-native/goaway.ts create mode 100644 test/interop/clients/rust/control.rs create mode 100644 test/interop/clients/rust/goaway.rs diff --git a/js/net/src/ietf/adapter.test.ts b/js/net/src/ietf/adapter.test.ts index cbd1ad88b7a..8eda62ac499 100644 --- a/js/net/src/ietf/adapter.test.ts +++ b/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((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 diff --git a/js/net/src/ietf/adapter.ts b/js/net/src/ietf/adapter.ts index 048eb4f0b6d..dc04ce6fc97 100644 --- a/js/net/src/ietf/adapter.ts +++ b/js/net/src/ietf/adapter.ts @@ -1054,8 +1054,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 diff --git a/quest/m1/interop-coverage/control.md b/quest/m1/interop-coverage/control.md index 23a6438d890..272dbe8c42d 100644 --- a/quest/m1/interop-coverage/control.md +++ b/quest/m1/interop-coverage/control.md @@ -18,3 +18,28 @@ datagrams; fix that comment, and add a datagram round to its moq-dev section if it is cheaper than a matrix case. Public API: none. Wire: none. + +## Progress + +The implementation PR adds relay datagrams in both Rust/native-JS directions on +all five drafts, JS wire TRACK_STATUS queries to Rust on all five, and Rust +query-before-subscribe to JS from draft 18. Draft-14 real answers exposed missing +TRACK_STATUS_OK/ERROR routing in the JS control adapter; the patch includes +success/refusal/success coverage. A JS peer advertising one usable track slot checks +Rust waiting for credit on drafts 14 and 16. Real Rust-to-JS GOAWAY with a new URI +retains a reader on drafts 18, 20, and 22, with the same publisher epoch and a +replacement-only data barrier before cutting the original transport. + +The initial datagram burst and GOAWAY readiness failures were fixture errors. +No production handover policy change is justified by those results. + +## Remaining + +- Decide what the GOAWAY requirement means on drafts 14 and 16. They cannot carry + publisher epochs, and the documented model restarts across an unproven route + rather than splicing a retained reader. Recommend testing application + re-request/re-subscribe there and keeping same-reader continuity on 18+. +- Add the reverse GOAWAY direction (JS server to Rust reconnecting client) and + request credit with JS as the subscriber. Keep small-window setup in test peers, + rather than adding a public configuration knob solely for coverage. +- Retain this quest until the remaining coverage and legacy acceptance are settled. diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index 2970c6dec35..5c49e48cbfa 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -176,3 +176,11 @@ harness = false [[example]] 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" diff --git a/test/interop/README.md b/test/interop/README.md index 22f0e019ad2..c8987bc7705 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -143,6 +143,31 @@ 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. +- 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. + Drafts 14 and 16 cannot carry epochs and require an application restart, so this + retained-reader lane does not claim to cover them. + +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, diff --git a/test/interop/clients/js-native/control.ts b/test/interop/clients/js-native/control.ts new file mode 100644 index 00000000000..7e5a385cf12 --- /dev/null +++ b/test/interop/clients/js-native/control.ts @@ -0,0 +1,114 @@ +/** Real JS wire peer for metadata queries and a deliberately small legacy request window. */ +import assert from "node:assert/strict"; +import { createInterface } from "node:readline"; +import * as Moq from "@moq/net"; +import { offered } from "../../../../js/net/src/connection/extensions.ts"; +import { exchangeSetup } from "../../../../js/net/src/connection/handshake.ts"; +import { ControlStreamAdapter, NativeSession } from "../../../../js/net/src/ietf/adapter.ts"; +import { Connection } from "../../../../js/net/src/ietf/connection.ts"; +import * as Ietf from "../../../../js/net/src/ietf/index.ts"; +import { TrackStatusOk, TrackStatusRequest } from "../../../../js/net/src/ietf/track.ts"; +import * as Lite from "../../../../js/net/src/lite/index.ts"; +import { Stream } from "../../../../js/net/src/stream.ts"; +import Qmux from "../../../../js/qmux/src/index.ts"; + +console.debug = console.error; +console.info = console.error; +const [mode, address, name] = process.argv.slice(2); +assert(address && name && ["status", "credit"].includes(mode), "usage: control.ts status|credit URL VERSION"); +const version = (0xff000000 + Number(name.split("-").at(-1))) as Ietf.IetfVersion; +const legacy = version === Ietf.Version.DRAFT_14 || version === Ietf.Version.DRAFT_16; +const protocol = version === Ietf.Version.DRAFT_14 ? "moql" : `moqt-${Number(version) - 0xff000000}`; +const url = new URL(address); +const quic = new Qmux(url, { + protocols: [protocol], + versions: { [protocol]: legacy ? "qmux-00" : "qmux-01" }, + requireProtocol: true, +}); +await quic.ready; +const extensions = offered({ auth: false, epoch: false, solicit: false }); +// The ordinary client offers a large window. This peer deliberately advertises one usable track +// slot: draft 16 also holds namespace discovery open on its native request stream. +const window = mode === "credit" ? (version === Ietf.Version.DRAFT_16 ? 2n : 1n) : Ietf.REQUEST_WINDOW; +let control: Stream; +let maxRequestId = 0n; +if (legacy) { + // Reuse the production codecs and adapter. Only the small SETUP grant differs from connect(). + control = await Stream.open(quic, { version }); + await control.writer.u53(Lite.StreamId.ClientCompat); + const parameters = new Ietf.SetupOptions(); + parameters.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(false, window)); + await new Ietf.ClientSetup({ versions: [version], parameters }).encode(control.writer, version); + assert.equal(await control.reader.u53(), Lite.StreamId.ServerCompat); + const setup = await Ietf.ServerSetup.decode(control.reader, version); + assert.equal(setup.version, version, "wrong negotiated draft"); + maxRequestId = setup.parameters.getVarint(Ietf.SetupOption.MaxRequestId) ?? 0n; +} else { + ({ control } = await exchangeSetup(quic, version, "interop-control", { client: true, extensions })); +} + +if (mode === "status") { + const session = legacy + ? new ControlStreamAdapter(quic, control, version, maxRequestId, true) + : new NativeSession(quic, version, true); + const driver = session instanceof ControlStreamAdapter ? session.run() : undefined; + try { + const requestId = await session.nextRequestId(); + assert(requestId !== undefined); + const stream = await session.openBi(); + await stream.writer.u53(TrackStatusRequest.id); + await new TrackStatusRequest({ + requestId, + trackNamespace: Moq.Path.from("control"), + trackName: "video", + }).encode(stream.writer, version); + assert.equal(await stream.reader.u53(), TrackStatusOk.id(version), "TRACK_STATUS was not answered"); + const status = await TrackStatusOk.decode(stream.reader, version); + if (legacy) assert.equal(status.requestId, requestId); + assert.deepEqual(status.largest, { groupId: 4n, objectId: 1n }, "wrong publisher location"); + if (!legacy) assert.equal(status.properties.priority, 253, "wrong publisher priority"); + stream.close(); + console.error("wire track status answered"); + } finally { + session.close(); + quic.close(); + await driver?.catch(() => {}); + } +} else { + assert(legacy, "request credit only exists on drafts 14 and 16"); + const origin = new Moq.Origin.Producer(); + const broadcast = origin.createBroadcast(Moq.Path.from("control")); + const first = broadcast.createTrack("first"); + const second = broadcast.createTrack("second"); + broadcast.announce(); + const connection = new Connection({ + url, + quic, + control, + version, + maxRequestId, + client: true, + publish: origin.consume(), + requestWindow: window, + extensions, + }); + const lines = createInterface({ input: process.stdin }); + const commands = lines[Symbol.asyncIterator](); + try { + assert.equal((await commands.next()).value, "probe", "missing queued-request barrier"); + const group = first.appendGroup(); + group.writeFrame({ payload: new Uint8Array([1]) }); + group.close(); + const demand = second.demand(); + while (demand.wanted.peek() === undefined) await demand.wanted.changed(); + const next = second.appendGroup(); + next.writeFrame({ payload: new Uint8Array([2]) }); + next.close(); + assert.equal((await commands.next()).value, "done"); + console.error("request credit resumed"); + } finally { + lines.close(); + connection.abort(); + origin.close(); + } +} diff --git a/test/interop/clients/js-native/goaway.ts b/test/interop/clients/js-native/goaway.ts new file mode 100644 index 00000000000..c771f0a6b23 --- /dev/null +++ b/test/interop/clients/js-native/goaway.ts @@ -0,0 +1,86 @@ +/** Keep a real subscription alive while the Rust server redirects the connection. */ +import assert from "node:assert/strict"; +import { Milli } from "@moq/clock"; +import * as Moq from "@moq/net"; +import { Reload } from "../../../../js/net/src/connection/reload.ts"; +import { wireOf } from "../../../../js/net/src/wire.ts"; + +console.debug = console.error; +console.info = console.error; +const [first, second, version] = process.argv.slice(2); +assert(first && second && version, "usage: goaway.ts FIRST_URL SECOND_URL VERSION"); +const origin = new Moq.Origin.Producer(); +const connection = new Reload({ + url: new URL(first), + versions: [version], + consume: origin, + websocket: { enabled: true, delay: Milli(0) }, + goaway: { redirect: "follow" }, +}); +let replacementProbe: Moq.Track.Subscriber | undefined; +const request = origin.request(Moq.Path.from("handover"), { announced: true }); +try { + let broadcast = request.active.peek(); + while (!broadcast) { + await request.active.changed(); + broadcast = request.active.peek(); + } + const track = broadcast.track("video").subscribe({ groups: { start: { included: 0 } } }); + try { + for (let sequence = 0; sequence < 3; sequence++) { + const group = await track.recvGroup(); + assert(group, "subscription ended during redirect"); + try { + assert.equal(group.sequence, sequence, "handover duplicated or skipped a group"); + const frame = await group.readFrame(); + assert(frame, "empty group"); + assert.deepEqual(frame.payload, new Uint8Array(32).fill(sequence)); + assert.equal(await group.readFrame(), undefined, "extra frame"); + } finally { + group.close(); + } + if (sequence === 0) process.stdout.write("first group\n"); + if (sequence === 1) { + let replacement = connection.established.peek(); + while (replacement?.url.href !== new URL(second).href) { + await connection.established.changed(); + replacement = connection.established.peek(); + } + const announcements = replacement.announced(Moq.Path.Pattern.literal("handover")); + try { + for (;;) { + const event = await announcements.next(); + assert(event, "replacement announcements ended"); + if (event.kind === "start" && event.prefix === "handover") break; + } + } finally { + announcements.close(); + } + // A direct read from the replacement proves its SUBSCRIBE/data path is ready, + // beyond SETUP or namespace discovery, while the original reader stays held. + replacementProbe = wireOf(replacement) + .consume(Moq.Path.from("handover")) + .track("video") + .subscribe({ groups: { start: { included: 1 } } }); + const probe = await replacementProbe.recvGroup(); + assert(probe, "replacement has no group"); + try { + assert.equal(probe.sequence, 1, "replacement served the wrong group"); + assert.deepEqual((await probe.readFrame())?.payload, new Uint8Array(32).fill(1)); + } finally { + probe.close(); + } + process.stdout.write("replacement ready\n"); + } + } + // The server only writes group 2 after this acknowledgement and closing `first`. + process.stdout.write("goaway resumed\n"); + } finally { + track.close(); + } +} finally { + replacementProbe?.close(); + request.close(); + connection.close(); + origin.close(); +} diff --git a/test/interop/clients/js-native/tail.ts b/test/interop/clients/js-native/tail.ts index cbeb8a60d2b..aeff7396ddd 100644 --- a/test/interop/clients/js-native/tail.ts +++ b/test/interop/clients/js-native/tail.ts @@ -18,9 +18,18 @@ const GROUPS = 4; const BYTES = 256 * 1024; const [role, url, path, version, mode] = process.argv.slice(2); const drop = mode === "drop"; -const join = mode === "join"; -if (!url || !path || !["publish", "subscribe"].includes(role) || ![undefined, "drop", "join"].includes(mode)) { - throw new Error("usage: tail.ts publish|subscribe URL BROADCAST [VERSION [drop|join]]"); +console.debug = console.error; +console.info = console.error; +const join = mode === "join" || mode === "status"; +const datagram = mode === "datagram"; +const status = mode === "status"; +if ( + !url || + !path || + !["publish", "subscribe"].includes(role) || + ![undefined, "drop", "join", "datagram", "status"].includes(mode) +) { + throw new Error("usage: tail.ts publish|subscribe URL BROADCAST [VERSION [drop|join|datagram|status]]"); } const origin = new Moq.Origin.Producer(); @@ -33,25 +42,35 @@ const connection = await Moq.Connection.connect({ try { if (role === "publish") { const broadcast = origin.createBroadcast(Moq.Path.from(path)); - const track = broadcast.createTrack("tail"); + const track = broadcast.createTrack("tail", status ? { priority: 2 } : undefined); broadcast.announce(); - const demand = track.demand(); - while (demand.wanted.peek() === undefined) await demand.wanted.changed(); - // The end precedes the last group's data; the transport still has to drain it. - track.finishAt({ group: GROUPS, frame: 0 }); - for (let sequence = 0; sequence < GROUPS; sequence++) { - const group = track.appendGroup(); - if (drop && sequence === 1) { - group.close(new Error("cut")); - continue; + const lines = createInterface({ input: process.stdin }); + const commands = lines[Symbol.asyncIterator](); + if (!status) { + const demand = track.demand(); + while (demand.wanted.peek() === undefined) await demand.wanted.changed(); + if (datagram) { + for (let sequence = 0; sequence < GROUPS; sequence++) { + assert.equal((await commands.next()).value, `datagram ${sequence}`, "missing datagram command"); + track.appendDatagram(undefined, new Uint8Array(900).fill(sequence)); + } + } else { + // The end precedes the last group's data; the transport still has to drain it. + track.finishAt({ group: GROUPS, frame: 0 }); + for (let sequence = 0; sequence < GROUPS; sequence++) { + const group = track.appendGroup(); + if (drop && sequence === 1) { + group.close(new Error("cut")); + continue; + } + group.writeFrame({ payload: new Uint8Array(BYTES).fill(sequence) }); + group.close(); + } + track.close(); } - group.writeFrame({ payload: new Uint8Array(BYTES).fill(sequence) }); - group.close(); } - track.close(); - const lines = createInterface({ input: process.stdin }); try { - const ack = await lines[Symbol.asyncIterator]().next(); + const ack = await commands.next(); assert.equal(ack.value, "clean end", "reader did not acknowledge a clean end"); } finally { lines.close(); @@ -71,32 +90,50 @@ try { .subscribe({ maxDelay: Milli(drop ? 60_000 : 5000), groups: { start: { included: 0 } } }); const seen: number[] = []; try { - for (;;) { - const group = await track.recvGroup(); - if (!group) break; - try { - assert(group.sequence < GROUPS, `unexpected group ${group.sequence}`); - const frame = await group.readFrame(); - assert(frame, "empty group"); - assert.equal(frame.payload.byteLength, BYTES, "truncated group"); + if (datagram) { + await track.info(); + process.stdout.write("datagram ready\n"); + for (let sequence = 0; sequence < GROUPS; sequence++) { + const frame = await track.recvDatagram(); + assert(frame, "missing datagram"); + assert.equal(frame.sequence, sequence, "wrong datagram sequence"); + assert.equal(frame.payload.byteLength, 900, "truncated datagram"); assert( - frame.payload.every((byte) => byte === group.sequence), - "corrupt group", + frame.payload.every((byte) => byte === sequence), + "corrupt datagram", ); - assert.equal(await group.readFrame(), undefined, "extra frame"); - seen.push(group.sequence); - } finally { - group.close(); + assert.equal(frame.timestamp, undefined, "unexpected IETF timestamp"); + process.stdout.write(`datagram ${sequence}\n`); + } + console.error("datagram clean groups=0,1,2,3 bytes=3600"); + } else { + for (;;) { + const group = await track.recvGroup(); + if (!group) break; + try { + assert(group.sequence < GROUPS, `unexpected group ${group.sequence}`); + const frame = await group.readFrame(); + assert(frame, "empty group"); + assert.equal(frame.payload.byteLength, BYTES, "truncated group"); + assert( + frame.payload.every((byte) => byte === group.sequence), + "corrupt group", + ); + assert.equal(await group.readFrame(), undefined, "extra frame"); + seen.push(group.sequence); + } finally { + group.close(); + } } + seen.sort((a, b) => a - b); + const first = join ? (seen[0] ?? 0) : 0; + const all = Array.from({ length: GROUPS - first }, (_, i) => first + i); + assert.deepEqual(seen, drop ? [0, 2, 3] : all); + assert.deepEqual(await track.finished(), { group: GROUPS, frame: 0 }); + if (join) console.error(`tail join clean end=4 groups=${seen.join(",")}`); + else if (drop) console.error("tail drop clean end=4 groups=0,2,3 bytes=786432"); + else console.error("tail clean end=4 groups=0,1,2,3 bytes=1048576"); } - seen.sort((a, b) => a - b); - const first = join ? (seen[0] ?? 0) : 0; - const all = Array.from({ length: GROUPS - first }, (_, i) => first + i); - assert.deepEqual(seen, drop ? [0, 2, 3] : all); - assert.deepEqual(await track.finished(), { group: GROUPS, frame: 0 }); - if (join) console.error(`tail join clean end=4 groups=${seen.join(",")}`); - else if (drop) console.error("tail drop clean end=4 groups=0,2,3 bytes=786432"); - else console.error("tail clean end=4 groups=0,1,2,3 bytes=1048576"); } finally { track.close(); } diff --git a/test/interop/clients/rust/control.rs b/test/interop/clients/rust/control.rs new file mode 100644 index 00000000000..762fee757ed --- /dev/null +++ b/test/interop/clients/rust/control.rs @@ -0,0 +1,93 @@ +//! Real Rust server for a JS TRACK_STATUS requester and small-window publisher. +use std::process::Stdio; + +use anyhow::{Context, ensure}; +use futures::FutureExt; +use moq_tokio::moq_net; +use tokio::io::AsyncWriteExt; + +#[tokio::main(flavor = "current_thread")] +async fn main() -> anyhow::Result<()> { + let args: Vec<_> = std::env::args().collect(); + ensure!(args.len() == 4, "usage: interop-control status|credit VERSION node|bun"); + let version: moq_net::Version = args[2].parse().map_err(anyhow::Error::msg)?; + let origin = moq_tokio::origin::spawn(); + let consumer = origin.consume(); + let broadcast = origin.create_broadcast("control")?; + let _published = if args[1] == "status" { + let track = broadcast.create_track("video", moq_net::track::Info::default().with_priority(2))?; + let mut group = track.create_group(moq_net::group::Info { sequence: 4 })?; + for byte in [1, 2] { + group.write_frame(None, vec![byte])?; + } + group.finish()?; + broadcast.announce(Default::default())?; + Some(track) + } else { + None + }; + let socket = moq_tokio::websocket::Listener::bind("127.0.0.1:0".parse()?).await?; + let url = format!("http://{}/", socket.local_addr()?); + let mut config = moq_tokio::server::Config::default(); + config.listen.version.push(version); + config.websocket = Some(socket); + if args[1] == "status" { + config.publisher = Some(origin.consume()); + } else { + config.subscriber = Some(origin); + } + let mut listener = config.init_streams()?.listen().await?; + let mut command = tokio::process::Command::new(&args[3]); + if args[3] == "node" { + command.args(["--import", "tsx"]); + } + let mut child = command + .args([ + concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../test/interop/clients/js-native/control.ts" + ), + &args[1], + &url, + &args[2], + ]) + .stdin(Stdio::piped()) + .kill_on_drop(true) + .spawn()?; + let (session, driver) = listener.accept().await.context("accept")?.ok().await?; + ensure!(session.version() == version, "wrong negotiated version"); + if args[1] == "credit" { + // The JS peer has one usable track slot beyond any namespace discovery. + let remote = consumer.routed_broadcast("control").await?; + let mut first = remote.track("first")?.subscribe(None).await?; + let second = remote.track("second")?; + let mut pending = Box::pin(second.subscribe(None)); + ensure!( + pending.as_mut().now_or_never().is_none(), + "second track bypassed request credit" + ); + let input = child.stdin.as_mut().context("child stdin")?; + input.write_all(b"probe\n").await?; + // This round trip keeps the first request alive while the second is queued. + // Sending it without credit would close the JS peer before UNSUBSCRIBE can free a slot. + let mut group = first.recv_group().await?.context("probe group")?; + ensure!( + group.read_frame().await?.context("probe frame")?.payload.as_ref() == [1], + "wrong probe" + ); + drop(group); + drop(first); + let mut second = pending.await?; + let mut group = second.recv_group().await?.context("second group")?; + ensure!( + group.read_frame().await?.context("second frame")?.payload.as_ref() == [2], + "wrong resumed track" + ); + input.write_all(b"done\n").await?; + } + ensure!(child.wait().await?.success(), "JS peer failed"); + session.abort(moq_net::Error::Cancel); + let _ = driver.await?; + eprintln!("{} interop passed {version}", args[1]); + Ok(()) +} diff --git a/test/interop/clients/rust/goaway.rs b/test/interop/clients/rust/goaway.rs new file mode 100644 index 00000000000..29dc7e34437 --- /dev/null +++ b/test/interop/clients/rust/goaway.rs @@ -0,0 +1,87 @@ +//! A real Rust server hands one JS subscription between two WebSocket endpoints. +use std::process::Stdio; + +use anyhow::{Context, ensure}; +use moq_tokio::moq_net; +use tokio::io::AsyncBufReadExt; + +async fn endpoint( + origin: &moq_net::origin::Producer, + version: moq_net::Version, +) -> anyhow::Result<(moq_tokio::server::Listener, String)> { + let socket = moq_tokio::websocket::Listener::bind("127.0.0.1:0".parse()?).await?; + let url = format!("http://{}/", socket.local_addr()?); + let mut config = moq_tokio::server::Config::default(); + config.listen.version.push(version); + config.websocket = Some(socket); + config.publisher = Some(origin.consume()); + Ok((config.init_streams()?.listen().await?, url)) +} + +fn write(track: &mut moq_net::track::Producer, sequence: u64) -> anyhow::Result<()> { + let mut group = track.append_group()?; + ensure!(group.sequence == sequence, "wrong local sequence"); + group.write_frame(None, vec![sequence as u8; 32])?; + group.finish()?; + Ok(()) +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let args: Vec<_> = std::env::args().collect(); + ensure!(args.len() == 3, "usage: interop-goaway VERSION node|bun"); + let version: moq_net::Version = args[1].parse().map_err(anyhow::Error::msg)?; + let origin = moq_tokio::origin::spawn(); + let broadcast = origin.create_broadcast("handover")?; + let mut track = broadcast.create_track("video", moq_net::track::Info::default())?; + broadcast.announce(moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()))?; + let (mut first, first_url) = endpoint(&origin, version).await?; + let (mut second, second_url) = endpoint(&origin, version).await?; + let script = concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../test/interop/clients/js-native/goaway.ts" + ); + let mut command = tokio::process::Command::new(&args[2]); + if args[2] == "node" { + command.args(["--import", "tsx"]); + } + let mut child = command + .args([script, &first_url, &second_url, &args[1]]) + .stdout(Stdio::piped()) + .kill_on_drop(true) + .spawn()?; + let mut output = tokio::io::BufReader::new(child.stdout.take().context("child stdout")?).lines(); + let (session, driver) = first.accept().await.context("first accept")?.ok().await?; + ensure!(session.version() == version, "wrong first negotiated version"); + let mut demand = track.demand(); + while demand.wanted_changed().await?.is_none() {} + write(&mut track, 0)?; + // A JS acknowledgement, not a timer, determines when the first group was read. + ensure!( + output.next_line().await?.as_deref() == Some("first group"), + "JS did not read first group" + ); + session.drain().send(moq_net::goaway::Goaway::redirect(&second_url))?; + let (replacement, replacement_driver) = second.accept().await.context("replacement accept")?.ok().await?; + ensure!(replacement.version() == version, "wrong replacement negotiated version"); + // First exercise the replacement while the drained route still serves. A JS + // acknowledgement of its namespace proves more than the SETUP handshake. + write(&mut track, 1)?; + ensure!( + output.next_line().await?.as_deref() == Some("replacement ready"), + "replacement route did not arrive" + ); + // Only the replacement can deliver the next group after this transport is cut. + session.abort(moq_net::Error::Cancel); + let _ = driver.await?; + write(&mut track, 2)?; + ensure!( + output.next_line().await?.as_deref() == Some("goaway resumed"), + "JS subscription did not resume" + ); + ensure!(child.wait().await?.success(), "JS client failed"); + replacement.abort(moq_net::Error::Cancel); + let _ = replacement_driver.await?; + eprintln!("goaway redirect resumed {version}"); + Ok(()) +} diff --git a/test/interop/clients/rust/tail.rs b/test/interop/clients/rust/tail.rs index a27b2d8a0f4..3fb49af8abf 100644 --- a/test/interop/clients/rust/tail.rs +++ b/test/interop/clients/rust/tail.rs @@ -30,22 +30,27 @@ fn write(track: &mut moq_net::track::Producer, drop: bool) -> anyhow::Result<()> Ok(()) } -#[tokio::main] +// Keep metadata acceptance and alias registration in one task turn, before the +// subscriber acknowledges readiness to the harness. +#[tokio::main(flavor = "current_thread")] async fn main() -> anyhow::Result<()> { let args: Vec<_> = std::env::args().collect(); ensure!( (4..=6).contains(&args.len()), - "usage: interop-tail publish|subscribe URL BROADCAST [VERSION [drop|join|cached]]" + "usage: interop-tail publish|subscribe URL BROADCAST [VERSION [drop|join|cached|datagram|status]]" ); let url: url::Url = args[2].parse()?; let mode = args.get(5).map(String::as_str); ensure!( - matches!(mode, None | Some("drop" | "join" | "cached")), + matches!(mode, None | Some("drop" | "join" | "cached" | "datagram" | "status")), "unknown mode: {mode:?}" ); let drop = mode == Some("drop"); - let join = mode == Some("join"); + let join = matches!(mode, Some("join" | "status")); let cached = mode == Some("cached"); + let datagram = mode == Some("datagram"); + let status = mode == Some("status"); + let mut commands = tokio::io::BufReader::new(tokio::io::stdin()).lines(); let mut config = moq_tokio::connect::Config::default(); // An empty version offers the defaults, so a mode can follow it. if let Some(version) = args.get(4).filter(|version| !version.is_empty()) { @@ -60,7 +65,7 @@ async fn main() -> anyhow::Result<()> { let broadcast = origin.create_broadcast(&args[3])?; let mut track = broadcast.create_track( "tail", - moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI), + moq_net::track::Info::default().with_timescale((!datagram).then_some(moq_net::Timescale::MILLI)), )?; broadcast.announce(Default::default())?; if cached { @@ -75,15 +80,29 @@ async fn main() -> anyhow::Result<()> { if !cached { let mut demand = track.demand(); while demand.wanted_changed().await?.is_none() {} - write(&mut track, drop)?; + if datagram { + // Keep the track open until the harness acknowledges: a reliable end + // must not overtake best-effort datagrams on their independent channel. + for sequence in 0..GROUPS { + ensure!( + commands.next_line().await?.as_deref() == Some(&format!("datagram {sequence}")), + "missing datagram command" + ); + track.append_datagram(None, vec![sequence as u8; 900])?; + } + } else { + write(&mut track, drop)?; + } } // A finished track is not a drained transport. Keep the session alive // until the harness confirms the subscriber checked every byte and EOF. - let mut ack = String::new(); - tokio::io::BufReader::new(tokio::io::stdin()) - .read_line(&mut ack) - .await?; - ensure!(ack == "clean end\n", "reader did not acknowledge a clean end"); + ensure!( + commands.next_line().await?.as_deref() == Some("clean end"), + "reader did not acknowledge a clean end" + ); + if datagram { + track.finish()?; + } broadcast.close(); session.close().await?; eprintln!("tail acknowledged"); @@ -97,10 +116,40 @@ async fn main() -> anyhow::Result<()> { .established() .await?; let broadcast = consumer.routed_broadcast(&args[3]).await?; + let remote = broadcast.track("tail")?; + if status { + // Query before subscribing so draft-18+ must use TRACK_STATUS, + // rather than learn metadata from SUBSCRIBE_OK as a side effect. + let info = remote.query().await?; + ensure!(info.timescale.is_none(), "IETF track should be untimed"); + ensure!(info.priority == 2, "TRACK_STATUS lost publisher priority"); + eprintln!("track status answered"); + session.close().await?; + client.close().await; + return Ok(()); + } let options = moq_net::track::Subscription::default() .with_max_delay(Duration::from_secs(if drop { 60 } else { 5 })) .with_groups(0..); - let mut track = broadcast.track("tail")?.subscribe(Some(options)).await?; + let mut track = remote.subscribe(Some(options)).await?; + if datagram { + println!("datagram ready"); + for sequence in 0..GROUPS { + let frame = track.recv_datagram().await?.context("missing datagram")?; + ensure!(frame.sequence == sequence, "wrong datagram sequence"); + ensure!(frame.payload.len() == 900, "truncated datagram"); + ensure!( + frame.payload.iter().all(|byte| *byte == sequence as u8), + "corrupt datagram" + ); + ensure!(frame.timestamp.is_none(), "unexpected IETF timestamp"); + println!("datagram {sequence}"); + } + eprintln!("datagram clean groups=0,1,2,3 bytes=3600"); + session.close().await?; + client.close().await; + return Ok(()); + } let mut seen = Vec::new(); while let Some(mut group) = track.recv_group().await? { let sequence = group.sequence; diff --git a/test/interop/interop.sh b/test/interop/interop.sh index 848b0170822..c50aaafcf1f 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -370,6 +370,18 @@ plan_pass() { needs "$runtime" || continue record "not run" case=tail publisher=rust subscriber="$runtime" record "not run" case=tail publisher="$runtime" subscriber=rust + if [[ "$VERSION" == moq-transport-* ]]; then + record "not run" case=datagram publisher=rust subscriber="$runtime" + record "not run" case=datagram publisher="$runtime" subscriber=rust + record "not run" case="wire track status" publisher=rust subscriber="$runtime" + case "$VERSION" in + moq-transport-14 | moq-transport-16) record "not run" case="request credit" publisher="$runtime" subscriber=rust ;; + *) + record "not run" case="track status" publisher="$runtime" subscriber=rust + record "not run" case=goaway publisher=rust subscriber="$runtime" + ;; + esac + fi done [[ "$VERSION" == auto ]] && needs js-native-node || return 0 for version in "${DROP_VERSIONS[@]}"; do @@ -447,7 +459,7 @@ build_relay_cli() { exit 1 } if [[ "$NEGATIVE" -eq 0 && "$MEDIA" -eq 0 ]]; then - (cd "$WORKSPACE" && cargo build --locked ${flag[@]+"${flag[@]}"} -p moq-cli --example interop-tail) + (cd "$WORKSPACE" && cargo build --locked ${flag[@]+"${flag[@]}"} -p moq-cli --example interop-tail --example interop-control --example interop-goaway) fi [[ -n "$RELAY" ]] || RELAY="$TARGET_BASE/$PROFILE/moq-relay" # The `moq-cli` crate ships its binary as `moq` (a `[[bin]]` override). @@ -1235,9 +1247,17 @@ run_tail_publisher() { run_tail_client "$@" <"$fifo" 9>&- } +# shellcheck disable=SC2329 # invoked indirectly via harness_spawn +run_tail_subscriber() { + local fifo="$1" + shift + run_tail_client "$@" >"$fifo" 8>&- +} + run_tail_pair() { - local pub="$1" sub="$2" name="tail-$1-$2" pid subscriber_pid started=$SECONDS from version read - local fifo="$LOGS/tail-ack" broadcast="tail-$1-$2-$$-$RANDOM" mode="" want='tail clean end=4 groups=0,1,2,3 bytes=1048576' + local pub="$1" sub="$2" kind="${3:-tail}" pid subscriber_pid started=$SECONDS from version read + local name="${kind// /-}-$1-$2" + local fifo="$LOGS/tail-ack" broadcast="$name-$$-$RANDOM" mode="" want='tail clean end=4 groups=0,1,2,3 bytes=1048576' version=$(pinned_version) # moq-net subscribes over moq-transport from the largest group, which can be past group 0 by # the time the relay answers, so the reader takes any contiguous run through the last group. @@ -1245,7 +1265,14 @@ run_tail_pair() { mode="join" want='tail join clean end=4 groups=[0-3](,[0-3])*' fi - local cell=(case=tail publisher="$pub" subscriber="$sub" log="$LOGS/$name-sub.log") + if [[ "$kind" == datagram ]]; then + mode=datagram + want='datagram clean groups=0,1,2,3 bytes=3600' + elif [[ "$kind" == "track status" ]]; then + mode=status + want="track status answered" + fi + local cell=(case="$kind" publisher="$pub" subscriber="$sub" log="$LOGS/$name-sub.log") if is_broken "$pub" || is_broken "$sub"; then echo " FAIL tail $pub -> $sub (client unavailable)" record fail "${cell[@]}" detail="client unavailable" @@ -1259,18 +1286,36 @@ run_tail_pair() { # Each side's token grants this broadcast alone, like a media round's. harness_spawn "$name-pub" "$LOGS/$name-pub.log" run_tail_publisher "$fifo" "$pub" publish "$broadcast" "$(token_url --publish "$(publish_grant "$broadcast")")" "$version" "$mode" pid="$HARNESS_PID" - harness_spawn "$name-sub" "$LOGS/$name-sub.log" run_tail_client "$sub" subscribe "$broadcast" "$(token_url --subscribe "$(subscribe_grant "$broadcast")")" "$version" "$mode" - subscriber_pid="$HARNESS_PID" + if [[ "$mode" == datagram ]]; then + # Subscription metadata, on the clients' serialized task runtimes, confirms the alias is bound. An acknowledgement + # per datagram avoids testing a best-effort receive buffer's burst capacity. + mkfifo "$LOGS/datagram-ack" + exec 8<>"$LOGS/datagram-ack" + harness_spawn "$name-sub" "$LOGS/$name-sub.log" run_tail_subscriber "$LOGS/datagram-ack" "$sub" subscribe "$broadcast" "$(token_url --subscribe "$(subscribe_grant "$broadcast")")" "$version" "$mode" + subscriber_pid="$HARNESS_PID" + local ack sequence + if read -r -t "$TIMEOUT" ack <&8 && [[ "$ack" == "datagram ready" ]]; then + for sequence in 0 1 2 3; do + printf 'datagram %s\n' "$sequence" >&9 + read -r -t "$TIMEOUT" ack <&8 && [[ "$ack" == "datagram $sequence" ]] || break + done + fi + exec 8>&- + rm "$LOGS/datagram-ack" + else + harness_spawn "$name-sub" "$LOGS/$name-sub.log" run_tail_client "$sub" subscribe "$broadcast" "$(token_url --subscribe "$(subscribe_grant "$broadcast")")" "$version" "$mode" + subscriber_pid="$HARNESS_PID" + fi harness_wait "$subscriber_pid" || true # Like run_native, a verified data result survives the NAPI addon's exit crash. - if read=$(grep -xE "$want" "$LOGS/$name-sub.log"); then + if read=$(grep -xE "$want" "$LOGS/$name-sub.log") && { [[ "$mode" != status ]] || grep -qx "track status answered" "$LOGS/$name-sub.log"; }; then printf 'clean end\n' >&9 harness_wait "$pid" || true if [[ -n "$version" ]] && ! check_versions "tail $pub -> $sub" "$from" "$version"; then record fail "${cell[@]}" seconds=$((SECONDS - started)) detail="a relay connection negotiated another version" overall=1 elif grep -qx 'tail acknowledged' "$LOGS/$name-pub.log"; then - echo " PASS tail $pub -> $sub (${read#tail })" + echo " PASS $kind $pub -> $sub (${read#tail })" record pass "${cell[@]}" seconds=$((SECONDS - started)) else echo " FAIL tail $pub -> $sub (publisher did not acknowledge)" @@ -1458,6 +1503,31 @@ run_matrix() { fi } +# Standalone control peers use the same pinned draft, with OS-assigned endpoint ports. +run_control_case() { + local runtime="$1" kind="$2" version name="$2-$1" started=$SECONDS + local cell=(case="$kind" publisher=rust subscriber="$runtime" log="$LOGS/$name.log") + version=$(pinned_version) + local command=("$TARGET_BASE/$PROFILE/examples/interop-control" "$kind" "$version" "${runtime##*-}") + case "$kind" in + status) cell=(case="wire track status" publisher=rust subscriber="$runtime" log="$LOGS/$name.log") ;; + credit) cell=(case="request credit" publisher="$runtime" subscriber=rust log="$LOGS/$name.log") ;; + goaway) command=("$TARGET_BASE/$PROFILE/examples/interop-goaway" "$version" "${runtime##*-}") ;; + esac + if is_broken "$runtime"; then + record fail "${cell[@]}" detail="client unavailable" + overall=1 + elif timeout -k 3 "$TIMEOUT" "${command[@]}" >"$LOGS/$name.log" 2>&1; then + echo " PASS $kind rust <-> $runtime" + record pass "${cell[@]}" seconds=$((SECONDS - started)) + else + echo " FAIL $kind rust <-> $runtime" + record fail "${cell[@]}" seconds=$((SECONDS - started)) detail="control peer failed" + cat "$LOGS/$name.log" + overall=1 + fi +} + # This checkout's CLI FETCH and finite track tails in this pass, then the lite drop lanes. run_tails() { local runtime @@ -1469,6 +1539,18 @@ run_tails() { if needs "$runtime"; then run_tail_pair rust "$runtime" run_tail_pair "$runtime" rust + if [[ "$VERSION" == moq-transport-* ]]; then + run_tail_pair rust "$runtime" datagram + run_tail_pair "$runtime" rust datagram + run_control_case "$runtime" status + case "$VERSION" in + moq-transport-14 | moq-transport-16) run_control_case "$runtime" credit ;; + *) + run_tail_pair "$runtime" rust "track status" + run_control_case "$runtime" goaway + ;; + esac + fi fi done # The drop lanes pin their own lite versions. diff --git a/test/smoke/cloudflare.sh b/test/smoke/cloudflare.sh index db4dff1001e..edc6ed39e1b 100755 --- a/test/smoke/cloudflare.sh +++ b/test/smoke/cloudflare.sh @@ -261,7 +261,6 @@ for transport in webtransport quic; do webtransport) url="$HTTPS_URL" ;; quic) url="$MOQT_URL" ;; esac - # moq-dev's IETF adapter currently forwards subgroup streams, not datagrams. # A moderate cadence avoids conflating lossy live-edge turnover with protocol # interoperability while still validating sustained multi-group delivery. run_round moq-dev "$transport" "$url" subgroup 65536 32 25 From f70b280be0f57943b104947e3290143e54a161bd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 05:37:52 -0700 Subject: [PATCH 3/7] Cover reverse GOAWAY and JS request-credit waits Keep both peers on real QMux sessions, use a replacement-only marker before cutting the old endpoint, and preserve the legacy epoch acceptance gap in the active quest. Keep compression fixture modes from current main. Co-Authored-By: GPT-6 --- quest/m1/interop-coverage/control.md | 14 +- rs/moq-cli/Cargo.toml | 4 + test/interop/README.md | 9 +- test/interop/clients/js-native/server.ts | 242 +++++++++++++++++++++ test/interop/clients/js-native/tail.ts | 13 +- test/interop/clients/rust/goaway-client.rs | 74 +++++++ test/interop/clients/rust/tail.rs | 14 +- test/interop/interop.sh | 28 ++- 8 files changed, 378 insertions(+), 20 deletions(-) create mode 100644 test/interop/clients/js-native/server.ts create mode 100644 test/interop/clients/rust/goaway-client.rs diff --git a/quest/m1/interop-coverage/control.md b/quest/m1/interop-coverage/control.md index 272dbe8c42d..ea9007ce8b7 100644 --- a/quest/m1/interop-coverage/control.md +++ b/quest/m1/interop-coverage/control.md @@ -19,7 +19,7 @@ section if it is cheaper than a matrix case. Public API: none. Wire: none. -## Progress +**Progress:** The implementation PR adds relay datagrams in both Rust/native-JS directions on all five drafts, JS wire TRACK_STATUS queries to Rust on all five, and Rust @@ -28,18 +28,20 @@ TRACK_STATUS_OK/ERROR routing in the JS control adapter; the patch includes success/refusal/success coverage. A JS peer advertising one usable track slot checks Rust waiting for credit on drafts 14 and 16. Real Rust-to-JS GOAWAY with a new URI retains a reader on drafts 18, 20, and 22, with the same publisher epoch and a -replacement-only data barrier before cutting the original transport. +replacement-only data barrier before cutting the original transport. The reverse +JS-server-to-Rust-client case uses a marker broadcast excluded from the original +endpoint's scope to prove replacement-only delivery. JS-subscriber credit uses a +controlled JS QMux peer with two slots (namespace discovery plus one held track); +the same peer codecs are checked cross-language by the Rust-subscriber case. +Small-window setup stays in test peers, with no production configuration knob. The initial datagram burst and GOAWAY readiness failures were fixture errors. No production handover policy change is justified by those results. -## Remaining +**Remaining:** - Decide what the GOAWAY requirement means on drafts 14 and 16. They cannot carry publisher epochs, and the documented model restarts across an unproven route rather than splicing a retained reader. Recommend testing application re-request/re-subscribe there and keeping same-reader continuity on 18+. -- Add the reverse GOAWAY direction (JS server to Rust reconnecting client) and - request credit with JS as the subscriber. Keep small-window setup in test peers, - rather than adding a public configuration knob solely for coverage. - Retain this quest until the remaining coverage and legacy acceptance are settled. diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index 5c49e48cbfa..e485de9fed2 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -184,3 +184,7 @@ 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" diff --git a/test/interop/README.md b/test/interop/README.md index e834cbf17a0..3a8b83edcbf 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -151,16 +151,21 @@ The in-tree control lanes also run with `--tail`: 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 + 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. + 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 cannot carry epochs and require an application restart, so this retained-reader lane does not claim to cover them. diff --git a/test/interop/clients/js-native/server.ts b/test/interop/clients/js-native/server.ts new file mode 100644 index 00000000000..2d0f16f2a85 --- /dev/null +++ b/test/interop/clients/js-native/server.ts @@ -0,0 +1,242 @@ +/** Controlled real QMux peers for reverse GOAWAY and the JS subscriber's credit wait. */ +import assert from "node:assert/strict"; +import { createInterface } from "node:readline"; +import { Milli } from "@moq/clock"; +import * as Moq from "@moq/net"; +import * as Extensions from "../../../../js/net/src/connection/extensions.ts"; +import { exchangeSetup } from "../../../../js/net/src/connection/handshake.ts"; +import { Connection } from "../../../../js/net/src/ietf/connection.ts"; +import { GoAway } from "../../../../js/net/src/ietf/goaway.ts"; +import * as Ietf from "../../../../js/net/src/ietf/index.ts"; +import * as Lite from "../../../../js/net/src/lite/index.ts"; +import { Stream } from "../../../../js/net/src/stream.ts"; +import Qmux, { selectSubprotocol } from "../../../../js/qmux/src/index.ts"; + +console.debug = console.error; +console.info = console.error; +const [mode, name] = process.argv.slice(2); +assert(name && ["goaway", "credit"].includes(mode), "usage: server.ts goaway|credit VERSION"); +const version = (0xff000000 + Number(name.split("-").at(-1))) as Ietf.IetfVersion; +const legacy = version === Ietf.Version.DRAFT_14 || version === Ietf.Version.DRAFT_16; +const alpn = version === Ietf.Version.DRAFT_14 ? "moql" : `moqt-${Number(version) - 0xff000000}`; +const extensions = Extensions.offered(); +const origin = new Moq.Origin.Producer(); +const broadcast = origin.createBroadcast(Moq.Path.from("handover")); +const firstTrack = broadcast.createTrack("first"); +const secondTrack = broadcast.createTrack("second"); +const video = broadcast.createTrack("video"); +broadcast.announce({ epoch: Moq.Epoch.mint() }); +const marker = origin.createBroadcast(Moq.Path.from("replacement")); +const ready = marker.createTrack("ready"); +const markerGroup = ready.appendGroup(); +markerGroup.writeFrame({ payload: new Uint8Array([42]) }); +markerGroup.close(); +marker.announce(); + +type Socket = { + protocol: string; + url: string; + receive?: (data: Uint8Array) => void; + end?: (code: number, reason: string) => void; +}; + +function listen(publish: Moq.Origin.Consumer, window?: bigint) { + const accepted = Promise.withResolvers<{ connection: Connection; control: Stream }>(); + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request, server) { + const protocol = selectSubprotocol(request.headers.get("sec-websocket-protocol"), { + protocols: [alpn], + versions: { [alpn]: legacy ? "qmux-00" : "qmux-01" }, + requireProtocol: true, + }); + if (!protocol) return new Response("no protocol", { status: 400 }); + if ( + server.upgrade(request, { + headers: { "Sec-WebSocket-Protocol": protocol }, + data: { protocol, url: request.url }, + }) + ) + return; + return new Response("upgrade refused", { status: 400 }); + }, + websocket: { + open(socket) { + const closed = Promise.withResolvers<{ closeCode: number; reason: string }>(); + const readable = new ReadableStream({ + start(controller) { + socket.data.receive = (data) => controller.enqueue(data); + socket.data.end = (closeCode, reason) => { + controller.close(); + closed.resolve({ closeCode, reason }); + }; + }, + }); + const writable = new WritableStream({ + write(data) { + socket.send(data); + }, + }); + const quic = Qmux.accept( + { + url: socket.data.url, + opened: Promise.resolve({ readable, writable, protocol: socket.data.protocol, extensions: "" }), + closed: closed.promise, + close() { + socket.close(); + }, + }, + { protocol: socket.data.protocol }, + ); + void accept(quic, new URL(socket.data.url), publish, window).then(accepted.resolve, accepted.reject); + }, + message(socket, data) { + assert(typeof data !== "string"); + socket.data.receive?.(new Uint8Array(data)); + }, + close(socket, code, reason) { + socket.data.end?.(code, reason); + }, + }, + }); + return { server, url: `http://127.0.0.1:${server.port}/`, accepted: accepted.promise }; +} + +async function accept(quic: WebTransport, url: URL, publish: Moq.Origin.Consumer, window?: bigint) { + if (!legacy) { + const setup = await exchangeSetup(quic, version, "interop-js-server", { client: false, extensions }); + return { + control: setup.control, + connection: new Connection({ + url, + quic, + version, + client: false, + maxRequestId: 0n, + publish, + extensions, + ...setup, + }), + }; + } + assert(window !== undefined, "legacy server needs an explicit test window"); + const control = await Stream.accept(quic, version); + assert(control, "missing SETUP"); + assert.equal(await control.reader.u53(), Lite.StreamId.ClientCompat); + const client = await Ietf.ClientSetup.decode(control.reader, version); + const parameters = new Ietf.SetupOptions(); + parameters.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(true, window)); + Extensions.intoSetup(parameters, extensions, version); + await control.writer.u53(Lite.StreamId.ServerCompat); + await new Ietf.ServerSetup({ version, parameters }).encode(control.writer, version); + return { + control, + connection: new Connection({ + url, + quic, + control, + version, + client: false, + publish, + extensions, + requestWindow: window, + maxRequestId: client.parameters.getVarint(Ietf.SetupOption.MaxRequestId) ?? 0n, + solicit: Ietf.solicitFromSetup(client.parameters), + }), + }; +} + +if (mode === "credit") { + assert(legacy, "request credit exists on drafts 14 and 16"); + // Namespace discovery holds one slot; the first track holds the other. + const peer = listen(origin.consume(), 2n); + const received = new Moq.Origin.Producer(); + const client = await Moq.Connection.connect({ + url: new URL(peer.url), + versions: [name], + websocket: { delay: Milli(0) }, + consume: received, + }); + const server = await peer.accepted; + const request = received.request(Moq.Path.from("handover"), { announced: true }); + try { + let remote = request.active.peek(); + while (!remote) { + await request.active.changed(); + remote = request.active.peek(); + } + const first = remote.track("first").subscribe(); + await first.info(); + const second = remote.track("second").subscribe(); + const queued = second.info(); + try { + const probe = firstTrack.appendGroup(); + probe.writeFrame({ payload: new Uint8Array([1]) }); + probe.close(); + const data = await first.recvGroup(); + assert(data); + assert.deepEqual((await data.readFrame())?.payload, new Uint8Array([1])); + data.close(); + first.close(); + await queued; + const next = secondTrack.appendGroup(); + next.writeFrame({ payload: new Uint8Array([2]) }); + next.close(); + const resumed = await second.recvGroup(); + assert(resumed); + assert.deepEqual((await resumed.readFrame())?.payload, new Uint8Array([2])); + resumed.close(); + console.error("js request credit resumed"); + } finally { + first.close(); + second.close(); + } + } finally { + request.close(); + await client.close(); + server.connection.abort(); + peer.server.stop(true); + received.close(); + origin.close(); + } +} else { + assert(!legacy, "retained-reader GOAWAY requires an epoch-carrying draft"); + const scoped = origin.scope(Moq.Path.empty(), new Moq.Path.Patterns([Moq.Path.Pattern.literal("handover")])); + const first = listen(scoped.consume()); + const second = listen(origin.consume()); + process.stdout.write(`${JSON.stringify([first.url, second.url])}\n`); + const lines = createInterface({ input: process.stdin }); + const commands = lines[Symbol.asyncIterator](); + let initial: Awaited | undefined; + let replacement: Awaited | undefined; + try { + initial = await first.accepted; + const demand = video.demand(); + while (demand.wanted.peek() === undefined) await demand.wanted.changed(); + const group = video.appendGroup(); + group.writeFrame({ payload: new Uint8Array(32) }); + group.close(); + assert.equal((await commands.next()).value, "first group"); + await initial.control.writer.u53(GoAway.id); + await new GoAway({ + newSessionUri: second.url, + requestId: version === Ietf.Version.DRAFT_18 ? 0n : undefined, + }).encode(initial.control.writer, version); + replacement = await second.accepted; + assert.equal((await commands.next()).value, "replacement ready"); + initial.connection.abort(); + const next = video.appendGroup(); + next.writeFrame({ payload: new Uint8Array(32).fill(1) }); + next.close(); + assert.equal((await commands.next()).value, "done"); + console.error("js goaway redirect resumed"); + } finally { + lines.close(); + initial?.connection.abort(); + replacement?.connection.abort(); + first.server.stop(true); + second.server.stop(true); + origin.close(); + } +} diff --git a/test/interop/clients/js-native/tail.ts b/test/interop/clients/js-native/tail.ts index 91ff334d9af..a3cdc4154ec 100644 --- a/test/interop/clients/js-native/tail.ts +++ b/test/interop/clients/js-native/tail.ts @@ -66,7 +66,7 @@ try { continue; } for (const size of compress ? [BYTES / 2, 0, BYTES / 2] : [BYTES]) { - group.writeFrame({ payload: new Uint8Array(size).fill(sequence) }); + group.writeFrame({ payload: new Uint8Array(size).fill(sequence) }); } group.close(); } @@ -118,10 +118,13 @@ try { try { assert(group.sequence < GROUPS, `unexpected group ${group.sequence}`); for (const size of compress ? [BYTES / 2, 0, BYTES / 2] : [BYTES]) { - const frame = await group.readFrame(); - assert(frame, "empty group"); - assert.equal(frame.payload.byteLength, size, "truncated group"); - assert(frame.payload.every((byte) => byte === group.sequence), "corrupt group"); + const frame = await group.readFrame(); + assert(frame, "empty group"); + assert.equal(frame.payload.byteLength, size, "truncated group"); + assert( + frame.payload.every((byte) => byte === group.sequence), + "corrupt group", + ); } assert.equal(await group.readFrame(), undefined, "extra frame"); seen.push(group.sequence); diff --git a/test/interop/clients/rust/goaway-client.rs b/test/interop/clients/rust/goaway-client.rs new file mode 100644 index 00000000000..9b55f65f70e --- /dev/null +++ b/test/interop/clients/rust/goaway-client.rs @@ -0,0 +1,74 @@ +//! Keep a Rust reader while a real JS QMux server redirects it to a replacement. +use std::process::Stdio; + +use anyhow::{Context, ensure}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt}; + +#[tokio::main(flavor = "current_thread")] +async fn main() -> anyhow::Result<()> { + let version = std::env::args() + .nth(1) + .context("usage: interop-goaway-client VERSION")?; + let mut child = tokio::process::Command::new("bun") + .args([ + concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../test/interop/clients/js-native/server.ts" + ), + "goaway", + &version, + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .kill_on_drop(true) + .spawn()?; + let mut output = tokio::io::BufReader::new(child.stdout.take().context("stdout")?).lines(); + let urls: Vec = serde_json::from_str(&output.next_line().await?.context("JS endpoint URLs")?)?; + ensure!(urls.len() == 2 && urls[0] != urls[1], "missing replacement endpoint"); + let mut config = moq_tokio::connect::Config::default(); + config.version.push(version.parse().map_err(anyhow::Error::msg)?); + let client = config.init(Default::default())?; + let origin = moq_tokio::origin::spawn(); + let consumer = origin.consume(); + let connection = client + .clone() + .with_subscriber(origin) + .connect(url::Url::parse(&urls[0])?) + .established() + .await?; + ensure!( + connection.version() == Some(version.parse().map_err(anyhow::Error::msg)?), + "wrong negotiated version" + ); + let broadcast = consumer.routed_broadcast("handover").await?; + let mut track = broadcast.track("video")?.subscribe(None).await?; + let input = child.stdin.as_mut().context("stdin")?; + for sequence in 0..2 { + let mut group = track.recv_group().await?.context("reader ended during GOAWAY")?; + ensure!(group.sequence == sequence, "wrong group after migration"); + ensure!( + group.read_frame().await?.context("empty group")?.payload.as_ref() == [sequence as u8; 32], + "corrupt group" + ); + ensure!(group.read_frame().await?.is_none(), "extra frame"); + if sequence == 0 { + input.write_all(b"first group\n").await?; + // The first endpoint's publisher scope excludes this broadcast, so its bytes + // prove that the replacement session is serving beyond its SETUP handshake. + let ready = consumer.routed_broadcast("replacement").await?; + let mut marker = ready.track("ready")?.subscribe(None).await?; + let mut marker = marker.recv_group().await?.context("replacement marker")?; + ensure!( + marker.read_frame().await?.context("marker frame")?.payload.as_ref() == [42], + "wrong marker" + ); + input.write_all(b"replacement ready\n").await?; + } + } + input.write_all(b"done\n").await?; + ensure!(child.wait().await?.success(), "JS server failed"); + let _ = connection.close().await; + client.close().await; + eprintln!("reverse goaway resumed {version}"); + Ok(()) +} diff --git a/test/interop/clients/rust/tail.rs b/test/interop/clients/rust/tail.rs index 4e8f77b1157..6ce1d06b186 100644 --- a/test/interop/clients/rust/tail.rs +++ b/test/interop/clients/rust/tail.rs @@ -50,7 +50,10 @@ async fn main() -> anyhow::Result<()> { let url: url::Url = args[2].parse()?; let mode = args.get(5).map(String::as_str); ensure!( - matches!(mode, None | Some("drop" | "join" | "cached" | "compress" | "datagram" | "status")), + matches!( + mode, + None | Some("drop" | "join" | "cached" | "compress" | "datagram" | "status") + ), "unknown mode: {mode:?}" ); let drop = mode == Some("drop"); @@ -74,7 +77,9 @@ async fn main() -> anyhow::Result<()> { let broadcast = origin.create_broadcast(&args[3])?; let mut track = broadcast.create_track( "tail", - moq_net::track::Info::default().with_timescale((!datagram).then_some(moq_net::Timescale::MILLI)).with_compress(compress), + moq_net::track::Info::default() + .with_timescale((!datagram).then_some(moq_net::Timescale::MILLI)) + .with_compress(compress), )?; broadcast.announce(Default::default())?; if cached { @@ -160,7 +165,10 @@ async fn main() -> anyhow::Result<()> { return Ok(()); } if compress { - ensure!(track.info().compress == args.get(4).is_some_and(|version| version == "moq-lite-07-wip"), "wrong compression metadata"); + ensure!( + track.info().compress == args.get(4).is_some_and(|version| version == "moq-lite-07-wip"), + "wrong compression metadata" + ); } let mut seen = Vec::new(); while let Some(mut group) = track.recv_group().await? { diff --git a/test/interop/interop.sh b/test/interop/interop.sh index 4a03f931e25..0642dd77217 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -377,10 +377,14 @@ plan_pass() { record "not run" case=datagram publisher="$runtime" subscriber=rust record "not run" case="wire track status" publisher=rust subscriber="$runtime" case "$VERSION" in - moq-transport-14 | moq-transport-16) record "not run" case="request credit" publisher="$runtime" subscriber=rust ;; + moq-transport-14 | moq-transport-16) + record "not run" case="request credit" publisher="$runtime" subscriber=rust + [[ "$runtime" != js-native-bun ]] || record "not run" case="request credit" publisher="$runtime" subscriber="$runtime" + ;; *) record "not run" case="track status" publisher="$runtime" subscriber=rust record "not run" case=goaway publisher=rust subscriber="$runtime" + [[ "$runtime" != js-native-bun ]] || record "not run" case=goaway publisher="$runtime" subscriber=rust ;; esac fi @@ -469,7 +473,7 @@ build_relay_cli() { exit 1 } if [[ "$NEGATIVE" -eq 0 && "$MEDIA" -eq 0 ]]; then - (cd "$WORKSPACE" && cargo build --locked ${flag[@]+"${flag[@]}"} -p moq-cli --example interop-tail --example interop-control --example interop-goaway) + (cd "$WORKSPACE" && cargo build --locked ${flag[@]+"${flag[@]}"} -p moq-cli --example interop-tail --example interop-control --example interop-goaway --example interop-goaway-client) fi [[ -n "$RELAY" ]] || RELAY="$TARGET_BASE/$PROFILE/moq-relay" # The `moq-cli` crate ships its binary as `moq` (a `[[bin]]` override). @@ -1526,11 +1530,23 @@ run_control_case() { status) cell=(case="wire track status" publisher=rust subscriber="$runtime" log="$LOGS/$name.log") ;; credit) cell=(case="request credit" publisher="$runtime" subscriber=rust log="$LOGS/$name.log") ;; goaway) command=("$TARGET_BASE/$PROFILE/examples/interop-goaway" "$version" "${runtime##*-}") ;; + reverse-goaway) + command=("$TARGET_BASE/$PROFILE/examples/interop-goaway-client" "$version") + cell=(case=goaway publisher="$runtime" subscriber=rust log="$LOGS/$name.log") + ;; + js-credit) + command=(bun "$CLIENTS/js-native/server.ts" credit "$version") + cell=(case="request credit" publisher="$runtime" subscriber="$runtime" log="$LOGS/$name.log") + ;; esac if is_broken "$runtime"; then record fail "${cell[@]}" detail="client unavailable" overall=1 - elif timeout -k 3 "$TIMEOUT" "${command[@]}" >"$LOGS/$name.log" 2>&1; then + return + fi + # Own the Rust launcher and its JS child together, including on a deadline or failure. + harness_spawn "$name" "$LOGS/$name.log" timeout -k 3 "$TIMEOUT" "${command[@]}" + if harness_wait "$HARNESS_PID"; then echo " PASS $kind rust <-> $runtime" record pass "${cell[@]}" seconds=$((SECONDS - started)) else @@ -1557,10 +1573,14 @@ run_tails() { run_tail_pair "$runtime" rust datagram run_control_case "$runtime" status case "$VERSION" in - moq-transport-14 | moq-transport-16) run_control_case "$runtime" credit ;; + moq-transport-14 | moq-transport-16) + run_control_case "$runtime" credit + [[ "$runtime" != js-native-bun ]] || run_control_case "$runtime" js-credit + ;; *) run_tail_pair "$runtime" rust "track status" run_control_case "$runtime" goaway + [[ "$runtime" != js-native-bun ]] || run_control_case "$runtime" reverse-goaway ;; esac fi From 22a10cfcab74e9190d2ba7fa71eac5ecf2ffb4ba Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 05:42:01 -0700 Subject: [PATCH 4/7] Quote the compression fixture mode for shellcheck Co-Authored-By: GPT-6 --- test/interop/interop.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/interop/interop.sh b/test/interop/interop.sh index 0642dd77217..60ebf5a3f17 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -1286,7 +1286,7 @@ run_tail_pair() { mode=status want="track status answered" elif [[ "$kind" == compress ]]; then - mode=compress + mode="compress" fi local cell=(case="$kind" publisher="$pub" subscriber="$sub" log="$LOGS/$name-sub.log") [[ -z "${4:-}" ]] || cell+=(version="$version") From cb690c61b1d597076c578dc2e95b70595c6537d5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 14:36:26 -0700 Subject: [PATCH 5/7] Cover GOAWAY re-subscribe on drafts 14 and 16 Drafts 14 and 16 carry no publisher epoch, so a redirect cannot prove the replacement serves the same publisher. The GOAWAY lanes now run on every IETF draft: on 14/16 they check that the held reader ends with the original transport and that requesting the path again subscribes through the new URI and reads the final group. The JS QMux peer also accepts Rust's moq-00 draft-14 ALPN for the reverse direction. Retire the control-cases quest. Co-Authored-By: Claude Opus 5.5 --- quest/m1/interop-coverage/README.md | 1 - quest/m1/interop-coverage/control.md | 47 -------------------- test/interop/README.md | 6 ++- test/interop/clients/js-native/goaway.ts | 51 +++++++++++++++------- test/interop/clients/js-native/server.ts | 9 ++-- test/interop/clients/rust/goaway-client.rs | 15 ++++++- test/interop/clients/rust/goaway.rs | 10 ++++- test/interop/interop.sh | 8 ++-- 8 files changed, 70 insertions(+), 77 deletions(-) delete mode 100644 quest/m1/interop-coverage/control.md diff --git a/quest/m1/interop-coverage/README.md b/quest/m1/interop-coverage/README.md index 6894bfbb899..f642cf96fc6 100644 --- a/quest/m1/interop-coverage/README.md +++ b/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 diff --git a/quest/m1/interop-coverage/control.md b/quest/m1/interop-coverage/control.md deleted file mode 100644 index ea9007ce8b7..00000000000 --- a/quest/m1/interop-coverage/control.md +++ /dev/null @@ -1,47 +0,0 @@ -# [M] Interop covers GOAWAY, TRACK_STATUS, datagrams, and request limits - -## Goal - -On the IETF matrix (`just test interop --versions ietf`), across -Rust and JS: a GOAWAY with a new URI moves the session and its -subscriptions resume there; TRACK_STATUS gets a real answer from the other -implementation; datagram objects cross moq-relay intact; and a subscriber -opening more concurrent tracks than its MAX_REQUEST_ID grant waits for -credit instead of failing (drafts 14 and 16; drafts 17 and later have no -MAX_REQUEST_ID). - -## Plan - -Decided with the maintainer on 2026-10-21: in-tree cases, not proposed to -the runner's spec. `test/smoke/cloudflare.sh` still says moq-relay does no -datagrams; fix that comment, and add a datagram round to its moq-dev -section if it is cheaper than a matrix case. - -Public API: none. Wire: none. - -**Progress:** - -The implementation PR adds relay datagrams in both Rust/native-JS directions on -all five drafts, JS wire TRACK_STATUS queries to Rust on all five, and Rust -query-before-subscribe to JS from draft 18. Draft-14 real answers exposed missing -TRACK_STATUS_OK/ERROR routing in the JS control adapter; the patch includes -success/refusal/success coverage. A JS peer advertising one usable track slot checks -Rust waiting for credit on drafts 14 and 16. Real Rust-to-JS GOAWAY with a new URI -retains a reader on drafts 18, 20, and 22, with the same publisher epoch and a -replacement-only data barrier before cutting the original transport. The reverse -JS-server-to-Rust-client case uses a marker broadcast excluded from the original -endpoint's scope to prove replacement-only delivery. JS-subscriber credit uses a -controlled JS QMux peer with two slots (namespace discovery plus one held track); -the same peer codecs are checked cross-language by the Rust-subscriber case. -Small-window setup stays in test peers, with no production configuration knob. - -The initial datagram burst and GOAWAY readiness failures were fixture errors. -No production handover policy change is justified by those results. - -**Remaining:** - -- Decide what the GOAWAY requirement means on drafts 14 and 16. They cannot carry - publisher epochs, and the documented model restarts across an unproven route - rather than splicing a retained reader. Recommend testing application - re-request/re-subscribe there and keeping same-reader continuity on 18+. -- Retain this quest until the remaining coverage and legacy acceptance are settled. diff --git a/test/interop/README.md b/test/interop/README.md index d4e074a2eab..3a8dc73a34f 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -182,8 +182,10 @@ The in-tree control lanes also run with `--tail`: 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 cannot carry epochs and require an application restart, so this - retained-reader lane does not claim to cover them. +- 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. diff --git a/test/interop/clients/js-native/goaway.ts b/test/interop/clients/js-native/goaway.ts index c771f0a6b23..4207e5ee912 100644 --- a/test/interop/clients/js-native/goaway.ts +++ b/test/interop/clients/js-native/goaway.ts @@ -1,4 +1,4 @@ -/** Keep a real subscription alive while the Rust server redirects the connection. */ +/** Follow a Rust server's GOAWAY redirect: retain the reader where an epoch proves continuity, else re-subscribe. */ import assert from "node:assert/strict"; import { Milli } from "@moq/clock"; import * as Moq from "@moq/net"; @@ -9,6 +9,9 @@ console.debug = console.error; console.info = console.error; const [first, second, version] = process.argv.slice(2); assert(first && second && version, "usage: goaway.ts FIRST_URL SECOND_URL VERSION"); +// Drafts 14 and 16 carry no epoch, so nothing proves the replacement serves the same +// publisher: the held reader ends with the old route and the application requests again. +const legacy = version === "moq-transport-14" || version === "moq-transport-16"; const origin = new Moq.Origin.Producer(); const connection = new Reload({ url: new URL(first), @@ -18,16 +21,29 @@ const connection = new Reload({ goaway: { redirect: "follow" }, }); let replacementProbe: Moq.Track.Subscriber | undefined; -const request = origin.request(Moq.Path.from("handover"), { announced: true }); -try { +const requests = [origin.request(Moq.Path.from("handover"), { announced: true })]; +async function active(request: (typeof requests)[number]) { let broadcast = request.active.peek(); while (!broadcast) { await request.active.changed(); broadcast = request.active.peek(); } - const track = broadcast.track("video").subscribe({ groups: { start: { included: 0 } } }); + return broadcast; +} +try { + let track = (await active(requests[0])).track("video").subscribe({ groups: { start: { included: 0 } } }); try { for (let sequence = 0; sequence < 3; sequence++) { + if (sequence === 2 && legacy) { + assert.equal(await track.recvGroup(), undefined, "an epochless route spliced the held reader"); + track.close(); + const request = origin.request(Moq.Path.from("handover"), { announced: true }); + requests.push(request); + track = (await active(request)).track("video").subscribe({ groups: { start: { included: 2 } } }); + await track.info(); + // The server only writes group 2 after this, and only the replacement remains. + process.stdout.write("resubscribed\n"); + } const group = await track.recvGroup(); assert(group, "subscription ended during redirect"); try { @@ -58,17 +74,20 @@ try { } // A direct read from the replacement proves its SUBSCRIBE/data path is ready, // beyond SETUP or namespace discovery, while the original reader stays held. - replacementProbe = wireOf(replacement) - .consume(Moq.Path.from("handover")) - .track("video") - .subscribe({ groups: { start: { included: 1 } } }); - const probe = await replacementProbe.recvGroup(); - assert(probe, "replacement has no group"); - try { - assert.equal(probe.sequence, 1, "replacement served the wrong group"); - assert.deepEqual((await probe.readFrame())?.payload, new Uint8Array(32).fill(1)); - } finally { - probe.close(); + // The legacy re-subscribe proves that itself, with its own SUBSCRIBE. + if (!legacy) { + replacementProbe = wireOf(replacement) + .consume(Moq.Path.from("handover")) + .track("video") + .subscribe({ groups: { start: { included: 1 } } }); + const probe = await replacementProbe.recvGroup(); + assert(probe, "replacement has no group"); + try { + assert.equal(probe.sequence, 1, "replacement served the wrong group"); + assert.deepEqual((await probe.readFrame())?.payload, new Uint8Array(32).fill(1)); + } finally { + probe.close(); + } } process.stdout.write("replacement ready\n"); } @@ -80,7 +99,7 @@ try { } } finally { replacementProbe?.close(); - request.close(); + for (const request of requests) request.close(); connection.close(); origin.close(); } diff --git a/test/interop/clients/js-native/server.ts b/test/interop/clients/js-native/server.ts index 2d0f16f2a85..e44b127ac69 100644 --- a/test/interop/clients/js-native/server.ts +++ b/test/interop/clients/js-native/server.ts @@ -18,7 +18,8 @@ const [mode, name] = process.argv.slice(2); assert(name && ["goaway", "credit"].includes(mode), "usage: server.ts goaway|credit VERSION"); const version = (0xff000000 + Number(name.split("-").at(-1))) as Ietf.IetfVersion; const legacy = version === Ietf.Version.DRAFT_14 || version === Ietf.Version.DRAFT_16; -const alpn = version === Ietf.Version.DRAFT_14 ? "moql" : `moqt-${Number(version) - 0xff000000}`; +// JS offers draft 14 behind `moql`, negotiated in SETUP; Rust offers its own `moq-00` ALPN. +const alpns = version === Ietf.Version.DRAFT_14 ? ["moql", "moq-00"] : [`moqt-${Number(version) - 0xff000000}`]; const extensions = Extensions.offered(); const origin = new Moq.Origin.Producer(); const broadcast = origin.createBroadcast(Moq.Path.from("handover")); @@ -47,8 +48,8 @@ function listen(publish: Moq.Origin.Consumer, window?: bigint) { port: 0, fetch(request, server) { const protocol = selectSubprotocol(request.headers.get("sec-websocket-protocol"), { - protocols: [alpn], - versions: { [alpn]: legacy ? "qmux-00" : "qmux-01" }, + protocols: alpns, + versions: Object.fromEntries(alpns.map((alpn) => [alpn, legacy ? "qmux-00" : "qmux-01"])), requireProtocol: true, }); if (!protocol) return new Response("no protocol", { status: 400 }); @@ -120,7 +121,6 @@ async function accept(quic: WebTransport, url: URL, publish: Moq.Origin.Consumer }), }; } - assert(window !== undefined, "legacy server needs an explicit test window"); const control = await Stream.accept(quic, version); assert(control, "missing SETUP"); assert.equal(await control.reader.u53(), Lite.StreamId.ClientCompat); @@ -201,7 +201,6 @@ if (mode === "credit") { origin.close(); } } else { - assert(!legacy, "retained-reader GOAWAY requires an epoch-carrying draft"); const scoped = origin.scope(Moq.Path.empty(), new Moq.Path.Patterns([Moq.Path.Pattern.literal("handover")])); const first = listen(scoped.consume()); const second = listen(origin.consume()); diff --git a/test/interop/clients/rust/goaway-client.rs b/test/interop/clients/rust/goaway-client.rs index 9b55f65f70e..5984d6bb272 100644 --- a/test/interop/clients/rust/goaway-client.rs +++ b/test/interop/clients/rust/goaway-client.rs @@ -1,4 +1,4 @@ -//! Keep a Rust reader while a real JS QMux server redirects it to a replacement. +//! Follow a real JS QMux server's redirect: retain the reader with an epoch, else re-subscribe. use std::process::Stdio; use anyhow::{Context, ensure}; @@ -40,10 +40,23 @@ async fn main() -> anyhow::Result<()> { connection.version() == Some(version.parse().map_err(anyhow::Error::msg)?), "wrong negotiated version" ); + // Drafts 14 and 16 carry no epoch, so nothing proves the replacement serves the same + // publisher: the held reader ends with the old route and the application requests again. + let legacy = ["moq-transport-14", "moq-transport-16"].contains(&version.as_str()); let broadcast = consumer.routed_broadcast("handover").await?; let mut track = broadcast.track("video")?.subscribe(None).await?; let input = child.stdin.as_mut().context("stdin")?; for sequence in 0..2 { + if sequence == 1 && legacy { + ensure!( + !matches!(track.recv_group().await, Ok(Some(_))), + "an epochless route spliced the held reader" + ); + // The server wrote this group just after cutting the original, so only the + // replacement, already its largest, can serve it. + let broadcast = consumer.routed_broadcast("handover").await?; + track = broadcast.track("video")?.subscribe(None).await?; + } let mut group = track.recv_group().await?.context("reader ended during GOAWAY")?; ensure!(group.sequence == sequence, "wrong group after migration"); ensure!( diff --git a/test/interop/clients/rust/goaway.rs b/test/interop/clients/rust/goaway.rs index 29dc7e34437..daa5ee74306 100644 --- a/test/interop/clients/rust/goaway.rs +++ b/test/interop/clients/rust/goaway.rs @@ -1,4 +1,4 @@ -//! A real Rust server hands one JS subscription between two WebSocket endpoints. +//! A real Rust server redirects one JS subscriber between two WebSocket endpoints. use std::process::Stdio; use anyhow::{Context, ensure}; @@ -31,6 +31,8 @@ async fn main() -> anyhow::Result<()> { let args: Vec<_> = std::env::args().collect(); ensure!(args.len() == 3, "usage: interop-goaway VERSION node|bun"); let version: moq_net::Version = args[1].parse().map_err(anyhow::Error::msg)?; + // Without an epoch, JS re-subscribes instead of retaining its reader across the move. + let legacy = ["moq-transport-14", "moq-transport-16"].contains(&args[1].as_str()); let origin = moq_tokio::origin::spawn(); let broadcast = origin.create_broadcast("handover")?; let mut track = broadcast.create_track("video", moq_net::track::Info::default())?; @@ -74,6 +76,12 @@ async fn main() -> anyhow::Result<()> { // Only the replacement can deliver the next group after this transport is cut. session.abort(moq_net::Error::Cancel); let _ = driver.await?; + if legacy { + ensure!( + output.next_line().await?.as_deref() == Some("resubscribed"), + "JS did not re-subscribe through the replacement" + ); + } write(&mut track, 2)?; ensure!( output.next_line().await?.as_deref() == Some("goaway resumed"), diff --git a/test/interop/interop.sh b/test/interop/interop.sh index e4ce18344e4..102d6e61500 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -412,6 +412,8 @@ plan_pass() { record "not run" case=datagram publisher=rust subscriber="$runtime" record "not run" case=datagram publisher="$runtime" subscriber=rust record "not run" case="wire track status" publisher=rust subscriber="$runtime" + record "not run" case=goaway publisher=rust subscriber="$runtime" + [[ "$runtime" != js-native-bun ]] || record "not run" case=goaway publisher="$runtime" subscriber=rust case "$VERSION" in moq-transport-14 | moq-transport-16) record "not run" case="request credit" publisher="$runtime" subscriber=rust @@ -419,8 +421,6 @@ plan_pass() { ;; *) record "not run" case="track status" publisher="$runtime" subscriber=rust - record "not run" case=goaway publisher=rust subscriber="$runtime" - [[ "$runtime" != js-native-bun ]] || record "not run" case=goaway publisher="$runtime" subscriber=rust ;; esac fi @@ -1767,6 +1767,8 @@ run_tails() { run_tail_pair rust "$runtime" datagram run_tail_pair "$runtime" rust datagram run_control_case "$runtime" status + run_control_case "$runtime" goaway + [[ "$runtime" != js-native-bun ]] || run_control_case "$runtime" reverse-goaway case "$VERSION" in moq-transport-14 | moq-transport-16) run_control_case "$runtime" credit @@ -1774,8 +1776,6 @@ run_tails() { ;; *) run_tail_pair "$runtime" rust "track status" - run_control_case "$runtime" goaway - [[ "$runtime" != js-native-bun ]] || run_control_case "$runtime" reverse-goaway ;; esac fi From 5a8aef5548bf1452efd659ce6d3ea1063ec69774 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 14:49:52 -0700 Subject: [PATCH 6/7] Expect the signed wire priority in the TRACK_STATUS lane Co-Authored-By: Claude Opus 5.5 --- test/interop/clients/js-native/control.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/test/interop/clients/js-native/control.ts b/test/interop/clients/js-native/control.ts index 7e5a385cf12..8493f3d4c4e 100644 --- a/test/interop/clients/js-native/control.ts +++ b/test/interop/clients/js-native/control.ts @@ -66,7 +66,8 @@ if (mode === "status") { const status = await TrackStatusOk.decode(stream.reader, version); if (legacy) assert.equal(status.requestId, requestId); assert.deepEqual(status.largest, { groupId: 4n, objectId: 1n }, "wrong publisher location"); - if (!legacy) assert.equal(status.properties.priority, 253, "wrong publisher priority"); + // Rust publishes priority 2, which moq-transport carries as 127 - 2. + if (!legacy) assert.equal(status.properties.priority, 125, "wrong publisher priority"); stream.close(); console.error("wire track status answered"); } finally { From eeda9cc4232e4bf3de10ea424412435950c583e0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 15:11:28 -0700 Subject: [PATCH 7/7] fix(net): a local session close declares no received end A JS session closed locally ended each received track with a clean close, which declares an end past the last group received. Since resumed readers bind any copy's declared end, a GOAWAY handover that retires the drained session refused the replacement's next group as Incompatible. A local close is a dead route, not finished content, as in Rust: end the copy cleanly without declaring an end, on lite and IETF alike. Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/subscriber.ts | 3 ++- js/net/src/integration.test.ts | 13 ++++++++----- js/net/src/internal.ts | 10 ++++++++++ js/net/src/lite/subscriber.ts | 7 ++++--- js/net/src/resume.test.ts | 29 +++++++++++++++++++++++++++++ js/net/src/track.ts | 9 ++++++++- quest/m1/public-model/deferred.md | 13 ++++++++----- 7 files changed, 69 insertions(+), 15 deletions(-) diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index ff46861e506..067750aa061 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -978,7 +978,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)}`, ); diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index 76cf549637e..a82ce591650 100644 --- a/js/net/src/integration.test.ts +++ b/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(); }); } diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 2bef0d5825a..1e97a39765f 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -403,6 +403,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; @@ -584,6 +591,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"); }, diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 5e44ff3464f..9cec1d16e14 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -1784,8 +1784,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 @@ -1794,7 +1794,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(); diff --git a/js/net/src/resume.test.ts b/js/net/src/resume.test.ts index c366ce31937..6641bb9bd11 100644 --- a/js/net/src/resume.test.ts +++ b/js/net/src/resume.test.ts @@ -1878,6 +1878,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(); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 4c6e06c9869..10eb46f64aa 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -809,6 +809,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(); @@ -1697,11 +1698,17 @@ 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; - if (abort === undefined && this.#state.end.peek() === undefined) { + if (declare && abort === undefined && this.#state.end.peek() === undefined) { this.#declareEnd({ group: this.#received, frame: 0 }); } // Nobody will finish these, so a subscriber that has not taken one yet never sees it. diff --git a/quest/m1/public-model/deferred.md b/quest/m1/public-model/deferred.md index ab1cc6bce6b..6287e4dc746 100644 --- a/quest/m1/public-model/deferred.md +++ b/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.