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/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 8630ac1c2c..66b6b7874d 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"; @@ -2324,6 +2327,260 @@ 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(); + } + }); +} + +/** 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", "token"] 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 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( + 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(); + } +}); + +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 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 { + 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"; + } + 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 e3789a1d16..47c2e633f7 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,12 +434,78 @@ export class Publisher { const unsubscribed = new Promise((resolve) => { unsubscribe = resolve; }); - void stream.reader.closed.then( + + // 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; + 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.subscriptionFilter !== undefined || + 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. + 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 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]); + 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 @@ -444,7 +513,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(); @@ -491,12 +560,10 @@ export class Publisher { }) : Promise.resolve(); - let publishError: Error | undefined; 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); } @@ -505,6 +572,14 @@ 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]); + // 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, filling]); + } // Draft 14 on has no end location in PUBLISH_DONE: an END_OF_TRACK object is what // tells the subscriber where the track ended. @@ -527,24 +602,35 @@ 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 ? PublishDoneStatus.INTERNAL_ERROR : PublishDoneStatus.TRACK_ENDED, - streamCount: BigInt(streams.opened), - reasonPhrase: 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. } - // 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) { @@ -809,6 +895,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) { @@ -880,8 +967,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; } @@ -1558,3 +1645,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); +}