From 030b83f99cf0b9e6d925116314b4007e7fdad588 Mon Sep 17 00:00:00 2001 From: Vojtech Zalesky Date: Tue, 4 Aug 2026 18:22:49 +0200 Subject: [PATCH] fix(webtransport): buffer track objects that arrive before their SUBSCRIBE_OK binds an alias The control stream (carrying SUBSCRIBE_OK) and unidirectional data streams (carrying Objects) are not ordered relative to each other (draft-ietf-moq-transport-16 section 10.4). A publisher can legitimately deliver a subscription data before the local subscriber has processed SUBSCRIBE_OK and bound the Track Alias in rawAliasMaps. routeToTrackSubscription previously returned false for any alias with no bound route yet, silently dropping the object. For tracks whose entire content is a single object at group 0 (MSF/CMAF catalogs, init segments) this loses the track permanently and the subscription hangs - nondeterministically, depending on relay/publisher timing. Buffer objects (and graceful subgroup-stream FINs) for an unbound alias while any subscribeTrack() call is still pending, capped at 256 per alias, and replay them in submission order once SUBSCRIBE_OK binds the alias. Buffers are cleared whenever no subscription is left pending to claim them (REQUEST_ERROR, subscribeTrack() send failure, or session teardown), so nothing lingers past its possible owner. Also adds an optional onSubgroupClosed callback to TrackSubscribeOptions - the only reliable end-of-subgroup signal available when a publisher does not set the subgroup header END_OF_GROUP flag, needed here so a graceful FIN racing SUBSCRIBE_OK can be buffered and replayed the same way as objects. Found and originally patched locally against v0.5.0 while integrating this library into a downstream player against a real relay; the race reproduces reliably enough in practice that it was carried as a submodule patch. Re-derived here against the current adapter.ts. --- packages/webtransport/src/adapter.test.ts | 49 ++++++++++++++ packages/webtransport/src/adapter.ts | 80 +++++++++++++++++++++++ 2 files changed, 129 insertions(+) diff --git a/packages/webtransport/src/adapter.test.ts b/packages/webtransport/src/adapter.test.ts index cce1b3c..50c6ae8 100644 --- a/packages/webtransport/src/adapter.test.ts +++ b/packages/webtransport/src/adapter.test.ts @@ -2623,6 +2623,55 @@ describe('MoqtConnection draft-14', () => { expect(adapter.session.state).not.toBe(SessionState.CLOSED); }); + // ─── §10.4: control and data streams are not ordered relative to each other. + // A publisher may deliver a subscription's objects before the subscriber has + // processed its own SUBSCRIBE_OK and bound the track alias. Without buffering, + // routeToTrackSubscription finds no route yet and the object is lost forever — + // fatal for tracks whose entire content is one object at group 0 (catalogs, + // CMAF init segments). ────────────────────────────────────────────────────── + it('draft-16: an object arriving before its SUBSCRIBE_OK binds the alias is buffered and delivered once bound', async () => { + const mock = createMockTransport(); + const adapter = await connectAdapter(mock); + const enc = (s: string) => new TextEncoder().encode(s); + const onObject = vi.fn(); + + const subP = adapter.subscribeTrack([enc('live')], enc('vid'), { onObject }); + await deepFlush(); + + const codec = createControlCodec(16); + let subReqId = -1n; + for (let i = mock.controlWritten.length - 1; i >= 0; i--) { + const { message } = codec.decode(mock.controlWritten[i]!, 0); + if (message.type === 'SUBSCRIBE') { subReqId = (message as { requestId: bigint }).requestId; break; } + } + expect(subReqId).not.toBe(-1n); + + const alias = 42n; + const header = makeSubgroupHeader({ trackAlias: varint(alias) }); + const obj = makeSubgroupObject({ objectId: varint(0), payload: new Uint8Array([0xAB]) }); + + // The object arrives on the data plane before SUBSCRIBE_OK is processed. + const stream = mock.addIncomingStream(); + stream.push(concat( + encodeSubgroupHeader(header), + encodeSubgroupObject(obj, header.hasExtensions, varint(0), true), + )); + await deepFlush(); + + expect(onObject).not.toHaveBeenCalled(); // buffered, not yet routable + + mock.pushControlBytes(encodeControlMessage({ + type: 'SUBSCRIBE_OK', requestId: subReqId, trackAlias: varint(alias), + parameters: new Map(), trackExtensions: [], + } as ControlMessage)); + await deepFlush(); + + const sub = await subP; + expect(sub.trackAlias).toBe(alias); + expect(onObject).toHaveBeenCalledOnce(); + expect(onObject.mock.calls[0]![0].payload).toEqual(obj.payload); + }); + // ─── §5.1: consistent crossed-response contract — a crossed REQUEST_ERROR (like // a crossed SUBSCRIBE_OK) is suppressed from the application onMessage but STILL // observed raw on the qlog channel. ────────────────────────────────────────── diff --git a/packages/webtransport/src/adapter.ts b/packages/webtransport/src/adapter.ts index ee2f02d..9314ccc 100644 --- a/packages/webtransport/src/adapter.ts +++ b/packages/webtransport/src/adapter.ts @@ -99,6 +99,12 @@ export interface TrackSubscribeOptions { readonly deliveryTimeout?: SubscribeOptions['deliveryTimeout']; /** Called for each object delivered on this subscription (stream-based only; datagrams excluded). */ onObject?: (obj: MoqtObject) => void; + /** + * Called when a subgroup data stream of this subscription ends gracefully + * (FIN). This is the only reliable end-of-subgroup signal when the + * publisher does not set the subgroup header's END_OF_GROUP flag. + */ + onSubgroupClosed?: (header: SubgroupHeader) => void; } /** @@ -114,6 +120,8 @@ export interface TrackSubscription { readonly trackAlias: bigint; /** Called for each object — mutable, read live on each delivery. */ onObject: ((obj: MoqtObject) => void) | null; + /** Called on graceful FIN of a subgroup data stream — mutable, read live. */ + onSubgroupClosed: ((header: SubgroupHeader) => void) | null; /** Unsubscribe and clean up. */ unsubscribe(): Promise; } @@ -323,6 +331,17 @@ export class MoqtConnection { private rawSubscriptions = new Map(); /** Track subscriptions by trackAlias (active, alias resolved). */ private rawAliasMaps = new Map(); + /** + * Objects that arrived on a data stream before the SUBSCRIBE_OK that binds + * their track alias (control and data streams are not ordered relative to + * each other, so a fast publisher can deliver before the subscriber has + * processed its own SUBSCRIBE_OK). Buffered while ANY subscribeTrack() is + * still pending, replayed once the matching alias binds. + */ + private pendingAliasObjects = new Map(); + private static readonly MAX_PENDING_OBJECTS_PER_ALIAS = 256; + /** Subgroup stream FINs racing SUBSCRIBE_OK, replayed after the buffered objects. */ + private pendingAliasCloses = new Map(); /** Inbound PUBLISH (draft-18 §10.10) stream contexts, keyed by Request ID. */ private inboundRequestContexts = new Map(); @@ -1730,6 +1749,8 @@ export class MoqtConnection { } this.rawSubscriptions.clear(); this.rawAliasMaps.clear(); + this.pendingAliasObjects.clear(); + this.pendingAliasCloses.clear(); } /** @@ -1929,6 +1950,7 @@ export class MoqtConnection { requestId: reqIdBigint, trackAlias: 0n, // placeholder, updated when SUBSCRIBE_OK arrives onObject: options?.onObject ?? null, + onSubgroupClosed: options?.onSubgroupClosed ?? null, unsubscribe: async () => { // Delegate to the single centralized path: it arms terminal alias // protection synchronously (from the raw entry's real alias — null-safe, @@ -1951,6 +1973,10 @@ export class MoqtConnection { const raw = this.rawSubscriptions.get(reqIdBigint); if (raw) { this.rawSubscriptions.delete(reqIdBigint); + if (!this.hasPendingRawSubscription()) { + this.pendingAliasObjects.clear(); + this.pendingAliasCloses.clear(); + } raw.reject?.(err instanceof Error ? err : new Error(String(err))); } } @@ -1975,9 +2001,48 @@ export class MoqtConnection { pub.onObject?.(obj); return true; } + // The publisher may deliver objects before its SUBSCRIBE_OK is processed + // (control and data streams are not ordered relative to each other, §10.4); + // dropping them would permanently starve the subscription of anything it + // published up front (init segments, catalogs). We cannot yet know WHICH + // pending subscribeTrack() this alias belongs to, so buffer tentatively + // while any is pending and replay once the alias binds (or is cleared once + // no subscription is left pending to claim it). + if (this.hasPendingRawSubscription()) { + const pending = this.pendingAliasObjects.get(alias) ?? []; + if (pending.length < MoqtConnection.MAX_PENDING_OBJECTS_PER_ALIAS) { + pending.push(obj); + this.pendingAliasObjects.set(alias, pending); + } + return true; + } + return false; + } + + private hasPendingRawSubscription(): boolean { + for (const raw of this.rawSubscriptions.values()) { + if (raw.resolve !== null) return true; + } return false; } + /** Notify the owning subscription that one of its subgroup streams ended gracefully. */ + private routeSubgroupClosed(header: SubgroupHeader): void { + const alias = BigInt(header.trackAlias); + const rawSub = this.rawAliasMaps.get(alias); + if (rawSub) { + rawSub.sub.onSubgroupClosed?.(header); + return; + } + if (this.hasPendingRawSubscription()) { + const pending = this.pendingAliasCloses.get(alias) ?? []; + if (pending.length < MoqtConnection.MAX_PENDING_OBJECTS_PER_ALIAS) { + pending.push(header); + this.pendingAliasCloses.set(alias, pending); + } + } + } + // ── receiver §10.11 terminal tracker (bounded, Stream-Count-driven) ── /** STOP_SENDING one incoming subgroup stream and drop its receiver state. */ @@ -2196,6 +2261,16 @@ export class MoqtConnection { raw.trackAlias = alias; (raw.sub as { trackAlias: bigint }).trackAlias = alias; this.rawAliasMaps.set(alias, raw); + const buffered = this.pendingAliasObjects.get(alias); + if (buffered) { + this.pendingAliasObjects.delete(alias); + for (const obj of buffered) raw.sub.onObject?.(obj); + } + const bufferedCloses = this.pendingAliasCloses.get(alias); + if (bufferedCloses) { + this.pendingAliasCloses.delete(alias); + for (const closedHeader of bufferedCloses) raw.sub.onSubgroupClosed?.(closedHeader); + } // §8/§10.11: record the EFFECTIVE delivery timeout for this alias so a later // teardown's guard outlives the window in which the publisher may still // deliver old-alias streams — combining the publisher's Track Properties @@ -2225,6 +2300,10 @@ export class MoqtConnection { raw.resolve = null; raw.reject = null; this.rawSubscriptions.delete(reqId); + if (!this.hasPendingRawSubscription()) { + this.pendingAliasObjects.clear(); + this.pendingAliasCloses.clear(); + } return true; } } @@ -5630,6 +5709,7 @@ export class MoqtConnection { this.onObject?.(streamId, eogGap); } } + this.routeSubgroupClosed(header); return; } buf = this.appendBuffer(buf, value);