From b6f7ff6d91f3bd9b169583effcc4dac79f5be3ff Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 18:08:41 -0700 Subject: [PATCH 1/3] quest: claim quest/m1/js-track-tail Co-Authored-By: Claude Opus 5.5 From b2e662f6bd250cc9cf7380368622406cb28e72a3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 18:42:11 -0700 Subject: [PATCH 2/3] wip: js track tail Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/adapter.ts | 6 +- js/net/src/ietf/object.ts | 23 +++- js/net/src/ietf/publish.ts | 40 +++++- js/net/src/ietf/publisher.test.ts | 76 ++++++++++- js/net/src/ietf/publisher.ts | 105 ++++++++++++-- js/net/src/ietf/subscriber.ts | 163 +++++++++++++++++----- js/net/src/ietf/tail.test.ts | 177 ++++++++++++++++++++++++ js/net/src/lite/publisher.test.ts | 13 +- js/net/src/lite/publisher.ts | 28 +++- js/net/src/lite/subscriber.ts | 118 ++++++++++++---- js/net/src/lite/tail.test.ts | 219 ++++++++++++++++++++++++++++++ js/net/src/tail.test.ts | 105 ++++++++++++++ js/net/src/tail.ts | 125 +++++++++++++++++ js/net/src/track.test.ts | 66 +++++++++ js/net/src/track.ts | 123 +++++++++++++---- 15 files changed, 1268 insertions(+), 119 deletions(-) create mode 100644 js/net/src/ietf/tail.test.ts create mode 100644 js/net/src/lite/tail.test.ts create mode 100644 js/net/src/tail.test.ts create mode 100644 js/net/src/tail.ts diff --git a/js/net/src/ietf/adapter.ts b/js/net/src/ietf/adapter.ts index 11d11dff5a..fd16a5349d 100644 --- a/js/net/src/ietf/adapter.ts +++ b/js/net/src/ietf/adapter.ts @@ -58,7 +58,7 @@ export class NativeSession implements Session { const Route = { NewRequest: 0, // Create virtual bidi stream, push initial message Response: 1, // Push message to existing stream (keep open) - ErrorResponse: 2, // Push message to existing stream, then close + ErrorResponse: 2, // Push a final message to existing stream, then close CloseStream: 3, // Close stream recv (no bytes pushed) FollowUp: 4, // Push follow-up message to existing stream MaxRequestId: 5, // Update flow control @@ -658,9 +658,9 @@ export class ControlStreamAdapter implements Session { return { route: Route.CloseStream, requestId }; } case 0x0b: { - // PublishDone + // PublishDone: the subscriber reads its status and stream count before the end. const requestId = await readRequestId(); - return { route: Route.CloseStream, requestId }; + return { route: Route.ErrorResponse, requestId }; } case 0x17: { // FetchCancel diff --git a/js/net/src/ietf/object.ts b/js/net/src/ietf/object.ts index 6599d52216..974b44706b 100644 --- a/js/net/src/ietf/object.ts +++ b/js/net/src/ietf/object.ts @@ -3,6 +3,7 @@ import { Timescale, Timestamp } from "../time.ts"; import { type IetfVersion, Version } from "./version.ts"; const GROUP_END = 0x03; +const END_OF_TRACK = 0x04; // MOQ Object Property ids, shared with draft-ietf-moq-loc-04. const PROP_TIMESCALE = 0x08n; @@ -259,14 +260,24 @@ export class Group { /** A moq-transport object inside a group stream. */ export class Frame { - /** The object payload, or `undefined` for the end of group marker. */ + /** The object payload, or `undefined` for an end of group or end of track marker. */ payload?: Uint8Array; /** The presentation timestamp carried in object properties, when present. */ timestamp?: Timestamp; + /** + * An END_OF_TRACK marker: no object at or past its location exists. At object 0 its group + * does not exist either, so the track ends at that group; later in a group it ends after it. + */ + endOfTrack: boolean; - constructor({ payload, timestamp }: { payload?: Uint8Array; timestamp?: Timestamp } = {}) { + constructor({ + payload, + timestamp, + endOfTrack = false, + }: { payload?: Uint8Array; timestamp?: Timestamp; endOfTrack?: boolean } = {}) { this.payload = payload; this.timestamp = timestamp; + this.endOfTrack = endOfTrack; } /** @@ -284,7 +295,10 @@ export class Frame { await w.write(extensions); } - if (this.payload !== undefined) { + if (this.endOfTrack) { + await w.u53(0); // length = 0 + await w.u53(END_OF_TRACK); + } else if (this.payload !== undefined) { await w.u53(this.payload.byteLength); if (this.payload.byteLength === 0) { @@ -334,6 +348,9 @@ export class Frame { const status = await r.u53(); + // Defined on every implemented draft, whether or not the header marks the group's end. + if (status === END_OF_TRACK) return new Frame({ endOfTrack: true }); + if (flags.hasEnd) { // Empty frame if (status === 0) return new Frame({ payload: new Uint8Array(0), timestamp }); diff --git a/js/net/src/ietf/publish.ts b/js/net/src/ietf/publish.ts index 200d19cf78..40f105338a 100644 --- a/js/net/src/ietf/publish.ts +++ b/js/net/src/ietf/publish.ts @@ -213,21 +213,53 @@ export class PublishError { } } +/** PUBLISH_DONE status codes this implementation distinguishes. Stable across drafts 14 through 22. */ +export const PublishDoneStatus = { + INTERNAL_ERROR: 0x0, + TRACK_ENDED: 0x2, + /** Removed in draft-20, where 0x3 is unassigned. */ + SUBSCRIPTION_ENDED: 0x3, +} as const; + +/** Whether a PUBLISH_DONE status ends the track cleanly rather than aborting it. */ +export function publishDoneClean(statusCode: number, version: IetfVersion): boolean { + if (statusCode === PublishDoneStatus.TRACK_ENDED) return true; + if (statusCode !== PublishDoneStatus.SUBSCRIPTION_ENDED) return false; + switch (version) { + case Version.DRAFT_14: + case Version.DRAFT_15: + case Version.DRAFT_16: + case Version.DRAFT_17: + case Version.DRAFT_18: + case Version.DRAFT_19: + return true; + default: + return false; + } +} + // In draft-14, this message is renamed from SUBSCRIBE_DONE to PUBLISH_DONE export class PublishDone { static readonly id = 0x0b; requestId: bigint | undefined; statusCode: number; + /** + * How many data streams the publisher opened for the subscription, fill streams included. + * A hint: a peer may send 0 or the "unknown" sentinel regardless. + */ + streamCount: bigint; reasonPhrase: string; constructor({ requestId, statusCode, + streamCount = 0n, reasonPhrase, - }: { requestId?: bigint; statusCode: number; reasonPhrase: string }) { + }: { requestId?: bigint; statusCode: number; streamCount?: bigint; reasonPhrase: string }) { this.requestId = requestId; this.statusCode = statusCode; + this.streamCount = streamCount; this.reasonPhrase = reasonPhrase; } @@ -237,7 +269,7 @@ export class PublishDone { await w.u62(this.requestId); } await w.u62(BigInt(this.statusCode)); - await w.u62(BigInt(0)); // stream_count = 0 (unsupported) + await w.u62(this.streamCount); await w.string(this.reasonPhrase); } @@ -255,9 +287,9 @@ export class PublishDone { ? await r.u62() : undefined; const statusCode = Number(await r.u62()); - await r.u62(); // ignore stream_count + const streamCount = await r.u62(); const reasonPhrase = await r.string(); - return new PublishDone({ requestId, statusCode, reasonPhrase }); + return new PublishDone({ requestId, statusCode, streamCount, reasonPhrase }); } } diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 0d0908fb2f..a16ce4ccbc 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -14,7 +14,7 @@ import { wireOf } from "../wire.ts"; import { NativeSession, type Session } from "./adapter.ts"; import type * as Cluster from "./cluster.ts"; import { FetchHeader } from "./fetch.ts"; -import { Group as GroupMessage } from "./object.ts"; +import { Frame, Group as GroupMessage } from "./object.ts"; import { PublishDone } from "./publish.ts"; import { PublishNamespace } from "./publish_namespace.ts"; import { Publisher } from "./publisher.ts"; @@ -908,6 +908,16 @@ async function readGroup(stream: ReadableStream): Promise): Promise { + const reader = new Reader(stream, undefined, V20); + const header = await GroupMessage.decode(reader, V20); + const frame = await Frame.decode(reader, header.flags, undefined, V20); + expect(frame.endOfTrack).toBe(true); + expect(await reader.done()).toBe(true); + return header.groupId; +} + /** * Read a fill's fetch stream to its end, reporting a reset rather than throwing. * @@ -1339,8 +1349,12 @@ test("draft-20: a clean close past a bounded filter's end still sends PUBLISH_DO expect(await client.reader.u53()).toBe(PublishDone.id); const done = await PublishDone.decode(client.reader, V20); expect(done.statusCode).toBe(TRACK_ENDED_STATUS); + expect(done.streamCount).toBe(2n); - // Only the in-range group was ever opened. + // Only the in-range group was ever served; the other stream marks the track's end. + const end = await nextUni(fx.uni); + if (!end) throw new Error("the track's end was never marked"); + expect(await readEndOfTrack(end)).toBe(2); expect(await nextUni(fx.uni)).toBeUndefined(); } finally { fx.close(); @@ -1454,3 +1468,61 @@ test("draft-20: a fill works on a dynamically requested track", async () => { client.close(); } }); + +// PUBLISH_DONE MUST wait until every stream the subscription will open is closed, so its +// Stream Count is final. A group still queued for a stream slot when the track ends is one. +test("draft-20: PUBLISH_DONE waits for a queued group and counts every stream", async () => { + const fx = fixture(); + const track = fx.broadcast.createTrack("video"); + + // Park the first stream open, the way a transport at its stream cap does. + const slot = Promise.withResolvers(); + const create = fx.pair.server.createUnidirectionalStream.bind(fx.pair.server); + let parked = false; + fx.pair.server.createUnidirectionalStream = async (options?: WebTransportSendStreamOptions) => { + if (!parked) { + parked = true; + await slot.promise; + } + return create(options); + }; + + const { client } = await runSubscribe( + fx, + new Subscribe({ + requestId: 7n, + trackNamespace: Path.from("test"), + trackName: "video", + subscriberPriority: 0, + filter: { kind: "absolute", startGroup: 0n, startObject: 0n }, + }), + ); + + try { + writeGroup(track, 1); + track.close(); + + // Nothing ends the subscription while the group waits for its slot. + const response = client.reader.u53(); + const idle = new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 20)); + expect(await Promise.race([response, idle])).toBe("pending"); + + slot.resolve(); + const served = await nextUni(fx.uni); + if (!served) throw new Error("the queued group was never served"); + expect((await readGroup(served)).sequence).toBe(0); + + expect(await response).toBe(PublishDone.id); + const done = await PublishDone.decode(client.reader, V20); + expect(done.statusCode).toBe(TRACK_ENDED_STATUS); + // The group's stream and the END_OF_TRACK marker's. + expect(done.streamCount).toBe(2n); + + const end = await nextUni(fx.uni); + if (!end) throw new Error("the track's end was never marked"); + expect(await readEndOfTrack(end)).toBe(1); + } finally { + fx.close(); + client.close(); + } +}); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 0074eef517..a14dbb4ce2 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -7,7 +7,7 @@ import { hiddenBelow, hooks } from "../internal.ts"; import type { Consumer as OriginConsumer } from "../origin.ts"; import * as Path from "../path.ts"; import { type Stream, Writer } from "../stream.ts"; -import { Milli, type Timescale } from "../time.ts"; +import { Milli, Timescale } from "../time.ts"; import type { Subscriber as TrackSubscriber } from "../track.ts"; import { TimeoutError, withTimeout } from "../util/timeout.ts"; import * as Varint from "../varint.ts"; @@ -20,7 +20,7 @@ import * as Filter from "./filter.ts"; import { FetchFrame, Frame, Group as GroupMessage } from "./object.ts"; import { fromWire, toWire } from "./priority.ts"; import * as Properties from "./properties.ts"; -import { PublishDone } from "./publish.ts"; +import { PublishDone, PublishDoneStatus } from "./publish.ts"; import { PublishNamespace, PublishNamespaceDone, PublishNamespaceOk } from "./publish_namespace.ts"; import { RequestError, RequestOk } from "./request.ts"; import { type Subscribe, SubscribeError, SubscribeOk } from "./subscribe.ts"; @@ -52,12 +52,6 @@ function sameAdvert(a: Advertised | undefined, b: Advertised | undefined): boole return a !== undefined && b !== undefined && a.identity === b.identity && routesEqual(a.route, b.route); } -/** PUBLISH_DONE statuses this implementation emits. Stable across drafts 14 through 19. */ -const PUBLISH_DONE_STATUS = { - INTERNAL_ERROR: 0x0, - TRACK_ENDED: 0x2, -} as const; - /** * How long one advertisement may take to be answered. Matches the Rust publisher, and the * peer accepting the stream is only half the exchange: one it never answers on holds the @@ -108,8 +102,14 @@ interface RunGroup { /** Settles when the subscriber leaves, dropping a group still queued for a stream slot. */ unsubscribed: Promise; + + /** The subscription's data stream count, which PUBLISH_DONE reports. */ + streams: StreamCount; } +/** How many data streams a subscription opened, fill streams included. */ +type StreamCount = { opened: number }; + /** What {@link Publisher.runFill} needs to serve one subscription's backfill. */ interface RunFill { /** The subscription's request ID, which the fetch stream names. */ @@ -141,6 +141,9 @@ interface RunFill { /** Settles when the subscriber leaves, releasing a fill still waiting on its group. */ unsubscribed: Promise; + + /** The subscription's data stream count, which PUBLISH_DONE reports. */ + streams: StreamCount; } /** @@ -349,6 +352,11 @@ export class Publisher { () => unsubscribe(), ); + // Every group started, until its stream finishes or resets, and the data streams + // opened for PUBLISH_DONE to report. + const groups = new Set>(); + const streams: StreamCount = { opened: 0 }; + // Serve track groups, racing with stream close (= Unsubscribe) const serving = (async () => { for (;;) { @@ -365,7 +373,7 @@ export class Publisher { continue; } - void this.#runGroup({ + const task = this.#runGroup({ requestId: msg.requestId, group, timescale, @@ -373,7 +381,10 @@ export class Publisher { stamped: msg.propertiesWanted, slice: groupSlice(range, group.sequence), unsubscribed, + streams, }); + groups.add(task); + void task.finally(() => groups.delete(task)); } })(); @@ -389,16 +400,38 @@ export class Publisher { timescale, stamped: msg.propertiesWanted, unsubscribed, + streams, }) : Promise.resolve(); let publishError: Error | undefined; + let ended = false; try { - await race([Promise.all([serving, filling]), stream.reader.closed]); + const served = Symbol("served"); + ended = + (await race([Promise.all([serving, filling]).then(() => served), stream.reader.closed])) === served; } catch (err: unknown) { publishError = error(err); } + // PUBLISH_DONE waits until every stream this subscription will open is closed, as + // the draft requires, so its count is final. The subscriber leaving cancels the + // ones still queued instead. + await race([Promise.all(groups), unsubscribed]); + + // Draft 14 on has no end location in PUBLISH_DONE: an END_OF_TRACK object is what + // tells the subscriber where the track ended. + const final = track.final(); + if (ended && !publishError && final !== undefined) { + await this.#runEndOfTrack({ + requestId: msg.requestId, + final, + publisherPriority, + unsubscribed, + streams, + }); + } + console.debug(`publish done: broadcast=${name} track=${track.name}`); if (publishError) { console.warn(`publish error: broadcast=${name} track=${track.name} error=${reason(publishError)}`); @@ -413,7 +446,8 @@ export class Publisher { version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16 ? msg.requestId : undefined, - statusCode: publishError ? PUBLISH_DONE_STATUS.INTERNAL_ERROR : PUBLISH_DONE_STATUS.TRACK_ENDED, + statusCode: publishError ? PublishDoneStatus.INTERNAL_ERROR : PublishDoneStatus.TRACK_ENDED, + streamCount: BigInt(streams.opened), reasonPhrase: publishError ? "internal error" : "track ended", }); await done.encode(stream.writer, version); @@ -442,7 +476,7 @@ export class Publisher { * Runs a group and sends its frames using ObjectStream (Subgroup delivery mode). */ async #runGroup(options: RunGroup) { - const { requestId, group, timescale, publisherPriority, stamped, slice, unsubscribed } = options; + const { requestId, group, timescale, publisherPriority, stamped, slice, unsubscribed, streams } = options; try { // One stream per group is faster than a peer at its limit can retire them, so this // is the one path that doesn't wait for a slot: the transport would serve the opens @@ -457,6 +491,7 @@ export class Publisher { group.close(new Error("no stream slot")); return; } + streams.opened += 1; const header = new GroupMessage({ trackAlias: requestId, @@ -525,6 +560,49 @@ export class Publisher { } } + /** + * Mark the track's end with an END_OF_TRACK object on its own stream, at object 0 of the + * group that will never exist. + * + * The last group's stream has usually finished before the track ends, so the marker cannot + * ride on it. A failure only costs the subscriber the early boundary. + */ + async #runEndOfTrack(options: { + requestId: bigint; + final: number; + publisherPriority: number; + unsubscribed: Promise; + streams: StreamCount; + }) { + const { requestId, final, publisherPriority, unsubscribed, streams } = options; + const version = this.#session.version; + const stream = await Writer.tryOpen(this.#quic, { cancel: unsubscribed, version }).catch(() => undefined); + if (!stream) return; + streams.opened += 1; + + try { + const header = new GroupMessage({ + trackAlias: requestId, + groupId: final, + subGroupId: 0, + publisherPriority, + flags: { + hasExtensions: false, + hasSubgroup: false, + hasSubgroupObject: false, + hasEnd: false, + hasPriority: true, + firstObject: true, + }, + }); + await header.encode(stream, version); + await new Frame({ endOfTrack: true }).encode(stream, header.flags, Timescale.MILLI, version); + stream.close(); + } catch (err: unknown) { + stream.reset(error(err)); + } + } + /** * Serve a draft-20 fill on its own fetch stream: the requested range, read from the * group cache, capped at the Largest Object snapshot. @@ -534,7 +612,7 @@ export class Publisher { * fill-failure signal. Nothing here touches the subscription either way. */ async #runFill(options: RunFill) { - const { requestId, fill, cache, timescale, stamped, unsubscribed } = options; + const { requestId, fill, cache, timescale, stamped, unsubscribed, streams } = options; const version = this.#session.version; // Everything is inside the try so the cache fork is released on every path out, @@ -548,6 +626,7 @@ export class Publisher { console.debug(`fill stream failed to open: fill=${requestId}`); return; } + streams.opened += 1; await stream.u53(FetchHeader.type); await new FetchHeader({ requestId }).encode(stream, version); diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 33e8595b92..b107b0ca23 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -8,6 +8,7 @@ import { Cost, type Route, routesEqual, UNKNOWN_HOP } from "../hop.ts"; import { hiddenBelow, hooks, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts"; import * as Path from "../path.ts"; import type { Reader, Stream } from "../stream.ts"; +import { TAIL_GRACE_MS, Tail } from "../tail.ts"; import { type Timescale, Timestamp } from "../time.ts"; import type * as track from "../track.ts"; import { TimeoutError, withTimeout } from "../util/timeout.ts"; @@ -18,7 +19,7 @@ import * as Cluster from "./cluster.ts"; import { requestReason, toRequestCode } from "./error.ts"; import { Frame, type Group as GroupMessage } from "./object.ts"; import { fromWire, toWire } from "./priority.ts"; -import { type Publish, PublishError } from "./publish.ts"; +import { type Publish, PublishDone, PublishError, publishDoneClean } from "./publish.ts"; import { type PublishNamespace, PublishNamespaceDone, @@ -44,6 +45,15 @@ import { Version } from "./version.ts"; // blocks. The timeout turns that into a clear error. const SUBSCRIBE_OK_TIMEOUT_MS = 10_000; +// A live subscription, as the track alias its data streams name resolves to. +type Subscription = { + // The write side incoming group streams are routed into. + track: track.Producer; + // The group streams received, so the subscription can wait for the ones PUBLISH_DONE + // says are still owed. + tail: Tail; +}; + // Out-parameter for #openSubscribe: lets the caller observe partial progress // (stream opened, trackAlias registered) so it can clean up on timeout even // before the setup promise settles. @@ -93,7 +103,7 @@ export class Subscriber { #cluster?: Cluster.Hops; // Publisher-chosen aliases used by incoming group streams. - #aliases = new TrackAliases(); + #aliases = new TrackAliases(); // Units for each track's object Timestamps, from the TIMESCALE Track Property in // SUBSCRIBE_OK. A track missing from this map declared no timeline, so the publisher @@ -481,12 +491,13 @@ export class Subscriber { // Keep the request pending until SUBSCRIBE_OK supplies immutable track metadata. // Group streams already wait on the alias, so early data stays behind this response. const producer = hooks.pendingTrackProducer(request); + const subscription: Subscription = { track: producer, tail: new Tail() }; // Open the stream and wait for SUBSCRIBE_OK under a timeout. State // flows back via `state` so the timeout path can clean up the stream // and any registration if setup eventually finishes. const state: SubscribeSetupState = {}; - const setup = this.#openSubscribe(state, broadcast, request, producer, requestId); + const setup = this.#openSubscribe(state, broadcast, request, subscription, requestId); // The publisher can be serving before it answers, so waiting only on the response // would miss the local side going away and leave it serving a track nobody reads. @@ -534,7 +545,7 @@ export class Subscriber { const cleanup = async (afterSetup: boolean) => { state.cancelled = true; - if (state.registeredAlias !== undefined && this.#aliases.retire(state.registeredAlias, producer)) { + if (state.registeredAlias !== undefined && this.#aliases.retire(state.registeredAlias, subscription)) { this.#timescales.delete(state.registeredAlias); } @@ -575,10 +586,10 @@ export class Subscriber { const localEnded = Symbol("local"); const idle = Symbol("idle"); - // Terminal conditions settle at most once (stream close = PublishDone, track close = - // local unsubscribe); race them once so the demand loop doesn't re-subscribe each pass. + // Terminal conditions settle at most once (PublishDone, track close = local + // unsubscribe); race them once so the demand loop doesn't re-subscribe each pass. const done = race([ - stream.reader.closed.then(() => publisherEnded), + this.#runPublishDone(stream, subscription).then(() => publisherEnded), producer.closed.then(() => localEnded), ]); @@ -615,8 +626,37 @@ export class Subscriber { } finally { // Only the owner tears down the alias metadata: a later subscription may have // reclaimed the alias and installed its own timescale. - if (this.#aliases.retire(trackAlias, producer)) this.#timescales.delete(trackAlias); + if (this.#aliases.retire(trackAlias, subscription)) this.#timescales.delete(trackAlias); + } + } + + /** + * Read the PUBLISH_DONE that ends a subscription, then wait for the data streams it counts. + * + * An error status aborts the track with it. A clean one leaves streams in flight, since + * QUIC does not order them, so wait until the Stream Count many have been read, or a + * bounded grace for the ones that never arrive (the draft says to use a timeout). The count + * is only a hint: a peer may send 0 regardless, so 0 waits out the grace. A request stream + * that ends without one ends the track the same way. + */ + async #runPublishDone(stream: Stream, subscription: Subscription): Promise { + const version = this.#session.version; + let count: bigint | undefined; + if (!(await stream.reader.done())) { + const typeId = await stream.reader.u53(); + if (typeId !== PublishDone.id) { + throw new ProtocolViolation(`unexpected message on a subscription: 0x${typeId.toString(16)}`); + } + const done = await PublishDone.decode(stream.reader, version); + if (!publishDoneClean(done.statusCode, version)) { + throw new Error(`publish done: status=0x${done.statusCode.toString(16)} reason=${done.reasonPhrase}`); + } + count = done.streamCount; } + + const { tail, track } = subscription; + const complete = () => count !== undefined && count > 0n && BigInt(tail.streams) >= count; + await tail.settle(complete, TAIL_GRACE_MS, track.closed); } /** @@ -648,7 +688,7 @@ export class Subscriber { state: SubscribeSetupState, broadcast: Path.Valid, request: track.Request, - producer: track.Producer, + subscription: Subscription, requestId: bigint, ): Promise<{ stream: Stream; alias: bigint }> { const version = this.#session.version; @@ -706,7 +746,7 @@ export class Subscriber { request.accept({ priority: fromWire(ok.properties.priority ?? 128) }); try { - this.#aliases.set(ok.trackAlias, producer, { broadcast, name: request.name }); + this.#aliases.set(ok.trackAlias, subscription, { broadcast, name: request.name }); const timescale = ok.properties.timescale; if (timescale !== undefined) { this.#timescales.set(ok.trackAlias, timescale); @@ -935,35 +975,62 @@ export class Subscriber { throw new Error("subgroups are not supported"); } - // FIRST_OBJECT clear says this stream starts partway through the group, which the - // draft lets a publisher do to answer a filter. Nothing above here can use it: the - // objects that would arrive are not decodable without the missing head, and a group - // is the unit an application resyncs on. Drop it and pick up at the next group, the - // same degradation as a publisher that no longer holds the head. - // - // This only saves reading a stream we would throw away. The bit is the publisher's - // claim, so what is enforced is the object ids themselves: `Frame.decode` holds every - // object to starting at 0 and incrementing by 1, whatever the header said and on the - // drafts that have no such bit to read. - if (!group.flags.firstObject) { - console.debug(`dropping a group with no head: alias=${group.trackAlias} group=${group.groupId}`); - stream.stop(new Error("a group must start at object 0")); + let subscription: Subscription; + try { + // The control message establishing this alias can arrive after the data stream. + subscription = await this.#aliases.get(group.trackAlias); + } catch (err: unknown) { + const e = error(err); + // Ours: we cancelled the subscription and the publisher has not stopped yet. + // Anything else on this alias is the publisher sending data for a track it never + // acknowledged, which is worth seeing. + if (e instanceof RetiredTrackAlias) { + console.debug(`dropping group for a cancelled subscription: alias=${group.trackAlias}`); + } + stream.stop(e); return; } - const producer = new netGroup.Producer(group.groupId); + const { track, tail } = subscription; + // Every data stream counts toward PUBLISH_DONE's Stream Count, even one dropped below. + const read = tail.open(group.groupId); + + // Created on the first object rather than the header: an END_OF_TRACK at object 0 + // means the group does not exist at all. + let producer: netGroup.Producer | undefined; + const open = () => { + if (!producer) { + producer = new netGroup.Producer(group.groupId); + track.writeGroup(producer); + } + return producer; + }; try { - // The control message establishing this alias can arrive after the data stream. - const track = await this.#aliases.get(group.trackAlias); + // FIRST_OBJECT clear says this stream starts partway through the group, which the + // draft lets a publisher do to answer a filter. Nothing above here can use it: the + // objects that would arrive are not decodable without the missing head, and a group + // is the unit an application resyncs on. Drop it and pick up at the next group, the + // same degradation as a publisher that no longer holds the head. + // + // This only saves reading a stream we would throw away. The bit is the publisher's + // claim, so what is enforced is the object ids themselves: `Frame.decode` holds every + // object to starting at 0 and incrementing by 1, whatever the header said and on the + // drafts that have no such bit to read. + if (!group.flags.firstObject) { + console.debug(`dropping a group with no head: alias=${group.trackAlias} group=${group.groupId}`); + stream.stop(new Error("a group must start at object 0")); + return; + } + // The alias binds after SUBSCRIBE_OK commits the track property; an omitted // header priority inherits it (draft-21 section 10.4). if (!group.flags.hasPriority) group.publisherPriority = toWire((await track.info()).priority); - track.writeGroup(producer); - for (;;) { - const done = await race([stream.done(), producer.closed, track.closed]); + // Only the group's own stream ends it: a track that closes first has already + // closed (or aborted) this group through its cache. + const done = await (producer ? race([stream.done(), producer.closed]) : stream.done()); if (done !== false) break; const frame = await Frame.decode( @@ -972,22 +1039,44 @@ export class Subscriber { this.#timescales.get(group.trackAlias), this.#session.version, ); + + if (frame.endOfTrack) { + // No object at or past this location exists: after the group's last object + // the track ends with it, and at object 0 it ends before it. + const end = producer ? group.groupId + 1 : group.groupId; + producer?.close(); + try { + track.finishAt(end); + } catch (err: unknown) { + throw new ProtocolViolation(`invalid END_OF_TRACK: ${reason(error(err))}`); + } + return; + } if (frame.payload === undefined) break; - producer.writeFrame({ payload: frame.payload, timestamp: frame.timestamp ?? Timestamp.now() }); + open().writeFrame({ payload: frame.payload, timestamp: frame.timestamp ?? Timestamp.now() }); } - producer.close(); + // A group with no objects still exists. + open().close(); } catch (err: unknown) { const e = error(err); - // Ours: we cancelled the subscription and the publisher has not stopped yet. - // Anything else on this alias is the publisher sending data for a track it never - // acknowledged, which is worth seeing. - if (e instanceof RetiredTrackAlias) { - console.debug(`dropping group for a cancelled subscription: alias=${group.trackAlias}`); + if (e instanceof ProtocolViolation) { + // The publisher broke the track's end, which no later group can repair. + producer?.close(e); + track.close(e); + } else { + // A stream that fails before its first object still names a group, which the + // reader sees fail rather than silently go missing. + try { + open().close(e); + } catch { + // The track has already closed or ended below this group. + } } - producer.close(e); stream.stop(e); + } finally { + read(); } } } diff --git a/js/net/src/ietf/tail.test.ts b/js/net/src/ietf/tail.test.ts new file mode 100644 index 0000000000..d7a505f642 --- /dev/null +++ b/js/net/src/ietf/tail.test.ts @@ -0,0 +1,177 @@ +import { expect, test } from "bun:test"; +import type { Consumer as GroupConsumer } from "../group.ts"; +import { createMockTransportPair } from "../mock.ts"; +import * as Path from "../path.ts"; +import { Reader, Stream } from "../stream.ts"; +import { TAIL_GRACE_MS } from "../tail.ts"; +import { Milli } from "../time.ts"; +import { NativeSession } from "./adapter.ts"; +import { type GroupFlags, Group as GroupMessage } from "./object.ts"; +import { PublishDone } from "./publish.ts"; +import { Subscribe, SubscribeOk } from "./subscribe.ts"; +import { Subscriber } from "./subscriber.ts"; +import { ALPN, Version } from "./version.ts"; + +const VERSION = Version.DRAFT_19; +const ALIAS = 9n; +const TRACK_ENDED = 0x2; +const INTERNAL_ERROR = 0x0; + +// A plain subgroup stream: no extensions, no subgroup id, end of group on FIN. +const FLAGS: GroupFlags = { + hasExtensions: false, + hasSubgroup: false, + hasSubgroupObject: false, + hasEnd: true, + hasPriority: true, + firstObject: true, +}; + +/** One object with a zero id delta. Every field is under 64, so each is a one-byte varint. */ +function object(payload: string): Uint8Array { + const bytes = new TextEncoder().encode(payload); + return new Uint8Array([0, bytes.byteLength, ...bytes]); +} + +/** An END_OF_TRACK object: zero length, then status 0x4. */ +const END_OF_TRACK = new Uint8Array([0, 0, 0x4]); + +/** A group stream the test writes by hand, handed to the subscriber as if it arrived. */ +function groupStream(subscriber: Subscriber, groupId: number) { + let controller!: ReadableStreamDefaultController; + const readable = new ReadableStream({ start: (c) => (controller = c) }); + const header = new GroupMessage({ trackAlias: ALIAS, groupId, subGroupId: 0, publisherPriority: 0, flags: FLAGS }); + const handled = subscriber.handleGroup(header, new Reader(readable, undefined, VERSION)); + return { + write: (bytes: Uint8Array) => controller.enqueue(bytes), + finish: () => controller.close(), + handled, + }; +} + +/** A subscriber with one track subscribed and answered; the test plays the publisher. */ +async function subscribed() { + const pair = createMockTransportPair(ALPN.DRAFT_19); + const session = new NativeSession(pair.server, VERSION, true); + const subscriber = new Subscriber({ session }); + const reader = subscriber + .consume(Path.from("room")) + .track("video") + .subscribe({ maxAge: Milli(60_000) }); + + const peer = await Stream.accept(pair.client, VERSION); + if (!peer) throw new Error("the subscriber never opened a subscribe stream"); + expect(await peer.reader.u53()).toBe(Subscribe.id); + const request = await Subscribe.decode(peer.reader, VERSION); + await peer.writer.u53(SubscribeOk.id); + await new SubscribeOk({ requestId: request.requestId, trackAlias: ALIAS }).encode(peer.writer, VERSION); + + return { + subscriber, + reader, + done: async (statusCode: number, streamCount: bigint) => { + await peer.writer.u53(PublishDone.id); + await new PublishDone({ statusCode, streamCount, reasonPhrase: "done" }).encode(peer.writer, VERSION); + peer.writer.close(); + }, + }; +} + +async function readAll(group: GroupConsumer | undefined): Promise { + if (!group) throw new Error("no group"); + const out: string[] = []; + for (;;) { + const next = await group.readString(); + if (next === undefined) return out; + out.push(next); + } +} + +test("a group stream that arrives after PUBLISH_DONE is delivered", async () => { + const { subscriber, reader, done } = await subscribed(); + const first = groupStream(subscriber, 0); + first.write(object("0.0")); + first.finish(); + await first.handled; + await done(TRACK_ENDED, 2n); + + // QUIC does not order streams, so the second one lands after PUBLISH_DONE. + const started = performance.now(); + const late = groupStream(subscriber, 1); + late.write(object("1.0")); + late.finish(); + + expect(await readAll(await reader.recvGroup())).toEqual(["0.0"]); + expect(await readAll(await reader.recvGroup())).toEqual(["1.0"]); + expect(await reader.recvGroup()).toBeUndefined(); + expect(await reader.closed).toBeNull(); + // The Stream Count was met, so nothing waited out the grace. + expect(performance.now() - started).toBeLessThan(TAIL_GRACE_MS); +}); + +test("a group read across PUBLISH_DONE is delivered whole", async () => { + const { subscriber, reader, done } = await subscribed(); + const group = groupStream(subscriber, 0); + group.write(object("0.0")); + await done(TRACK_ENDED, 1n); + + const received = await reader.recvGroup(); + expect(await received?.readString()).toBe("0.0"); + group.write(object("0.1")); + group.finish(); + expect(await readAll(received)).toEqual(["0.1"]); + expect(await reader.recvGroup()).toBeUndefined(); + expect(await reader.closed).toBeNull(); +}); + +test("a Stream Count of 0 is a hint, so a late stream within the grace is still delivered", async () => { + const { subscriber, reader, done } = await subscribed(); + const started = performance.now(); + await done(TRACK_ENDED, 0n); + + const late = groupStream(subscriber, 0); + late.write(object("0.0")); + late.finish(); + + expect(await readAll(await reader.recvGroup())).toEqual(["0.0"]); + expect(await reader.recvGroup()).toBeUndefined(); + expect(await reader.closed).toBeNull(); + expect(performance.now() - started).toBeGreaterThanOrEqual(TAIL_GRACE_MS - 5); +}); + +test("a PUBLISH_DONE with an error status aborts the track", async () => { + const { reader, done } = await subscribed(); + await done(INTERNAL_ERROR, 0n); + expect(await reader.closed).toBeInstanceOf(Error); +}); + +test("END_OF_TRACK after a group's last object ends the track after that group", async () => { + const { subscriber, reader, done } = await subscribed(); + const group = groupStream(subscriber, 4); + group.write(object("4.0")); + group.write(END_OF_TRACK); + group.finish(); + + expect(await reader.finished()).toBe(5); + expect(await readAll(await reader.recvGroup())).toEqual(["4.0"]); + + await done(TRACK_ENDED, 1n); + expect(await reader.recvGroup()).toBeUndefined(); + expect(reader.final()).toBe(5); +}); + +test("END_OF_TRACK at object 0 ends the track before its group, which never exists", async () => { + const { subscriber, reader, done } = await subscribed(); + const group = groupStream(subscriber, 0); + group.write(object("0.0")); + group.finish(); + const end = groupStream(subscriber, 2); + end.write(END_OF_TRACK); + end.finish(); + + expect(await reader.finished()).toBe(2); + await done(TRACK_ENDED, 2n); + expect((await reader.recvGroup())?.sequence).toBe(0); + expect(await reader.recvGroup()).toBeUndefined(); + expect(reader.final()).toBe(2); +}); diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index d8092a083f..30b1f5bbce 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -1325,22 +1325,27 @@ test("lite draft-05: a group waiting for a stream slot is dropped when the subsc // The publisher FINs the subscribe stream itself once a track ends, which must not be // mistaken for the subscriber leaving: SUBSCRIBE_END counts those queued groups as -// delivered, so dropping them here would strand the tail of every finite track. +// delivered, so dropping them here would strand the tail of every finite track. The FIN +// tells the subscriber every group is accounted for, so it waits for the queued group. test("lite draft-05: a group waiting for a stream slot survives the track finishing", async () => { const { client, track, freeSlot, outcome, close } = await saturatedGroup(); track.close(); - // Read to the FIN the publisher sends after SUBSCRIBE_END. That FIN is the moment a - // cancel keyed on our own close would fire, so the slot must not free up before it. + // SUBSCRIBE_END goes out while the group is still waiting for its slot. for (;;) { const resp = await decodeSubscribeResponse(client.reader, Version.DRAFT_05); if ("end" in resp) break; } - await client.reader.closed; + + // The FIN holds until the queued group is on the wire. + const fin = client.reader.closed.then(() => "fin" as const); + const idle = new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 20)); + expect(await Promise.race([fin, idle])).toBe("pending"); freeSlot(); expect(await outcome).toBe("sent"); + expect(await fin).toBe("fin"); close(); }); diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index bff615b0e9..b9fb8dc098 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -245,6 +245,11 @@ class SubscriptionControls { return false; } + /** Settles once the stream is over: `null` when it ended cleanly, or the failure. */ + get ended(): Promise { + return this.#ended; + } + #finish(end: Error | null) { if (this.#end !== undefined) return; this.#end = end; @@ -755,6 +760,9 @@ export class Publisher { // One ranking for the whole subscription, shared by every group it serves. const priority = new Priority(track); + // Every group this subscription started serving, until its stream finishes or resets. + const groups = new Set>(); + // Cancels groups still queued for a stream slot. Only the subscriber leaving counts: // a track that ran out of groups still has to flush the ones already queued, and the // caller FINs the subscribe stream to say so. @@ -802,6 +810,12 @@ export class Publisher { case "error": throw recv.error; case "idle": + // An end declared ahead of the live edge goes out as soon as it is + // known, while the remaining groups are still being produced. + if (!endSent && track.final() !== undefined) { + if (!(await sendEnd())) return; + continue; + } await waitForSubscription(controls, track); continue; case "boundary": @@ -813,13 +827,21 @@ export class Publisher { } await waitForSubscription(controls, track); continue; - case "done": + case "done": { if (!endSent) { if (!(await sendEnd())) return; continue; } + // The FIN tells the subscriber every group is accounted for, so it waits + // until each group stream finished or reset. The subscriber leaving + // instead cancels whatever is still queued. + 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; finished = true; return; + } } const group = recv.group; @@ -846,7 +868,7 @@ export class Publisher { return; } - void this.#runGroup({ + const task = this.#runGroup({ sub, group, timescale, @@ -855,6 +877,8 @@ export class Publisher { start: range.start, end: range.end, }); + groups.add(task); + void task.finally(() => groups.delete(task)); } } finally { if (!finished) unsubscribe(); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 67b4c34240..6b6f97f18b 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -9,6 +9,7 @@ import { Cost, type Hop, MAX_HOPS, type Route, routesEqual, UNKNOWN_HOP } from " import { groupBounds, hiddenBelow, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts"; import * as Path from "../path.ts"; import { type Reader, Stream } from "../stream.ts"; +import { TAIL_GRACE_MS, Tail } from "../tail.ts"; import * as Time from "../time.ts"; import type * as track from "../track.ts"; import { TimeoutError, withTimeout } from "../util/timeout.ts"; @@ -69,6 +70,13 @@ interface SubscribeEntry { // runGroup must consume to stay in sync; group streams block on it before decoding, // since a group's QUIC stream can race ahead of the subscribe stream. timescale: Signal; + // The group streams received, so the subscription can wait for the ones still owed + // after the publisher ends it. + tail: Tail; + // The first group the publisher serves (SUBSCRIBE_START) and the track's exclusive end + // (SUBSCRIBE_END), once it declares them. + start?: number; + end?: number; } /** @@ -531,7 +539,7 @@ export class Subscriber { const state: { stream?: Stream } = {}; const setup = this.#openSubscribe(state, msg, request, id, timescale); - let opened: { stream: Stream; producer: track.Producer }; + let opened: { stream: Stream; entry: SubscribeEntry }; try { opened = await withTimeout( setup, @@ -556,16 +564,20 @@ export class Subscriber { return; } - const { stream, producer } = opened; + const { stream, entry } = opened; + const producer = entry.track; try { // Watch for subscription changes and send SUBSCRIBE_UPDATE. Lite01/Lite02 // 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; - // drain them (we don't drive delivery off the resolved range) so the FIN is - // observed. Older drafts just wait for the stream to close. - const closed = supportsTrackStream(this.version) ? this.#drainResponses(stream) : stream.reader.closed; + // 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. + const responses = supportsTrackStream(this.version) + ? this.#runResponses(stream, entry) + : stream.reader.closed; + const closed = responses.then(() => this.#settleTail(entry)); const subscriptionUpdates = this.version === Version.DRAFT_01 || this.version === Version.DRAFT_02 ? undefined @@ -573,8 +585,10 @@ export class Subscriber { // Terminal conditions (stream end, track close, a failed subscription update) settle at most // once; race them into one stable promise so the demand loop doesn't re-subscribe each pass. + // Updates stop quietly at the FIN, which can land before the responses ahead of it are + // decoded, so only their failure is terminal on its own. const terminal: PromiseLike[] = [closed, producer.closed]; - if (subscriptionUpdates !== undefined) terminal.push(subscriptionUpdates); + if (subscriptionUpdates !== undefined) terminal.push(subscriptionUpdates.then(() => closed)); const done = race(terminal); // Serve until a terminal condition fires or the last local subscriber leaves. The unused @@ -615,7 +629,7 @@ export class Subscriber { request: track.Request, id: bigint, timescale: Signal, - ): Promise<{ stream: Stream; producer: track.Producer }> { + ): Promise<{ stream: Stream; entry: SubscribeEntry }> { let producer: track.Producer; let drainOk = false; @@ -632,7 +646,8 @@ export class Subscriber { } // Register before opening SUBSCRIBE so a racing GROUP stream finds the entry. - this.#subscribes.set(id, { track: producer, timescale }); + const entry: SubscribeEntry = { track: producer, timescale, tail: new Tail() }; + this.#subscribes.set(id, entry); state.stream = await Stream.open(this.#quic); await state.stream.writer.u53(StreamId.Subscribe); @@ -646,7 +661,7 @@ export class Subscriber { } } - return { stream: state.stream, producer }; + return { stream: state.stream, entry }; } // Opens a TRACK stream, reads the single TRACK_INFO, and FINs. Lite-05+ only. @@ -792,21 +807,68 @@ export class Subscriber { } } - // Drains SUBSCRIBE_START/END/DROP on the subscribe stream until FIN (lite-05+). - // The resolved range is informational here; the producer already orders groups. - // Resolves (never rejects) on FIN or on the stream being reset out from under it, - // so it's safe to drop from a race without an unhandled rejection. - async #drainResponses(stream: Stream): Promise { - try { - for (;;) { - const resp = await decodeSubscribeResponseMaybe(stream.reader, this.version); - if (!resp) return; + // 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. Resolves on FIN or on the + // stream being reset out from under it; rejects only on a response that breaks the range. + async #runResponses(stream: Stream, entry: SubscribeEntry): Promise { + for (;;) { + let resp: Awaited>; + try { + resp = await decodeSubscribeResponseMaybe(stream.reader, this.version); + } catch { + // Stream closed or reset; nothing more to read. + return; + } + if (!resp) return; + + if ("start" in resp) { + entry.start = resp.start.group; + } else if ("end" in resp) { + if (entry.end !== undefined) throw new ProtocolViolation("duplicate SUBSCRIBE_END"); + entry.end = resp.end.group; + // A local close can win the race with the response; there is nothing left to end. + if (entry.track.closed.peek() !== undefined) continue; + try { + entry.track.finishAt(entry.end); + } catch (err) { + throw new ProtocolViolation(`invalid SUBSCRIBE_END: ${reason(error(err))}`); + } + } else if ("drop" in resp) { + entry.tail.account(resp.drop.start, resp.drop.end + 1); } - } catch { - // Stream closed or reset; nothing more to drain. } } + // Wait for the group streams the publisher still owes once it has ended the subscription. + // + // Its FIN says every group below the end is accounted for, but QUIC does not order streams, + // so one can still be in flight. Wait until each group from SUBSCRIBE_START to the end has + // a stream (read to its end) or a SUBSCRIBE_DROP. A group reset before its header arrived + // never shows up, so give up on missing groups after the subscription's effective max age, + // then end cleanly with them skipped like any stale group. That is a wall-clock stopgap for + // a presentation-time budget; a publisher sending SUBSCRIBE_DROP for every group it reset + // would account for them with no timer at all. + #settleTail(entry: SubscribeEntry): Promise { + const { tail, track } = entry; + // Already the smaller of the subscriber's and the track's max age. + const maxAge = track.subscription.peek()?.maxAge ?? Time.Milli.zero; + const grace = maxAge > 0 ? maxAge : TAIL_GRACE_MS; + + const complete = () => { + // Without SUBSCRIBE_END (older drafts) nothing says which groups are owed. + if (entry.end === undefined) return false; + // Without SUBSCRIBE_START the publisher served no group at all. + if (entry.start === undefined) return true; + const bounds = groupBounds(track.subscription.peek()?.groups ?? {}); + const start = Math.max(entry.start, bounds.start); + const end = bounds.end === undefined ? entry.end : Math.min(entry.end, bounds.end); + return tail.covers(start, end); + }; + + return tail.settle(complete, grace, track.closed); + } + /** * Send SUBSCRIBE_UPDATE messages whenever the track's aggregate subscription changes. * @@ -891,11 +953,13 @@ export class Subscriber { return; } - const { track, timescale } = entry; + const { track, timescale, tail } = entry; const producer = new netGroup.Producer(group.sequence); - track.writeGroup(producer); + const read = tail.open(group.sequence); try { + track.writeGroup(producer); + // Block until the timescale is known; the group's stream can arrive before // TRACK_INFO (or implicit defaults) resolves it on the subscribe stream. let scale = timescale.peek(); @@ -916,7 +980,9 @@ export class Subscriber { let prevTs = 0n; for (;;) { - const done = await race([stream.done(), track.closed, producer.closed]); + // Only the group's own stream ends it: a track that closes first has already + // closed (or aborted) this group through its cache. + const done = await race([stream.done(), producer.closed]); if (done !== false) break; let timestamp: Time.Timestamp; @@ -940,6 +1006,8 @@ export class Subscriber { const e = error(err); producer.close(e); stream.stop(e); + } finally { + read(); } } @@ -1002,6 +1070,8 @@ export class Subscriber { if (!scale) return; const timestamp = new Time.Timestamp(dg.timestamp, Time.Timescale(scale)); + // A datagram's sequence is never owed a stream, so it never holds the tail open. + entry.tail.account(dg.sequence, dg.sequence + 1); entry.track.insertDatagram(dg.sequence, timestamp, dg.payload); } diff --git a/js/net/src/lite/tail.test.ts b/js/net/src/lite/tail.test.ts new file mode 100644 index 0000000000..b74636c55c --- /dev/null +++ b/js/net/src/lite/tail.test.ts @@ -0,0 +1,219 @@ +import { expect, test } from "bun:test"; +import type { Consumer as GroupConsumer } from "../group.ts"; +import { randomHop } from "../hop.ts"; +import { createMockTransportPair } from "../mock.ts"; +import * as Path from "../path.ts"; +import { Reader, Stream } from "../stream.ts"; +import { Milli } from "../time.ts"; +import { Group as GroupMessage } from "./group.ts"; +import { StreamId } from "./stream.ts"; +import { + encodeSubscribeResponse, + Subscribe, + SubscribeDrop, + SubscribeEnd, + type SubscribeResponse, + SubscribeStart, +} from "./subscribe.ts"; +import { Subscriber } from "./subscriber.ts"; +import { TrackInfo, Track as TrackMessage } from "./track.ts"; +import { ALPN_05, Version } from "./version.ts"; + +const VERSION = Version.DRAFT_05; + +// The subscription's max age, which is also how long it waits for a group that never arrives. +const GRACE = Milli(100); + +/** One lite-05 frame: a zero timestamp delta, then the length-prefixed payload. */ +function frame(payload: string): Uint8Array { + const bytes = new TextEncoder().encode(payload); + // Every field is under 64, so each is a one-byte varint. + return new Uint8Array([0, bytes.byteLength, ...bytes]); +} + +/** A group stream the test writes by hand, handed to the subscriber as if it arrived. */ +function groupStream(subscriber: Subscriber, sequence: number) { + let controller!: ReadableStreamDefaultController; + const readable = new ReadableStream({ start: (c) => (controller = c) }); + const handled = subscriber.runGroup( + new GroupMessage({ subscribe: 0n, sequence }), + new Reader(readable, undefined, undefined), + ); + return { + write: (payload: string) => controller.enqueue(frame(payload)), + finish: () => controller.close(), + reset: () => controller.error(new Error("reset")), + handled, + }; +} + +/** + * A lite-05 subscriber with one track subscribed, whose publisher the test plays by hand: + * it answers TRACK_INFO, then writes whatever responses the test asks for on the subscribe + * stream and FINs it when told. + */ +async function subscribed(maxAge = GRACE) { + const pair = createMockTransportPair(ALPN_05); + const subscriber = new Subscriber(pair.client, VERSION, randomHop()); + const reader = subscriber.consume(Path.from("room")).track("video").subscribe({ maxAge }); + + const info = await Stream.accept(pair.server); + if (!info) throw new Error("the subscriber never asked for TRACK_INFO"); + expect(await info.reader.u53()).toBe(StreamId.Track); + await TrackMessage.decode(info.reader, VERSION); + await new TrackInfo({ maxAge: 60_000 }).encode(info.writer, VERSION); + info.close(); + + const sub = await Stream.accept(pair.server); + if (!sub) throw new Error("the subscriber never subscribed"); + expect(await sub.reader.u53()).toBe(StreamId.Subscribe); + await Subscribe.decode(sub.reader, VERSION); + + return { + subscriber, + reader, + respond: (resp: SubscribeResponse) => encodeSubscribeResponse(sub.writer, resp, VERSION), + fin: () => sub.writer.close(), + }; +} + +/** Read one group to its end, or the error it ended with. */ +async function readAll(group: GroupConsumer | undefined): Promise { + if (!group) throw new Error("no group"); + const payloads: string[] = []; + for (;;) { + const frame = await group.readString(); + if (frame === undefined) return payloads; + payloads.push(frame); + } +} + +/** Whether `promise` settles within `ms`. */ +async function settlesWithin(promise: Promise, ms: number): Promise { + let timer: ReturnType | undefined; + const pending = new Promise((resolve) => { + timer = setTimeout(() => resolve(false), ms); + }); + try { + return await Promise.race([promise.then(() => true), pending]); + } finally { + clearTimeout(timer); + } +} + +test("a group stream that arrives after the subscribe stream's FIN is delivered", async () => { + const { subscriber, reader, respond, fin } = await subscribed(); + await respond({ start: new SubscribeStart(0) }); + const first = groupStream(subscriber, 0); + first.write("0.0"); + first.finish(); + await respond({ end: new SubscribeEnd(2) }); + await fin(); + + // The end is known before the last group arrives. + expect(await reader.finished()).toBe(2); + + // QUIC does not order streams, so group 1 lands after the FIN. + const late = groupStream(subscriber, 1); + late.write("1.0"); + late.finish(); + + expect(await readAll(await reader.recvGroup())).toEqual(["0.0"]); + expect(await readAll(await reader.recvGroup())).toEqual(["1.0"]); + expect(await reader.recvGroup()).toBeUndefined(); + expect(await reader.closed).toBeNull(); + expect(reader.final()).toBe(2); +}); + +test("a group read across the subscribe stream's FIN is delivered whole", async () => { + const { subscriber, reader, respond, fin } = await subscribed(); + await respond({ start: new SubscribeStart(0) }); + const group = groupStream(subscriber, 0); + group.write("0.0"); + + await respond({ end: new SubscribeEnd(1) }); + await fin(); + const received = await reader.recvGroup(); + expect(await received?.readString()).toBe("0.0"); + + // The FIN does not end the group: only its own stream does. + group.write("0.1"); + group.finish(); + expect(await readAll(received)).toEqual(["0.1"]); + expect(await reader.recvGroup()).toBeUndefined(); + expect(await reader.closed).toBeNull(); +}); + +test("a group reset after the subscribe stream's FIN is not presented as complete", async () => { + const { subscriber, reader, respond, fin } = await subscribed(); + await respond({ start: new SubscribeStart(0) }); + const group = groupStream(subscriber, 0); + group.write("0.0"); + await respond({ end: new SubscribeEnd(1) }); + await fin(); + + const received = await reader.recvGroup(); + expect(await received?.readString()).toBe("0.0"); + group.reset(); + await expect(received?.readString() ?? Promise.resolve()).rejects.toThrow(); + + // The track still ends cleanly: the group was accounted for, as a reset. + expect(await reader.recvGroup()).toBeUndefined(); + expect(await reader.closed).toBeNull(); +}); + +test("a group that never arrives is given up on after the subscription's max age", async () => { + const { subscriber, reader, respond, fin } = await subscribed(); + await respond({ start: new SubscribeStart(0) }); + const group = groupStream(subscriber, 1); + group.write("1.0"); + group.finish(); + await respond({ end: new SubscribeEnd(2) }); + await fin(); + + // Group 0 was reset before its header arrived, so nothing ever accounts for it. + expect((await reader.recvGroup())?.sequence).toBe(1); + const started = performance.now(); + expect(await reader.recvGroup()).toBeUndefined(); + expect(performance.now() - started).toBeGreaterThanOrEqual(GRACE - 5); + expect(await reader.closed).toBeNull(); + expect(reader.final()).toBe(2); +}); + +test("a subscription ends without waiting once every group is accounted for", async () => { + // A max age far past the test's patience: only the accounting may end it. + const { subscriber, reader, respond, fin } = await subscribed(Milli(60_000)); + await respond({ start: new SubscribeStart(0) }); + await respond({ drop: new SubscribeDrop({ start: 0, end: 0, error: 0 }) }); + const group = groupStream(subscriber, 1); + group.write("1.0"); + group.finish(); + await respond({ end: new SubscribeEnd(2) }); + await fin(); + + expect((await reader.recvGroup())?.sequence).toBe(1); + expect(await settlesWithin(reader.recvGroup(), 1000)).toBe(true); + expect(await reader.closed).toBeNull(); +}); + +test("a subscription that served nothing ends at SUBSCRIBE_END", async () => { + const { reader, respond, fin } = await subscribed(Milli(60_000)); + await respond({ end: new SubscribeEnd(0) }); + await fin(); + + expect(await settlesWithin(reader.recvGroup(), 1000)).toBe(true); + expect(await reader.closed).toBeNull(); + expect(reader.final()).toBe(0); +}); + +test("a SUBSCRIBE_END below a group already received aborts the track", async () => { + const { subscriber, reader, respond } = await subscribed(); + await respond({ start: new SubscribeStart(0) }); + const group = groupStream(subscriber, 3); + group.finish(); + await group.handled; + await respond({ end: new SubscribeEnd(2) }); + + const closed = await reader.closed; + expect(closed).toBeInstanceOf(Error); +}); diff --git a/js/net/src/tail.test.ts b/js/net/src/tail.test.ts new file mode 100644 index 0000000000..9daa4eaa56 --- /dev/null +++ b/js/net/src/tail.test.ts @@ -0,0 +1,105 @@ +import { expect, test } from "bun:test"; +import { accept, connect } from "./connection/index.ts"; +import * as Ietf from "./ietf/index.ts"; +import * as Lite from "./lite/index.ts"; +import { createMockTransportPair } from "./mock.ts"; +import { Producer as OriginProducer } from "./origin.ts"; +import * as Path from "./path.ts"; +import { TAIL_GRACE_MS } from "./tail.ts"; +import { Milli } from "./time.ts"; +import type { Ordered } from "./track.ts"; +import { wireOf } from "./wire.ts"; + +const url = new URL("https://localhost:4443/test"); + +// Long enough that no group is skipped as stale, and the moq-lite grace for the one group the +// IETF case never produces. +const MAX_AGE = Milli(100); + +async function session(protocol: string) { + const pair = createMockTransportPair(protocol); + const origin = new OriginProducer(); + const [client, server] = await Promise.all([ + connect({ url, transport: pair.client }), + accept({ transport: pair.server, url, publish: origin.consume() }), + ]); + const broadcast = origin.createBroadcast(Path.from("test")); + broadcast.announce(); + const video = broadcast.createTrack("video"); + const remote = wireOf(client).consume(Path.from("test")); + const reader = remote.track("video").subscribe({ maxAge: MAX_AGE }).ordered(); + + return { + video, + reader, + close: () => { + broadcast.close(); + remote.close(); + client.close(); + server.close(); + }, + }; +} + +async function readAll(reader: Ordered): Promise { + const out: string[] = []; + for (;;) { + const next = await reader.readString(); + if (next === undefined) return out; + out.push(next); + } +} + +// moq-lite carries the end in SUBSCRIBE_END as soon as it is declared, so a subscriber +// learns it while the last group is still to come. +test.each([Lite.ALPN_05, Lite.ALPN_06])( + "%s: an end declared ahead of the live edge reaches the subscriber", + async (alpn) => { + const { video, reader, close } = await session(alpn); + try { + video.writeString("0"); + expect(await reader.readString()).toBe("0"); + video.writeString("1"); + expect(await reader.readString()).toBe("1"); + + video.finishAt(3); + expect(await reader.finished()).toBe(3); + + video.writeString("2"); + video.close(); + expect(await readAll(reader)).toEqual(["2"]); + expect(await reader.closed).toBeNull(); + expect(reader.final()).toBe(3); + } finally { + close(); + } + }, +); + +// moq-transport carries the end in an END_OF_TRACK object, so it survives a track that +// declared an end past the groups it produced, and the Stream Count lets the subscriber stop +// waiting for streams at once. +test.each([Ietf.ALPN.DRAFT_16, Ietf.ALPN.DRAFT_17, Ietf.ALPN.DRAFT_20])( + "%s: END_OF_TRACK carries the declared end", + async (alpn) => { + const { video, reader, close } = await session(alpn); + try { + video.writeString("0"); + expect(await reader.readString()).toBe("0"); + + video.finishAt(4); + video.writeString("1"); + video.writeString("2"); + const ended = performance.now(); + video.close(); + + expect(await readAll(reader)).toEqual(["1", "2"]); + expect(await reader.closed).toBeNull(); + expect(reader.final()).toBe(4); + // Every counted stream arrived, so nothing waited out the grace. + expect(performance.now() - ended).toBeLessThan(TAIL_GRACE_MS); + } finally { + close(); + } + }, +); diff --git a/js/net/src/tail.ts b/js/net/src/tail.ts new file mode 100644 index 0000000000..e516f1ea08 --- /dev/null +++ b/js/net/src/tail.ts @@ -0,0 +1,125 @@ +import { type GetPromise, Signal } from "@moq/signals"; +import { Milli } from "./time.ts"; + +/** + * How long a subscriber waits for a group stream it cannot account for once the publisher + * has ended the subscription. + * + * A group reset before its header arrived leaves no trace, and QUIC does not order streams, + * so a stream opened before the end can still be in flight after it. This bounds the wait on + * IETF, and on moq-lite when the subscription has no max age to bound it with. + */ +export const TAIL_GRACE_MS = Milli(1000); + +// setTimeout truncates a longer delay to a signed 32-bit int and fires at once. +const MAX_TIMEOUT_MS = 2 ** 31 - 1; + +/** + * The group streams a subscription has received, so its end can wait for the ones still owed. + * + * A publisher ends a subscription before every group stream it opened has necessarily + * arrived. This records which sequences are accounted for (a stream's header arrived, or + * the publisher dropped them) and how many streams are still being read, and {@link settle} + * waits on them. + * + * @internal + */ +export class Tail { + // Disjoint, sorted, exclusive-end ranges of accounted sequences. A gap splits a range, so + // this stays as small as the number of gaps rather than the number of groups. + #accounted: [number, number][] = []; + // Group streams whose header arrived, and those still being read. + #streams = 0; + #active = 0; + #changed = new Signal(0); + + /** Group streams whose header arrived, whether they finished or were reset. */ + get streams(): number { + return this.#streams; + } + + /** + * Record a group stream whose header arrived. Returns the call that marks it read to + * its end, which is idempotent. + */ + open(sequence: number): () => void { + this.#streams += 1; + this.#active += 1; + this.#account(sequence, sequence + 1); + + let closed = false; + return () => { + if (closed) return; + closed = true; + this.#active -= 1; + this.#bump(); + }; + } + + /** Record sequences `[start, end)` as accounted for without a stream: dropped, or a datagram. */ + account(start: number, end: number): void { + this.#account(start, end); + } + + /** Whether every sequence in `[start, end)` is accounted for. */ + covers(start: number, end: number): boolean { + if (start >= end) return true; + // Ranges are merged on insert, so one range covers the span or none does. + return this.#accounted.some(([lo, hi]) => lo <= start && end <= hi); + } + + /** + * Wait until every stream is read to its end and `complete()` holds, or `grace` + * milliseconds pass with nothing left being read, or `closed` settles. + * + * A stream still being read is always waited for: a group ends on its own stream's FIN + * or reset, never because its track ended. The grace only gives up on streams that never + * arrived. + */ + async settle(complete: () => boolean, grace: Milli, closed: GetPromise): Promise { + let expired = false; + const timer = setTimeout( + () => { + expired = true; + this.#bump(); + }, + Math.min(grace, MAX_TIMEOUT_MS), + ); + try { + while (closed.peek() === undefined) { + if (this.#active === 0 && (expired || complete())) return; + await Signal.race(this.#changed, closed); + } + } finally { + clearTimeout(timer); + } + } + + #account(start: number, end: number): void { + if (start >= end) return; + + const merged: [number, number][] = []; + let lo = start; + let hi = end; + let placed = false; + for (const range of this.#accounted) { + if (range[1] < lo) { + merged.push(range); + } else if (hi < range[0]) { + if (!placed) merged.push([lo, hi]); + placed = true; + merged.push(range); + } else { + lo = Math.min(lo, range[0]); + hi = Math.max(hi, range[1]); + } + } + if (!placed) merged.push([lo, hi]); + this.#accounted = merged; + this.#bump(); + } + + #bump(): void { + this.#changed.update((revision) => revision + 1); + } +} diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index b5bdcd4af1..60836121ce 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -1568,3 +1568,69 @@ test("malformed group bounds do not partially advance the cursor", () => { expect(track.tryRecvGroup()?.sequence).toBe(0); track.close(); }); + +test("finishAt refuses an end at or below a produced sequence", () => { + const producer = new TrackProducer("test"); + producer.writeGroup(new GroupProducer(5)); + + // Exclusive, so it must be above the highest produced sequence. + expect(() => producer.finishAt(5)).toThrow(); + expect(() => producer.finishAt(4)).toThrow(); + producer.finishAt(6); + + // A second end is refused, as is any group at or past the first. + expect(() => producer.finishAt(7)).toThrow(); + expect(() => producer.writeGroup(new GroupProducer(6))).toThrow(); + expect(() => producer.appendGroup()).toThrow(); + expect(() => producer.insertDatagram(6, Timestamp.now(), new Uint8Array(1))).toThrow(); + producer.close(); +}); + +test("finishAt declares an end ahead of the live edge without ending the track", async () => { + const producer = new TrackProducer("test"); + const track = producer.subscribe({ maxAge: Milli(60_000) }); + producer.writeGroup(new GroupProducer(5)); + const first = await track.recvGroup(); + expect(first?.sequence).toBe(5); + + producer.finishAt(8); + expect(track.final()).toBe(8); + expect(await track.finished()).toBe(8); + + // Not terminal: groups below the end still arrive, including a straggler. + const late = new GroupProducer(7); + late.close(); + producer.writeGroup(late); + const straggler = new GroupProducer(6); + straggler.close(); + producer.writeGroup(straggler); + expect((await track.recvGroup())?.sequence).toBe(6); + expect((await track.recvGroup())?.sequence).toBe(7); + + // A clean close keeps the declared end rather than the live edge. + producer.close(); + expect(await track.recvGroup()).toBeUndefined(); + expect(track.final()).toBe(8); + + // A late subscriber sees the same end. + expect(producer.subscribe().final()).toBe(8); +}); + +test("finished rejects when the track aborts or closes without an end", async () => { + const aborted = new TrackProducer("test"); + const pending = aborted.subscribe().finished(); + aborted.close(new Error("boom")); + await expect(pending).rejects.toThrow("boom"); + + const producer = new TrackProducer("test"); + const track = producer.subscribe(); + track.close(); + await expect(track.finished()).rejects.toThrow(); + + // A clean close is an end, so it resolves. + const clean = new TrackProducer("test"); + const reader = clean.subscribe(); + clean.appendGroup().close(); + clean.close(); + expect(await reader.finished()).toBe(1); +}); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index e94f0343fc..913d1ce70f 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -302,11 +302,12 @@ class TrackState { datagrams = new Signal([]); latest?: number; /** - * The exclusive final boundary, stamped when the producer closes cleanly: one past the - * highest sequence produced. Groups and datagrams share the namespace, so this can - * exceed `latest + 1` (which only tracks groups). Mirrors the Rust `final_sequence`. + * The exclusive final boundary, declared by {@link Producer.finishAt} or stamped by a + * clean close as one past the highest sequence produced. Groups and datagrams share the + * namespace, so this can exceed `latest + 1` (which only tracks groups). Mirrors the + * Rust `final_sequence`. */ - final?: number; + final = new Signal(undefined); closed = new Once(); update: Signal; /** Resolved once the producer commits the immutable properties. */ @@ -549,10 +550,8 @@ export class Producer { this.#prune(); for (const entry of this.#cache) this.#mirror(entry, sink); - if (closed !== undefined) { - sink.final = this.#state.final; - closeTrackState(sink, closed instanceof Error ? closed : undefined); - } + sink.final.set(this.#state.final.peek()); + if (closed !== undefined) closeTrackState(sink, closed instanceof Error ? closed : undefined); } // Recompute from every live sink because an update or close can narrow as well as widen @@ -669,11 +668,19 @@ export class Producer { this.#prune(); } - /** Append a new group with the next sequence number. */ - appendGroup(): GroupProducer { + // Refuse a write once the track is closed, or at or past its declared end. + #writable(sequence: number): void { if (this.#state.closed.peek() !== undefined) throw new Error("track is closed"); + const final = this.#state.final.peek(); + if (final !== undefined && sequence >= final) { + throw new Error(`sequence ${sequence} is at or past the track's end ${final}`); + } + } + /** Append a new group with the next sequence number. */ + appendGroup(): GroupProducer { const sequence = this.#sequence; + this.#writable(sequence.next); const group = new GroupProducer(sequence.next); sequence.next = group.sequence + 1; this.#publish(group); @@ -690,7 +697,7 @@ export class Producer { * entry is already gone, so a long-evicted sequence is accepted as new. */ writeGroup(group: GroupProducer) { - if (this.#state.closed.peek() !== undefined) throw new Error("track is closed"); + this.#writable(group.sequence); const existing = this.#cache.findIndex((entry) => entry.group.sequence === group.sequence); if (existing >= 0) { @@ -735,11 +742,11 @@ export class Producer { * relay preserving upstream numbering uses {@link insertDatagram}. */ appendDatagram(timestamp: Timestamp, payload: Uint8Array): number { - if (this.#state.closed.peek() !== undefined) throw new Error("track is closed"); - if (payload.byteLength > MAX_DATAGRAM_BYTES) throw new Error("datagram payload too large"); - const counter = this.#sequence; const sequence = counter.next; + this.#writable(sequence); + if (payload.byteLength > MAX_DATAGRAM_BYTES) throw new Error("datagram payload too large"); + counter.next = sequence + 1; this.#publishDatagram({ sequence, timestamp, payload }); return sequence; @@ -753,7 +760,7 @@ export class Producer { * apply. Most origin publishers want {@link appendDatagram} instead. */ insertDatagram(sequence: number, timestamp: Timestamp, payload: Uint8Array) { - if (this.#state.closed.peek() !== undefined) throw new Error("track is closed"); + this.#writable(sequence); if (payload.byteLength > MAX_DATAGRAM_BYTES) throw new Error("datagram payload too large"); const counter = this.#sequence; @@ -763,12 +770,41 @@ export class Producer { this.#publishDatagram({ sequence, timestamp, payload }); } - /** Close the track and every subscriber, mirroring the abort to their groups. Idempotent. */ + /** + * Declare the track's exclusive end, possibly ahead of the live edge, mirroring the Rust + * `finish_at`. + * + * `final` is the first sequence that will never be produced, so a track whose last group + * is 89 finishes at 90. Groups and datagrams below it are still accepted; anything at or + * above it is refused. Unlike {@link close} it is not terminal: call `close()` once the + * remaining groups are written. Throws if the track is closed, already has an end, or + * `final` is at or below a sequence already produced. + */ + finishAt(final: number): void { + if (this.#state.closed.peek() !== undefined) throw new Error("track is closed"); + if (!Number.isSafeInteger(final) || final < 0) throw new RangeError(`invalid track end: ${final}`); + const declared = this.#state.final.peek(); + if (declared !== undefined) throw new Error(`track already ends at ${declared}`); + if (final < this.#sequence.next) { + throw new Error(`track end ${final} is below the next sequence ${this.#sequence.next}`); + } + this.#declareFinal(final); + } + + #declareFinal(final: number): void { + this.#state.final.set(final); + for (const sink of this.#sinks) sink.final.set(final); + } + + /** + * Close the track and every subscriber, mirroring the abort to their groups. Idempotent. + * + * A clean close keeps the end {@link finishAt} declared, or declares one past the highest + * sequence produced; an abort ends without one. + */ close(abort?: Error) { - // A clean close declares the final boundary; an abort ends without one. - if (abort === undefined && this.#state.closed.peek() === undefined) { - this.#state.final = this.#sequence.next; - for (const sink of this.#sinks) sink.final = this.#state.final; + if (abort === undefined && this.#state.closed.peek() === undefined && this.#state.final.peek() === undefined) { + this.#declareFinal(this.#sequence.next); } closeTrackState(this.#state, abort); clearTimeout(this.#pruneTimer); @@ -999,13 +1035,35 @@ export class Subscriber { } /** - * The track's exclusive final boundary, known once the producer closes cleanly: - * one past the highest sequence produced, or 0 for a track that produced none. - * Groups and datagrams share the sequence namespace, so this can exceed - * `latest() + 1`. Undefined while the track is live or after an abort. + * The track's exclusive final boundary: the end {@link Producer.finishAt} declared, which + * can be ahead of the live edge, or one past the highest sequence produced once the + * producer closes cleanly (0 for a track that produced none). Groups and datagrams share + * the sequence namespace, so this can exceed `latest() + 1`. Undefined until declared, + * and after an abort that declared none. */ final(): number | undefined { - return this.#state.final; + return this.#state.final.peek(); + } + + /** + * Resolve with the track's exclusive final boundary once it is known, mirroring the Rust + * `finished`. + * + * Resolves as soon as the end is declared, which may be ahead of the live edge, so it + * says nothing about every group having arrived: read until the cursor returns + * `undefined` for that. Rejects with the abort, or if the track closes without an end. + */ + async finished(): Promise { + for (;;) { + const final = this.#state.final.peek(); + if (final !== undefined) return final; + + const closed = this.#state.closed.peek(); + if (closed instanceof Error) throw closed; + if (closed !== undefined) throw new Error("track closed before its end was known"); + + await Signal.race(this.#state.final, this.#state.closed); + } } /** @@ -1158,9 +1216,15 @@ export class Subscriber { } // Package-internal readiness half of recvGroup. Each registration fires at most once, and - // the caller disposes the losers after whichever source wakes it. + // the caller disposes the losers after whichever source wakes it. A declared end wakes it + // too, so a publisher can forward the end before the live edge reaches it. #groupChanged(fn: () => void): Dispose { - const dispose = [this.#state.groups.changed(fn), this.#cursor.changed(fn), this.#state.closed.changed(fn)]; + const dispose = [ + this.#state.groups.changed(fn), + this.#cursor.changed(fn), + this.#state.closed.changed(fn), + this.#state.final.changed(fn), + ]; return () => { for (const close of dispose) close(); }; @@ -1410,6 +1474,11 @@ export class Ordered { return this.#subscriber.final(); } + /** Resolve with the track's exclusive final boundary once known; see {@link Subscriber.finished}. */ + finished(): Promise { + return this.#subscriber.finished(); + } + /** Limit subsequent reads to these groups and return this reader for chaining. */ withGroups(groups: Groups): this { this.setGroups(groups); From 994d27b140554f6f71c1bcf5fb9868f081b5feb6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 18:55:03 -0700 Subject: [PATCH 3/3] docs: record the settled track tail choices Co-Authored-By: Claude Opus 5.5 --- doc/lib/js/net.md | 1 + quest/m1/README.md | 1 - quest/m1/js-track-tail.md | 111 -------------------------------- quest/m1/quic/reliable-reset.md | 4 +- quest/m1/rust-track-tail.md | 30 +++++---- quest/m1/session-death-error.md | 3 +- 6 files changed, 20 insertions(+), 130 deletions(-) delete mode 100644 quest/m1/js-track-tail.md diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 3b6b3c693a..f086c11cd9 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -52,6 +52,7 @@ for (;;) { - **Bandwidth** (`Bandwidth.Allocator`) divides the connection's send-rate estimate by track priority, max-min fair within a tier. An idle track claims nothing. The receive side is untouched. - **Discovery** by any pattern scope (`origin.announced(scope)`, such as `room/*/chat`; default everything). Each event's `prefix` is the covered prefix relative to the origin, `captures` reports what the scope's wildcards matched when the prefix pins them, and `kind` says whether it was announced, updated, or retracted. The consumer is an async iterable. `origin.broadcasts(scope)` is a live `Getter>` of the same covered prefixes for UIs that need the current set. A borrowed `Connection.origin` also exposes `dynamic(prefix, route)` for serving paths on demand. - **Subscriptions** carry a priority, a `Time.Milli` max age, and optional `groups` bounds. Groups arrive out of order and are read frame by frame, with `Error.TooFarBehind` when a reader asks for a frame the group never held and `Error.GroupTooLarge` when a write exceeds the cache budget and aborts the group. +- **Track ends**: `close()` ends a track at its live edge, while `finishAt(n)` declares the exclusive end ahead of it and still accepts the groups below. A subscriber reads the end with `final()` or awaits `finished()`. A remote track ends only once every group below its end has arrived or was dropped; one reset before its header arrived is skipped after the subscription's max age on moq-lite (one second without one), or after one second on IETF. - **Datagrams** on moq-lite 05+ and fetch-by-sequence for history. - **Errors** live under one namespace: a stream reset throws `Error.Stream` with a `StreamCode`, while a session close gives `Error.Session` with a `SessionCode`. The registries are disjoint, so the same number means different things in each, and 64+ is yours. Named conditions such as `Error.TooFarBehind`, `Error.FrameTooLarge`, and `Error.GroupTooLarge` subclass `Error.Stream`, so one `code` check handles a condition raised here or reported by the peer. IETF streams use their own mapping: cancellation sends CANCELLED, other local failures send INTERNAL\_ERROR, and received codes remain opaque. - **Paths** with `Path.relative` for the cross-broadcast catalog references hang uses. Path patterns (`Path.Pattern`, `Path.Patterns`) are re-exported from [`@moq/pattern`](https://www.npmjs.com/package/@moq/pattern). Literal `Path` stays a coordinate. diff --git a/quest/m1/README.md b/quest/m1/README.md index 60c9c1c16f..d4fd61560b 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -20,7 +20,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Play harness](/quest/m1/play-harness.md) - moq play's tune-in, rendition-switch, and drain logic runs in per-PR CI without a device - [Missing fetch group](/quest/m1/fetch-missing-group.md) - HTTP /fetch answers 404 and `moq fetch` fails cleanly for a group the track lacks - [libmoq hidden opt-in](/quest/m1/libmoq-hidden.md) - `moq_origin_announced` takes a `hidden` flag so C callers can list `.`-named broadcasts -- [JS track tail](/quest/m1/js-track-tail.md) - a `@moq/net` subscriber delivers every group up to the declared end over lite and IETF, and JS publishers drain their groups before ending a subscription - [lite-07 stream count](/quest/m1/lite-stream-count.md) - moq-lite-07 replaces SUBSCRIBE_DROP with a group-stream count in SUBSCRIBE_END, like moq-transport - [Rust track tail](/quest/m1/rust-track-tail.md) - a moq-net subscriber accepts groups that arrive after the subscription's end, and PublishDone carries the real stream count - [Session death error](/quest/m1/session-death-error.md) - a dying session ends its tracks with its own error in Rust and JS, never a clean end, `Dropped`, or `Cancel` diff --git a/quest/m1/js-track-tail.md b/quest/m1/js-track-tail.md deleted file mode 100644 index 8fb114d8b4..0000000000 --- a/quest/m1/js-track-tail.md +++ /dev/null @@ -1,111 +0,0 @@ -# [L] JS track tail - -## Goal - -A `@moq/net` subscriber delivers every stream-delivered group of a track up to -the publisher's declared end, over moq-lite and IETF alike, then ends cleanly. A group cut off -mid-read is never presented as complete. JS publishers finish their group -streams before they end a subscription, as the drafts require. A browser -publisher can also declare a track's end ahead of the live edge, and a browser -consumer can await it, matching Rust's `finish_at` and `finished()`. - -## Plan - -Today the tail of a track can be lost in the browser: - -- The lite subscriber discards SUBSCRIBE_END and closes the track for good - when the subscribe stream FINs. A group stream that arrives later is - dropped, and a group still being read is closed cleanly, so a truncated - group looks whole. `final()` is stamped from the highest group received, - not the declared end. The IETF subscriber does the same on PublishDone. -- The lite publisher FINs the subscribe stream while its group streams are - still being written (`void this.#runGroup`), against the moq-lite draft: - "The publisher closes the stream (FIN) only once every group from start to - end has been accounted for". The IETF publisher sends PublishDone the same - way. Rust drains its group tasks first. -- Even a compliant publisher races: QUIC does not order streams, so a group - below the boundary can arrive after the FIN. - -Build the primitive first, which absorbs #2318's remaining work: - -- `Track.Producer.finishAt(final)`, mirroring Rust's `finish_at`: the boundary - must exceed the highest produced sequence; groups below it are still - accepted and groups at or above it are refused. Unlike `close()`, it is not - terminal. -- Feed the boundary into the consumer's `final()`, so a remote clean end is - observable before the live edge reaches it. -- An awaitable `finished()` twin of `final()`, mirroring Rust: it resolves - with the boundary once known and rejects on abort. - -Then the subscribers, lite and IETF: - -- The declared end calls `finishAt`, and the subscribe stream's FIN no longer - closes the track. On moq-lite it is SUBSCRIBE_END. On IETF it depends on the - draft: draft-07's SUBSCRIBE_DONE carries a Final Group and Object, while on - drafts 14-22 PUBLISH_DONE carries no location, so the boundary comes from the - END_OF_TRACK object. Verify this against each draft. -- A PublishDone whose status is an error (INTERNAL_ERROR or similar) aborts - the track with it; only a clean status (TRACK_ENDED or equivalent) ends it - cleanly. -- Keep accepting groups below the boundary until each is accounted for: - completed, reset, dropped via SUBSCRIBE_DROP, or covered by the stream - count. Then end cleanly. -- A group reset before its header arrived can never be accounted for, so - after the boundary is known the subscriber gives up on missing groups after - a grace, then ends cleanly, skipping them like any stale group: - - moq-lite: the subscription's effective `max_age` (the smaller of the - subscriber's and the track's), used as a wall-clock duration. This is the - wrong clock on purpose, as a stopgap: `max_age` measures presentation-time - drift and elsewhere never adds wall-clock delay. Define a fallback for a - subscription and track with no `max_age`, since a zero grace reintroduces - the race. The correct fix is the publisher sending SUBSCRIBE_DROP for - every group it reset or never finished, which accounts for every group - with no timer; reliable reset is probably better still. - - IETF: a bounded wall-clock wait, since moq-transport has no per-group drop - and itself says subscribers SHOULD use a timeout here. Settle its value. - - The reliable-reset quest removes both waits once a reset group stream - keeps its header. -- The complete-tail guarantee covers groups delivered on streams only. - Datagrams are unreliable by design and a lost one leaves no signal, so the - subscriber never waits for a datagram-delivered sequence, and a - datagram-only subscription ends at the boundary at once. -- A group ends on its own stream's FIN or reset, never because its track - ended. A reset aborts the group. - -And the publishers: - -- Lite FINs the subscribe stream only after every group stream it started has - finished or been reset. -- IETF sends PublishDone after the same drain, with the real number of data - streams it opened instead of the hardcoded 0. -- IETF on drafts 14-22 writes the END_OF_TRACK object at the boundary; today - it writes only GROUP_END (`js/net/src/ietf/object.ts`), so no in-repo - subscriber could learn the end. Assert the boundary end to end. -- A received stream count is a hint: stop waiting once that many streams are - accounted for, but accept a late stream below the boundary within the grace. - A published peer's 0 then behaves like today's lite FIN plus the grace. - -Confirm Stream Count's meaning for the implemented IETF drafts (07, 14-22) -before relying on it, and bring any draft that disagrees back as a question. - -Reproduce each race deterministically before fixing it. The mock transport -(`js/net/src/mock.ts`) delivers streams in creation order, so drive the -publisher by hand: answer TRACK_INFO, write SUBSCRIBE_START, open a group -stream and write part of it, then write SUBSCRIBE_END and FIN, then finish -the group or open another one. `gateWrites` in -`js/net/src/lite/publisher.test.ts` holds a stream's writes. Cover a late -group, a mid-read group, a reset group, a missing group resolved by the -grace, and both publishers draining. - -Additive on `@moq/net`, so it targets `main`. The wire fix to the publishers -and to stream_count follows the published drafts, which already required it. - -## Closes - -- [#2318](https://github.com/moq-dev/moq/issues/2318) - close this issue when the quest finishes - -## Related - -- [Rust track tail](/quest/m1/rust-track-tail.md) - the same rule in moq-net, so local and remote readers match -- [Session death error](/quest/m1/session-death-error.md) - the other way a JS track ends wrong: cleanly instead of with the error -- [Reliable stream reset](/quest/m1/quic/reliable-reset.md) - removes the grace once a reset group stream keeps its header diff --git a/quest/m1/quic/reliable-reset.md b/quest/m1/quic/reliable-reset.md index aebca66342..ed01ee6620 100644 --- a/quest/m1/quic/reliable-reset.md +++ b/quest/m1/quic/reliable-reset.md @@ -66,8 +66,8 @@ provisional codepoints if the document changes before release. ## Related -- [JS track tail](/quest/m1/js-track-tail.md) and [Rust track tail](/quest/m1/rust-track-tail.md) - - wait a grace for a group whose reset lost its header, until this lands +- [Rust track tail](/quest/m1/rust-track-tail.md) - waits a grace for a group + whose reset lost its header until this lands, as `@moq/net` already does - [qmux on the QUIC stream state machine](/quest/m1/quic/qmux.md) - consumes the same reset state without a parallel implementation - The removed quiche backend was the one stack that had this, so it is the diff --git a/quest/m1/rust-track-tail.md b/quest/m1/rust-track-tail.md index 5af1ff663d..2da144893a 100644 --- a/quest/m1/rust-track-tail.md +++ b/quest/m1/rust-track-tail.md @@ -22,24 +22,27 @@ remaining gap is the subscriber's bookkeeping: `Error::Cancel` (`ietf/subscriber.rs`, `Alias::Retired`). Retire it only once the streams are accounted for or the grace expires. - Grace: a group reset before its header arrived can never be accounted for, - so give up after the same grace as the JS quest and end cleanly, skipping - the missing group as stale: the effective `max_age` as a wall-clock stopgap - on moq-lite (with a fallback when none is set), and a bounded wall-clock - wait on IETF. The reliable-reset quest removes both. + so give up after the same grace as `@moq/net` (`js/net/src/tail.ts`) and end + cleanly, skipping the missing group as stale: the effective `max_age` as a + wall-clock stopgap on moq-lite, 1s when it is zero, and 1s on IETF. The + reliable-reset quest removes both. - Only stream-delivered groups are waited for; datagrams are never. - IETF publisher: send the real number of data streams opened in PublishDone instead of `stream_count: 0`. On receipt, treat the count as a hint: stop waiting once that many are accounted for, but accept a late stream below the boundary within the grace, so a published peer's 0 keeps working. -- IETF publisher on drafts 14-22: write the END_OF_TRACK object at the - boundary, which moq-net does not send today, and assert the - boundary end to end. - -Confirm Stream Count's meaning for the implemented IETF drafts first, and -where each draft carries the track's end (draft-07's SUBSCRIBE_DONE Final -Group and Object, or the END_OF_TRACK object on drafts 14-22), as the JS quest -does; both must agree. A PublishDone with an error status aborts the track -rather than ending it cleanly. +- IETF on drafts 14-22: write the END_OF_TRACK object at the boundary, and + decode it. `@moq/net` already sends it on its own stream at object 0 of + the group `final`, after the group streams drain; moq-net's subscriber + rejects status 0x4 as `Unsupported` today, so it aborts a bogus group + `final` at the end of every JS-published IETF track until this lands. + +`@moq/net` settled the draft reading: on 14-22 Stream Count counts every data +stream opened, fill streams included (20+), with a 2^62-1 or 2^64-1 unknown +sentinel. PUBLISH_DONE carries no end location on any of them, so the end is +the END_OF_TRACK object: at object 0 of group G the track ends at G, +otherwise at G+1. Only TRACK_ENDED (and SUBSCRIPTION_ENDED before 20) ends a +track cleanly; an error status aborts it. Reproduce each case before fixing it: a group header decoded after the subscribe stream's FIN, and a late stream after PublishDone, over the mock @@ -53,6 +56,5 @@ in flight. ## Related -- [JS track tail](/quest/m1/js-track-tail.md) - the same rule in `@moq/net` - [Session death error](/quest/m1/session-death-error.md) - tracks ending wrong when the session dies - [Reliable stream reset](/quest/m1/quic/reliable-reset.md) - removes the grace diff --git a/quest/m1/session-death-error.md b/quest/m1/session-death-error.md index 11df708a9a..223bfbaab6 100644 --- a/quest/m1/session-death-error.md +++ b/quest/m1/session-death-error.md @@ -56,5 +56,4 @@ controls: a finished track still ends clean while its session lives. ## Related -- [JS track tail](/quest/m1/js-track-tail.md) - the clean-end half: a track that did end is delivered whole -- [Rust track tail](/quest/m1/rust-track-tail.md) - the same in moq-net +- [Rust track tail](/quest/m1/rust-track-tail.md) - the clean-end half: a track that did end is delivered whole