diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 670612e71b6..69801db63ee 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -28,7 +28,7 @@ subset drops. What a peer actually observes against this implementation: - **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined, unless they answer our `SUBSCRIBE_TRACKS` (below); announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited. A client asks only for the prefixes its readers watch, and a server or relay link for every prefix it may discover. We never hold two overlapping requests at a peer that did not declare solicit: a broader one replaces the narrower ones once they are closed. The solicit `SETUP` option makes us wait to be asked. Without it, a peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We send `PUBLISH` only to answer a `SUBSCRIBE_TRACKS` (below). - **`SUBSCRIBE_TRACKS` lists, it never pushes.** It lists every track of every broadcast under its prefix, later ones included. With `FORWARD=0`, each track gets a paused `PUBLISH` carrying its Track Properties, and a `PUBLISH_DONE` once the track is no longer listed: `TRACK_ENDED` if it finished, even just before its broadcast ended, and an error otherwise, since the track may go on elsewhere. A `REQUEST_UPDATE` with `FORWARD=1` resumes one, served from the cache like a `SUBSCRIBE` with the update's filter and answered with `LARGEST_OBJECT`; `FORWARD=0` pauses it again. Each paused `PUBLISH` opens only as the peer's stream credit admits it, so a peer that grants none holds none. A listing that omits `FORWARD` or sets it to 1 asks for every track's objects: on draft 22 each track gets `PUBLISH_SKIPPED`, and the subscriber `SUBSCRIBE`s to what it wants; draft 18 has no such skip, so the request is refused with `NOT_SUPPORTED`. A prefix overlapping another listing gets `PREFIX_OVERLAP`, a `TRACK_PROPERTY_FILTER` gets `NOT_SUPPORTED`, and a `REQUEST_UPDATE` on the listing itself ends it. On draft 18, a `SUBSCRIBE` to a track we `PUBLISH`ed gets `DUPLICATE_SUBSCRIPTION`. Cancelling the listing stops new `PUBLISH`es, while those sent live on. A broadcast is listed only while our publish grant covers it, from scratch once it does. Rust sends it upstream for a listing (below); JavaScript does not. - **A Rust TRACKS listing sends `SUBSCRIBE_TRACKS` upstream, only on demand.** Listing a broadcast from a moq-transport peer sends `SUBSCRIBE_TRACKS` with `FORWARD=0` while someone lists it, never on connect, announce, or subscribe, so a peer that refuses, resets, or ignores it affects only that listing. Each listed track's paused `PUBLISH` is its upstream copy, which a reader resumes rather than subscribing again. Only a `PUBLISH_DONE` with `TRACK_ENDED` removes a listed track, and it lists as finished; any other end, including the error our own answerer sends for a track that left unfinished, fails the listing, and readers list again. -- **History is fetched in ranges.** In Rust, Object ID gaps must be marked by Prior Object ID Gap properties for a group to be fetchable. A `FETCH` returns a range of groups from the cache, which a relay fills upstream with one `FETCH` per run of missing groups, or on draft 18 the saved prefix of the group a new subscription just joined. The range ends at Largest Object's group and reads it whole, so a `FETCH` of a group still being written waits for it to finish. That departs from draft 22's cap at Largest Object: a cache cannot know an open group's real Largest Object, and a cut group would be cached downstream as complete. On draft 18, `FETCH_OK` End Object 0 covers the entire End Group, matching the request; Rust also accepts the legacy answer naming the next group with Object 0. On draft 22, `FETCH_OK`'s End Location is the requested end, capped at Largest Object's group: a range holding no objects gets an empty fetch stream, and only a start past Largest Object is refused, with `INVALID_RANGE`. A Rust relay retains the upstream Largest Object from `TRACK_STATUS`, so even a fetch-only copy can resolve a draft 22 `LOCATION_FILTER` relative to it; without a publisher report or live feed, it refuses that filter as `NOT_SUPPORTED`. JavaScript publishing answers a `FETCH` from its cache alone, and a joining `FETCH` only for the head of its subscription's own group. It honors `FILL_TIMEOUT` on a `FETCH` or a fill as a total budget for waiting on objects that have not arrived, zero meaning cache only; once it runs out, the rest of each group still arriving is marked an Unknown gap on draft 18 and a Timed-Out gap on draft 22, never left implying the objects do not exist. It bounds the wait, not how old a cached object is. Rust still refuses `FILL_TIMEOUT` with `NOT_SUPPORTED`. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; a reader whose publisher refuses `TRACK_STATUS` as `NOT_SUPPORTED` still subscribes. +- **History is fetched in ranges.** In Rust, Object ID gaps must be marked by Prior Object ID Gap properties for a group to be fetchable. A `FETCH` returns a range of groups from the cache, which a relay fills upstream with one `FETCH` per run of missing groups, or on draft 18 the saved prefix of the group a new subscription just joined. The range ends at Largest Object's group and reads it whole, so a `FETCH` of a group still being written waits for it to finish. That departs from draft 22's cap at Largest Object: a cache cannot know an open group's real Largest Object, and a cut group would be cached downstream as complete. On draft 18, `FETCH_OK` End Object 0 covers the entire End Group, matching the request; Rust also accepts the legacy answer naming the next group with Object 0. On draft 22, `FETCH_OK`'s End Location is the requested end, capped at Largest Object's group: a range holding no objects gets an empty fetch stream, and only a start past Largest Object is refused, with `INVALID_RANGE`. A Rust relay retains the upstream Largest Object from `TRACK_STATUS`, so even a fetch-only copy can resolve a draft 22 `LOCATION_FILTER` relative to it; without a publisher report or live feed, it refuses that filter as `NOT_SUPPORTED`. JavaScript publishing answers a `FETCH` from its cache alone, and a joining `FETCH` only for the head of its subscription's own group. It honors `FILL_TIMEOUT` on a `FETCH` or a fill as a total budget for waiting on objects that have not arrived, zero meaning cache only; once it runs out, the rest of each group still arriving is marked an Unknown gap on draft 18 and a Timed-Out gap on draft 22, never left implying the objects do not exist. It bounds the wait, not how old a cached object is. Rust still refuses `FILL_TIMEOUT` with `NOT_SUPPORTED`. A JavaScript subscription joins at the current group: on draft 18 it fetches the group's head with a Relative Joining `FETCH`, and on draft 22 asks for it with a fill. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; a reader whose publisher refuses `TRACK_STATUS` as `NOT_SUPPORTED` still subscribes. - **Compression.** Rust peers exchange compressed tracks when both offer the [Flate extension](/draft/moq-flate). Other peers receive one shared inflated copy. A publisher that seals the compression declaration makes that fallback unavailable: an incompatible peer is refused, preserving the sealed metadata. JavaScript does not offer the IETF extension. - **Timing.** A track whose `SUBSCRIBE_OK` declares no `TIMESCALE` is untimed. A Rust reader that never subscribed takes the units from `TRACK_STATUS`. - **Strict SETUP and parameters.** Repeated unknown `SETUP` options, GREASE included, are accepted if well-formed; a repeated known option is rejected. A `GROUP_ORDER` other than Ascending (1) or Descending (2) closes the session. An unknown message parameter, or one defined only for another message, closes the session. Other malformed control input closes the session with the draft's code too, such as a server `SETUP` carrying `PATH` or `AUTHORITY`, a namespace field ending in `/` (it would map to a path with an empty segment), a request cut short by FIN, or a second `GOAWAY` on one stream; a first `GOAWAY` on a request stream is legal. diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 98bf82d7a4b..72f64cffcab 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -527,8 +527,11 @@ export class Connection implements Established { return; } - // We never FETCH, so a fetch response answers nothing of ours. - if (type === BigInt(FetchHeader.type)) throw new Error("unexpected fetch stream"); + // The head of a join, answering our draft-18 joining FETCH or draft-22 fill. + if (type === BigInt(FetchHeader.type)) { + await this.#subscriber.handleFetch(stream); + return; + } // Anything else is unknown, and a second SETUP is a violation too. throw new ProtocolViolation(`unknown uni stream type: 0x${type.toString(16)}`); diff --git a/js/net/src/ietf/filter.test.ts b/js/net/src/ietf/filter.test.ts index 0bad3b5c8ad..41d91bb331d 100644 --- a/js/net/src/ietf/filter.test.ts +++ b/js/net/src/ietf/filter.test.ts @@ -9,11 +9,14 @@ import { Version } from "./version.ts"; const OLD = Version.DRAFT_18; const NEW = Version.DRAFT_22; -test("legacy live joins do not exclude the prefix without a joining FETCH", () => { - for (const version of [Version.DRAFT_18]) { - expect([...Filter.encode(joinFilter(version), version)]).toEqual([3, 0, 0]); - } - expect([...Filter.encode(joinFilter(NEW), NEW)]).toEqual([0x01, 0x01]); +test("a live join is Next Object, its head left to a joining FETCH or a StartGroup=1 fill", () => { + const old = joinFilter(OLD); + expect([...Filter.encode(old.filter, OLD)]).toEqual([0x02]); + expect(old.fill).toBeUndefined(); + + const draft22 = joinFilter(NEW); + expect([...Filter.encode(draft22.filter, NEW)]).toEqual([0x05]); + expect(draft22.fill).toEqual({ filter: { kind: "relative", groups: 1n }, rangeFilters: false }); }); function roundTrip(filter: Filter.Filter, version: typeof OLD | typeof NEW): Filter.Filter { diff --git a/js/net/src/ietf/join.test.ts b/js/net/src/ietf/join.test.ts new file mode 100644 index 00000000000..075412d14f8 --- /dev/null +++ b/js/net/src/ietf/join.test.ts @@ -0,0 +1,593 @@ +import { expect, jest, onTestFinished, test } from "bun:test"; +import * as Connection from "../connection/index.ts"; +import { GRACE } from "../ledger.ts"; +import { createMockTransportPair, textFrame } from "../mock.ts"; +import { Producer as OriginProducer } from "../origin.ts"; +import * as Path from "../path.ts"; +import { Reader, Stream, Writer } from "../stream.ts"; +import { Tail } from "../tail.ts"; +import { Timescale } from "../time.ts"; +import * as track from "../track.ts"; +import { wireOf } from "../wire.ts"; +import { Fetch, FetchOk } from "./fetch.ts"; +import { Join } from "./join.ts"; +import { decodeFetchObject, FetchFrame, type GroupFlags, Group as GroupMessage } from "./object.ts"; +import type { MessageLocation } from "./parameters.ts"; +import { PublishDone, PublishDoneStatus } from "./publish.ts"; +import { RequestError } from "./request.ts"; +import { NativeSession } from "./session.ts"; +import { Subscribe, SubscribeOk } from "./subscribe.ts"; +import { Subscriber } from "./subscriber.ts"; +import { ALPN, type IetfVersion, Version } from "./version.ts"; + +const VERSION = Version.DRAFT_18; +const DRAFT_22 = Version.DRAFT_22; +const ALIAS = 9n; +/** The group SUBSCRIBE_OK names as Largest in most cases below. */ +const GROUP = 3; + +const text = (s: string) => new TextEncoder().encode(s); +const untext = (b: Uint8Array | undefined) => (b ? new TextDecoder().decode(b) : undefined); + +/** Accept the next bidi stream the subscriber opens, or undefined if none comes soon. */ +async function nextStream( + transport: WebTransport, + wait = 200, + version: IetfVersion = VERSION, +): Promise { + const reader = + transport.incomingBidirectionalStreams.getReader() as ReadableStreamDefaultReader; + let timer: ReturnType | undefined; + try { + const next = await Promise.race([ + reader.read(), + new Promise((resolve) => { + timer = setTimeout(() => resolve(undefined), wait); + }), + ]); + if (!next || next.done) return undefined; + return new Stream({ readable: next.value.readable, writable: next.value.writable, version }); + } finally { + clearTimeout(timer); + reader.releaseLock(); + } +} + +/** A subscriber with one track, and the SUBSCRIBE it sent, still unanswered. */ +async function subscribed(version: IetfVersion = VERSION) { + const pair = createMockTransportPair(version === VERSION ? ALPN.DRAFT_18 : ALPN.DRAFT_22); + const session = new NativeSession(pair.server, version, true); + const subscriber = new Subscriber({ session }); + const consumer = subscriber.consume(Path.from("room")).track("video").subscribe(); + onTestFinished(() => { + consumer.close(); + session.close(); + }); + + const peer = await nextStream(pair.client, 200, version); + if (!peer) throw new Error("the subscriber never subscribed"); + expect(await peer.reader.u53()).toBe(Subscribe.id); + const subscribe = await Subscribe.decode(peer.reader, version); + const answer = async (largest: MessageLocation | null) => { + await peer.writer.u53(SubscribeOk.id); + await new SubscribeOk({ trackAlias: ALIAS, largest: largest ?? undefined }).encode(peer.writer, version); + }; + return { subscriber, consumer, subscribe, peer, answer, client: pair.client }; +} + +/** + * A draft-18 subscriber with one track answered with Largest `largest`, and the joining + * FETCH it sent in response, still unanswered. + */ +async function join(largest: MessageLocation | null = { groupId: BigInt(GROUP), objectId: 1n }) { + const subscription = await subscribed(); + await subscription.answer(largest); + + const fetching = await nextStream(subscription.client); + let fetch: Fetch | undefined; + if (fetching) { + expect(await fetching.reader.u53()).toBe(Fetch.id); + fetch = await Fetch.decode(fetching.reader, VERSION); + } + return { ...subscription, fetching, fetch }; +} + +async function accept(fetching: Stream | undefined) { + if (!fetching) throw new Error("no joining FETCH"); + await fetching.writer.u53(FetchOk.id); + await new FetchOk({ endOfTrack: false, endLocation: { group: BigInt(GROUP), object: 2n } }).encode( + fetching.writer, + VERSION, + ); +} + +async function refuse(fetching: Stream | undefined) { + if (!fetching) throw new Error("no joining FETCH"); + await fetching.writer.u53(RequestError.id); + await new RequestError({ errorCode: 0x4, reasonPhrase: "not here" }).encode(fetching.writer, VERSION); +} + +/** A fetch stream after its type: the header naming `requestId`, then `payloads` from object 0. */ +async function head(requestId: bigint, payloads: string[], group = GROUP, version: IetfVersion = VERSION) { + const bytes = await Writer.collect(version, async (w) => { + await w.u62(requestId); + for (const [object, payload] of payloads.entries()) { + await new FetchFrame({ payload: text(payload) }).encode( + w, + { group, object, first: object === 0 }, + undefined, + version, + ); + } + }); + return new Reader(undefined, bytes, version); +} + +function flags(firstObject: boolean): GroupFlags { + return { + hasExtensions: false, + hasSubgroup: false, + hasSubgroupObject: false, + hasEnd: true, + hasPriority: true, + firstObject, + }; +} + +/** A live subgroup stream for `group` starting at Object ID `start`. */ +function liveStream( + group: number, + start: number, + payloads: string[], + version: IetfVersion = VERSION, +): [GroupMessage, Reader] { + const bytes: number[] = []; + for (const [i, payload] of payloads.entries()) { + const p = text(payload); + // Every field here is under 64, so each is a one-byte varint. + bytes.push(i === 0 ? start : 0, p.byteLength, ...p); + } + const header = new GroupMessage({ + trackAlias: ALIAS, + groupId: group, + subGroupId: 0, + publisherPriority: 0, + flags: flags(start === 0), + }); + return [header, new Reader(undefined, new Uint8Array(bytes), version)]; +} + +async function frames(group: { readFrame(): Promise<{ payload: Uint8Array } | undefined> } | undefined) { + const out: (string | undefined)[] = []; + for (;;) { + const frame = await group?.readFrame(); + if (!frame) return out; + out.push(untext(frame.payload)); + } +} + +test("a draft-18 join subscribes at Largest Object and fetches Largest's group", async () => { + const { subscribe, fetch } = await join(); + expect(subscribe.filter).toEqual({ kind: "nextObject" }); + expect(fetch?.type).toEqual({ kind: "relativeJoining", subscription: subscribe.requestId, offset: 0n }); + expect(fetch?.groupOrder).toBe("ascending"); + expect(fetch?.requestId).not.toBe(subscribe.requestId); +}); + +test("an empty track sends no joining FETCH", async () => { + const { fetching } = await join(null); + expect(fetching).toBeUndefined(); +}); + +test("a head that lands before its tail is continued by it", async () => { + const { subscriber, consumer, fetching, fetch } = await join(); + if (!fetch) throw new Error("no fetch"); + await accept(fetching); + + await subscriber.handleFetch(await head(fetch.requestId, ["h0", "h1"])); + await subscriber.handleGroup(...liveStream(GROUP, 2, ["t2", "t3"])); + + const group = await consumer.ordered().nextGroup(); + expect(group?.sequence).toBe(GROUP); + expect(await frames(group)).toEqual(["h0", "h1", "t2", "t3"]); +}); + +test("a tail that lands before its head waits for it", async () => { + const { subscriber, consumer, fetching, fetch } = await join(); + if (!fetch) throw new Error("no fetch"); + + const tail = subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"])); + await accept(fetching); + await subscriber.handleFetch(await head(fetch.requestId, ["h0", "h1"])); + await tail; + + const group = await consumer.ordered().nextGroup(); + expect(await frames(group)).toEqual(["h0", "h1", "t2"]); +}); + +test("a catalog written once is readable from the head alone", async () => { + const { subscriber, consumer, fetching, fetch } = await join({ groupId: 0n, objectId: 0n }); + if (!fetch) throw new Error("no fetch"); + await accept(fetching); + + // No live stream ever comes: the group was written before the subscription. + await subscriber.handleFetch(await head(fetch.requestId, ['{"video":{}}'], 0)); + + const group = await consumer.ordered().nextGroup(); + expect(group?.sequence).toBe(0); + expect(untext((await group?.readFrame())?.payload)).toBe('{"video":{}}'); +}); + +test("an empty tail finishes a group the head already holds whole", async () => { + const { subscriber, consumer, fetching, fetch } = await join(); + if (!fetch) throw new Error("no fetch"); + await accept(fetching); + + await subscriber.handleFetch(await head(fetch.requestId, ["h0", "h1"])); + const [header] = liveStream(GROUP, 2, []); + header.flags = flags(false); + await subscriber.handleGroup(header, new Reader(undefined, new Uint8Array(), VERSION)); + + const group = await consumer.ordered().nextGroup(); + expect(await frames(group)).toEqual(["h0", "h1"]); + expect(await group?.closed).toBeNull(); +}); + +for (const outcome of ["a refused joining FETCH", "an empty joining FETCH"] as const) { + test(`${outcome} costs only the group's head`, async () => { + const { subscriber, consumer, fetching, fetch, client } = await join(); + if (!fetch) throw new Error("no fetch"); + if (outcome === "an empty joining FETCH") { + await accept(fetching); + await subscriber.handleFetch(await head(fetch.requestId, [])); + } else { + await refuse(fetching); + } + // No standalone FETCH follows: a relay that refuses the join has a bug of its own. + expect(await nextStream(client)).toBeUndefined(); + + // The mid-group tail has nothing to stitch onto, so it is dropped; the next group plays. + await subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"])); + await subscriber.handleGroup(...liveStream(GROUP + 1, 0, ["n0"])); + + const group = await consumer.ordered().nextGroup(); + expect(group?.sequence).toBe(GROUP + 1); + expect(await frames(group)).toEqual(["n0"]); + }); +} + +test("a tail parked on a refused joining FETCH is dropped", async () => { + const { subscriber, consumer, fetching } = await join(); + + const tail = subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"])); + await refuse(fetching); + await tail; + await subscriber.handleGroup(...liveStream(GROUP + 1, 0, ["n0"])); + + const group = await consumer.ordered().nextGroup(); + expect(group?.sequence).toBe(GROUP + 1); +}); + +test("a joining FETCH has no deadline of its own", async () => { + jest.useFakeTimers(); + onTestFinished(() => jest.useRealTimers()); + const { subscriber, consumer, fetching, fetch, client } = await join(); + if (!fetch) throw new Error("no fetch"); + + const tail = subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"])); + // Well past any request deadline: the FETCH is still out, and nothing else was sent. + jest.advanceTimersByTime(60_000); + jest.useRealTimers(); + expect(await nextStream(client)).toBeUndefined(); + + await accept(fetching); + await subscriber.handleFetch(await head(fetch.requestId, ["h0", "h1"])); + await tail; + + const group = await consumer.ordered().nextGroup(); + expect(group?.sequence).toBe(GROUP); + expect(await frames(group)).toEqual(["h0", "h1", "t2"]); +}); + +test("a joining FETCH reset after FETCH_OK costs only the head", async () => { + const { subscriber, consumer, fetching, peer } = await join(); + const started = performance.now(); + + const tail = subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"])); + await accept(fetching); + // No fetch stream ever comes. + fetching?.writer.reset(new Error("gone")); + await tail; + await subscriber.handleGroup(...liveStream(GROUP + 1, 0, ["n0"])); + await peer.writer.u53(PublishDone.id); + const done = new PublishDone({ statusCode: PublishDoneStatus.TRACK_ENDED, streamCount: 2n, reasonPhrase: "done" }); + await done.encode(peer.writer, VERSION); + + // The parked tail was dropped and nothing holds the end open. + const group = await consumer.recvGroup(); + expect(group?.sequence).toBe(GROUP + 1); + expect(await consumer.recvGroup()).toBeUndefined(); + expect(await consumer.closed).toBeNull(); + expect(performance.now() - started).toBeLessThan(GRACE); +}); + +test("SUBSCRIPTION_ENDED cancels a head nothing claimed", async () => { + const { subscriber, consumer, fetching, fetch, peer } = await join(); + if (!fetch) throw new Error("no fetch"); + await accept(fetching); + await subscriber.handleFetch(await head(fetch.requestId, ["h0", "h1"])); + const group = await consumer.ordered().nextGroup(); + expect(group?.sequence).toBe(GROUP); + + // The group may run past the head upstream, so it must not read as whole. + await subscriber.handleGroup(...liveStream(GROUP + 1, 0, ["n0"])); + await peer.writer.u53(PublishDone.id); + const done = new PublishDone({ + statusCode: PublishDoneStatus.SUBSCRIPTION_ENDED, + streamCount: 1n, + reasonPhrase: "done", + }); + await done.encode(peer.writer, VERSION); + expect(await consumer.closed).toBeInstanceOf(Error); + await expect(frames(group)).rejects.toThrow(); +}); + +test("a whole group that precedes the head wins the sequence", async () => { + const { subscriber, consumer, fetching, fetch } = await join(); + if (!fetch) throw new Error("no fetch"); + await accept(fetching); + + await subscriber.handleGroup(...liveStream(GROUP, 0, ["w0", "w1", "w2"])); + await subscriber.handleFetch(await head(fetch.requestId, ["h0", "h1"])); + + const group = await consumer.ordered().nextGroup(); + expect(await frames(group)).toEqual(["w0", "w1", "w2"]); +}); + +test("a whole group for a headed sequence is refused, with its head", async () => { + const { subscriber, consumer, fetching, fetch } = await join(); + if (!fetch) throw new Error("no fetch"); + await accept(fetching); + + await subscriber.handleFetch(await head(fetch.requestId, ["h0", "h1"])); + await subscriber.handleGroup(...liveStream(GROUP, 0, ["w0"])); + + const group = await consumer.ordered().nextGroup(); + expect(group?.sequence).toBe(GROUP); + expect(await group?.closed).toBeInstanceOf(Error); +}); + +test("a head that does not meet its tail is refused", async () => { + const { subscriber, consumer, fetching, fetch } = await join(); + if (!fetch) throw new Error("no fetch"); + await accept(fetching); + + await subscriber.handleFetch(await head(fetch.requestId, ["h0", "h1"])); + // Object 2 is missing. + await subscriber.handleGroup(...liveStream(GROUP, 3, ["t3"])); + + const group = await consumer.ordered().nextGroup(); + expect(await group?.closed).toBeInstanceOf(Error); +}); + +/** A head whose objects after the first spell Object ID Delta `delta` (flag 0x04, no Group ID). */ +async function deltaHead(requestId: bigint, payloads: string[], delta: number) { + const bytes = await Writer.collect(VERSION, async (w) => { + await w.u62(requestId); + for (const [object, payload] of payloads.entries()) { + if (object === 0) { + await new FetchFrame({ payload: text(payload) }).encode( + w, + { group: GROUP, object, first: true }, + undefined, + VERSION, + ); + continue; + } + const p = text(payload); + await w.write(new Uint8Array([0x04, delta, p.byteLength, ...p])); + } + }); + return new Reader(undefined, bytes, VERSION); +} + +test("a head that spells Object ID Delta 1 is contiguous", async () => { + const { subscriber, consumer, fetching, fetch } = await join(); + if (!fetch) throw new Error("no fetch"); + await accept(fetching); + + await subscriber.handleFetch(await deltaHead(fetch.requestId, ["h0", "h1"], 1)); + await subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"])); + + const group = await consumer.ordered().nextGroup(); + expect(await frames(group)).toEqual(["h0", "h1", "t2"]); +}); + +test("a head that repeats an Object ID is dropped", async () => { + const { subscriber, consumer, fetching, fetch } = await join(); + if (!fetch) throw new Error("no fetch"); + await accept(fetching); + + await subscriber.handleFetch(await deltaHead(fetch.requestId, ["h0", "h0"], 0)); + + const group = await consumer.ordered().nextGroup(); + expect(await group?.closed).toBeInstanceOf(Error); +}); + +test("a tail that is only an End of Group finishes the head", async () => { + const { subscriber, consumer, fetching, fetch } = await join(); + if (!fetch) throw new Error("no fetch"); + await accept(fetching); + + await subscriber.handleFetch(await head(fetch.requestId, ["h0", "h1"])); + const [header] = liveStream(GROUP, 2, []); + header.flags = flags(false); + // Object ID 2, an empty payload, then status End of Group. + await subscriber.handleGroup(header, new Reader(undefined, new Uint8Array([2, 0, 3]), VERSION)); + + const group = await consumer.ordered().nextGroup(); + expect(await frames(group)).toEqual(["h0", "h1"]); + expect(await group?.closed).toBeNull(); +}); + +test("a joining FETCH spanning another group drops the head", async () => { + const { subscriber, consumer, fetching, fetch } = await join(); + if (!fetch) throw new Error("no fetch"); + await accept(fetching); + + const bytes = await Writer.collect(VERSION, async (w) => { + await w.u62(fetch.requestId); + await new FetchFrame({ payload: text("h0") }).encode( + w, + { group: GROUP, object: 0, first: true }, + undefined, + VERSION, + ); + await new FetchFrame({ payload: text("x0") }).encode( + w, + { group: GROUP + 1, object: 0, first: false, previousGroup: GROUP }, + undefined, + VERSION, + ); + }); + await subscriber.handleFetch(new Reader(undefined, bytes, VERSION)); + + const group = await consumer.ordered().nextGroup(); + expect(await group?.closed).toBeInstanceOf(Error); +}); + +test("a draft-22 join subscribes at Next Object with a StartGroup=1 fill, and sends no FETCH", async () => { + const { subscribe, answer, client } = await subscribed(DRAFT_22); + expect(subscribe.filter).toEqual({ kind: "nextObject" }); + expect(subscribe.fill).toEqual({ filter: { kind: "relative", groups: 1n }, rangeFilters: false }); + await answer({ groupId: BigInt(GROUP), objectId: 1n }); + expect(await nextStream(client, 200, DRAFT_22)).toBeUndefined(); +}); + +test("a draft-22 fill and its tail stitch into one group", async () => { + const { subscriber, consumer, subscribe, answer } = await subscribed(DRAFT_22); + await answer({ groupId: BigInt(GROUP), objectId: 1n }); + + const tail = subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"], DRAFT_22)); + // The fill names the SUBSCRIBE's own request. + await subscriber.handleFetch(await head(subscribe.requestId, ["h0", "h1"], GROUP, DRAFT_22)); + await tail; + + const group = await consumer.ordered().nextGroup(); + expect(group?.sequence).toBe(GROUP); + expect(await frames(group)).toEqual(["h0", "h1", "t2"]); +}); + +test("a draft-22 fill that overtakes SUBSCRIBE_OK waits for it", async () => { + const { subscriber, consumer, subscribe, answer } = await subscribed(DRAFT_22); + const fill = subscriber.handleFetch(await head(subscribe.requestId, ["h0", "h1"], GROUP, DRAFT_22)); + await answer({ groupId: BigInt(GROUP), objectId: 1n }); + await fill; + await subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"], DRAFT_22)); + + const group = await consumer.ordered().nextGroup(); + expect(await frames(group)).toEqual(["h0", "h1", "t2"]); +}); + +test("a draft-22 fill counts toward PUBLISH_DONE's Stream Count", async () => { + const { subscriber, consumer, subscribe, peer, answer } = await subscribed(DRAFT_22); + await answer({ groupId: 0n, objectId: 0n }); + const started = performance.now(); + await subscriber.handleFetch(await head(subscribe.requestId, ["catalog"], 0, DRAFT_22)); + await peer.writer.u53(PublishDone.id); + const done = new PublishDone({ statusCode: PublishDoneStatus.TRACK_ENDED, streamCount: 1n, reasonPhrase: "done" }); + await done.encode(peer.writer, DRAFT_22); + + // The fill was the one stream counted, so the head is the whole group and the track ends. + const group = await consumer.recvGroup(); + expect(group?.sequence).toBe(0); + expect(await frames(group)).toEqual(["catalog"]); + expect(await consumer.recvGroup()).toBeUndefined(); + expect(await consumer.closed).toBeNull(); + expect(performance.now() - started).toBeLessThan(GRACE); +}); + +test("a clean end finishes a head nothing claimed", () => { + const t = new Tail(); + const j = new Join(0, t); + const producer = new track.Producer("catalog").createGroup(0); + expect(j.open(() => producer)).toBe(producer); + producer.writeFrame({ payload: text("c") }); + j.wrote(); + j.finish(); + expect(j.outstanding).toBe(true); + j.end(); + expect(producer.closed.peek()).toBeNull(); + expect(j.outstanding).toBe(false); +}); + +test("fetch objects decode as FetchFrame wrote them", async () => { + const bytes = await Writer.collect(VERSION, async (w) => { + await new FetchFrame({ payload: text("a") }).encode( + w, + { group: 5, object: 0, first: true }, + undefined, + VERSION, + ); + await new FetchFrame({ payload: new Uint8Array() }).encode( + w, + { group: 5, object: 1, first: false }, + undefined, + VERSION, + ); + }); + const reader = new Reader(undefined, bytes, VERSION); + const first = await reader.decode((c) => decodeFetchObject(c, undefined)); + expect(first).toMatchObject({ kind: "object", group: 5, object: 0, priority: 0, subgroup: "zero" }); + const second = await reader.decode((c) => decodeFetchObject(c, undefined)); + expect(second).toMatchObject({ kind: "object", group: undefined, object: undefined, payload: new Uint8Array() }); +}); + +// End to end against the JS publisher, which answers the joining FETCH from its cache and +// serves a fill split from the subscription at Largest, as a strict relay does. +async function publishing(version: IetfVersion) { + const origin = new OriginProducer(); + const pair = createMockTransportPair(version === VERSION ? ALPN.DRAFT_18 : ALPN.DRAFT_22); + const url = new URL("https://localhost:4443/test"); + const [client, server] = await Promise.all([ + Connection.connect({ url, transport: pair.client }), + Connection.accept({ transport: pair.server, url, publish: origin.consume() }), + ]); + onTestFinished(() => { + client.abort(); + server.abort(); + }); + + const publication = origin.createBroadcast(Path.from("room")); + publication.announce(); + const broadcast = publication.broadcast; + // Written once, before anyone subscribed, and left open: only the head can deliver it. + const catalog = broadcast.createTrack("catalog.json", { timescale: Timescale.MILLI }); + catalog.appendGroup().writeFrame(textFrame("catalog")); + const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); + const group = video.appendGroup(); + group.writeFrame(textFrame("v0")); + group.writeFrame(textFrame("v1")); + + const sent = jest.spyOn(Fetch.prototype, "encode"); + onTestFinished(() => sent.mockRestore()); + return { remote: wireOf(client).consume(Path.from("room")), group, sent }; +} + +for (const version of [VERSION, DRAFT_22]) { + const draft = version === VERSION ? "draft 18" : "draft 22"; + test(`${draft}: a late JS subscriber gets a JS publisher's current groups from object 0`, async () => { + const { remote, group, sent } = await publishing(version); + const catalogGroup = await remote.track("catalog.json").subscribe().recvGroup(); + expect(catalogGroup?.sequence).toBe(0); + expect(untext((await catalogGroup?.readFrame())?.payload)).toBe("catalog"); + + const videoGroup = await remote.track("video").subscribe().recvGroup(); + expect(videoGroup?.sequence).toBe(0); + group.writeFrame(textFrame("v2")); + group.finish(); + expect(await frames(videoGroup)).toEqual(["v0", "v1", "v2"]); + // A joining FETCH per track on draft 18; draft 22 asks for the head with a fill. + expect(sent.mock.calls).toHaveLength(version === VERSION ? 2 : 0); + }); +} diff --git a/js/net/src/ietf/join.ts b/js/net/src/ietf/join.ts new file mode 100644 index 00000000000..02b4fe77dc6 --- /dev/null +++ b/js/net/src/ietf/join.ts @@ -0,0 +1,242 @@ +import type * as netGroup from "../group.ts"; +import type { Tail } from "../tail.ts"; + +/** The head and the Object ID its live tail must start at. */ +export interface Head { + producer: netGroup.Producer; + next: number; +} + +type State = + /** The joining FETCH or the fill is out, or about to be: a head may still come. */ + | { kind: "pending" } + /** The fetch stream is writing the head into the group. */ + | { kind: "writing"; producer: netGroup.Producer; next: number } + /** The head is written and waits for the live tail to take the group over. */ + | { kind: "ready"; producer: netGroup.Producer; next: number } + /** No head is coming: refused, empty, failed, claimed, or the subscription went. */ + | { kind: "done" } + /** The track ended, so no tail is coming and a head is the whole group. */ + | { kind: "ended" }; + +/** + * The join of one live group: the head a draft-18 joining FETCH or a draft-22 fill brings and + * the live tail the subscription delivers, stitched into one group. + * + * A Next Object subscription starts after the Largest SUBSCRIBE_OK names, `{G, k}`, and the + * head's fetch stream ends at it, so the head carries objects `0..=k` and the tail starts at + * `k + 1`. The track has one producer per group, so the fetch stream creates group G and + * writes the head as it arrives (readers see it at once: a catalog written once needs no + * tail at all), and the tail takes that producer over here. A tail that arrives first waits. + * + * Mirrors the Rust subscriber's `Fill`. A refused, empty, or failed head costs only G's + * head: its mid-group tail is dropped like any headless stream, and later groups are whole. + * + * @internal + */ +export class Join { + /** The group this join brings the head of: SUBSCRIBE_OK's Largest group. */ + readonly sequence: number; + + #state: State = { kind: "pending" }; + // Tails parked on the head, woken synchronously so they hold the end before the head lets go. + #waiters: (() => void)[] = []; + // The subscription's end stays open while a head may still come, unless the Stream Count + // already waits for the head's stream. + #release: () => void; + #tail: Tail; + #settled: Promise; + #settle!: () => void; + // A head still being written when the track ended, finished with its stream. + #ended?: netGroup.Producer; + + /** + * `counted` says the head's stream is one of the subscription's own, which PUBLISH_DONE's + * Stream Count includes: a draft-22 fill. A draft-18 joining FETCH is a request of its own, + * so the join holds the end open for it instead. + */ + constructor(sequence: number, tail: Tail, { counted = false }: { counted?: boolean } = {}) { + this.sequence = sequence; + this.#tail = tail; + this.#release = counted ? () => {} : tail.hold(); + this.#settled = new Promise((resolve) => { + this.#settle = resolve; + }); + } + + /** Whether a head may still arrive or waits to be claimed. */ + get outstanding(): boolean { + const { kind } = this.#state; + return kind === "pending" || kind === "writing" || kind === "ready"; + } + + /** Whether no head has started yet: nothing was refused, and no object arrived. */ + get pending(): boolean { + return this.#state.kind === "pending"; + } + + /** Resolves once the head is written or none is coming. */ + get settled(): Promise { + return this.#settled; + } + + /** + * The fetch stream's first object arrived: create the group for the head with `create`. + * + * Returns undefined when nothing should be written: a whole group already took the + * sequence, the join gave up, or a second stream answers the one request. + */ + open(create: () => netGroup.Producer): netGroup.Producer | undefined { + const { kind } = this.#state; + // After the track ended the head still lands, as the whole group. + if (kind !== "pending" && kind !== "ended") return undefined; + const producer = create(); + if (kind === "ended") { + this.#ended = producer; + return producer; + } + this.#state = { kind: "writing", producer, next: 0 }; + return producer; + } + + /** The head wrote its next object. */ + wrote(): void { + if (this.#state.kind === "writing") this.#state.next += 1; + } + + /** The fetch stream ended cleanly: the head waits for its tail, or none came at all. */ + finish(): void { + if (this.#ended) { + this.#ended.finish(); + this.#ended = undefined; + return; + } + const state = this.#state; + if (state.kind === "writing") { + this.#transition({ kind: "ready", producer: state.producer, next: state.next }); + } else if (state.kind === "pending") { + // An empty fetch stream: there is no head. + this.#transition({ kind: "done" }); + } + } + + /** The fetch stream failed: a half-written head is a group with no end. */ + fail(err: Error): void { + if (this.#ended) { + this.#ended.abort(err); + this.#ended = undefined; + return; + } + const state = this.#state; + if (state.kind === "writing" || state.kind === "ready") state.producer.abort(err); + if (state.kind !== "ended") this.#transition({ kind: "done" }); + } + + /** + * The joining FETCH was refused or its request failed. A head already written stands; + * one still being written cannot be trusted to finish. + */ + refuse(err: Error): void { + const state = this.#state; + if (state.kind === "writing") { + state.producer.abort(err); + this.#transition({ kind: "done" }); + } else if (state.kind === "pending") { + this.#transition({ kind: "done" }); + } + } + + /** + * A live stream for this group starts at object 0, so it carries the whole group. + * + * Before the head it wins the sequence, and the head is dropped when it lands. After it, + * the publisher served those objects twice and the track has one producer per group, so + * both are refused rather than cache a cut group as complete. Returns whether the whole + * group may stand. Matches the Rust subscriber. + */ + whole(): boolean { + const state = this.#state; + if (state.kind === "pending") { + this.#transition({ kind: "done" }); + return true; + } + if (state.kind === "writing" || state.kind === "ready") { + console.warn(`a whole group arrived for one its fetch stream already headed: group=${this.sequence}`); + state.producer.abort(new Error("the publisher served the group's head twice")); + this.#transition({ kind: "done" }); + return false; + } + return true; + } + + /** + * Take the head over for a live tail starting at Object ID `start`, or `undefined` for a + * tail with no objects, which ends the group wherever the head did. + * + * Waits while the head is being fetched, meanwhile not holding the subscription's end open + * through `suspend`: the head holds it instead. Returns undefined when there is no head to + * continue, and drops a head that ends somewhere other than where the tail starts, since + * that hole is something the model cannot express. + */ + async claim(start: number | undefined, suspend?: () => () => void): Promise { + const { kind } = this.#state; + if (kind === "pending" || kind === "writing") { + const resume = suspend?.(); + await new Promise((resolve) => { + this.#waiters.push(() => { + resume?.(); + resolve(); + }); + }); + } + + const state = this.#state; + if (state.kind !== "ready") return undefined; + if (start !== undefined && start !== state.next) { + console.warn( + `the head does not meet the live tail: group=${this.sequence} head=${state.next} tail=${start}`, + ); + state.producer.abort(new Error("the head and the live tail do not meet")); + this.#transition({ kind: "done" }); + return undefined; + } + this.#transition({ kind: "done" }); + return { producer: state.producer, next: state.next }; + } + + /** + * The track ended and the subscription's streams are accounted for, so no tail is coming: + * a head nothing claimed is the whole group, and one still being written will be. + */ + end(): void { + const state = this.#state; + if (state.kind === "ready") state.producer.finish(); + if (state.kind === "writing") this.#ended = state.producer; + if (state.kind !== "done") this.#transition({ kind: "ended" }); + } + + /** + * The subscription went without the track ending. Abort a head nothing claimed rather than + * finish it: the group may still be open upstream, and a later join can serve it whole. + */ + cancel(err: Error): void { + const state = this.#state; + if (state.kind === "writing" || state.kind === "ready") state.producer.abort(err); + this.#ended?.abort(err); + this.#ended = undefined; + if (state.kind !== "done" && state.kind !== "ended") this.#transition({ kind: "done" }); + } + + #transition(next: State): void { + this.#state = next; + if (next.kind === "writing") return; + // Out of pending or writing: the head is settled, so wake its tails before letting go + // of the end, which they then hold themselves. + const waiters = this.#waiters; + this.#waiters = []; + for (const wake of waiters) wake(); + this.#settle(); + this.#release(); + this.#tail.touch(); + } +} diff --git a/js/net/src/ietf/object.ts b/js/net/src/ietf/object.ts index 42699d82caa..4b8034f67b5 100644 --- a/js/net/src/ietf/object.ts +++ b/js/net/src/ietf/object.ts @@ -289,8 +289,14 @@ export class Frame { if (payload !== undefined && payload.byteLength > 0) await w.write(payload); } - /** Decode a frame using the group flags, at the cursor's negotiated IETF version. */ - static decode(c: Cursor, flags: GroupFlags, timescale: Timescale | undefined): Frame { + /** + * Decode a frame using the group flags, at the cursor's negotiated IETF version. + * + * `offset` lets a stream's first object start past zero, reporting its absolute Object ID + * in `idDelta` rather than throwing {@link ObjectIdGap}: that is the live tail of a group + * whose head a joining FETCH or a fill brings. + */ + static decode(c: Cursor, flags: GroupFlags, timescale: Timescale | undefined, offset = false): Frame { // The first object's delta is its absolute Object ID; every later one is the prior ID // plus the delta plus one. moq-lite groups start at object 0 and never skip one, so // a sequential group is a zero delta throughout, and any other value means the group @@ -326,9 +332,11 @@ export class Frame { const payloadLength = c.u53(); if (payloadLength > 0) { - if (delta !== 0) throw new ObjectIdGap(delta); + if (delta !== 0 && !offset) throw new ObjectIdGap(delta); const payload = c.read(payloadLength); - return new Frame({ payload, timestamp }); + const frame = new Frame({ payload, timestamp }); + frame.idDelta = delta; + return frame; } const status = c.u53(); @@ -343,12 +351,20 @@ export class Frame { frame.hasProperties = hasProperties; return frame; } - if (delta !== 0) throw new ObjectIdGap(delta); + if (delta !== 0 && !(offset && (status === 0 || status === GROUP_END))) throw new ObjectIdGap(delta); // Allowed even when the header marks the group's end: that bit only lets a FIN imply it, // and imquic sends both. - if (status === GROUP_END) return new Frame(); + if (status === GROUP_END) { + const frame = new Frame(); + frame.idDelta = delta; + return frame; + } - if (status === 0) return new Frame({ payload: new Uint8Array(0), timestamp }); + if (status === 0) { + const frame = new Frame({ payload: new Uint8Array(0), timestamp }); + frame.idDelta = delta; + return frame; + } throw new Error(`Unsupported object status: ${status}`); } @@ -484,3 +500,93 @@ export class FetchEndOfRange { await w.u62(this.object); } } + +const END_OF_RANGE: readonly number[] = Object.values(EndOfRange); + +// Serialization Flags at or above this are named values, not a set of flags. +const FETCH_NAMED = 0x80; +// The flag of an object published as a datagram, whose low bits then mean nothing. +const FETCH_DATAGRAM = 0x40; +// The low bits, which spell the Subgroup ID. +const FETCH_SUBGROUP = 0x03; + +/** How a fetched object names its Subgroup ID. */ +export type FetchSubgroup = "zero" | "prior" | "next" | "explicit" | "datagram"; + +/** One object read off a fetch stream, the inverse of {@link FetchFrame.encode}. */ +export type FetchedObject = + | { + kind: "object"; + /** The Group ID field: absolute on the first object, a delta to a different group after. */ + group?: number; + /** How the Subgroup ID is spelled. */ + subgroup: FetchSubgroup; + /** The explicit Subgroup ID, for `subgroup: "explicit"`. */ + subgroupId?: number; + /** The Object ID field: absolute with `group`, otherwise a delta from the prior one. */ + object?: number; + /** The Publisher Priority, absent when it repeats the prior object's. */ + priority?: number; + /** The Timestamp in the track's units, when it declared some. */ + timestamp?: Timestamp; + /** The object payload; empty for an empty object. */ + payload: Uint8Array; + } + | { kind: "endOfRange"; reason: EndOfRange; group: number; object: number }; + +/** + * Decode the next object on a fetch stream, in the track's `timescale`, or not at all when + * it declared none. + */ +export function decodeFetchObject(c: Cursor, timescale: Timescale | undefined): FetchedObject { + const flags = c.u53(); + if (flags >= FETCH_NAMED) { + if (!END_OF_RANGE.includes(flags)) { + throw new ProtocolViolation(`unknown fetch serialization flags: 0x${flags.toString(16)}`); + } + return { kind: "endOfRange", reason: flags as EndOfRange, group: c.u53(), object: c.u53() }; + } + + // Wire order: Group ID, Subgroup ID, Object ID, Priority, Properties. + const group = flags & FETCH_GROUP_ID ? c.u53() : undefined; + let subgroup: FetchSubgroup; + let subgroupId: number | undefined; + if (flags & FETCH_DATAGRAM) { + subgroup = "datagram"; + } else { + switch (flags & FETCH_SUBGROUP) { + case 0: + subgroup = "zero"; + break; + case 1: + subgroup = "prior"; + break; + case 2: + subgroup = "next"; + break; + default: + subgroup = "explicit"; + subgroupId = c.u53(); + } + } + const object = flags & FETCH_OBJECT_ID ? c.u53() : undefined; + const priority = flags & FETCH_PRIORITY ? c.u8() : undefined; + + let timestamp: Timestamp | undefined; + if (flags & FETCH_PROPERTIES) { + const length = c.u53(); + if (length > MAX_OBJECT_EXTENSIONS) { + throw new StreamError(StreamCode.MalformedTrack, { message: "object extensions exceed 64 KiB" }); + } + if (timescale !== undefined) { + timestamp = c.exact(length, (e) => decodeObjectTime(e, timescale)); + } else { + c.read(length); + } + } + + // A fetch stream carries no Object Status: a zero length is simply an empty object. + const length = c.u53(); + const payload = length > 0 ? c.read(length) : new Uint8Array(0); + return { kind: "object", group, subgroup, subgroupId, object, priority, timestamp, payload }; +} diff --git a/js/net/src/ietf/subscribe.ts b/js/net/src/ietf/subscribe.ts index 1668402d816..6203b77909b 100644 --- a/js/net/src/ietf/subscribe.ts +++ b/js/net/src/ietf/subscribe.ts @@ -46,14 +46,19 @@ function decodeGroupOrder(value: number): Order { } /** - * The filter this implementation joins a live track with. + * How this implementation joins a live track: at the current group, from its first object, + * as the Rust subscriber does. * - * moq-lite joins at the start of the current group, which is a decodable point, and - * draft-22's relative form is the first that can name it without knowing Largest Object. - * Earlier drafts request unfiltered delivery because this client does not send joining FETCH. + * The subscription itself starts at the next object (draft 18's Largest Object), and the + * group's head comes on a fetch stream. Draft 22 asks for it with a fill one group back from + * the next, its current-group join (section 3.5); earlier drafts have no fill, so the + * subscriber sends a Relative Joining FETCH once SUBSCRIBE_OK names Largest. Spelling the + * filter keeps a peer from reading an absent one its own way. */ -export function joinFilter(version: IetfVersion): Filter.Filter { - return Filter.isDraft22(version) ? { kind: "relative", groups: 1n } : { kind: "unfiltered" }; +export function joinFilter(version: IetfVersion): { filter: Filter.Filter; fill?: Filter.Fill } { + const filter: Filter.Filter = { kind: "nextObject" }; + if (!Filter.isDraft22(version)) return { filter }; + return { filter, fill: { filter: { kind: "relative", groups: 1n }, rangeFilters: false } }; } export class Subscribe { diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 827954c3496..e256f2ca588 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -46,9 +46,19 @@ import { DuplicateTrackAlias, RetiredTrackAlias, TrackAliases } from "./aliases. import * as Cluster from "./cluster.ts"; import { ObjectDatagram } from "./datagram.ts"; import { requestReason } from "./error.ts"; +import { Fetch, FetchHeader, FetchOk } from "./fetch.ts"; +import { isDraft22 } from "./filter.ts"; +import { Join } from "./join.ts"; import * as Message from "./message.ts"; import * as Namespace from "./namespace.ts"; -import { decodeObjectTime, EndOfTrackProperties, Frame, type Group as GroupMessage, ObjectIdGap } from "./object.ts"; +import { + decodeFetchObject, + decodeObjectTime, + EndOfTrackProperties, + Frame, + type Group as GroupMessage, + ObjectIdGap, +} from "./object.ts"; import { Parameters } from "./parameters.ts"; import { DEFAULT_WIRE, fromWire, toWire } from "./priority.ts"; import { type Publish, PublishDone, PublishDoneStatus, publishDoneClean } from "./publish.ts"; @@ -65,6 +75,14 @@ import { MalformedAuthToken } from "./token.ts"; // Cost::DRAIN: cost is the whole mechanism, not a separate state. const DRAIN_COST: Cost = 2n ** 62n - 1n; +/** A fetch stream the head cannot be stitched from, which costs only that head. */ +class JoinUnsupported extends Error { + constructor(message: string) { + super(message); + this.name = "JoinUnsupported"; + } +} + // A live subscription, as the track alias its data streams name resolves to. type Subscription = { // The write side incoming group streams are routed into. @@ -74,6 +92,13 @@ type Subscription = { tail: Tail; // The track's exclusive end, once an END_OF_TRACK declares it. end?: number; + // The join of SUBSCRIBE_OK's Largest group, whose head a draft-18 joining FETCH or a + // draft-22 fill brings. + join?: Join; + // The track's TIMESCALE, which the head's fetch objects are stamped in too. + timescale?: Timescale; + // The request IDs the head's fetch stream may name, registered until the subscription ends. + fetchIds: bigint[]; }; // Out-parameter for #openSubscribe: lets the caller observe partial progress @@ -102,6 +127,8 @@ type SubscribeSetupState = { * as a protocol violation. */ rejected?: boolean; + /** Hands a draft-22 fill stream that overtook SUBSCRIBE_OK the subscription it heads, or nothing. */ + fill?: (subscription: Subscription | undefined) => void; }; /** An announced path: each live advertisement referencing it, oldest first, and the route it holds. */ @@ -148,6 +175,11 @@ export class Subscriber { // arrive untimed. #timescales = new Map(); + // The requests a fetch stream may answer, mapped to the subscription whose head it brings: + // a draft-18 joining FETCH's own request ID, or a draft-22 SUBSCRIBE's, whose fill can + // overtake SUBSCRIBE_OK and so waits on it. A fetch stream needs no track alias. + #fetches = new Map>(); + // Dedup consumed broadcasts per path: repeat consume() calls share one subscription. #consumes = new BroadcastCache(); @@ -830,6 +862,11 @@ export class Subscriber { const cleanup = async (afterSetup: boolean) => { state.cancelled = true; + // A fill stream waiting on SUBSCRIBE_OK gets nothing, and a head already started goes. + state.fill?.(undefined); + this.#fetches.delete(requestId); + state.subscription?.join?.cancel(e); + if ( state.registeredAlias !== undefined && state.subscription && @@ -868,7 +905,13 @@ export class Subscriber { const producer = subscription.track; let disposeGrant: Dispose | undefined; + // Ends the joining FETCH with the subscription. + const joining = new AbortController(); try { + if (subscription.join && !isDraft22(this.#session.version)) { + void this.#runJoiningFetch(subscription, requestId, priority, joining.signal); + } + // Which terminal fired decides how the track closes, so tag them rather than // racing bare promises. const publisherEnded = Symbol("publisher"); @@ -942,6 +985,10 @@ export class Subscriber { `subscribe error: id=${requestId} broadcast=${broadcast} track=${request.name} error=${reason(e)}`, ); } finally { + joining.abort(); + // A track end already settled the head; anything else leaves it for a later join. + subscription.join?.cancel(new Error("subscription ended")); + for (const id of subscription.fetchIds) this.#fetches.delete(id); disposeGrant?.(); // Only the owner tears down the alias metadata: a later subscription may have // reclaimed the alias and installed its own timescale. @@ -993,14 +1040,19 @@ export class Subscriber { } const count = done.streamCount; - const { tail, track } = subscription; + const { tail, track, join } = subscription; const complete = () => count > 0n && BigInt(tail.streams) >= count; await tail.settle(complete, hooks.trackReleased(track)); // SUBSCRIPTION_ENDED ends just this subscription, so the copy fails rather than declare an // end the track may run past: a later reader subscribes afresh, and its groups continue. + // A head nothing claimed may be run past the same way, so it is cancelled, not finished. if (done.statusCode !== PublishDoneStatus.TRACK_ENDED) { throw new Error(`publish done: subscription ended reason=${done.reasonPhrase}`); } + // The track ended and every stream is accounted for, so no tail is coming: a head nothing + // claimed is its whole group. A draft-18 joining FETCH is not in the Stream Count; its + // head held the settle. + if (hooks.trackReleased(track).peek() === undefined) join?.end(); } /** @@ -1081,6 +1133,88 @@ export class Subscriber { } } + /** + * Fetch the head of a draft-18 join's group, and keep the request open until the head is + * in or the subscription ends. + * + * It asks for the group SUBSCRIBE_OK names as Largest, from object 0 up to Largest, with a + * Relative Joining FETCH (offset 0) on its own request, outside PUBLISH_DONE's Stream Count. + * The objects arrive on a fetch stream that {@link handleFetch} writes into the + * {@link Join}. As in the Rust subscriber, a refusal or a failed request costs only the + * head, and the request has no deadline of its own: it lives as long as the subscription. + * A malformed answer closes the session, as on any request. + */ + async #runJoiningFetch(subscription: Subscription, subscribeId: bigint, priority: number, signal: AbortSignal) { + const join = subscription.join; + if (!join) return; + const version = this.#session.version; + let request: RequestStream | undefined; + try { + const fetchId = await this.#session.nextRequestId(signal); + if (fetchId === undefined) throw await this.#closedSession(); + // Registered before the FETCH goes out: its stream can overtake FETCH_OK. It stays + // until the subscription ends, so a head that lands after a whole group won the + // sequence is read away rather than refused. + this.#fetches.set(fetchId, Promise.resolve(subscription)); + subscription.fetchIds.push(fetchId); + + request = new RequestStream( + await untilAborted(Promise.resolve(this.#session.openBi(requestOptions)), signal), + this.#session, + this.#quic, + ); + const msg = new Fetch({ + requestId: fetchId, + type: { kind: "relativeJoining", subscription: subscribeId, offset: 0n }, + subscriberPriority: priority, + groupOrder: "ascending", + }); + await Message.write(request.writer, Fetch.id, msg, version); + console.debug(`joining fetch: id=${fetchId} group=${join.sequence}`); + + const response = await untilAborted(request.next(), signal); + if (response === undefined) throw new Error("joining FETCH ended before its response"); + if (response.type === RequestError.id) { + const err = await RequestError.decode(response.reader, version); + const why = requestReason(err.errorCode, err.reasonPhrase, "fetch", version); + throw new Error(`joining FETCH refused: ${why}`); + } + if (response.type !== FetchOk.id) { + throw new ProtocolViolation(`unexpected answer to FETCH: 0x${response.type.toString(16)}`); + } + try { + await FetchOk.decode(response.reader, version); + } catch (err) { + throw new ProtocolViolation(`malformed FETCH_OK: ${reason(error(err))}`); + } + + // The answer is in; the head arrives on the fetch stream, or never. The request stays + // read meanwhile: a reset means its fetch stream cannot be trusted to come, while a + // FIN only ends the publisher's half. + const ended = request.next().then((framed) => { + if (framed !== undefined) { + throw new ProtocolViolation(`unexpected message after FETCH_OK: 0x${framed.type.toString(16)}`); + } + return join.settled; + }); + await untilAborted(Promise.race([join.settled, ended]), signal); + request.close(); + } catch (err) { + const e = error(err); + if (e instanceof ProtocolViolation) { + request?.abort(e); + this.#fail(e.code, e.message); + } else if (signal.aborted) { + request?.cancel(); + request?.close(); + } else { + request?.abort(e); + console.debug(`joining FETCH failed, joining at the next whole group: ${reason(e)}`); + } + join.refuse(e); + } + } + // Opens the subscribe stream, sends SUBSCRIBE, and reads the response. // `state` is populated as soon as the stream opens and again when the // trackAlias is registered, so the caller can clean both up on timeout @@ -1121,15 +1255,18 @@ export class Subscriber { maxAge: this.#maxAge ? (request.subscription.maxDelay ?? 0) : undefined, // Fixed for the subscription's life: no REQUEST_UPDATE parameter carries it. groupOrder: request.subscription.order, - // No fill is requested: the fill's fetch stream and the live subscription split - // the group across two streams, and this subscriber does not reassemble them into - // one group yet. A lenient publisher (like ours) replays the whole in-range group - // on the subscription instead; a strict one only delivers from the next published - // object, and that mid-group stream is dropped, degrading the join to the next - // group boundary. - filter: joinFilter(version), + // Joins at the current group, from its first object: the subscription delivers the + // objects after Largest, and the group's head before it comes on a fetch stream + // (see {@link Join}). Fixed for the subscription's life. + ...joinFilter(version), epoch, }); + if (msg.fill) { + // The fill names this request and can overtake SUBSCRIBE_OK, so it waits on it here. + const accepted = Promise.withResolvers(); + this.#fetches.set(requestId, accepted.promise); + state.fill = accepted.resolve; + } await Message.write(state.stream.writer, Subscribe.id, msg, version); state.sent = true; @@ -1176,7 +1313,15 @@ export class Subscriber { ietf: version, }); const producer = staged.producer; - const subscription: Subscription = { track: producer, tail: new Tail() }; + const tail = new Tail(); + const subscription: Subscription = { track: producer, tail, timescale: ok.properties.timescale, fetchIds: [] }; + // The subscription starts after Largest, so the joining FETCH or the fill brings its + // group's head. With no Largest the track is empty and every group whole. A draft-22 fill + // is one of the subscription's own streams, so the Stream Count already waits for it. + if (ok.largest) { + subscription.join = new Join(Number(ok.largest.groupId), tail, { counted: msg.fill !== undefined }); + } + if (msg.fill) subscription.fetchIds.push(requestId); state.subscription = subscription; if (ok.largest) { // Object IDs can start past zero or skip, while readers see frame ordinals, so only @@ -1193,6 +1338,7 @@ export class Subscriber { requested: groups?.start === undefined ? undefined : groupBounds(groups).start, }); staged.commit(); + state.fill?.(subscription); try { this.#aliases.set(ok.trackAlias, subscription, { broadcast, name: request.name, epoch }); @@ -1425,7 +1571,11 @@ export class Subscriber { if (!group.flags.hasPriority) group.publisherPriority = toWire((await track.info()).priority); const timescale = this.#timescales.get(group.trackAlias); - const decode = (c: Cursor) => Frame.decode(c, group.flags, timescale); + // A stream for the group a join is fetching the head of may start partway + // through it, at the Object ID the head ends before (see Join). + const join = subscription.join?.sequence === group.groupId ? subscription.join : undefined; + let first = join?.outstanding === true; + const decode = (c: Cursor) => Frame.decode(c, group.flags, timescale, first); let nextObject = 0; for (;;) { // Every object already buffered is written without an await, so the reader wakes @@ -1439,6 +1589,39 @@ export class Subscriber { stream.stop(frame ?? new StreamError(StreamCode.Cancel)); return; } + if (first && join) { + first = false; + const suspend = () => tail.suspend(); + // No producer yet, so a closing group or track already returned above. + const object = frame instanceof Error ? undefined : (frame ?? undefined); + // The first object's ID is absolute: where in the group this stream starts. An End + // of Group there names where the group ends, past the head's last object. + const start = + object?.payload !== undefined || object?.endOfTrack || (object?.idDelta ?? 0) > 0 + ? object?.idDelta + : undefined; + if (start === 0 && object?.payload !== undefined) { + // The whole group: it stands alone unless the head already took the sequence. + if (!join.whole()) { + stream.stop(new Error("the group's head was served twice")); + return; + } + } else if (start === undefined || start > 0) { + // A tail with no objects ends the group wherever the head did. + const head = await join.claim(start, suspend); + if (head) { + producer = head.producer; + nextObject = head.next; + if (object) object.idDelta = 0; + } else if (start !== undefined && !object?.endOfTrack) { + console.debug( + `dropping a group with no head: alias=${group.trackAlias} group=${group.groupId} start=${start}`, + ); + stream.stop(new Error("no head to stitch a mid-group stream onto")); + return; + } + } + } if (!frame || frame instanceof Error) { // FIRST_OBJECT clear on a stream that ends with no object: it starts partway // through its group and carries none of it, so drop it like any headless @@ -1529,6 +1712,108 @@ export class Subscriber { } } + /** + * Handles a fetch stream, with its type already read: the head of a join, answering a + * draft-18 joining FETCH or a draft-22 fill. + * + * Either asked for one group from object 0, so the first object must spell + * `{G, 0}` and every later one follow it in that group. The head is created as soon as its + * first object arrives and written as the objects do, so readers see it at once; the live + * tail takes the group over through the {@link Join}. Anything the head cannot stitch, + * another group or a gap among them, drops it, which costs only that group's head. + * + * @internal + */ + async handleFetch(stream: Reader) { + const version = this.#session.version; + const { requestId } = await FetchHeader.decode(stream, version); + const subscription = await this.#fetches.get(requestId); + // Nothing of ours asked: the publisher is answering a FETCH or fill we never sent. + if (!subscription?.join) throw new Error(`unexpected fetch stream: request=${requestId}`); + const { join, track, tail, timescale } = subscription; + // A draft-22 fill is one of the subscription's own data streams, which PUBLISH_DONE + // counts. A draft-18 joining FETCH is a request of its own; the join holds the end for it. + const read = isDraft22(version) ? tail.open(join.sequence) : undefined; + + let producer: netGroup.Producer | undefined; + // A whole group took the sequence first, or the join gave up: read the rest away. + let discarding = false; + let next = 0; + try { + const decode = (c: Cursor) => decodeFetchObject(c, timescale); + for (let first = true; ; first = false) { + const object = + stream.tryDecode(decode) ?? + (await race([stream.decodeMaybe(decode), producer?.closed ?? hooks.trackReleased(track)])); + if (object === null || object instanceof Error) { + join.fail(object ?? new Error("the track closed under the head")); + stream.stop(object ?? new StreamError(StreamCode.Cancel)); + return; + } + if (object === undefined) break; + if (discarding) continue; + + if (object.kind === "endOfRange") { + throw new JoinUnsupported("a head with an End of Range cannot be stitched"); + } + const inherits = + object.group === undefined || + object.object === undefined || + object.priority === undefined || + object.subgroup === "prior" || + object.subgroup === "next"; + if (first && inherits) throw new ProtocolViolation("the first fetch object refers to a prior object"); + if (object.subgroup === "datagram") throw new JoinUnsupported("a datagram group is not fetchable"); + if (object.subgroup === "next" || (object.subgroup === "explicit" && object.subgroupId !== 0)) { + throw new JoinUnsupported("subgroups are not supported"); + } + + // The first object names its group and Object ID outright; a later Group ID names + // another group, which a head of one group does not have. A later Object ID Delta + // is added to the prior ID, so only 1 is the next object. + if (first) { + if (object.group !== join.sequence) { + throw new JoinUnsupported(`the head answered group ${object.group}, not ${join.sequence}`); + } + if (object.object !== 0) throw new JoinUnsupported("the head does not start at object 0"); + } else if (object.group !== undefined) { + throw new JoinUnsupported("a head spanning several groups cannot be stitched"); + } else if (object.object !== undefined && object.object !== 1) { + throw new JoinUnsupported("a gap in the head cannot be stitched"); + } + + if (timescale !== undefined && object.timestamp === undefined) { + throw new StreamError(StreamCode.MalformedTrack, { + message: `object without a Timestamp on a track with TIMESCALE: group=${join.sequence}`, + }); + } + + if (first) { + producer = join.open(() => track.createGroup(join.sequence)); + if (!producer) { + discarding = true; + continue; + } + } + producer?.writeFrame({ payload: object.payload, timestamp: object.timestamp }); + join.wrote(); + next += 1; + } + console.debug(`join head: group=${join.sequence} objects=${next}`); + join.finish(); + } catch (err: unknown) { + const e = await sessionCause(this.#quic, err); + if (e instanceof JoinUnsupported) console.warn(`${e.message}, joining at the next whole group`); + join.fail(e); + if (e instanceof ProtocolViolation) this.#fail(e.code, e.message); + // As for a subgroup object: the track is malformed, not just its head. + if (e instanceof StreamError && e.code === StreamCode.MalformedTrack) track.abort(e); + stream.stop(e); + } finally { + read?.(); + } + } + /** * Receive QUIC datagrams, each an OBJECT_DATAGRAM for one of our subscriptions. * diff --git a/js/net/src/order.test.ts b/js/net/src/order.test.ts index 79e1e89bc63..8d31bac0e6e 100644 --- a/js/net/src/order.test.ts +++ b/js/net/src/order.test.ts @@ -9,6 +9,7 @@ import { expect, test } from "bun:test"; import { Milli } from "@moq/clock"; import { accept, connect, type Established } from "./connection/index.ts"; import type { Producer as GroupProducer } from "./group.ts"; +import { FetchHeader } from "./ietf/fetch.ts"; import * as Ietf from "./ietf/index.ts"; import * as Lite from "./lite/index.ts"; import { createMockTransportPair, type MockTransport, textFrame } from "./mock.ts"; @@ -33,10 +34,42 @@ function expected(order: Order, count: number): number[] { ); } +// The positions in `sendStreams.uni` of each transport's fetch streams. +const fetchStreams = new WeakMap>(); + +// Note which uni streams open with a fetch header. A draft-18 subscriber that joins after +// Largest fetches the group's head on one, ranked with that group, beside its group streams. +function noteFetchStreams(transport: MockTransport): void { + const fetches = new Set(); + fetchStreams.set(transport, fetches); + const create = transport.createUnidirectionalStream.bind(transport); + transport.createUnidirectionalStream = async (options) => { + const stream = await create(options); + // Kept by position, since a wrapper such as holdFins replaces the logged stream. + const index = transport.sendStreams.uni.indexOf(stream as SendStream); + const getWriter = stream.getWriter.bind(stream); + stream.getWriter = () => { + const writer = getWriter(); + const write = writer.write.bind(writer); + let first = true; + writer.write = (chunk) => { + if (first && chunk?.[0] === FetchHeader.type) fetches.add(index); + first = false; + return write(chunk); + }; + return writer; + }; + return stream; + }; +} + // The send orders of a transport's group streams. Protocol streams, such as the setup stream, -// open without one. +// open without one, and fetch streams are left out. function sendOrders(transport: MockTransport): number[] { - return transport.sendStreams.uni.flatMap((stream) => (stream.sendOrder === undefined ? [] : [stream.sendOrder])); + const fetches = fetchStreams.get(transport); + return transport.sendStreams.uni.flatMap((stream, index) => + stream.sendOrder === undefined || fetches?.has(index) ? [] : [stream.sendOrder], + ); } // A group with one frame, left open. @@ -75,6 +108,7 @@ async function session(protocol: string, origin: OriginProducer): Promise; release: () => void } { let release!: () => void; @@ -178,10 +212,13 @@ function holdFins(transport: MockTransport): { closing: Promise; release: const create = transport.createUnidirectionalStream.bind(transport); transport.createUnidirectionalStream = async (options) => { const inner = await create(options); + const index = transport.sendStreams.uni.indexOf(inner as SendStream); const writer = inner.getWriter(); const outer: SendStream = new WritableStream({ write: (chunk) => writer.write(chunk), close: async () => { + // A join's fetch stream closes as usual; only group streams are held. + if (fetchStreams.get(transport)?.has(index)) return writer.close(); closed(); await released; await writer.close(); @@ -191,7 +228,7 @@ function holdFins(transport: MockTransport): { closing: Promise; release: outer.sendOrder = options?.sendOrder; // Record the stream the publisher re-ranks. const log = transport.sendStreams.uni; - log[log.indexOf(inner)] = outer; + log[index] = outer; return outer; }; return { closing, release }; diff --git a/js/net/src/tail.ts b/js/net/src/tail.ts index 209ff90d924..ec410c58a01 100644 --- a/js/net/src/tail.ts +++ b/js/net/src/tail.ts @@ -57,6 +57,43 @@ export class Tail { }; } + /** + * Hold the end open for something being read that is not one of the counted streams, + * such as a joining FETCH's head, or let go of a counted stream while it waits on one. + * Returns the call that lets go, which is idempotent. + */ + hold(): () => void { + this.#active += 1; + let released = false; + return () => { + if (released) return; + released = true; + this.#active -= 1; + this.#bump(); + }; + } + + /** + * Stop holding the end open for a counted stream while it waits on something that holds + * it instead. Returns the call that holds it again, which must come before the stream's + * own close. + */ + suspend(): () => void { + this.#active -= 1; + this.#bump(); + let resumed = false; + return () => { + if (resumed) return; + resumed = true; + this.#active += 1; + }; + } + + /** Re-check what {@link settle} waits on, after something outside the ledger changed. */ + touch(): void { + 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); diff --git a/quest/m1/README.md b/quest/m1/README.md index dc4f950225f..0a6c05a3fdd 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -156,7 +156,6 @@ removed). - [On-demand JavaScript FETCH](/quest/m1/js-fetch-on-demand.md) - browser publishers answer IETF FETCH for evicted or uncached groups through the JS ranges request surface - [Rust FETCH parity](/quest/m1/ietf-fetch-parity.md) - Rust finishes an answered joining FETCH's head, refuses a lost head with DOES_NOT_EXIST, and sends no timestamps without TIMESCALE, as the JS publisher does - [SUBSCRIBE honors the properties opt-out](/quest/m1/subscribe-timestamps.md) - the Rust publisher sends untimed objects to a SUBSCRIBE that declines Track Properties, and the Rust subscriber asks for them when it needs timestamps -- [JS joins at the current group](/quest/m1/ietf-js-join.md) - a late JS moq-transport subscriber gets the group in progress whole through a joining FETCH or fill, as Rust does - [Rust marks a lost listed group unknown](/quest/m1/ietf-publisher-unknown-range.md) - the Rust IETF publisher sends End of Unknown Range for a listed group it can't serve, after the publisher's open quests land - [Identity and recovery guide](/quest/m1/identity-guide.md) - explain epochs, pinned catalogs, lazy jobs, restart and version compatibility as they exist; takeover adds its own section - [Native rung disable](/quest/m1/simulcast-rung-disable.md) - OBS stops sending the top renditions its grant cannot fund, as the Rust/FFI publishers do, and a shaped uplink through moq-ffi proves the resume diff --git a/quest/m1/ietf-js-join.md b/quest/m1/ietf-js-join.md deleted file mode 100644 index b4fdc0ed942..00000000000 --- a/quest/m1/ietf-js-join.md +++ /dev/null @@ -1,63 +0,0 @@ -# [M] The JS moq-transport subscriber joins at the current group - -## Goal - -A late JS subscriber behind a spec-strict relay (such as moqx) receives the -group in progress whole, as the Rust subscriber does. A catalog or snapshot -written once arrives, and the first media group plays from its head. It works -on draft 18 and on draft 22. - -## Plan - -Decided 2026-10-28 (maintainer), from [#7088](https://github.com/moq-dev/moq/issues/7088): - -- **Mirror Rust exactly.** On draft 18, SUBSCRIBE with Largest Object. After - a SUBSCRIBE_OK with Largest `{G, k}`, send a Relative Joining FETCH - (offset 0) as its own request, outside PUBLISH_DONE's Stream Count, and - stitch the head and the live tail into one group producer. On draft 22, - `relative(1)` carries the matching fill. Today it sends none, so the same - gap exists there. -- **No join knob.** Every subscription joins at the current group. Rejected: - `join: "current" | "next"` and @moq/watch joining media at the next group. - [Fill refill](/quest/m2/ietf-fill-refill.md) already rejected the next group - ("costs a GOP of startup"). Lite and Rust have no such anchor, and how far - behind live to play is the player's choice through `maxDelay`. A "next" - anchor, if ever wanted, belongs in - [JS ranges](/quest/m1/subscribe-ranges/js.md) shared with Rust and lite. -- **No standalone-FETCH fallback.** A relay that refuses the joining FETCH - for a lone subscriber (moqx: "Session has no active subscribe", reported as - [openmoq/moqx#883](https://github.com/openmoq/moqx/issues/883)) has a bug of - its own. A refusal costs only that group's head, as in Rust - ([#6407](https://github.com/moq-dev/moq/pull/6407)). -- **No FETCH_OK deadline**, matching Rust: the joining FETCH lives as long as - the subscription. Our publishers accept and ignore FILL_TIMEOUT - ([one group](/quest/m1/ietf-publisher-one-group.md)). -- **Rust's strictness on duplicates.** A whole group from object 0 that - arrives before the head wins and the FETCH is discarded. One that arrives - after the head is refused. -- **Implementation:** take over [#7089](https://github.com/moq-dev/moq/pull/7089) - (michaelstuurman, maintainer edits allowed). Trim it on its branch: drop - `Track.Join`, `Subscription.join` and the @moq/watch `MEDIA_JOIN` change, - drop the standalone fallback and the 10 s per-request deadline, and add the - draft-22 fill. The author keeps credit, and the review history stays with - the PR. - -Update the note this makes stale: the interop README's claim that a JS -subscriber requests neither a joining FETCH nor a fill. - -Tests: the unit tests in #7089 minus the knob, plus a draft-22 fill. The -interop late-join lane passes with a JS subscriber against a strict publisher -(a lane that refuses to replay the group). `just ci` runs green, since #7089 -has not yet run CI. - -Public API: none. Wire: the JS subscriber sends Largest Object and a joining -FETCH on draft 18, and a fill on draft 22. - -## Closes - -- [#7088](https://github.com/moq-dev/moq/issues/7088) - a late draft-18 JS subscriber never receives a catalog written once - -## Related - -- [JS publisher answers a late join](/quest/m2/ietf-js-joining-fetch.md) - the publisher half after a quick finish -- [Headless moq-transport groups](/quest/m2/ietf-headless-group.md) - what a subscriber does with a tail whose head never comes diff --git a/quest/m1/interop-coverage/README.md b/quest/m1/interop-coverage/README.md index 3581bbd7d18..36d24b8f589 100644 --- a/quest/m1/interop-coverage/README.md +++ b/quest/m1/interop-coverage/README.md @@ -46,7 +46,7 @@ coverage on `main` at `3ab6cbf81d`. - [JS namespace lifecycle](/quest/m1/interop-coverage/js-namespace-done.md) - the runner's moq-dev-js client moves to an `@moq/net` that withdraws namespaces by reset, so subscribe-namespace-lifecycle passes on Cloudflare draft-18 and aiomoqt - [JS subgroup data through imquic](/quest/m1/interop-coverage/js-imquic-objects.md) - the runner's moq-dev-js client moves to an `@moq/net` patch that writes data stream headers in one write, so data-subgroup-basic passes against imquic -- [JS subscribers FETCH](/quest/m1/interop-coverage/js-ietf-fetch.md) - `@moq/net` fetches past groups and recovers a mid-group join over moq-transport, and joins the CLI FETCH round +- [JS subscribers FETCH](/quest/m1/interop-coverage/js-ietf-fetch.md) - `@moq/net` fetches past groups over moq-transport, and joins the CLI FETCH round ## Related diff --git a/quest/m1/interop-coverage/js-ietf-fetch.md b/quest/m1/interop-coverage/js-ietf-fetch.md index 87058f7baf9..80d2cf0633c 100644 --- a/quest/m1/interop-coverage/js-ietf-fetch.md +++ b/quest/m1/interop-coverage/js-ietf-fetch.md @@ -2,27 +2,25 @@ ## Goal -`@moq/net`'s moq-transport subscriber fetches a past group, and recovers the -start of the current group when it joins mid-group, so a strict third-party -relay no longer cuts a JS player's join to the next group boundary. JS -subscribers join the CLI FETCH interop round beside Rust. +`@moq/net`'s moq-transport subscriber fetches a past group with a standalone +FETCH, so `track.Consumer.fetchGroup()` works on moq-transport. JS subscribers +join the CLI FETCH interop round beside Rust. ## Plan Found by [moq-dev/moq#7015](https://github.com/moq-dev/moq/pull/7015). Today -`track.Consumer.fetchGroup()` is rejected on moq-transport, and a JS -subscriber asks for neither a joining FETCH (draft 18) nor a fill (draft 22); -see the SUBSCRIBE filter comment in `js/net/src/ietf/subscriber.ts`. The -in-tree relay replays the group, so only third-party relays show the cut. +`track.Consumer.fetchGroup()` is rejected on moq-transport. The mid-group +join already recovers the current group's head, with a joining FETCH on +draft 18 or a fill on draft 22 +([moq-dev/moq#7089](https://github.com/moq-dev/moq/pull/7089)), and its +fetch-stream decoder and head stitching are what a standalone FETCH reuses. Decided 2026-10-28 (maintainer): one [M] quest in m1, after [JS ranges](/quest/m1/subscribe-ranges/js.md), which turns `fetchGroup` into -a one-range subscription. Map a past range to a standalone FETCH, and a -mid-group join to a joining FETCH on draft 18 or a fill on draft 22, the way +a one-range subscription. Map a past range to a standalone FETCH, the way Rust's subscriber does. Then add the native JS runtimes as subscribers to the -`run_fetch_round` cells in the tail suite, and a late-join case through a -relay that does not replay the group. Our publishers answer only one group -([one group](/quest/m1/ietf-publisher-one-group.md), 2026-10-28), so a +`run_fetch_round` cells in the tail suite. Our publishers answer only one +group ([one group](/quest/m1/ietf-publisher-one-group.md), 2026-10-28), so a past-range FETCH against them is refused. Test ranges against a third-party publisher, and one group against ours. diff --git a/test/interop/README.md b/test/interop/README.md index 841db23c19b..1d7d4a8a2ae 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -60,10 +60,9 @@ writes the first two objects of group 0 before any subscriber exists and leaves open. The subscriber joins at the latest group, as a player does, and reports once it has read those two objects; only then does the harness release the rest of the group and group 1. A join that skips the head cannot pass by racing the publisher. On -moq-transport a Rust subscriber, and the relay toward its publisher, recover the head -with a joining FETCH before draft 22 and a fill from it. A JS subscriber requests -neither: on draft 22 it subscribes from the current group's start, and before that the -Rust relay replays the group on the subscription. +moq-transport the subscription starts after Largest Object, so the relay does not replay +the group on it: a subscriber, Rust or JS, and the relay toward its publisher, recover the +head with a joining FETCH before draft 22 and a fill from it. `--tail` runs the FETCH round, the late joins, and the five finite-tail lanes. The wire-compat lanes, which swap in released relays and clients, skip them.