From 2b83e272772465ff19986818b0398fee15f2714b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 09:31:31 -0700 Subject: [PATCH 1/6] chore: claim JavaScript request FIN backport Co-Authored-By: GPT-6 From c082edaa6cdf24850cabcd48e256fdafdffa2daa Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 09:51:47 -0700 Subject: [PATCH 2/6] fix(net): honor requester FIN semantics on release JS Co-Authored-By: GPT-6 --- js/net/src/ietf/publish.ts | 1 + js/net/src/ietf/publisher.test.ts | 61 ++++++++++++++++++++++++ js/net/src/ietf/publisher.ts | 79 ++++++++++++++++++++++++++----- js/net/src/ietf/request_stream.ts | 23 +++++++++ 4 files changed, 153 insertions(+), 11 deletions(-) create mode 100644 js/net/src/ietf/request_stream.ts diff --git a/js/net/src/ietf/publish.ts b/js/net/src/ietf/publish.ts index 40f105338a..cb49048801 100644 --- a/js/net/src/ietf/publish.ts +++ b/js/net/src/ietf/publish.ts @@ -217,6 +217,7 @@ export class PublishError { export const PublishDoneStatus = { INTERNAL_ERROR: 0x0, TRACK_ENDED: 0x2, + UPDATE_FAILED: 0x8, /** Removed in draft-20, where 0x3 is unassigned. */ SUBSCRIPTION_ENDED: 0x3, } as const; diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 9f94be6bdc..02f0ae9d66 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -2323,3 +2323,64 @@ for (const version of [Version.DRAFT_15, Version.DRAFT_19] as const) { } } } + +for (const version of [ + Version.DRAFT_17, + Version.DRAFT_18, + Version.DRAFT_19, + Version.DRAFT_20, + Version.DRAFT_21, + Version.DRAFT_22, +]) { + test(`requester FIN is version gated for subscription ${version}`, async () => { + const pair = createMockTransportPair(ALPN.DRAFT_19); + const session = new NativeSession(pair.server, version, true); + const { pub, origin } = publisher(pair.server, { session }); + const broadcast = publish(origin, Path.from("test")); + const track = broadcast.createTrack("video"); + const client = await Stream.open(pair.client, { version }); + const server = await Stream.accept(pair.server, version); + if (!server) throw new Error("missing stream"); + const serving = pub.runSubscribe( + new Subscribe({ + requestId: 0n, + trackNamespace: Path.from("test"), + trackName: "video", + subscriberPriority: 128, + }), + server, + ); + const uni = pair.client.incomingUnidirectionalStreams.getReader(); + try { + expect(await client.reader.u53()).toBe(SubscribeOk.id); + await SubscribeOk.decode(client.reader, version); + client.writer.close(); + await client.writer.closed; + if (version <= Version.DRAFT_18) { + await serving; + expect(track.subscription.peek()).toBeUndefined(); + } else { + const group = new GroupProducer(0); + group.writeFrame({ + payload: new TextEncoder().encode("after FIN"), + timestamp: Timestamp.fromMillis(0), + }); + group.close(); + track.writeGroup(group); + const next = await uni.read(); + expect(next.done).toBeFalse(); + if (next.done) throw new Error("no group after FIN"); + const reader = new Reader(next.value, undefined, version); + expect((await GroupMessage.decode(reader, version)).groupId).toBe(0); + client.reader.stop(new Error("cancel")); + await serving; + } + } finally { + client.close(); + track.close(); + broadcast.close(); + origin.close(); + uni.releaseLock(); + } + }); +} diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 5d39644a22..4fa6aa1098 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -17,7 +17,9 @@ import * as Cluster from "./cluster.ts"; import { type RequestKind, requestReason, toRequestCode } from "./error.ts"; import { type Fetch, FetchError, FetchHeader } from "./fetch.ts"; import * as Filter from "./filter.ts"; +import * as Message from "./message.ts"; import { FetchFrame, Frame, Group as GroupMessage } from "./object.ts"; +import { Parameters } from "./parameters.ts"; import { fromWire, toWire } from "./priority.ts"; import * as Properties from "./properties.ts"; import { PublishDone, PublishDoneStatus } from "./publish.ts"; @@ -28,6 +30,7 @@ import { PublishNamespaceUpdate, } from "./publish_namespace.ts"; import { RequestError, RequestOk } from "./request.ts"; +import { cancelled, finCancels } from "./request_stream.ts"; import { type Subscribe, SubscribeError, SubscribeOk } from "./subscribe.ts"; import { type SubscribeNamespace, @@ -431,7 +434,47 @@ export class Publisher { const unsubscribed = new Promise((resolve) => { unsubscribe = resolve; }); - void stream.reader.closed.then( + const updates = (async () => { + if (version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16) + return stream.reader.closed; + for (;;) { + if (await stream.reader.done()) { + if (finCancels(version)) return; + return stream.writer.closed; + } + if ((await stream.reader.u53()) !== 0x02) throw new Error("unexpected message on subscribe stream"); + const params = await Message.decode(stream.reader, async (r) => { + await r.u62(); + if (version === Version.DRAFT_17) await r.u62(); + return Parameters.decode(r, version); + }); + const unsupported = + params.forward === false || + params.bytes.size > 0 || + params.rangeFilters || + params.trackPropertyFilter || + [...params.vars.keys()].some((key) => key !== 0x10n && key !== 0x20n); + if (unsupported) { + await stream.writer.u53(RequestError.id); + await new RequestError({ + requestId: undefined, + errorCode: toRequestCode("not_supported", "subscribe", version), + reasonPhrase: "REQUEST_UPDATE parameters not supported", + }).encode(stream.writer, version); + throw new UpdateFailed(); + } + // update() replaces every option, so carry the rest over as Rust does. + if (params.subscriberPriority !== undefined) + track.update({ ...track.subscription.peek(), priority: fromWire(params.subscriberPriority) }); + await stream.writer.u53(RequestOk.id); + await new RequestOk({ requestId: undefined }).encode(stream.writer, version); + } + })(); + const requestEnded = + version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16 + ? updates + : race([updates, stream.writer.closed]); + void requestEnded.then( () => { if (!finished) unsubscribe(); }, @@ -444,7 +487,7 @@ export class Publisher { const groups = new Set>(); const streams: StreamCount = { opened: 0 }; - // Serve track groups, racing with stream close (= Unsubscribe) + // Serve groups until the track ends or the requester cancels. const serving = (async () => { for (;;) { const group = await track.recvGroup(); @@ -495,8 +538,7 @@ export class Publisher { let ended = false; try { const served = Symbol("served"); - ended = - (await race([Promise.all([serving, filling]).then(() => served), stream.reader.closed])) === served; + ended = (await race([Promise.all([serving, filling]).then(() => served), requestEnded])) === served; } catch (err: unknown) { publishError = error(err); } @@ -533,18 +575,26 @@ export class Publisher { version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16 ? msg.requestId : undefined, - statusCode: publishError ? PublishDoneStatus.INTERNAL_ERROR : PublishDoneStatus.TRACK_ENDED, + statusCode: + publishError instanceof UpdateFailed + ? PublishDoneStatus.UPDATE_FAILED + : publishError + ? PublishDoneStatus.INTERNAL_ERROR + : PublishDoneStatus.TRACK_ENDED, streamCount: BigInt(streams.opened), - reasonPhrase: publishError ? "internal error" : "track ended", + reasonPhrase: + publishError instanceof UpdateFailed + ? "update failed" + : publishError + ? "internal error" + : "track ended", }); await done.encode(stream.writer, version); } catch { // Stream might already be closed by peer. } - // Only now is the close below ours. Claiming it any earlier would read a peer FIN - // that lands while PublishDone is still going out as our own completion, leaving - // queued groups to open for a subscriber that has already left. + // Our close follows PUBLISH_DONE; it must not cancel data still queued for delivery. finished = true; stream.close(); } catch (err: unknown) { @@ -812,6 +862,7 @@ export class Publisher { // Draft-14/15: the open PUBLISH_NAMESPACE request per advertised suffix. const requests: Requests = new Map(); + const ending = cancelled(stream, version); try { // Send OK response if (version === Version.DRAFT_14) { @@ -884,8 +935,8 @@ export class Publisher { // Wait for the next change, or for the peer to unsubscribe. const next = await (retry - ? race([changed, stream.reader.closed, retryAfter(retry).then(() => advertised)]) - : race([changed, stream.reader.closed])); + ? race([changed, ending, retryAfter(retry).then(() => advertised)]) + : race([changed, ending])); dispose(); if (!next) break; } @@ -1562,3 +1613,9 @@ function fillRange(fill: Filter.Fill, subscription: Filter.Filter, largest?: Loc until: end.object === undefined ? undefined : end.object + 1n, }; } + +class UpdateFailed extends Error { + constructor() { + super("subscription update failed"); + } +} diff --git a/js/net/src/ietf/request_stream.ts b/js/net/src/ietf/request_stream.ts new file mode 100644 index 0000000000..85fcab4efc --- /dev/null +++ b/js/net/src/ietf/request_stream.ts @@ -0,0 +1,23 @@ +import { race } from "@moq/signals"; +import type { Stream } from "../stream.ts"; +import { type IetfVersion, Version } from "./version.ts"; + +/** Whether the draft treats the requester finishing its messages as cancellation. */ +export function finCancels(version: IetfVersion): boolean { + return ( + version === Version.DRAFT_14 || + version === Version.DRAFT_15 || + version === Version.DRAFT_16 || + version === Version.DRAFT_17 || + version === Version.DRAFT_18 + ); +} + +/** Wait for request cancellation, including STOP_SENDING after a clean requester FIN. */ +export async function cancelled(stream: Stream, version: IetfVersion): Promise { + const read = stream.reader.closed.then(() => (finCancels(version) ? undefined : stream.writer.closed)); + // Legacy virtual response writers finish while the adapted control request lives. + if (version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16) + return read.catch(() => undefined); + await race([read, stream.writer.closed]).catch(() => undefined); +} From 8fc14fe87a7a74db94f8bbc6554aabcd674c383a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 16:32:05 -0700 Subject: [PATCH 3/6] fix(net): retain unsupported subscription update failures Co-Authored-By: GPT-6 Astra --- js/net/src/ietf/publisher.test.ts | 94 ++++++++++++++++++++++++++++++- js/net/src/ietf/publisher.ts | 11 +++- 2 files changed, 101 insertions(+), 4 deletions(-) diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 02f0ae9d66..fa40b38ed8 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -13,9 +13,12 @@ import type { Producer as TrackProducer } from "../track.ts"; import { wireOf } from "../wire.ts"; import { NativeSession, type Session } from "./adapter.ts"; import type * as Cluster from "./cluster.ts"; +import { toRequestCode } from "./error.ts"; import { Fetch, FetchHeader } from "./fetch.ts"; +import * as Message from "./message.ts"; import { Frame, Group as GroupMessage } from "./object.ts"; -import { PublishDone } from "./publish.ts"; +import { Parameters } from "./parameters.ts"; +import { PublishDone, PublishDoneStatus } from "./publish.ts"; import { PublishNamespace, PublishNamespaceUpdate } from "./publish_namespace.ts"; import { Publisher } from "./publisher.ts"; import { RequestError, RequestOk } from "./request.ts"; @@ -2384,3 +2387,92 @@ for (const version of [ } }); } + +/** Send an update on an established subscription without closing the requester. */ +async function sendSubscriptionUpdate(client: Stream, params: Parameters, version: IetfVersion) { + await client.writer.u53(0x02); + await Message.encode(client.writer, async (writer) => { + await writer.u62(1n); + if (version === Version.DRAFT_17) await writer.u62(7n); + await params.encode(writer, version); + }); +} + +for (const version of [ + Version.DRAFT_17, + Version.DRAFT_18, + Version.DRAFT_19, + Version.DRAFT_20, + Version.DRAFT_21, + Version.DRAFT_22, +]) { + for (const field of ["filter", "largest"] as const) { + test(`unsupported subscription update ${field} is refused in ${version}`, async () => { + const fx = fixture(version); + const track = fx.broadcast.createTrack("video"); + const { client } = await runSubscribe( + fx, + new Subscribe({ + requestId: 7n, + trackNamespace: Path.from("test"), + trackName: "video", + subscriberPriority: 0, + }), + ); + try { + const params = new Parameters(); + if (field === "filter") params.subscriptionFilter = { kind: "nextObject" }; + else params.largest = { groupId: 1n, objectId: 0n }; + await sendSubscriptionUpdate(client, params, version); + expect(await client.reader.u53()).toBe(RequestError.id); + expect((await RequestError.decode(client.reader, version)).errorCode).toBe( + toRequestCode("not_supported", "subscribe", version), + ); + expect(await client.reader.u53()).toBe(PublishDone.id); + expect((await PublishDone.decode(client.reader, version)).statusCode).toBe( + PublishDoneStatus.UPDATE_FAILED, + ); + } finally { + track.close(); + client.close(); + fx.close(); + } + }); + } +} + +test("unsupported subscription update remains UPDATE_FAILED while the last group drains", async () => { + const fx = fixture(); + const track = fx.broadcast.createTrack("video"); + const queued = Promise.withResolvers(); + const slot = Promise.withResolvers(); + const create = fx.pair.server.createUnidirectionalStream.bind(fx.pair.server); + fx.pair.server.createUnidirectionalStream = async (options?: WebTransportSendStreamOptions) => { + queued.resolve(); + await slot.promise; + return create(options); + }; + const { client } = await runSubscribe( + fx, + new Subscribe({ requestId: 7n, trackNamespace: Path.from("test"), trackName: "video", subscriberPriority: 0 }), + ); + try { + writeGroup(track, 1); + await queued.promise; + track.close(); + const params = new Parameters(); + params.forward = false; + await sendSubscriptionUpdate(client, params, V20); + expect(await client.reader.u53()).toBe(RequestError.id); + expect((await RequestError.decode(client.reader, V20)).errorCode).toBe( + toRequestCode("not_supported", "subscribe", V20), + ); + expect(await client.reader.u53()).toBe(PublishDone.id); + expect((await PublishDone.decode(client.reader, V20)).statusCode).toBe(PublishDoneStatus.UPDATE_FAILED); + } finally { + slot.resolve(); + track.close(); + client.close(); + fx.close(); + } +}); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 4fa6aa1098..d073299ba6 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -451,6 +451,8 @@ export class Publisher { const unsupported = params.forward === false || params.bytes.size > 0 || + params.subscriptionFilter !== undefined || + params.largest !== undefined || params.rangeFilters || params.trackPropertyFilter || [...params.vars.keys()].some((key) => key !== 0x10n && key !== 0x20n); @@ -474,12 +476,16 @@ export class Publisher { version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16 ? updates : race([updates, stream.writer.closed]); + let publishError: Error | undefined; void requestEnded.then( () => { if (!finished) unsubscribe(); }, - // A reset is always the peer. - () => unsubscribe(), + // Updates can fail after the serving loop ends while its last groups drain. + (err: unknown) => { + publishError = error(err); + unsubscribe(); + }, ); // Every group started, until its stream finishes or resets, and the data streams @@ -534,7 +540,6 @@ export class Publisher { }) : Promise.resolve(); - let publishError: Error | undefined; let ended = false; try { const served = Symbol("served"); From e8941e03e4da3ed68122163b4fb2a167f0db6f24 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 08:24:10 -0700 Subject: [PATCH 4/6] fix(net): serialize subscription update replies with PUBLISH_DONE REQUEST_OK and REQUEST_ERROR come from the update loop while PUBLISH_DONE comes from the serving task, and each message is written in several pieces. Chain the writes so each message goes out whole and in order, nothing follows PUBLISH_DONE, and an update refused ahead of it still reports UPDATE_FAILED. Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/publisher.test.ts | 47 ++++++++++++++++++ js/net/src/ietf/publisher.ts | 82 ++++++++++++++++++++----------- 2 files changed, 99 insertions(+), 30 deletions(-) diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 301d5943e2..b46281ba1e 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -2477,6 +2477,53 @@ test("unsupported subscription update remains UPDATE_FAILED while the last group } }); +test("PUBLISH_DONE waits for an update reply that is partly written", async () => { + const fx = fixture(); + const track = fx.broadcast.createTrack("video"); + const client = await Stream.open(fx.pair.client, { version: V20 }); + const server = await Stream.accept(fx.pair.server, V20); + if (!server) throw new Error("missing stream"); + const serving = fx.pub.runSubscribe( + new Subscribe({ requestId: 7n, trackNamespace: Path.from("test"), trackName: "video", subscriberPriority: 0 }), + server, + ); + try { + expect(await client.reader.u53()).toBe(SubscribeOk.id); + await SubscribeOk.decode(client.reader, V20); + + // Hold the REQUEST_OK after its type, before its length and body. + const paused = Promise.withResolvers(); + const resume = Promise.withResolvers(); + const write = server.writer.write.bind(server.writer); + let writes = 0; + server.writer.write = async (v: Uint8Array) => { + if (++writes === 2) { + paused.resolve(); + await resume.promise; + } + await write(v); + }; + + const params = new Parameters(); + params.subscriberPriority = 1; + await sendSubscriptionUpdate(client, params, V20); + await paused.promise; + track.close(); + await flush(); + resume.resolve(); + + expect(await client.reader.u53()).toBe(RequestOk.id); + await RequestOk.decode(client.reader, V20); + expect(await client.reader.u53()).toBe(PublishDone.id); + expect((await PublishDone.decode(client.reader, V20)).statusCode).toBe(PublishDoneStatus.TRACK_ENDED); + await serving; + } finally { + track.close(); + client.close(); + fx.close(); + } +}); + for (const version of [ Version.DRAFT_16, Version.DRAFT_17, diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 3cf36afb50..37da399272 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -434,6 +434,19 @@ export class Publisher { const unsubscribed = new Promise((resolve) => { unsubscribe = resolve; }); + + // Update replies and PUBLISH_DONE come from separate tasks but share this stream, and a + // message is written in several pieces. Chain them so each goes out whole, in order, + // and nothing follows PUBLISH_DONE. + let replies: Promise = Promise.resolve(); + let replied = false; + const reply = (write: () => Promise): Promise => { + const sent = replies.then(() => (replied ? undefined : write())); + replies = sent.catch(() => undefined); + return sent; + }; + let publishError: Error | undefined; + const updates = (async () => { if (version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16) return stream.reader.closed; @@ -457,26 +470,32 @@ export class Publisher { params.trackPropertyFilter || [...params.vars.keys()].some((key) => key !== 0x10n && key !== 0x20n); if (unsupported) { - await stream.writer.u53(RequestError.id); - await new RequestError({ - requestId: undefined, - errorCode: toRequestCode("not_supported", "subscribe", version), - reasonPhrase: "REQUEST_UPDATE parameters not supported", - }).encode(stream.writer, version); - throw new UpdateFailed(); + // Recorded before the reply, so a PUBLISH_DONE queued behind it reports the failure. + const failed = new UpdateFailed(); + publishError = failed; + await reply(async () => { + await stream.writer.u53(RequestError.id); + await new RequestError({ + requestId: undefined, + errorCode: toRequestCode("not_supported", "subscribe", version), + reasonPhrase: "REQUEST_UPDATE parameters not supported", + }).encode(stream.writer, version); + }); + throw failed; } // update() replaces every option, so carry the rest over as Rust does. if (params.subscriberPriority !== undefined) track.update({ ...track.subscription.peek(), priority: fromWire(params.subscriberPriority) }); - await stream.writer.u53(RequestOk.id); - await new RequestOk({ requestId: undefined }).encode(stream.writer, version); + await reply(async () => { + await stream.writer.u53(RequestOk.id); + await new RequestOk({ requestId: undefined }).encode(stream.writer, version); + }); } })(); const requestEnded = version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16 ? updates : race([updates, stream.writer.closed]); - let publishError: Error | undefined; void requestEnded.then( () => { if (!finished) unsubscribe(); @@ -574,27 +593,30 @@ export class Publisher { // PUBLISH_DONE is required for every supported draft. The peer may have already // closed its side to unsubscribe, in which case there is nowhere to send it. try { - await stream.writer.u53(PublishDone.id); - const done = new PublishDone({ - requestId: - version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16 - ? msg.requestId - : undefined, - statusCode: - publishError instanceof UpdateFailed - ? PublishDoneStatus.UPDATE_FAILED - : publishError - ? PublishDoneStatus.INTERNAL_ERROR - : PublishDoneStatus.TRACK_ENDED, - streamCount: BigInt(streams.opened), - reasonPhrase: - publishError instanceof UpdateFailed - ? "update failed" - : publishError - ? "internal error" - : "track ended", + await reply(async () => { + replied = true; + await stream.writer.u53(PublishDone.id); + const done = new PublishDone({ + requestId: + version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16 + ? msg.requestId + : undefined, + statusCode: + publishError instanceof UpdateFailed + ? PublishDoneStatus.UPDATE_FAILED + : publishError + ? PublishDoneStatus.INTERNAL_ERROR + : PublishDoneStatus.TRACK_ENDED, + streamCount: BigInt(streams.opened), + reasonPhrase: + publishError instanceof UpdateFailed + ? "update failed" + : publishError + ? "internal error" + : "track ended", + }); + await done.encode(stream.writer, version); }); - await done.encode(stream.writer, version); } catch { // Stream might already be closed by peer. } From 1fc8cc0ffd7316e2b457d6bce741e13a9f4e5e02 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 08:38:23 -0700 Subject: [PATCH 5/6] fix(net): refuse token updates and reset groups a refused update ends A REQUEST_UPDATE carrying AUTHORIZATION_TOKEN decoded into the repeatable parameter map, so the update loop acknowledged it; refuse it like every other non-priority update, as Rust does. A refused update also left open group streams to finish cleanly after PUBLISH_DONE, claiming a complete group; abort them with the failure and wait for them before PUBLISH_DONE. Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/parameters.ts | 5 ++++ js/net/src/ietf/publisher.test.ts | 47 +++++++++++++++++++++++++++++-- js/net/src/ietf/publisher.ts | 8 ++++++ 3 files changed, 58 insertions(+), 2 deletions(-) diff --git a/js/net/src/ietf/parameters.ts b/js/net/src/ietf/parameters.ts index 1d316a048e..bda886f1c7 100644 --- a/js/net/src/ietf/parameters.ts +++ b/js/net/src/ietf/parameters.ts @@ -332,6 +332,11 @@ export class Parameters { return this.#repeated.has(MSG_PARAM_TRACK_PROPERTY_FILTER); } + /** Whether the message carried an AUTHORIZATION_TOKEN. */ + get authorizationToken(): boolean { + return this.#repeated.has(MSG_PARAM_AUTHORIZATION_TOKEN); + } + // --- Numeric accessors --- get subscriberPriority(): number | undefined { diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index b46281ba1e..3d1341cf02 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -2406,7 +2406,7 @@ for (const version of [ Version.DRAFT_21, Version.DRAFT_22, ]) { - for (const field of ["filter", "largest"] as const) { + for (const field of ["filter", "largest", "token"] as const) { test(`unsupported subscription update ${field} is refused in ${version}`, async () => { const fx = fixture(version); const track = fx.broadcast.createTrack("video"); @@ -2422,7 +2422,9 @@ for (const version of [ try { const params = new Parameters(); if (field === "filter") params.subscriptionFilter = { kind: "nextObject" }; - else params.largest = { groupId: 1n, objectId: 0n }; + else if (field === "largest") params.largest = { groupId: 1n, objectId: 0n }; + // AUTHORIZATION_TOKEN, which only decodes as a repeatable parameter. + else params.bytes.set(0x03n, new Uint8Array([1])); await sendSubscriptionUpdate(client, params, version); expect(await client.reader.u53()).toBe(RequestError.id); expect((await RequestError.decode(client.reader, version)).errorCode).toBe( @@ -2524,6 +2526,47 @@ test("PUBLISH_DONE waits for an update reply that is partly written", async () = } }); +test("a refused update resets an open group instead of finishing it", async () => { + const fx = fixture(); + const track = fx.broadcast.createTrack("video"); + const { client } = await runSubscribe( + fx, + new Subscribe({ requestId: 7n, trackNamespace: Path.from("test"), trackName: "video", subscriberPriority: 0 }), + ); + try { + const group = track.appendGroup(); + group.writeFrame({ payload: new TextEncoder().encode("a"), timestamp: Timestamp.now() }); + const uni = await fx.uni.read(); + if (uni.done) throw new Error("no group"); + const raw = uni.value.getReader(); + await raw.read(); + const params = new Parameters(); + params.forward = false; + await sendSubscriptionUpdate(client, params, V20); + expect(await client.reader.u53()).toBe(RequestError.id); + await RequestError.decode(client.reader, V20); + expect(await client.reader.u53()).toBe(PublishDone.id); + expect((await PublishDone.decode(client.reader, V20)).statusCode).toBe(PublishDoneStatus.UPDATE_FAILED); + let ended = "open"; + try { + for (;;) { + const next = await raw.read(); + if (next.done) { + ended = "fin"; + break; + } + } + } catch { + ended = "reset"; + } + expect(ended).toBe("reset"); + } finally { + track.close(); + client.close(); + fx.close(); + } +}); + for (const version of [ Version.DRAFT_16, Version.DRAFT_17, diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 37da399272..7273033527 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -468,6 +468,7 @@ export class Publisher { params.largest !== undefined || params.rangeFilters || params.trackPropertyFilter || + params.authorizationToken || [...params.vars.keys()].some((key) => key !== 0x10n && key !== 0x20n); if (unsupported) { // Recorded before the reply, so a PUBLISH_DONE queued behind it reports the failure. @@ -571,6 +572,13 @@ export class Publisher { // the draft requires, so its count is final. The subscriber leaving cancels the // ones still queued instead. await race([Promise.all(groups), unsubscribed]); + // A refused update ends the subscription on our side while the requester still + // listens, so its open groups reset rather than finish as if complete, and + // PUBLISH_DONE still follows them. + if (publishError instanceof UpdateFailed) { + track.close(publishError); + await Promise.all(groups); + } // Draft 14 on has no end location in PUBLISH_DONE: an END_OF_TRACK object is what // tells the subscriber where the track ended. From 35479b6954a9ac65643be7885ad59879c49dc500 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 08:48:31 -0700 Subject: [PATCH 6/6] fix(net): settle every stream a locally ended subscription opened A malformed update ends the subscription with an error of our own just as a refused one does, and a live fill is a data stream too. Abort and await both groups and the fill before PUBLISH_DONE whenever the error is ours, leaving the requester-cancel shortcut for stream errors. Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/publisher.test.ts | 86 ++++++++++++++++++------------- js/net/src/ietf/publisher.ts | 11 ++-- 2 files changed, 56 insertions(+), 41 deletions(-) diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 3d1341cf02..66b6b7874d 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -2526,46 +2526,60 @@ test("PUBLISH_DONE waits for an update reply that is partly written", async () = } }); -test("a refused update resets an open group instead of finishing it", async () => { - const fx = fixture(); - const track = fx.broadcast.createTrack("video"); - const { client } = await runSubscribe( - fx, - new Subscribe({ requestId: 7n, trackNamespace: Path.from("test"), trackName: "video", subscriberPriority: 0 }), - ); - try { - const group = track.appendGroup(); - group.writeFrame({ payload: new TextEncoder().encode("a"), timestamp: Timestamp.now() }); - const uni = await fx.uni.read(); - if (uni.done) throw new Error("no group"); - const raw = uni.value.getReader(); - await raw.read(); - const params = new Parameters(); - params.forward = false; - await sendSubscriptionUpdate(client, params, V20); - expect(await client.reader.u53()).toBe(RequestError.id); - await RequestError.decode(client.reader, V20); - expect(await client.reader.u53()).toBe(PublishDone.id); - expect((await PublishDone.decode(client.reader, V20)).statusCode).toBe(PublishDoneStatus.UPDATE_FAILED); - let ended = "open"; +for (const kind of ["refused", "malformed"] as const) { + test(`a ${kind} update resets an open group instead of finishing it`, async () => { + const fx = fixture(); + const track = fx.broadcast.createTrack("video"); + const { client } = await runSubscribe( + fx, + new Subscribe({ + requestId: 7n, + trackNamespace: Path.from("test"), + trackName: "video", + subscriberPriority: 0, + }), + ); try { - for (;;) { - const next = await raw.read(); - if (next.done) { - ended = "fin"; - break; + const group = track.appendGroup(); + group.writeFrame({ payload: new TextEncoder().encode("a"), timestamp: Timestamp.now() }); + const uni = await fx.uni.read(); + if (uni.done) throw new Error("no group"); + const raw = uni.value.getReader(); + await raw.read(); + if (kind === "refused") { + const params = new Parameters(); + params.forward = false; + await sendSubscriptionUpdate(client, params, V20); + expect(await client.reader.u53()).toBe(RequestError.id); + await RequestError.decode(client.reader, V20); + } else { + // Not REQUEST_UPDATE, the only message a requester may send here. + await client.writer.u53(0x03); + } + expect(await client.reader.u53()).toBe(PublishDone.id); + expect((await PublishDone.decode(client.reader, V20)).statusCode).toBe( + kind === "refused" ? PublishDoneStatus.UPDATE_FAILED : PublishDoneStatus.INTERNAL_ERROR, + ); + let ended = "open"; + try { + for (;;) { + const next = await raw.read(); + if (next.done) { + ended = "fin"; + break; + } } + } catch { + ended = "reset"; } - } catch { - ended = "reset"; + expect(ended).toBe("reset"); + } finally { + track.close(); + client.close(); + fx.close(); } - expect(ended).toBe("reset"); - } finally { - track.close(); - client.close(); - fx.close(); - } -}); + }); +} for (const version of [ Version.DRAFT_16, diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 7273033527..47c2e633f7 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -572,12 +572,13 @@ export class Publisher { // the draft requires, so its count is final. The subscriber leaving cancels the // ones still queued instead. await race([Promise.all(groups), unsubscribed]); - // A refused update ends the subscription on our side while the requester still - // listens, so its open groups reset rather than finish as if complete, and - // PUBLISH_DONE still follows them. - if (publishError instanceof UpdateFailed) { + // An error of our own (a refused or malformed update, a failed track) ends the + // subscription while the requester may still listen, so its open streams reset + // rather than finish as if complete, and PUBLISH_DONE still follows them. A stream + // error is the requester cancelling, which the shortcut above already covers. + if (publishError && !(publishError instanceof StreamError)) { track.close(publishError); - await Promise.all(groups); + await Promise.all([...groups, filling]); } // Draft 14 on has no end location in PUBLISH_DONE: an END_OF_TRACK object is what