diff --git a/js/net/src/ietf/object.ts b/js/net/src/ietf/object.ts index 3f6f86f6a3e..af7aaaacf94 100644 --- a/js/net/src/ietf/object.ts +++ b/js/net/src/ietf/object.ts @@ -25,7 +25,8 @@ const PROP_TIMESTAMP_DRAFT03 = 0x06n; // object 0 is still the head, and any other first ID is dropped. const FIRST_OBJECT_BIT = 0x40; -function hasFirstObjectBit(version: IetfVersion): boolean { +/** Whether the subgroup header carries FIRST_OBJECT: draft-18 and later, not drafts 14-17. */ +export function hasFirstObjectBit(version: IetfVersion): boolean { switch (version) { case Version.DRAFT_14: case Version.DRAFT_15: diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index ab3048e54bc..ebd3cbc4ce3 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -15,7 +15,7 @@ import { RequestError, RequestOk } from "./request.ts"; import { Subscribe, SubscribeOk, Unsubscribe } from "./subscribe.ts"; import { SubscribeNamespace, SubscribeNamespaceEntry, SubscribeNamespaceEntryDone } from "./subscribe_namespace.ts"; import { Subscriber } from "./subscriber.ts"; -import { ALPN, Version } from "./version.ts"; +import { ALPN, type IetfVersion, Version } from "./version.ts"; const VERSION = Version.DRAFT_19; @@ -860,20 +860,24 @@ function encodeObjects(deltas: number[]): Uint8Array { * A subscriber with one track subscribed and answered, which is what registers {@link ALIAS} * and lets a group stream naming it be handled. */ -async function subscribeTrack(): Promise<{ subscriber: Subscriber; track: track.Subscriber }> { - const pair = createMockTransportPair(ALPN.DRAFT_19); - const session = new NativeSession(pair.server, VERSION, true); +async function subscribeTrack( + version: IetfVersion = VERSION, +): Promise<{ subscriber: Subscriber; track: track.Subscriber }> { + const pair = createMockTransportPair(version === Version.DRAFT_16 ? ALPN.DRAFT_16 : ALPN.DRAFT_19); + const session = new NativeSession(pair.server, version, true); const subscriber = new Subscriber({ session }); const track = subscriber.consume(Path.from("room")).track("video").subscribe(); const peer = await nextStream(pair.client); if (!peer) throw new Error("the subscriber never opened a subscribe stream"); + peer.reader.version = version; + peer.writer.version = version; expect(await peer.reader.u53()).toBe(Subscribe.id); - const request = await Subscribe.decode(peer.reader, VERSION); + 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); + await new SubscribeOk({ requestId: request.requestId, trackAlias: ALIAS }).encode(peer.writer, version); return { subscriber, track }; } @@ -1030,6 +1034,49 @@ test("a clear FIRST_OBJECT past object 0 is dropped", async () => { track.close(); }); +/** + * Drafts 14-17 have no FIRST_OBJECT bit, so a subgroup that starts at the live edge + * arrives with `firstObject` forced on and a non-zero first delta. That stream is the + * in-progress group: drop it, keep the subscription, and deliver the next group, which + * starts at object 0. A gap after an object was delivered still fails that group. + */ +test("a draft without FIRST_OBJECT drops a subgroup that starts mid-group", async () => { + const version = Version.DRAFT_16; + const { subscriber, track } = await subscribeTrack(version); + + // The header cannot say otherwise on this draft: decode reports firstObject. + const whole = groupFlags(true); + await subscriber.handleGroup( + new GroupMessage({ trackAlias: ALIAS, groupId: 3, subGroupId: 0, publisherPriority: 0, flags: whole }), + new Reader(undefined, encodeObjects([2, 0]), version), + ); + expect(track.latest()).toBeUndefined(); + expect(track.closed.peek()).toBeUndefined(); + + await subscriber.handleGroup( + new GroupMessage({ trackAlias: ALIAS, groupId: 4, subGroupId: 0, publisherPriority: 0, flags: whole }), + new Reader(undefined, encodeObjects([0, 0]), version), + ); + + const ordered = track.ordered(); + const group = await ordered.nextGroup(); + expect(group?.sequence).toBe(4); + expect(await group?.readString()).toBe("object 0"); + + await subscriber.handleGroup( + new GroupMessage({ trackAlias: ALIAS, groupId: 5, subGroupId: 0, publisherPriority: 0, flags: whole }), + new Reader(undefined, encodeObjects([0, 5]), version), + ); + const gapped = await ordered.nextGroup(); + expect(gapped?.sequence).toBe(5); + expect(await gapped?.readString()).toBe("object 0"); + await expect(gapped?.readFrameSequence()).rejects.toThrow(/object IDs must start at 0/); + expect(track.closed.peek()).toBeUndefined(); + + ordered.close(); + track.close(); +}); + /** * FIRST_OBJECT is the publisher's claim, and the object ids are what actually happened. A * peer that sets the bit and then starts at object 5 is contradicting itself, so the group diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 473af22e40b..f8784cf637c 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -17,7 +17,7 @@ import type { Session } from "./adapter.ts"; import { DuplicateTrackAlias, RetiredTrackAlias, TrackAliases } from "./aliases.ts"; import * as Cluster from "./cluster.ts"; import { requestReason, toRequestCode } from "./error.ts"; -import { Frame, type Group as GroupMessage } from "./object.ts"; +import { Frame, type Group as GroupMessage, hasFirstObjectBit } from "./object.ts"; import { fromWire, toWire } from "./priority.ts"; import { type Publish, PublishDone, PublishError, publishDoneClean } from "./publish.ts"; import { @@ -1046,18 +1046,19 @@ export class Subscriber { // or a stream with no object, has a hole at the front: drop it and pick up at // the next group. // - // 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) { + // Drafts before the bit cannot say this in the header, so the first ID is the only + // signal: a non-zero one is the same hole and is dropped the same way, but an empty + // stream there is still a group. A later gap, or a header that claimed the group + // starts at object 0, still fails it: `Frame.decode` refuses every non-zero delta. + const legacy = !hasFirstObjectBit(this.#session.version); + if (!group.flags.firstObject || legacy) { let id: bigint | undefined; try { id = await stream.peekU62(); } catch (err: unknown) { if (!(err instanceof UnexpectedEnd)) throw err; } - if (id !== 0n) { + if (id !== 0n && !(legacy && id === undefined)) { 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;