From eccd008c86c2e25853dbca21e9cb150d2d71cd0b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 17:20:42 -0700 Subject: [PATCH 1/5] chore(quest): claim quest/m1/broadcast-lifecycle/js-close-fin.md Quest-Claim: 65071a8e756010dc From 033f31ce6508906530c1f2bedf11829908abfbc5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 17:34:06 -0700 Subject: [PATCH 2/5] test(net): reproduce legacy FIN and session-close ordering Co-Authored-By: GPT-6 --- js/net/src/lite/connection.test.ts | 21 +++++++++++++------- quest/m1/broadcast-lifecycle/js-close-fin.md | 15 ++++++++++++++ 2 files changed, 29 insertions(+), 7 deletions(-) diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index 863fdd23c8b..a0dad6df5c3 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -23,7 +23,7 @@ import { SubscribeStart, } from "./subscribe.ts"; import { TrackInfo, Track as TrackMessage } from "./track.ts"; -import { ALPN_04, ALPN_05, ALPN_06, ALPN_07_WIP, Version } from "./version.ts"; +import { ALPN, ALPN_03, ALPN_04, ALPN_05, ALPN_06, ALPN_07_WIP, Version, versionName } from "./version.ts"; /** A transport whose `getStats` behaves as described, or is absent entirely. */ function transport(getStats?: () => Promise): WebTransport { @@ -217,9 +217,17 @@ async function settle() { for (let i = 0; i < 5; i++) await new Promise((resolve) => setTimeout(resolve, 0)); } -for (const alpn of [ALPN_05, ALPN_06, ALPN_07_WIP]) { +for (const [alpn, version] of [ + [ALPN, Version.DRAFT_01], + [ALPN, Version.DRAFT_02], + [ALPN_03, Version.DRAFT_03], + [ALPN_04, Version.DRAFT_04], + [ALPN_05, Version.DRAFT_05], + [ALPN_06, Version.DRAFT_06], + [ALPN_07_WIP, Version.DRAFT_07], +] as const) { // A publisher that finishes a track and closes must not cut the final group short. - test(`close delivers a finished track's final group over ${alpn}`, async () => { + test(`close delivers a finished track's final group over ${versionName(version)}`, async () => { const pair = createMockTransportPair(alpn); const fin = holdFins(pair.server, "uni"); const origin = new Producer(); @@ -228,10 +236,8 @@ for (const alpn of [ALPN_05, ALPN_06, ALPN_07_WIP]) { broadcastPublication.announce(); const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const url = new URL("https://localhost/test"); - const [client, server] = await Promise.all([ - connect({ url, transport: pair.client }), - accept({ url, transport: pair.server, publish: origin.consume() }), - ]); + const client = new Connection({ url, quic: pair.client, version }); + const server = new Connection({ url, quic: pair.server, version, publish: origin.consume(), client: false }); const remote = wireOf(client).consume(Path.from("room")); const track = remote.track("video").subscribe().ordered(); let closed = false; @@ -258,6 +264,7 @@ for (const alpn of [ALPN_05, ALPN_06, ALPN_07_WIP]) { fin.release(); await closing; expect(closed).toBe(true); + expect(await track.nextGroup()).toBeUndefined(); } finally { fin.release(); track.close(); diff --git a/quest/m1/broadcast-lifecycle/js-close-fin.md b/quest/m1/broadcast-lifecycle/js-close-fin.md index 19ea400ce0e..332acf32eae 100644 --- a/quest/m1/broadcast-lifecycle/js-close-fin.md +++ b/quest/m1/broadcast-lifecycle/js-close-fin.md @@ -16,3 +16,18 @@ js/net has the same ordering; if so, reproduce with a two-`Connection` test and mirror the fix. If not, delete this quest with the evidence. Public API: none. Wire: none. + +Reproduced 2026-10-27 in +[moq-dev/moq#6969](https://github.com/moq-dev/moq/pull/6969): extending the +two-Connection final-group drain test to assert a clean terminal read fails on +lite-01 through lite-04 with `SessionCode.Cancel` after the peer's clean close. +Lite-05, lite-06, and lite-07 pass the same assertion. + +The driver calls `Subscriber.close(closeError(...))` as soon as the transport +closes, before the legacy SUBSCRIBE FIN and tail completion promise chain can +finish the track. Adding an implicit `Producer.finish()` after `#settleTail` +alone does not fix the race and was discarded. The remaining work needs an +explicit, bounded drain of already-delivered group and subscribe responses +before that teardown, analogous to Rust's `poll_drain`, rather than a sleep or +extra microtask turns. The PR keeps the reproducer for that work; no +production workaround is included. From 674916eae23b9711bcf74cb7f10688a8deaeaf40 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 22:24:42 -0700 Subject: [PATCH 3/5] fix(net): finish legacy tracks whose FIN races a graceful close Before lite-05 the subscribe stream's FIN is a track's only end, but js/net released the copy without declaring one, and the driver aborted every subscription as soon as the transport closed. Mirror moq-net: a FIN finishes the track once its tail settles, and a closing session first drains the group streams and subscribe responses the closed transport already delivered, so a track its publisher ended finishes rather than aborts. Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/connection.test.ts | 105 ++++++++++++++++++- js/net/src/lite/connection.ts | 22 +++- js/net/src/lite/subscriber.test.ts | 39 ++++++- js/net/src/lite/subscriber.ts | 36 ++++++- quest/m1/README.md | 1 - quest/m1/broadcast-lifecycle/README.md | 56 ---------- quest/m1/broadcast-lifecycle/js-close-fin.md | 33 ------ quest/m2/ietf-finish.md | 3 +- 8 files changed, 193 insertions(+), 102 deletions(-) delete mode 100644 quest/m1/broadcast-lifecycle/README.md delete mode 100644 quest/m1/broadcast-lifecycle/js-close-fin.md diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index 2157017b1c7..676af6e6a05 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -23,7 +23,17 @@ import { SubscribeStart, } from "./subscribe.ts"; import { TrackInfo, Track as TrackMessage } from "./track.ts"; -import { ALPN, ALPN_03, ALPN_04, ALPN_05, ALPN_06, ALPN_07_WIP, Version, versionName } from "./version.ts"; +import { + ALPN, + ALPN_03, + ALPN_04, + ALPN_05, + ALPN_06, + ALPN_07_WIP, + hasTrackStream, + Version, + versionName, +} from "./version.ts"; /** A transport whose `getStats` behaves as described, or is absent entirely. */ function transport(getStats?: () => Promise): WebTransport { @@ -203,6 +213,47 @@ function holdFins(transport: MockTransport, streams: "uni" | "incoming-bidi") { }; } +// Hand `transport` each incoming unidirectional stream only once its FIN arrived, buffered outside +// the transport so a session close cannot discard it. QMux keeps a finished stream's unread bytes +// across a close this way, so the reads still owed land after the close itself. +function deliverWhole(transport: MockTransport) { + Object.defineProperty(transport, "incomingUnidirectionalStreams", { + value: transport.incomingUnidirectionalStreams.pipeThrough( + new TransformStream, ReadableStream>({ + transform(stream, controller) { + controller.enqueue( + new ReadableStream({ + async start(whole) { + const chunks = await Array.fromAsync(stream); + for (const chunk of chunks) whole.enqueue(chunk); + whole.close(); + }, + }), + ); + }, + }), + ), + }); +} + +// Deliver the bytes of every bidirectional stream `transport` opens at once, but each FIN only a task +// after the session closed, so every reaction to the close runs before a SUBSCRIBE FIN sent first. +function finAfterClose(transport: MockTransport) { + const create = transport.createBidirectionalStream.bind(transport); + transport.createBidirectionalStream = async (options) => { + const stream = await create(options); + const readable = stream.readable.pipeThrough( + new TransformStream({ + async flush() { + await transport.closed; + await new Promise((resolve) => setTimeout(resolve, 0)); + }, + }), + ); + return { readable, writable: stream.writable }; + }; +} + // Holds the one-second close deadline on a manual clock, so a test fires it by hand instead of waiting. function mockDeadline() { const clock = new Manual(); @@ -275,6 +326,58 @@ for (const [alpn, version] of [ origin.close(); } }); + + // The peer's close can overtake the final group or, where it is the track's only end, the + // SUBSCRIBE FIN it sent first; the closing session reads them. + for (const overtaken of hasTrackStream(version) ? ["group"] : ["group", "fin"]) { + test(`a peer close drains the overtaken ${overtaken} over ${versionName(version)}`, async () => { + const pair = createMockTransportPair(alpn); + deliverWhole(pair.client); + if (overtaken === "fin") finAfterClose(pair.client); + const origin = new Producer(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); + const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); + const url = new URL("https://localhost/test"); + const client = new Connection({ url, quic: pair.client, version }); + const server = new Connection({ + url, + quic: pair.server, + version, + publish: origin.consume(), + client: false, + }); + const remote = wireOf(client).consume(Path.from("room")); + const track = remote.track("video").subscribe().ordered(); + try { + const first = producer.appendGroup(); + first.writeFrame(textFrame("first")); + first.finish(); + expect(await (await track.nextGroup())?.readString()).toBe("first"); + + const frames = Array.from({ length: 32 }, (_, i) => `frame ${i}`); + const last = producer.appendGroup(); + for (const frame of frames) last.writeFrame(textFrame(frame)); + last.finish(); + producer.finish(); + producer.close(); + await server.close(); + + const group = await track.nextGroup(); + for (const frame of frames) expect(await group?.readString()).toBe(frame); + expect(await group?.readString()).toBeUndefined(); + expect(await track.nextGroup()).toBeUndefined(); + } finally { + track.close(); + remote.close(); + client.abort(); + server.abort(); + broadcast.close(); + origin.close(); + } + }); + } } /** A lite-07 publisher serving a finished track to a raw subscriber that has read up to the FIN. */ diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index f1842f4871a..64a3467279c 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -104,6 +104,9 @@ export class Connection implements Established { // Module for distributing tracks. #subscriber: Subscriber; + // The received unidirectional streams still being read, which a closing session drains. + #unis = new Set>(); + /** The peer's PROBE estimates; see {@link Established.probe}. */ readonly probe: Getter; @@ -265,6 +268,10 @@ export class Connection implements Established { this.#publisher.close(); this.#subscriber.close(); + this.#closeTransport(); + } + + #closeTransport(): void { try { // TODO: For whatever reason, this try/catch doesn't seem to work..? this.#quic.close(); @@ -310,9 +317,16 @@ export class Connection implements Established { probes.close(); // A session that ends before the peer's SETUP never learns its implementation. this.#peerImplementation.resolve(undefined); + const err = fatal ?? (await closeError(this.#quic)); + // A failed task can end the session over a live transport. Closing it first ends + // every read the drain waits on. + this.#closeTransport(); + // The peer's close can overtake the group streams and SUBSCRIBE FIN it sent first. + // Read what already arrived, so a track its publisher ended finishes rather than aborts. + await Promise.all([...this.#unis, this.#subscriber.drain()]); // The session died under every track it was receiving, so they end with its // error. A deliberate close() already ended them cleanly, which makes this a no-op. - this.#subscriber.close(fatal ?? (await closeError(this.#quic))); + this.#subscriber.close(err); this.abort(); } } @@ -442,14 +456,16 @@ export class Connection implements Established { const stream = await readers.next(); if (!stream) break; - this.#runUni(stream) + const task = this.#runUni(stream) .then(() => { stream.stop(StreamCode.Cancel); }) .catch((err: unknown) => { stream.stop(err); if (err instanceof ProtocolViolation) closeViolation(this.#quic, err); - }); + }) + .finally(() => this.#unis.delete(task)); + this.#unis.add(task); } } diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index e61a68f46f3..638ccaa7bce 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -30,12 +30,13 @@ import { encodeSubscribeResponse, SubscribeDrop, SubscribeEnd, + SubscribeOk, type SubscribeResponse, SubscribeStart, } from "./subscribe.ts"; import { SUBSCRIBE_SETUP_TIMEOUT_MS, Subscriber } from "./subscriber.ts"; import { Track, TrackInfo } from "./track.ts"; -import { Version } from "./version.ts"; +import { Version, versionName } from "./version.ts"; test("closing the subscriber suppresses probe stream warnings", async () => { let readable!: ReadableStreamDefaultController; @@ -2255,6 +2256,42 @@ test("a lite-07 publisher FIN ends a covered bounded subscription", async () => } }); +// Before lite-05 the publisher FINs the subscribe stream only once the track is over, so that FIN +// finishes the track once the tail settles, rather than releasing it for another route. +test.each( + [Version.DRAFT_01, Version.DRAFT_02, Version.DRAFT_03, Version.DRAFT_04].map((v) => [versionName(v), v] as const), +)("a %s publisher FIN finishes the track", async (_, version) => { + const clock = new Manual(); + const installed = clock.install(); + const { quic, streams, opened, closes } = fakeSession(); + const subscriber = new Subscriber(quic, version, HopSchema.parse(1n)); + const broadcast = subscriber.consume(Path.from("room")); + const track = broadcast.track("video").subscribe(); + try { + const stream = await opened(0); + await stream.reading; + const chunks: Uint8Array[] = []; + const writer = new Writer( + new WritableStream({ write: (chunk) => void chunks.push(new Uint8Array(chunk)) }), + version, + ); + await encodeSubscribeResponse(writer, { ok: new SubscribeOk({}) }, version); + for (const chunk of chunks) streams[0].inbound.enqueue(chunk); + streams[0].inbound.close(); + await drainReactions(); + + await clock.advance(Nano.fromMilli(GRACE)); + expect(await track.finished()).toEqual({ group: 0, frame: 0 }); + expect(await track.closed).toBeNull(); + expect(closes).toEqual([]); + } finally { + track.close(); + broadcast.close(); + subscriber.close(); + installed.close(); + } +}); + // The FIN may answer the bounded range before the publisher read an update that widened it, // so the track fails rather than ending clean short of the demand. test.each([ diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 7b3debc7861..eef3fbc4ba1 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -138,6 +138,11 @@ interface SubscribeEntry { // (SUBSCRIBE_END), once it declares them. start?: number; end?: number; + // Set by the subscribe stream's FIN before lite-05, the track's only end there. + finished?: boolean; + // Settles once the subscribe stream's responses are read to its FIN or reset, which + // Subscriber.drain waits on. + responses?: Promise; // Whether the stream ever asked for a bounded range, which a lite-07 publisher FINs without // SUBSCRIBE_END once it covers it. bounded: boolean; @@ -965,7 +970,12 @@ export class Subscriber { // its FIN; older drafts just close it. Either way group streams can still be in // flight, so the track ends only once the tail is accounted for. A reset rejects // instead, so the track ends with that error rather than a clean tail. - const responses = hasTrackStream(this.version) ? this.runResponses(stream, entry) : stream.reader.closed; + const responses = hasTrackStream(this.version) + ? this.runResponses(stream, entry) + : stream.reader.closed.then(() => { + entry.finished = true; + }); + entry.responses = responses.catch(() => {}); let tailSettled = false; // Set when the publisher ended the subscription short of the demand. let short: Error | undefined; @@ -1008,10 +1018,14 @@ export class Subscriber { // A publisher that covered the range before reading an update widening it sent // fewer groups than the track has, so it fails rather than ending clean. Otherwise - // only a received SUBSCRIBE_END declares the track's end: a range covered or a - // subscription nobody wants anymore is a released copy, not finished content. + // only a received SUBSCRIBE_END, or before lite-05 a FIN, declares the track's end: + // a range covered or a subscription nobody wants anymore is a released copy, not + // finished content. if (short) producer.abort(short); - else hooks.releaseTrack(producer); + else { + if (tailSettled && entry.finished) producer.finish(); + hooks.releaseTrack(producer); + } // A settled lite07 tail acknowledges completion with FIN, without cancelling // the publisher's already-finished receive half. if (tailSettled && waitsForSubscriberFin(this.version)) stream.writer.close(); @@ -1843,9 +1857,20 @@ export class Subscriber { } } + /** + * Wait until every subscribe stream's responses are read to its FIN or reset. Bounded only + * once the transport is closed, so that its reads end. + * + * @internal + */ + async drain(): Promise { + await Promise.all(Array.from(this.#subscribes.values(), (entry) => entry.responses)); + } + /** * 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. + * A track whose publisher already ended it finishes either way. */ close(err?: Error) { // A fetch or setup exchange cut off by the session is incomplete even on a deliberate @@ -1853,7 +1878,8 @@ export class Subscriber { const cut = err ?? new StreamError(StreamCode.SessionClosed, { message: "session closed" }); this.#closed.abort(cut); - for (const { track } of this.#subscribes.values()) { + for (const { track, finished } of this.#subscribes.values()) { + if (finished) track.finish(); if (err) track.abort(err); else hooks.releaseTrack(track); } diff --git a/quest/m1/README.md b/quest/m1/README.md index 9c356f35460..2f6e2d73e2a 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -130,7 +130,6 @@ surface and zero-copy texture import, land before the cut. - [Finalize moq-lite-07](/quest/m1/lite07-finalize.md) - when the maintainer cuts it, lite-07 negotiates as `moq-lite-07` and the next release ships it - [Moxygen minimal interop](/quest/m1/smoke-moxygen.md) - waiting on Moxygen to send a registered filter type, then the smoke lane stops hiding its failure - [Moxygen pin bump](/quest/m1/moxygen-pin-bump.md) - waiting on moxygen#264, then the TRACK_STATUS patch is dropped -- [Broadcast lifecycle](/quest/m1/broadcast-lifecycle/README.md) - broadcasts are online or offline with no finished state in moq-net, finished-broadcast retention is a fixed moq-mux policy, and each linger timer has one meaning - [A smaller, safer API than release](/quest/m1/api-audit/README.md) - every surface added since release has a consumer or is gone, overlapping operations are collapsed, names mirror across Rust, JS, and the bindings, and every break says how to migrate - [moq-transport 18 and 22 only](/quest/m1/ietf-drop/README.md) - drafts 14-17 and 19-21 are deleted from Rust, JS, tests, interop lanes, and docs, removing the legacy control-stream adapter and its request limit - [moq-quic's API improves on quinn's](/quest/m1/quic-api/README.md) - moq-quic and both QUIC drivers ship one mirrored API: `send::`/`recv::` namespaces, one error per direction, consuming terminal operations, plain configs, no quinn baggage diff --git a/quest/m1/broadcast-lifecycle/README.md b/quest/m1/broadcast-lifecycle/README.md deleted file mode 100644 index a2451c570cc..00000000000 --- a/quest/m1/broadcast-lifecycle/README.md +++ /dev/null @@ -1,56 +0,0 @@ -# Broadcasts are online or offline, and retention is app policy - -## Goal - -A moq-net broadcast has no lifecycle state of its own: it is online while -announced and offline once closed. The app marks the end of its content with -a track FIN (hang: finish the catalog). Keeping a finished broadcast up for -late readers is a fixed-duration app policy in moq-mux, not a deadline moq-net -derives from track max-age. Every other timer called "linger" has one meaning -per name. - -## Plan - -Decided with the maintainer on 2026-10-23, from an audit of every linger -timer on main at 1582fdc07d: - -- `broadcast::Producer::finish` exists only in moq-net. Nothing reaches the - wire: ANNOUNCE active/ended already says online/offline, and a track FIN - already says a track's content ended. A BROADCAST wire message was - considered and rejected; it adds no behavior over those two and would need a - moq-transport mapping. -- The finished linger was inconsistent and partly accidental: - - Its duration is the max of track max-ages, which describe how long a - superseded group stays fresh and never the newest one. The JS publish - catalog declares `Number.MAX_SAFE_INTEGER` max-age, so every browser - publisher lingered the full 30s; Rust catalogs declare none. - - It counted from different events: Rust from the first poll of `wait`, - JS from the `wait` call, UniFFI from `finish`, and C from the caller's - own clock. - - Only Rust ended it when a newer epoch appeared at the path. -- Closing a broadcast already leaves tracks handed out playing to their end, - and session drain delivers their tail to attached subscribers. Retention - only serves readers that arrive after the end. -- A retaining publisher ignores a newer instance at its path: if the newer - one leaves, the old one is announced again, and readers see its finished - catalog and stop. Simpler than watching for supersession, and acceptable - under online/offline. -- The other timers stay. Each is a cache or debounce with its own name and - owner: the Rust idle track copy (30s), discovery (2s), the JS connection - pool (2s), and the relay's `[stats] linger` (5m). They are not cross-language - lifecycle guarantees. -- After this epic, user-facing "linger" means only publisher retention - (`moq` publishing subcommands, moqsink) and the relay's stats setting. The - export-side wait becomes `--follow`. -- Docs change inline in each quest; no separate lifecycle page. - -## Required - -- [js/net close FIN](/quest/m1/broadcast-lifecycle/js-close-fin.md) - js/net finishes tracks whose FIN arrives with a graceful close - -## Related - -- [SRT on the shared socket](/quest/m2/one-port/srt-demux.md) - an SRT ingest whose peer goes silent aborts instead of finishing as if it had shut down - -- [moq-transport finish interop](/quest/m2/ietf-finish.md) - the interop `finish` case, whose late readers rely on retention -- [CLI and gst flags](https://github.com/moq-dev/moq/pull/6522) - removed the opt-in `linger` booleans, leaving completion to this epic diff --git a/quest/m1/broadcast-lifecycle/js-close-fin.md b/quest/m1/broadcast-lifecycle/js-close-fin.md deleted file mode 100644 index 332acf32eae..00000000000 --- a/quest/m1/broadcast-lifecycle/js-close-fin.md +++ /dev/null @@ -1,33 +0,0 @@ -# [S] js/net finishes tracks whose FIN arrives with the close - -## Goal - -When a peer closes a session gracefully right after finishing its tracks, a -js/net relay or subscriber finishes those tracks instead of aborting them, on -every lite version, as Rust does since -[moq-dev/moq#6741](https://github.com/moq-dev/moq/pull/6741). - -## Plan - -In Rust the close could reach the subscriber before it read an already -buffered SUBSCRIBE FIN, so the track ended `Unroutable`. The fix polls the -sources once after the closed transport's group streams drain. Check whether -js/net has the same ordering; if so, reproduce with a two-`Connection` test and -mirror the fix. If not, delete this quest with the evidence. - -Public API: none. Wire: none. - -Reproduced 2026-10-27 in -[moq-dev/moq#6969](https://github.com/moq-dev/moq/pull/6969): extending the -two-Connection final-group drain test to assert a clean terminal read fails on -lite-01 through lite-04 with `SessionCode.Cancel` after the peer's clean close. -Lite-05, lite-06, and lite-07 pass the same assertion. - -The driver calls `Subscriber.close(closeError(...))` as soon as the transport -closes, before the legacy SUBSCRIBE FIN and tail completion promise chain can -finish the track. Adding an implicit `Producer.finish()` after `#settleTail` -alone does not fix the race and was discarded. The remaining work needs an -explicit, bounded drain of already-delivered group and subscribe responses -before that teardown, analogous to Rust's `poll_drain`, rather than a sleep or -extra microtask turns. The PR keeps the reproducer for that work; no -production workaround is included. diff --git a/quest/m2/ietf-finish.md b/quest/m2/ietf-finish.md index bdd9e7f072c..9c214702090 100644 --- a/quest/m2/ietf-finish.md +++ b/quest/m2/ietf-finish.md @@ -4,8 +4,7 @@ The interop `finish` case passes on every even moq-transport draft, not only moq-lite: each reader that joins a finished broadcast while its publisher retains it -([broadcast lifecycle](/quest/m1/broadcast-lifecycle/README.md)) gets -the whole ten-frame tail, natural media and catalog EOF, the exact broadcast's +(moq-mux's fixed linger) gets the whole ten-frame tail, natural media and catalog EOF, the exact broadcast's withdrawal, and the publisher's later health broadcast through the same session. Until then `interop.sh` records those cells as skipped; delete that skip in `run_finishes` and its README note. From 4022ae7399fe228a16ef020f8fd6fad9cc6e511a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 08:08:23 -0700 Subject: [PATCH 4/5] test(net): finish the first group in the final-group close test Since #6978, closing a track truncates its unfinished groups, so the test's never-finished first group made lite-01..04 abort at session close. The test checks delivery of the final group, so finish the first one explicitly. Also repoint the cut-audit quest's link to the deleted js-close-fin quest at this PR. Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/connection.test.ts | 4 +++- quest/m1/cut-audit/js-close-cause.md | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index 676af6e6a05..29f5b21b3d0 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -296,7 +296,9 @@ for (const [alpn, version] of [ closed = true; }); try { - producer.appendGroup().writeFrame(textFrame("first")); + const first = producer.appendGroup(); + first.writeFrame(textFrame("first")); + first.finish(); expect(await (await track.nextGroup())?.readString()).toBe("first"); fin.enable(); diff --git a/quest/m1/cut-audit/js-close-cause.md b/quest/m1/cut-audit/js-close-cause.md index 933df20c237..47311d8951c 100644 --- a/quest/m1/cut-audit/js-close-cause.md +++ b/quest/m1/cut-audit/js-close-cause.md @@ -31,7 +31,7 @@ Choose whether to retain that clean local route-close behavior or abort the received track with CANCELLED as well. Recommend retaining the mirrored Rust behavior, with cancellation at the unfinished-group boundary. No broader session redesign is included; the independent FIN-and-peer-close ordering -work remains in [JS close FIN](/quest/m1/broadcast-lifecycle/js-close-fin.md). +work lands in [moq-dev/moq#6969](https://github.com/moq-dev/moq/pull/6969). Public API: breaking in the prepared PR. Callers pass an error to `Request.reject`, and finish groups before closing when they need a complete From a8299c023c5ee05868f4527fffb31da9042496a9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 08:49:36 -0700 Subject: [PATCH 5/5] fix(net): read legacy subscribe streams to their FIN Before lite-05 the subscriber only awaited the subscribe stream's reader.closed after SUBSCRIBE_OK. A QMux receive stream closes only on a read, so that promise never settled there: the track never finished and the closing session's drain waited forever. Run runResponses on every version instead, which reads to the FIN, applies a lite-03/04 SUBSCRIBE_DROP like Rust does, and rejects a second SUBSCRIBE_OK. Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/connection.test.ts | 66 ++++++++++++++++++++++++++++++ js/net/src/lite/subscriber.ts | 31 +++++++------- 2 files changed, 83 insertions(+), 14 deletions(-) diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index 29f5b21b3d0..211bde08a52 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -1,5 +1,6 @@ import { expect, spyOn, test } from "bun:test"; import { Manual, Milli, Nano } from "@moq/clock"; +import Qmux from "@moq/qmux"; import { accept } from "../connection/accept.ts"; import { connect } from "../connection/connect.ts"; import { SessionCode, SessionError } from "../error.ts"; @@ -254,6 +255,29 @@ function finAfterClose(transport: MockTransport) { }; } +// A qmux session pair over an in-memory socket. Unlike the mock, its receive streams are +// pull-driven, so a stream's FIN surfaces only to a read. +function qmuxPair(): { client: WebTransport; server: WebTransport } { + const socket = (readable: ReadableStream, writable: WritableStream) => { + const closed = Promise.withResolvers<{ closeCode: number; reason: string }>(); + return { + url: "", + opened: Promise.resolve({ readable, writable, extensions: "", protocol: "qmux-01" }), + closed: closed.promise, + close: () => closed.resolve({ closeCode: 1000, reason: "" }), + }; + }; + // Copied, since a sender may reuse its buffer once a write resolves. + const pipe = () => + new TransformStream({ transform: (chunk, c) => c.enqueue(chunk.slice()) }); + const up = pipe(); + const down = pipe(); + const config = { idleTimeout: null }; + const client = new Qmux(socket(down.readable, up.writable), { config }); + const server = Qmux.accept(socket(up.readable, down.writable), { config }); + return { client, server }; +} + // Holds the one-second close deadline on a manual clock, so a test fires it by hand instead of waiting. function mockDeadline() { const clock = new Manual(); @@ -380,6 +404,48 @@ for (const [alpn, version] of [ } }); } + + // The subscriber reads a legacy subscribe stream to its FIN, since QMux closes a stream only + // on a read; awaiting the FIN instead would hold the closing session's drain forever. + if (!hasTrackStream(version)) { + test(`a peer close finishes a FIN'd track over qmux ${versionName(version)}`, async () => { + const pair = qmuxPair(); + const origin = new Producer(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); + const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); + const url = new URL("https://localhost/test"); + const client = new Connection({ url, quic: pair.client, version }); + const server = new Connection({ + url, + quic: pair.server, + version, + publish: origin.consume(), + client: false, + }); + const remote = wireOf(client).consume(Path.from("room")); + const track = remote.track("video").subscribe().ordered(); + try { + const group = producer.appendGroup(); + group.writeFrame(textFrame("only")); + group.finish(); + expect(await (await track.nextGroup())?.readString()).toBe("only"); + + producer.finish(); + producer.close(); + await server.close(); + expect(await track.nextGroup()).toBeUndefined(); + } finally { + track.close(); + remote.close(); + client.abort(); + server.abort(); + broadcast.close(); + origin.close(); + } + }); + } } /** A lite-07 publisher serving a finished track to a raw subscriber that has read up to the FIN. */ diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index eef3fbc4ba1..db145b4a6c0 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -966,15 +966,10 @@ export class Subscriber { // don't carry SUBSCRIBE_UPDATE on the wire, so skip the watcher there // and just wait on the stream/track like before. // - // On lite-05+ the publisher sends SUBSCRIBE_START/END/DROP on this stream until - // its FIN; older drafts just close it. Either way group streams can still be in - // flight, so the track ends only once the tail is accounted for. A reset rejects - // instead, so the track ends with that error rather than a clean tail. - const responses = hasTrackStream(this.version) - ? this.runResponses(stream, entry) - : stream.reader.closed.then(() => { - entry.finished = true; - }); + // The publisher sends responses on this stream until its FIN. Group streams can + // still be in flight, so the track ends only once the tail is accounted for. A reset + // rejects instead, so the track ends with that error rather than a clean tail. + const responses = this.runResponses(stream, entry); entry.responses = responses.catch(() => {}); let tailSettled = false; // Set when the publisher ended the subscription short of the demand. @@ -1449,11 +1444,13 @@ export class Subscriber { } } - // Reads SUBSCRIBE_START/END/DROP on the subscribe stream until FIN (lite-05+), recording - // the range the tail is accounted against. SUBSCRIBE_END declares the track's end right - // away, so a consumer learns it before the last groups arrive. The publisher must declare - // the end before FIN unless we cancelled locally, or on lite-07 it covered a bounded range; - // resets and malformed responses reject. + // Reads the subscribe stream's responses until FIN, recording the range the tail is + // accounted against: SUBSCRIBE_START/END/DROP on lite-05+, and any SUBSCRIBE_DROP after the + // OK before then, where the FIN itself ends the track. Reading to the FIN, rather than + // awaiting it, matters: a pull-driven transport stream closes only on a read. + // SUBSCRIBE_END declares the track's end right away, so a consumer learns it before the + // last groups arrive. The publisher must declare the end before FIN unless we cancelled + // locally, or on lite-07 it covered a bounded range; resets and malformed responses reject. private async runResponses(stream: Stream, entry: SubscribeEntry): Promise { for (;;) { const resp = await decodeSubscribeResponseMaybe(stream.reader, this.version); @@ -1463,6 +1460,11 @@ export class Subscriber { entry.answered?.(); entry.answered = undefined; if (!resp) { + // Before lite-05 the FIN is the track's only end. + if (!hasTrackStream(this.version)) { + entry.finished = true; + return; + } // A bounded range may have widened since, in an update the publisher never read. const covered = hasRanges(this.version) && entry.bounded; if (entry.end === undefined && !covered && hooks.trackReleased(entry.track).peek() === undefined) @@ -1470,6 +1472,7 @@ export class Subscriber { return; } + if ("ok" in resp) throw new ProtocolViolation("duplicate SUBSCRIBE_OK"); if ("start" in resp) { entry.start = resp.start.group; hooks.startTrack(entry.track, { start: entry.start, requested: entry.requested });