diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 8fc4143fc9..dfb6359b12 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -117,7 +117,12 @@ 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. -It rejects if delivery fails or times out, and closes the transport either way. +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 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 4cc3748703..99ed41e2b3 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -1,12 +1,18 @@ -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 { Fetch } from "./fetch.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 +134,342 @@ for (const [alpn, version] of [ } }); } + +// 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 wrap = (writable: WritableStream) => { + const writer = writable.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; + }, + }); + }; + 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: () => { + 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 = holdFins(pair.server, "uni"); + 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(); + } +}); + +// 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 () => { + 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/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..5d2fe2727a 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 [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; } 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