From 19706e4a310f0b952641a19191129a36cbbe0d60 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 4 Oct 2026 20:43:29 -0700 Subject: [PATCH 1/4] quest: claim js-close-drain Co-Authored-By: Claude Opus 5.5 From 120ae7eebb4d59324e8396b9b426be95f1679f2f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 4 Oct 2026 20:56:32 -0700 Subject: [PATCH 2/4] fix(net): JS lite close drains served subscriptions Connection.close() now waits, within its one-second deadline, for every served SUBSCRIBE, FETCH, and TRACK request to end, mirroring Rust's owed count. Group, FETCH, TRACK, and subscribe streams wait for their FIN to be acknowledged, and on lite-07 a served SUBSCRIBE also waits for the subscriber's FIN. Co-Authored-By: Claude Opus 5.5 --- doc/lib/js/net.md | 4 + js/net/src/lite/connection.test.ts | 217 ++++++++++++++++++++++++++++- js/net/src/lite/connection.ts | 36 +++-- js/net/src/lite/publisher.ts | 78 ++++++++--- js/net/src/lite/subscriber.ts | 3 +- js/net/src/lite/version.ts | 20 +++ quest/m1/README.md | 1 - quest/m1/js-close-drain.md | 36 ----- 8 files changed, 331 insertions(+), 64 deletions(-) delete mode 100644 quest/m1/js-close-drain.md diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 8fc4143fc9..add335ed4a 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -117,6 +117,10 @@ are [hidden](/concept/moq-lite#hidden-broadcasts) unless `announced(scope, { hid An established connection's `await connection.close()` withdraws its announcements and waits up to one second for delivery before disconnecting. +On moq-lite it also waits for the requests it serves: a subscription to a +finished track delivers its remaining groups, and on moq-lite-07 waits for the +subscriber's FIN, which says it read the tail. A live track never ends on its +own, so its subscription holds the close until the deadline. It rejects if delivery fails or times out, and closes the transport either way. Use `connection.abort()` for immediate shutdown. Reconnecting connection handles keep their synchronous `close()` disposal behavior. IETF drafts 14 diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index 4cc3748703..071b377c1a 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -1,12 +1,17 @@ -import { expect, test } from "bun:test"; +import { expect, spyOn, test } from "bun:test"; +import { accept } from "../connection/accept.ts"; +import { connect } from "../connection/connect.ts"; import { SessionCode } from "../error.ts"; -import { createMockTransportPair } from "../mock.ts"; +import { createMockTransportPair, type MockTransport } from "../mock.ts"; +import { Producer } from "../origin.ts"; +import * as Path from "../path.ts"; import { Stream, Writer } from "../stream.ts"; import { wireOf } from "../wire.ts"; import { Connection, probeLevel } from "./connection.ts"; import { Goaway } from "./goaway.ts"; import { ProbeLevel, Setup } from "./setup.ts"; import { DataType, StreamId } from "./stream.ts"; +import { decodeSubscribeResponse, Subscribe } from "./subscribe.ts"; import { ALPN_04, ALPN_05, ALPN_06, ALPN_07_WIP, Version } from "./version.ts"; /** A transport whose `getStats` behaves as described, or is absent entirely. */ @@ -128,3 +133,211 @@ for (const [alpn, version] of [ } }); } + +// Hold the FIN acknowledgement of every unidirectional stream `transport` opens once enabled, +// after the bytes and FIN reached the peer: a group stream whose tail is still in flight. +function holdUniFins(transport: MockTransport) { + const reached = Promise.withResolvers(); + const release = Promise.withResolvers(); + let enabled = false; + const create = transport.createUnidirectionalStream.bind(transport); + transport.createUnidirectionalStream = async (options) => { + const writer = (await create(options)).getWriter(); + return new WritableStream({ + write: (bytes) => writer.write(bytes), + abort: (reason) => writer.abort(reason), + async close() { + await writer.close(); + if (!enabled) return; + reached.resolve(); + await release.promise; + }, + }); + }; + return { + reached: reached.promise, + enable: () => { + enabled = true; + }, + release: release.resolve, + }; +} + +// Captures the one-second close deadline so a test fires it by hand instead of waiting. +function mockDeadline() { + let expire: (() => void) | undefined; + const original = globalThis.setTimeout; + const timer = spyOn(globalThis, "setTimeout").mockImplementation(((fn: () => void, ms?: number) => { + if (ms !== 1000) return original(fn, ms); + expire = fn; + return 0 as unknown as ReturnType; + }) as typeof setTimeout); + return { + expire: () => { + if (!expire) throw new Error("close armed no deadline"); + expire(); + }, + restore: () => timer.mockRestore(), + }; +} + +/** Runs every task queued so far, so a test can assert that something is still pending. */ +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]) { + // 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 () => { + const pair = createMockTransportPair(alpn); + const fin = holdUniFins(pair.server); + const origin = new Producer(); + const broadcast = origin.createBroadcast(Path.from("room")); + broadcast.announce(); + const producer = broadcast.createTrack("video"); + 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 remote = wireOf(client).consume(Path.from("room")); + const track = remote.track("video").subscribe().ordered(); + let closed = false; + void pair.server.closed.then(() => { + closed = true; + }); + try { + producer.appendGroup().writeString("first"); + expect(await (await track.nextGroup())?.readString()).toBe("first"); + + fin.enable(); + const last = producer.appendGroup(); + last.writeString("last"); + last.close(); + producer.close(); + const closing = server.close(); + + expect(await (await track.nextGroup())?.readString()).toBe("last"); + await fin.reached; + await settle(); + expect(closed).toBe(false); + + fin.release(); + await closing; + expect(closed).toBe(true); + } finally { + fin.release(); + 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. */ +async function servedRawSubscription() { + const version = Version.DRAFT_07; + const pair = createMockTransportPair(ALPN_07_WIP); + const origin = new Producer(); + const broadcast = origin.createBroadcast(Path.from("room")); + broadcast.announce(); + const producer = broadcast.createTrack("video"); + const server = new Connection({ + url: new URL("https://relay.example/"), + quic: pair.server, + version, + publish: origin.consume(), + }); + + const group = producer.appendGroup(); + group.writeString("last"); + group.close(); + + const subscriber = await Stream.open(pair.client, { version }); + await subscriber.writer.u53(StreamId.Subscribe); + await new Subscribe({ + id: 0n, + broadcast: Path.from("room"), + track: "video", + priority: 0, + maxAge: 10_000, + startGroup: 0, + }).encode(subscriber.writer, version); + + // Finish the track only once it is served: a track closed before anyone subscribed is gone. + expect("start" in (await decodeSubscribeResponse(subscriber.reader, version))).toBe(true); + producer.close(); + // SUBSCRIBE_END, then the publisher's FIN. + await subscriber.reader.readAll(); + + let closed = false; + void pair.server.closed.then(() => { + closed = true; + }); + return { + server, + subscriber, + closed: () => closed, + cleanup: () => { + server.abort(); + broadcast.close(); + origin.close(); + }, + }; +} + +// The publisher's FIN says nothing about whether the subscriber read the tail; on lite-07 the +// subscriber's FIN does, so close waits for it. +test("a lite-07 subscriber withholding FIN holds close until the deadline", async () => { + const deadline = mockDeadline(); + const served = await servedRawSubscription(); + try { + const closing = served.server.close(); + await settle(); + expect(served.closed()).toBe(false); + + deadline.expire(); + await expect(closing).rejects.toThrow("session close timed out"); + expect(served.closed()).toBe(true); + } finally { + deadline.restore(); + served.cleanup(); + } +}); + +test("a lite-07 subscriber FIN releases close", async () => { + const deadline = mockDeadline(); + const served = await servedRawSubscription(); + try { + const closing = served.server.close(); + await settle(); + expect(served.closed()).toBe(false); + + served.subscriber.writer.close(); + await closing; + expect(served.closed()).toBe(true); + } finally { + deadline.restore(); + served.cleanup(); + } +}); + +test("abort during the close drain ends the session at once", async () => { + const deadline = mockDeadline(); + const served = await servedRawSubscription(); + try { + const closing = served.server.close(); + await settle(); + expect(served.closed()).toBe(false); + + served.server.abort(); + await closing; + expect(served.closed()).toBe(true); + } finally { + deadline.restore(); + served.cleanup(); + } +}); diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index 6e3faf45b2..d7e0cbe538 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -152,9 +152,16 @@ export class Connection implements Established { void this.#run(); } - /** Withdraw announcements and wait up to one second for delivery before closing. */ + /** + * Withdraw announcements and finish the requests this session serves, waiting up to one + * second for delivery before closing. + * + * A served subscription first delivers its remaining group streams, and on lite-07 also + * waits for the subscriber's FIN, which says it read the tail. A live track never ends on + * its own, so it holds the close until the deadline. + */ close(): Promise { - this.#closing ??= withTimeout(this.#publisher.withdraw(), 1000, "session close timed out").finally(() => + this.#closing ??= withTimeout(this.#publisher.drain(), 1000, "session close timed out").finally(() => this.abort(), ); return this.#closing; @@ -271,14 +278,27 @@ export class Connection implements Established { const msg = await AnnounceRequest.decode(stream.reader, this.#version); await this.#publisher.runAnnounce(msg, stream); } else if (typ === StreamId.Subscribe) { - const msg = await Subscribe.decode(stream.reader, this.#version); - await this.#publisher.runSubscribe(msg, stream); + // Owed from dispatch, so a close that lands mid-decode still waits for the request. + await this.#publisher.owe( + (async () => { + const msg = await Subscribe.decode(stream.reader, this.#version); + await this.#publisher.runSubscribe(msg, stream); + })(), + ); } else if (typ === StreamId.Fetch) { - const msg = await Fetch.decode(stream.reader, this.#version); - await this.#publisher.runFetch(msg, stream); + await this.#publisher.owe( + (async () => { + const msg = await Fetch.decode(stream.reader, this.#version); + await this.#publisher.runFetch(msg, stream); + })(), + ); } else if (typ === StreamId.Track) { - const msg = await TrackMessage.decode(stream.reader, this.#version); - await this.#publisher.runTrackInfo(msg, stream); + await this.#publisher.owe( + (async () => { + const msg = await TrackMessage.decode(stream.reader, this.#version); + await this.#publisher.runTrackInfo(msg, stream); + })(), + ); } else if (typ === StreamId.Probe) { await this.#publisher.runProbe(stream); } else if (typ === StreamId.Goaway) { diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 8034151214..203043b35d 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -38,6 +38,7 @@ import { hasStreamCount, resolvesStart, Version, + waitsForSubscriberFin, } from "./version.ts"; const PROBE_INTERVAL = 100; // ms @@ -50,6 +51,14 @@ function zigzag(delta: bigint): bigint { return delta >= 0n ? delta << 1n : (-delta << 1n) - 1n; } +/** + * Settles once the peer acknowledged everything written before the FIN, or the stream failed. + * Closing the session sooner would discard bytes still in flight. + */ +function acknowledged(writer: Writer): Promise { + return writer.closed.catch(() => {}); +} + /** What {@link Publisher.openGroup} and {@link Publisher.serveGroup} need to serve one group. */ interface RunGroup { /** The subscription ID. */ @@ -245,7 +254,9 @@ class SubscriptionControls { async #decode(reader: Reader, version: Version, apply: (update: SubscribeUpdate) => void) { try { - while (this.#end === undefined) { + // Runs until the peer's half ends, even past our own FIN: a lite-07 publisher waits + // for that end, and every other teardown stops the reader. + for (;;) { const update = await SubscribeUpdate.decodeMaybe(reader, version); if (!update) break; apply(update); @@ -326,6 +337,11 @@ function positionCursor(track: track.Subscriber, version: Version, startGroup: n */ export class Publisher { #withdrawal = new Withdrawal(); + + // Served SUBSCRIBE, FETCH, and TRACK requests, from dispatch until they end, which a + // draining close waits for. + #owed = new Set>(); + // The version of the connection. readonly version: Version; @@ -639,7 +655,7 @@ export class Publisher { }); }, }); - await this.#runTrack(track, stream.writer, controls, { + const finished = await this.#runTrack(track, stream.writer, controls, { sub: msg.id, broadcast: msg.broadcast, timescale, @@ -652,10 +668,14 @@ export class Publisher { }); console.debug(`publish done: broadcast=${msg.broadcast} track=${track.name}`); - stream.close(); + // A lite-07 subscriber FINs once it has read the tail, so leave its half open and + // wait for that. Older versions stop it here, which ends the decoder. + if (waitsForSubscriberFin(this.version)) stream.writer.close(); + else stream.close(); + // Ends the datagram loop. track.close(); - // Closing the stream ends the decoder and track.close ends the datagram loop. - await Promise.all([datagrams, controls.decoding]); + // A subscriber that left is owed nothing more, and its stream may already be reset. + await Promise.all([datagrams, controls.decoding, finished && acknowledged(stream.writer)]); } catch (err: unknown) { const e = error(err); console.warn(`publish error: broadcast=${msg.broadcast} track=${track.name} error=${reason(e)}`); @@ -709,6 +729,7 @@ export class Publisher { console.debug(`fetch done: broadcast=${msg.broadcast} track=${msg.track} group=${msg.group}`); stream.close(); group.close(); + await acknowledged(stream.writer); } catch (err: unknown) { const e = error(err); console.warn( @@ -725,6 +746,8 @@ export class Publisher { * @param broadcast - The broadcast name * @param track - The track to run * @param stream - The stream to write to + * @returns Whether the track finished with every group stream done, rather than the + * subscriber leaving * * @internal */ @@ -733,7 +756,7 @@ export class Publisher { stream: Writer, controls: SubscriptionControls, serving: { sub: bigint; broadcast: Path.Valid; timescale: Timescale; bounds: FrameBounds }, - ) { + ): Promise { const { sub, broadcast, timescale, bounds } = serving; // Lite-05+ resolves the range on the subscribe stream: SUBSCRIBE_START once the // first group is known, SUBSCRIBE_END when the track finishes. @@ -797,7 +820,7 @@ export class Publisher { case "done": // The subscriber left. Its queued groups are pointless now, which // the finally below acts on since `finished` stays false. - return; + return false; case "error": throw control.error; case "update": { @@ -846,7 +869,7 @@ export class Publisher { if ( !(await controls.response(encodeSubscribeResponse(stream, { start }, this.version))) ) - return; + return false; continue; } } @@ -854,7 +877,7 @@ export class Publisher { // soon as it is known, while the remaining groups are still being // produced. The lite-07 count is not final until those groups open. if (!endSent && !countStreams && track.final() !== undefined) { - if (!(await sendEnd())) return; + if (!(await sendEnd())) return false; continue; } await waitForSubscription(controls, track); @@ -863,14 +886,14 @@ export class Publisher { // The producer finished but is still holding groups above the cap. // Declare the boundary, then wait for an update to release them. if (!endSent && !countStreams) { - if (!(await sendEnd())) return; + if (!(await sendEnd())) return false; continue; } await waitForSubscription(controls, track); continue; case "done": { if (!endSent) { - if (!(await sendEnd())) return; + if (!(await sendEnd())) return false; continue; } // The FIN tells the subscriber every group is accounted for, so it waits @@ -879,9 +902,9 @@ export class Publisher { const drained = Symbol("drained"); const end = await Promise.race([Promise.all(groups).then(() => drained), controls.ended]); if (end instanceof Error) throw end; - if (end !== drained) return; + if (end !== drained) return false; finished = true; - return; + return true; } } @@ -906,7 +929,7 @@ export class Publisher { ), )) ) - return; + return false; } const options: RunGroup = { @@ -954,6 +977,7 @@ export class Publisher { await info.encode(stream.writer, this.version); console.debug(`track info: broadcast=${msg.broadcast} track=${msg.track}`); stream.close(); + await acknowledged(stream.writer); } catch (err) { console.debug(`track unknown: broadcast=${msg.broadcast} track=${msg.track}`); stream.writer.reset(error(err)); @@ -1155,6 +1179,7 @@ export class Publisher { stream.close(); group.close(); + await acknowledged(stream); } catch (err: unknown) { const e = error(err); stream.reset(e); @@ -1253,8 +1278,29 @@ export class Publisher { } } - withdraw(): Promise { - return this.#withdrawal.close(); + /** + * Owes the peer `task`, a served SUBSCRIBE, FETCH, or TRACK request, until it ends. + * + * @internal + */ + owe(task: Promise): Promise { + const tracked = task.finally(() => this.#owed.delete(tracked)); + this.#owed.add(tracked); + return tracked; + } + + /** + * Withdraws announcements and waits for every owed request to end, including any that + * arrive meanwhile. Rejects if a withdrawal failed; an owed request ends either way. + * + * @internal + */ + async drain(): Promise { + const settle = async () => { + while (this.#owed.size > 0) await Promise.allSettled(this.#owed); + }; + const [withdrawn] = await Promise.allSettled([this.#withdrawal.close(), settle()]); + if (withdrawn.status === "rejected") throw withdrawn.reason; } close() { diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 0f9d24cb5f..f7417f3ca0 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -49,6 +49,7 @@ import { hasStreamCount, restartSupported, Version, + waitsForSubscriberFin, } from "./version.ts"; // Bound on how long stream-open plus the first response (SUBSCRIBE_OK on older @@ -625,7 +626,7 @@ export class Subscriber { producer.close(); // A settled lite07 tail acknowledges completion with FIN, without cancelling // the publisher's already-finished receive half. - if (tailSettled && this.version === Version.DRAFT_07) stream.writer.close(); + if (tailSettled && waitsForSubscriberFin(this.version)) stream.writer.close(); else stream.close(); console.debug(`subscribe close: id=${id} broadcast=${broadcast} track=${request.name}`); } catch (err) { diff --git a/js/net/src/lite/version.ts b/js/net/src/lite/version.ts index f9b1d3fd65..a69e8fea0d 100644 --- a/js/net/src/lite/version.ts +++ b/js/net/src/lite/version.ts @@ -249,6 +249,26 @@ export function hasStreamCount(version: Version): boolean { } } +/** + * Whether a served SUBSCRIBE completes only once the subscriber FINs or resets its half of the + * Subscribe Stream. Added in lite-07, where a subscriber FINs once its tail accounting settles, + * since a transport ACK does not say the application read the tail. + */ +export function waitsForSubscriberFin(version: Version): boolean { + // Explicitly list older versions so future versions keep the lite-07+ behavior. + switch (version) { + case Version.DRAFT_01: + case Version.DRAFT_02: + case Version.DRAFT_03: + case Version.DRAFT_04: + case Version.DRAFT_05: + case Version.DRAFT_06: + return false; + default: + return true; + } +} + /** Whether ANNOUNCE_START and ANNOUNCE_UPDATE may copy a path head or hop-chain tail from a live announcement. Added in lite-07. */ export function hasAnnounceCompression(version: Version): boolean { // Explicitly list older versions so future versions keep the lite-07+ behavior. diff --git a/quest/m1/README.md b/quest/m1/README.md index 49c666dbc5..d5b7da5d45 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -29,7 +29,6 @@ QUIC studies there on that rule. - [One route cost](/quest/m1/route-cost.md) - Warm and Cold collapse to one static route cost - [Delete removed cluster flags](/quest/m1/cluster-shims.md) - once a release has carried their refusals, `mesh` and `linger` leave `cluster::Config` - [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop` -- [JS close drain](/quest/m1/js-close-drain.md) - a JS `close()` waits for served subscriptions, and on lite-07 for the subscriber FIN, as Rust does - [Demand everywhere](/quest/m1/demand-everywhere.md) - requests and producers at every level watch subscribers through `demand()` alone, ahead of the bindings exposing it - [Request linger](/quest/m1/request-linger.md) - an upstream FETCH or SUBSCRIBE outlives its last reader by a short linger, so quick re-requests don't churn upstream - [Merge queue](/quest/m1/merge-queue.md) - the required checks run on `merge_group`, so a stale green check can no longer break main diff --git a/quest/m1/js-close-drain.md b/quest/m1/js-close-drain.md deleted file mode 100644 index f19b82c48f..0000000000 --- a/quest/m1/js-close-drain.md +++ /dev/null @@ -1,36 +0,0 @@ -# [M] JS close drains served subscriptions - -## Goal - -`Connection.close()` in `@moq/net` waits for the requests it serves to -finish before closing, as Rust's `Session::close()` does, within the same -one-second deadline. On every lite version it waits for each served SUBSCRIBE -to deliver its group streams, and for served FETCH and TRACK requests -(lite-05+); on lite-07 a served SUBSCRIBE also waits for the subscriber's FIN -or reset on the Subscribe Stream. Today `close()` waits only for announce -withdrawals, so a JS publisher that finishes a track and closes can cut the -final group short. - -## Plan - -- Full parity with Rust, not lite-07 alone (decided 2026-10-04): the FIN wait - means nothing while `close()` skips served subscriptions entirely. -- Mirror Rust's shape: count served SUBSCRIBE, FETCH, and TRACK requests from - dispatch until completion or cancellation (`owed` in - `rs/moq-net/src/lite/publisher.rs`), and gate the lite-07 wait, SUBSCRIBE - only, on the same version check as Rust's - `Version::waits_for_subscriber_fin()` (#4628). JS subscribers already FIN - on lite-07 after settling the track tail. -- Keep `abort()` immediate and the deadline at one second. -- Regressions with mocked time: a finished track's final group arrives before - close on lite-05/06/07, a lite-07 subscriber withholding FIN holds close - until the deadline, and abort during the drain ends at once. -- Update the `close()` doc comment in `js/net/src/lite/connection.ts` and the - close paragraph in `doc/lib/js/net.md`. - -Public API: none (behavior of `close()`). Wire: none; lite-07 FIN already -specified. - -## Related - -- [Track tail interop](/quest/m1/track-tail-interop.md) - proves the close across languages through a relay, which this makes hold for JS publishers From 0899fb27505366b3bfe30246a215db8f7d1ce2e7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 4 Oct 2026 21:52:28 -0700 Subject: [PATCH 3/4] fix(net): JS lite close drains requests that arrive during withdrawal The owed drain could finish before the withdrawals did, so a request served in between was cut off when close aborted the transport. Settle owed requests after the withdrawals, as Rust's drained() checks both together. Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/connection.test.ts | 95 +++++++++++++++++++++++++++--- js/net/src/lite/publisher.ts | 8 +-- 2 files changed, 92 insertions(+), 11 deletions(-) diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index 071b377c1a..f0717f3b77 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -134,15 +134,15 @@ for (const [alpn, version] of [ }); } -// Hold the FIN acknowledgement of every unidirectional stream `transport` opens once enabled, -// after the bytes and FIN reached the peer: a group stream whose tail is still in flight. -function holdUniFins(transport: MockTransport) { +// Once enabled, hold the FIN acknowledgement of every stream `transport` writes on (unidirectional +// streams it opens, or bidirectional streams the peer opens) after the bytes and FIN reached the +// peer: a group stream or withdrawal whose tail is still in flight. +function holdFins(transport: MockTransport, streams: "uni" | "incoming-bidi") { const reached = Promise.withResolvers(); const release = Promise.withResolvers(); let enabled = false; - const create = transport.createUnidirectionalStream.bind(transport); - transport.createUnidirectionalStream = async (options) => { - const writer = (await create(options)).getWriter(); + const wrap = (writable: WritableStream) => { + const writer = writable.getWriter(); return new WritableStream({ write: (bytes) => writer.write(bytes), abort: (reason) => writer.abort(reason), @@ -154,6 +154,20 @@ function holdUniFins(transport: MockTransport) { }, }); }; + if (streams === "uni") { + const create = transport.createUnidirectionalStream.bind(transport); + transport.createUnidirectionalStream = async (options) => wrap(await create(options)); + } else { + Object.defineProperty(transport, "incomingBidirectionalStreams", { + value: transport.incomingBidirectionalStreams.pipeThrough( + new TransformStream({ + transform(stream, controller) { + controller.enqueue({ readable: stream.readable, writable: wrap(stream.writable) }); + }, + }), + ), + }); + } return { reached: reached.promise, enable: () => { @@ -190,7 +204,7 @@ for (const alpn of [ALPN_05, ALPN_06, ALPN_07_WIP]) { // 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 () => { const pair = createMockTransportPair(alpn); - const fin = holdUniFins(pair.server); + const fin = holdFins(pair.server, "uni"); const origin = new Producer(); const broadcast = origin.createBroadcast(Path.from("room")); broadcast.announce(); @@ -341,3 +355,70 @@ test("abort during the close drain ends the session at once", async () => { served.cleanup(); } }); + +// A request that arrives while the withdrawals are still in flight is owed too, even though +// nothing was owed when close began. +test("close waits for a request served while withdrawals are in flight", async () => { + const deadline = mockDeadline(); + const version = Version.DRAFT_07; + const pair = createMockTransportPair(ALPN_07_WIP); + const fin = holdFins(pair.server, "incoming-bidi"); + const source = new Producer(); + const destination = new Producer(); + const broadcast = source.createBroadcast(Path.from("room")); + broadcast.announce(); + const producer = broadcast.createTrack("video"); + const group = producer.appendGroup(); + group.writeString("last"); + group.close(); + const url = new URL("https://localhost/test"); + const [client, server] = await Promise.all([ + connect({ url, transport: pair.client, consume: destination }), + accept({ url, transport: pair.server, publish: source.consume() }), + ]); + const announced = destination.announced(); + let closed = false; + void pair.server.closed.then(() => { + closed = true; + }); + try { + expect((await announced.next())?.kind).toBe("start"); + expect((await announced.next())?.kind).toBe("live"); + + fin.enable(); + const closing = server.close(); + await fin.reached; + + const subscriber = await Stream.open(pair.client, { version }); + await subscriber.writer.u53(StreamId.Subscribe); + await new Subscribe({ + id: 0n, + broadcast: Path.from("room"), + track: "video", + priority: 0, + maxAge: 10_000, + startGroup: 0, + }).encode(subscriber.writer, version); + expect("start" in (await decodeSubscribeResponse(subscriber.reader, version))).toBe(true); + producer.close(); + await subscriber.reader.readAll(); + + // The withdrawal completes, but the subscription still waits for the subscriber's FIN. + fin.release(); + await settle(); + expect(closed).toBe(false); + + subscriber.writer.close(); + await closing; + expect(closed).toBe(true); + } finally { + deadline.restore(); + fin.release(); + client.abort(); + server.abort(); + announced.close(); + broadcast.close(); + source.close(); + destination.close(); + } +}); diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 203043b35d..5d2fe2727a 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -1296,10 +1296,10 @@ export class Publisher { * @internal */ async drain(): Promise { - const settle = async () => { - while (this.#owed.size > 0) await Promise.allSettled(this.#owed); - }; - const [withdrawn] = await Promise.allSettled([this.#withdrawal.close(), settle()]); + const [withdrawn] = await Promise.allSettled([this.#withdrawal.close()]); + // Only after the withdrawals, so a request that arrived while they were in flight is + // waited for too. Owed requests run on their own meanwhile, so this adds no delay. + while (this.#owed.size > 0) await Promise.allSettled(this.#owed); if (withdrawn.status === "rejected") throw withdrawn.reason; } From 04efbcd59c1a368a48840e418f1cd7397adebaf0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 4 Oct 2026 22:45:43 -0700 Subject: [PATCH 4/4] test(net): cover the JS lite FETCH ack wait on close Also say when close() actually rejects: owed requests swallow their own failures, as in Rust, so only a failed withdrawal or the deadline rejects. Co-Authored-By: Claude Opus 5.5 --- doc/lib/js/net.md | 3 +- js/net/src/lite/connection.test.ts | 51 ++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index add335ed4a..dfb6359b12 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -121,7 +121,8 @@ On moq-lite it also waits for the requests it serves: a subscription to a finished track delivers its remaining groups, and on moq-lite-07 waits for the subscriber's FIN, which says it read the tail. A live track never ends on its own, so its subscription holds the close until the deadline. -It rejects if delivery fails or times out, and closes the transport either way. +It rejects if an announcement withdrawal fails or the deadline passes, and +closes the transport either way. Use `connection.abort()` for immediate shutdown. Reconnecting connection handles keep their synchronous `close()` disposal behavior. IETF drafts 14 through 16 send their withdrawals without waiting. diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index f0717f3b77..99ed41e2b3 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -8,6 +8,7 @@ import * as Path from "../path.ts"; import { Stream, Writer } from "../stream.ts"; import { wireOf } from "../wire.ts"; import { Connection, probeLevel } from "./connection.ts"; +import { Fetch } from "./fetch.ts"; import { Goaway } from "./goaway.ts"; import { ProbeLevel, Setup } from "./setup.ts"; import { DataType, StreamId } from "./stream.ts"; @@ -356,6 +357,56 @@ test("abort during the close drain ends the session at once", async () => { } }); +// A served FETCH stays owed until the peer acknowledges its FIN, so close cannot discard its tail. +test("close waits for a served FETCH to be acknowledged", async () => { + const deadline = mockDeadline(); + const version = Version.DRAFT_05; + const pair = createMockTransportPair(ALPN_05); + const fin = holdFins(pair.server, "incoming-bidi"); + const origin = new Producer(); + const broadcast = origin.createBroadcast(Path.from("room")); + broadcast.announce(); + const producer = broadcast.createTrack("video"); + const group = producer.appendGroup(); + group.writeString("last"); + group.close(); + const server = new Connection({ + url: new URL("https://relay.example/"), + quic: pair.server, + version, + publish: origin.consume(), + }); + let closed = false; + void pair.server.closed.then(() => { + closed = true; + }); + try { + fin.enable(); + const fetch = await Stream.open(pair.client, { version }); + await fetch.writer.u53(StreamId.Fetch); + await new Fetch({ broadcast: Path.from("room"), track: "video", priority: 0, group: 0 }).encode( + fetch.writer, + version, + ); + await fetch.reader.readAll(); + await fin.reached; + + const closing = server.close(); + await settle(); + expect(closed).toBe(false); + + fin.release(); + await closing; + expect(closed).toBe(true); + } finally { + deadline.restore(); + fin.release(); + server.abort(); + broadcast.close(); + origin.close(); + } +}); + // A request that arrives while the withdrawals are still in flight is owed too, even though // nothing was owed when close began. test("close waits for a request served while withdrawals are in flight", async () => {