From 16743d6f7183f9d8c2e317fce3fb156bbe261607 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 00:03:49 -0700 Subject: [PATCH 1/5] chore(quest): claim lite-stream-count Co-Authored-By: Claude Opus 5.5 From 3563eb796db0d3d609ba5124b7037033c3cd8d97 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 00:22:29 -0700 Subject: [PATCH 2/5] feat(net): count lite-07 group streams in SUBSCRIBE_END moq-lite-07's SUBSCRIBE_END gains Stream Count, the number of group streams the publisher opened for the subscription, and SUBSCRIBE_DROP is removed from lite-07. The Rust and JS publishers count the streams they open and send SUBSCRIBE_END once no served group is still opening its stream. Earlier versions are unchanged. Co-Authored-By: Claude Opus 5.5 --- drafts/draft-lcurley-moq-lite.md | 47 +++---- js/net/src/lite/publisher.test.ts | 111 ++++++++++++++-- js/net/src/lite/publisher.ts | 204 ++++++++++++++++++------------ js/net/src/lite/subscribe.test.ts | 24 +++- js/net/src/lite/subscribe.ts | 35 +++-- js/net/src/lite/version.ts | 19 +++ quest/m1/README.md | 2 +- quest/m1/announce-compression.md | 2 - quest/m1/lite-count-settle.md | 29 +++++ quest/m1/lite-stream-count.md | 43 ------- rs/moq-net/src/lite/publisher.rs | 161 ++++++++++++++++++++++- rs/moq-net/src/lite/subscribe.rs | 72 +++++++++-- rs/moq-net/src/lite/version.rs | 19 ++- 13 files changed, 565 insertions(+), 203 deletions(-) create mode 100644 quest/m1/lite-count-settle.md delete mode 100644 quest/m1/lite-stream-count.md diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 455a587917..679b2056d6 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -447,7 +447,7 @@ A receiver SHOULD NOT cache refusals; rate limiting is the advertiser's concern. A subscriber opens Subscribe Streams to request a Track. The subscriber MUST start a Subscribe Stream with a SUBSCRIBE message followed by any number of SUBSCRIBE_UPDATE messages. -The publisher replies with a SUBSCRIBE_OK message once the start group is resolved, followed by any number of SUBSCRIBE_END and SUBSCRIBE_DROP messages. +The publisher replies with a SUBSCRIBE_OK message once the start group is resolved, followed by a SUBSCRIBE_END message once the subscription ends. For a live track the publisher MAY withhold SUBSCRIBE_OK until the first matching group resolves the start; if the track has already ended with no matching groups, it sends SUBSCRIBE_END with no preceding SUBSCRIBE_OK. A rejection is a stream reset: a publisher that cannot serve the subscription (no such track, an ended broadcast, or any other refusal) MUST promptly reset the stream rather than leave it pending, so a subscriber distinguishes "pending" from "refused" by the reset, not by a timeout. A route claims capability rather than inventory, so a subscription for a covered path that names nothing is refused this way too. @@ -455,9 +455,8 @@ A route claims capability rather than inventory, so a subscription for a covered The track's immutable publisher properties are not carried here; they are fetched once via a [Track Stream](#track-stream). The subscriber needs the track's TRACK_INFO (notably its timescale) to interpret FRAME messages, and MAY open the Track and Subscribe streams concurrently, buffering frames until it arrives. -The publisher sends SUBSCRIBE_OK once the absolute start position is resolved, and SUBSCRIBE_END once no further groups will be produced (see [SUBSCRIBE_OK](#subscribe-ok) and [SUBSCRIBE_END](#subscribe-end)). -The publisher closes the stream (FIN) only once every group from start to end has been accounted for, either via a Group Stream (completed or reset) or a SUBSCRIBE_DROP message. -This MAY occur after SUBSCRIBE_END, since stragglers within the range can still be dropped. +The publisher sends SUBSCRIBE_OK once the absolute start position is resolved, and SUBSCRIBE_END once no further groups will be produced and every Group Stream it opens for the subscription has been opened (see [SUBSCRIBE_OK](#subscribe-ok) and [SUBSCRIBE_END](#subscribe-end)). +The publisher closes the stream (FIN) after SUBSCRIBE_END, once every counted Group Stream has finished or been reset. Unbounded subscriptions stay open until SUBSCRIBE_END, and either endpoint MAY reset the stream at any time. ### Fetch @@ -1116,7 +1115,7 @@ Set to 0x0 to indicate a SUBSCRIBE_OK message. **Group**: The absolute sequence number of the first group that will be delivered. -It MUST be greater than or equal to the requested start group; any groups in between are unavailable and implicitly dropped, with no separate SUBSCRIBE_DROP required. +It MUST be greater than or equal to the requested start group; any groups in between are unavailable. A subscriber that requested the latest group learns the resolved sequence here. There is no matching frame field, because the start frame is never in doubt: a partial group is only delivered when it was asked for, so the subscription starts either exactly where it asked or at the beginning of a later group (see [Positions](#positions)). @@ -1136,6 +1135,7 @@ SUBSCRIBE_END Message { Type (i) = 0x1 Message Length (i) Group (i) + Stream Count (i) } ~~~ @@ -1147,34 +1147,14 @@ The exclusive end of the range: the absolute sequence number of the first group A value of 0 means the track ended before producing any groups. The subscriber MUST NOT wait for any group at or after this sequence. -SUBSCRIBE_END bounds the range but does not by itself end the stream: the publisher MAY still send SUBSCRIBE_DROP for groups below this sequence that it cannot deliver, and FINs the stream only once every group below this sequence has been accounted for. +**Stream Count**: +The number of Group Streams the publisher opened for this subscription, whether they finished or were reset. +A group that was skipped, never produced, delivered only as a datagram, or given up before its stream opened is not counted. +A relay counts the Group Streams it opened itself, never the count it received upstream. -## SUBSCRIBE_DROP -A SUBSCRIBE_DROP message is sent by the publisher on the Subscribe Stream when groups cannot be served. -It MAY arrive at any point after the subscription is opened, including after SUBSCRIBE_END for stragglers within the resolved range (a leading range is instead dropped implicitly by SUBSCRIBE_OK). - -~~~ -SUBSCRIBE_DROP Message { - Type (i) = 0x2 - Message Length (i) - Group Start (i) - Group End (i) - Error Code (i) -} -~~~ - -**Type**: -Set to 0x2 to indicate a SUBSCRIBE_DROP message. - -**Group Start**: -The first absolute group sequence in the dropped range. - -**Group End**: -The last absolute group sequence in the dropped range (inclusive). - -**Error Code**: -An application-specific error code. -A value of 0 indicates no error; the groups are simply unavailable. +The publisher MUST NOT send SUBSCRIBE_END until every Group Stream it will open for the subscription has been opened, so the count is final; it does not wait for them to finish. +The subscriber has received every Group Stream once it has read the header of `Stream Count` of them, which MAY happen after SUBSCRIBE_END or the FIN since streams are not ordered. +A Group Stream reset before its header arrived is never seen, so a subscriber SHOULD bound how long it waits for the rest, for example by `Subscriber Max Age`. ## FETCH FETCH is sent by a subscriber to request a single group from a track. @@ -1332,6 +1312,9 @@ The `Message Length` describes the payload size on the wire. - Assigned `moq-lite-07-wip` as this draft's protocol identifier until it is finalized as `moq-lite-07`. - Hid routes with a `.`-prefixed segment below the requested prefix from announce discovery, and added the ANNOUNCE_REQUEST `Hidden` field to opt in. +- Added `Stream Count` to SUBSCRIBE_END: the number of Group Streams opened for the subscription. SUBSCRIBE_END is now sent once every counted Group Stream has opened, rather than as soon as the final group is known. +- Removed SUBSCRIBE_DROP and its type 0x2; a group without a Group Stream is not counted. +- The Subscribe Stream FIN now follows once every counted Group Stream has finished or been reset. ## moq-lite-06 diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 30b1f5bbce..125e1f223a 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -15,8 +15,8 @@ import { Group as GroupMessage } from "./group.ts"; import { sendOrder } from "./priority.ts"; import { Probe as ProbeMessage } from "./probe.ts"; import { Publisher } from "./publisher.ts"; -import { decodeSubscribeResponse, Subscribe, SubscribeUpdate } from "./subscribe.ts"; -import { ALPN_05, ALPN_06, Version } from "./version.ts"; +import { decodeSubscribeResponse, Subscribe, type SubscribeEnd, SubscribeUpdate } from "./subscribe.ts"; +import { ALPN_05, ALPN_06, ALPN_07_WIP, Version } from "./version.ts"; function publish(origin: OriginProducer, path: Path.Valid) { const broadcast = origin.createBroadcast(path); @@ -85,11 +85,11 @@ test.each([Version.DRAFT_01, Version.DRAFT_03, Version.DRAFT_06])( ); // Delivers `sequences` in the given order, finishes the track, and returns the -// SUBSCRIBE_END boundary the publisher put on the wire. -async function subscribeEnd(sequences: number[]): Promise { - const pair = createMockTransportPair(ALPN_05); +// SUBSCRIBE_END the publisher put on the wire. +async function subscribeEnd(sequences: number[], version = Version.DRAFT_05): Promise { + const pair = createMockTransportPair(version === Version.DRAFT_07 ? ALPN_07_WIP : ALPN_05); const origin = new OriginProducer(); - const publisher = new Publisher(pair.server, Version.DRAFT_05, randomHop(), origin.consume()); + const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); const broadcast = publish(origin, Path.from("test")); const track = broadcast.createTrack("video"); @@ -118,8 +118,8 @@ async function subscribeEnd(sequences: number[]): Promise { try { for (;;) { - const resp = await decodeSubscribeResponse(client.reader, Version.DRAFT_05); - if ("end" in resp) return resp.end.group; + const resp = await decodeSubscribeResponse(client.reader, version); + if ("end" in resp) return resp.end; } } finally { publisher.close(); @@ -972,19 +972,108 @@ test("lite draft-05: teardown unwinds with an undelivered update queued", async // A Rust subscriber feeds this value straight into `track::Producer::finish_at`, which is // exclusive, so an inclusive bound here silently truncates the final group across languages. test("lite draft-05: subscribe end is the exclusive boundary", async () => { - expect(await subscribeEnd([0, 1, 2])).toBe(3); + expect((await subscribeEnd([0, 1, 2])).group).toBe(3); }); // recvGroup is arrival-ordered, so the boundary has to clear the max sequence delivered, // not the last one seen. Otherwise the boundary lands on a group already on the wire. test("lite draft-05: subscribe end clears the max sequence when groups arrive out of order", async () => { - expect(await subscribeEnd([0, 2, 1])).toBe(3); + expect((await subscribeEnd([0, 2, 1])).group).toBe(3); }); // 0 is the only encoding for "no groups at all"; an inclusive bound cannot express it // without colliding with a track whose sole group was sequence 0. test("lite draft-05: subscribe end is 0 when no groups were produced", async () => { - expect(await subscribeEnd([])).toBe(0); + expect((await subscribeEnd([])).group).toBe(0); +}); + +// The count is of group streams opened, not of groups below the end: a group the track +// never produced has no stream and is not counted. +test("lite draft-07: subscribe end counts the group streams opened", async () => { + const end = await subscribeEnd([0, 2], Version.DRAFT_07); + expect([end.group, end.streams]).toEqual([3, 2]); +}); + +test("lite draft-07: subscribe end counts zero streams when no groups were produced", async () => { + const end = await subscribeEnd([], Version.DRAFT_07); + expect([end.group, end.streams]).toEqual([0, 0]); +}); + +// Serves one group with its stream open held until `open(ok)`, and returns the pending +// SUBSCRIBE_END plus the call that lets the open succeed or fail. +async function heldOpenEnd() { + const pair = createMockTransportPair(ALPN_07_WIP); + const origin = new OriginProducer(); + const publisher = new Publisher(pair.server, Version.DRAFT_07, randomHop(), origin.consume()); + const broadcast = publish(origin, Path.from("test")); + const track = broadcast.createTrack("video"); + + let open!: (ok: boolean) => void; + const opened = new Promise((resolve) => { + open = resolve; + }); + const createUni = pair.server.createUnidirectionalStream.bind(pair.server); + spyOn(pair.server, "createUnidirectionalStream").mockImplementation(async (options) => { + if (!(await opened)) throw new Error("no stream credit"); + return createUni(options); + }); + + const client = await Stream.open(pair.client); + const server = await Stream.accept(pair.server); + if (!server) throw new Error("publisher never accepted the subscribe stream"); + void publisher.runSubscribe( + new Subscribe({ id: 0n, broadcast: Path.from("test"), track: "video", priority: 0 }), + server, + ); + + const group = new GroupProducer(0); + group.writeString("hello"); + group.close(); + track.writeGroup(group); + track.close(); + + const start = await decodeSubscribeResponse(client.reader, Version.DRAFT_07); + expect("start" in start).toBe(true); + const end = decodeSubscribeResponse(client.reader, Version.DRAFT_07); + + return { + end, + open, + close() { + publisher.close(); + client.close(); + }, + }; +} + +// The count is final only once no served group is still waiting for its stream, so +// SUBSCRIBE_END waits for the open. +test("lite draft-07: subscribe end waits for every group stream to open", async () => { + const held = await heldOpenEnd(); + try { + const early = await Promise.race([held.end, new Promise((resolve) => setTimeout(resolve, IDLE_MS))]); + expect(early).toBeUndefined(); + + held.open(true); + const resp = await held.end; + if (!("end" in resp)) throw new Error("expected SUBSCRIBE_END"); + expect([resp.end.group, resp.end.streams]).toEqual([1, 1]); + } finally { + held.close(); + } +}); + +// A group that never gets a stream owes the subscriber nothing, so it is not counted. +test("lite draft-07: a group whose stream never opened is not counted", async () => { + const held = await heldOpenEnd(); + try { + held.open(false); + const resp = await held.end; + if (!("end" in resp)) throw new Error("expected SUBSCRIBE_END"); + expect([resp.end.group, resp.end.streams]).toEqual([1, 0]); + } finally { + held.close(); + } }); /** One group stream the publisher put on the wire. */ diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index b9fb8dc098..c7e8ee2030 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -27,7 +27,15 @@ import { SubscribeUpdate, } from "./subscribe.ts"; import { TrackInfo as TrackInfoMessage, type Track as TrackMessage } from "./track.ts"; -import { hasAnnounceId, hasAnnounceOk, hasDatagrams, hasProbeRtt, resolvesStart, Version } from "./version.ts"; +import { + hasAnnounceId, + hasAnnounceOk, + hasDatagrams, + hasProbeRtt, + hasStreamCount, + resolvesStart, + Version, +} from "./version.ts"; // Where each originated route lands under the requested prefix: its suffix beneath // the prefix, or the empty suffix for a route above it, where the most specific @@ -64,7 +72,7 @@ function zigzag(delta: bigint): bigint { return delta >= 0n ? delta << 1n : (-delta << 1n) - 1n; } -/** What {@link Publisher.runGroup} needs to serve one group. */ +/** What {@link Publisher.openGroup} and {@link Publisher.serveGroup} need to serve one group. */ interface RunGroup { /** The subscription ID. */ sub: bigint; @@ -737,6 +745,13 @@ export class Publisher { let startSent = false; let endSent = false; + // Lite-07+ counts the group streams in SUBSCRIBE_END, so it goes out only once every + // served group has opened its stream or given up, and a cap holding groups back + // delays it until they are released. + const countStreams = hasStreamCount(this.version); + let streams = 0; + const opening = new Set>(); + // The track's exclusive final boundary. A Rust subscriber feeds SUBSCRIBE_END // straight into finish_at, so it must name the track's boundary (which counts // datagram sequences too), not the delivered range: a subscription cap can hold @@ -744,17 +759,20 @@ export class Publisher { // before the producer declared it. const boundary = () => track.final() ?? (track.latest() ?? -1) + 1; - // SUBSCRIBE_END names that boundary, which a cap can hold groups back from, so it goes - // out as soon as the producer finishes and the subscription keeps serving whatever a - // later cap raise releases (see the Rust publisher's Recv::Boundary). + // Before lite-07, SUBSCRIBE_END names that boundary, which a cap can hold groups back + // from, so it goes out as soon as the producer finishes and the subscription keeps + // serving whatever a later cap raise releases (see the Rust publisher's Recv::Boundary). const sendEnd = async (): Promise => { endSent = true; - if (emitRange) { - return controls.response( - encodeSubscribeResponse(stream, { end: new SubscribeEnd(boundary()) }, this.version), - ); - } - return true; + if (!emitRange) return true; + return controls.response( + (async () => { + // A group that gives up before its stream opens is never counted. + if (countStreams) while (opening.size > 0) await Promise.all(opening); + const end = new SubscribeEnd(boundary(), streams); + await encodeSubscribeResponse(stream, { end }, this.version); + })(), + ); }; // One ranking for the whole subscription, shared by every group it serves. @@ -821,7 +839,7 @@ export class Publisher { case "boundary": // The producer finished but is still holding groups above the cap. // Declare the boundary, then wait for an update to release them. - if (!endSent) { + if (!endSent && !countStreams) { if (!(await sendEnd())) return; continue; } @@ -868,7 +886,7 @@ export class Publisher { return; } - const task = this.#runGroup({ + const options: RunGroup = { sub, group, timescale, @@ -876,9 +894,20 @@ export class Publisher { unsubscribed, start: range.start, end: range.end, + }; + // `opening` settles when the stream opens so the lite-07 count can be sent. + // `groups` covers the serve too, so the FIN still waits for every stream to + // finish or reset, including one that has not opened yet. + const opened = this.#openGroup(options); + const task = opened.then(async (writer) => { + if (!writer) return; + streams += 1; + await this.#serveGroup(writer, options); }); groups.add(task); void task.finally(() => groups.delete(task)); + opening.add(opened); + void opened.finally(() => opening.delete(opened)); } } finally { if (!finished) unsubscribe(); @@ -930,7 +959,7 @@ export class Publisher { // relays re-serve with the same window. maxAge: info.maxAge, // Lite05 mandates per-frame timestamps. Advertise the track's timescale; - // `#runGroup` emits each frame converted to it. + // `#serveGroup` emits each frame converted to it. timescale: info.timescale, }); })(); @@ -959,7 +988,7 @@ export class Publisher { const datagram = await track.recvDatagram(); if (!datagram) return; // Track finished; #runTrack tears the subscription down. - // Convert the timestamp to the track's advertised timescale, matching #runGroup. + // Convert the timestamp to the track's advertised timescale, matching #serveGroup. const ts = Math.round(datagram.timestamp.as(timescale)); const body = new DatagramMessage(sub, datagram.sequence, ts, datagram.payload).encode(); @@ -1009,15 +1038,13 @@ export class Publisher { } /** - * Serves one group on its own unidirectional stream. + * Opens the unidirectional stream for one group, or closes the group and resolves + * `undefined` when it cannot get one. * * @internal */ - async #runGroup(options: RunGroup) { - const { sub, group, timescale, priority, unsubscribed, start: startFrame, end: endFrame } = options; - // This model holds whole groups, so frame `startFrame` is always reachable unless - // the group ends first. Declaring it up front keeps the stream self-describing. - const msg = new GroupMessage({ subscribe: sub, sequence: group.sequence, frameStart: startFrame }); + async #openGroup(options: RunGroup): Promise { + const { group, priority, unsubscribed } = options; try { // The transport drains streams by send order, so this is what makes a high-priority // track (and a newer group within it) win the link when there isn't room for both. @@ -1031,79 +1058,88 @@ export class Publisher { cancel: unsubscribed, waitUntilAvailable: false, }); - if (!stream) { - group.close(new Error("no stream slot")); - return; - } - - // Everything past this point runs inside the cleanup scope, so a failure never leaves - // a finished group's stream being ranked. - try { - // A SUBSCRIBE_UPDATE re-ranks the subscription, so a group already on the wire - // follows it too rather than keeping a stale rank until it finishes. - priority.add(stream, group.sequence); + if (!stream) group.close(new Error("no stream slot")); + return stream; + } catch (err: unknown) { + group.close(error(err)); + return undefined; + } + } - await hooks.guardGroup( - group, - (async () => { - await stream.u53(0); // stream type - await msg.encode(stream, this.version); - })(), - ); + /** + * Serves one group on the stream {@link #openGroup} opened for it. + * + * @internal + */ + async #serveGroup(stream: Writer, options: RunGroup) { + const { sub, group, timescale, priority, start: startFrame, end: endFrame } = options; + // This model holds whole groups, so frame `startFrame` is always reachable unless + // the group ends first. Declaring it up front keeps the stream self-describing. + const msg = new GroupMessage({ subscribe: sub, sequence: group.sequence, frameStart: startFrame }); + // Everything past this point runs inside the cleanup scope, so a failure never leaves + // a finished group's stream being ranked. + try { + // A SUBSCRIBE_UPDATE re-ranks the subscription, so a group already on the wire + // follows it too rather than keeping a stale rank until it finishes. + priority.add(stream, group.sequence); + + await hooks.guardGroup( + group, + (async () => { + await stream.u53(0); // stream type + await msg.encode(stream, this.version); + })(), + ); - // Lite05+ prefixes every frame with a zigzag-delta timestamp at the track's - // advertised timescale; older drafts omit it. - const timestamps = supportsTrackStream(this.version); - let prevTs = 0n; - // Whether the cursor ever reached the requested start, which decides how the - // end of the group is read below. - let reached = startFrame === 0; - - for (;;) { - const read = await race([hooks.readGroupFrame(group), stream.closed]); - if (!read) { - // The group ended before the frame the subscriber asked to start - // at, so this publisher can't serve the range at all. FINning here - // would claim an empty group under that index; reset so it reads - // as the gap it is. - if (!reached) throw new Error(`group ended before frame ${startFrame}`); - break; - } + // Lite05+ prefixes every frame with a zigzag-delta timestamp at the track's + // advertised timescale; older drafts omit it. + const timestamps = supportsTrackStream(this.version); + let prevTs = 0n; + // Whether the cursor ever reached the requested start, which decides how the + // end of the group is read below. + let reached = startFrame === 0; - try { - // A group that ends exactly at the start is a valid, empty range. - if (read.sequence + 1 >= startFrame) reached = true; - // Frames below the requested start were excluded, and the receiver - // numbers what it gets from `startFrame`. - if (read.sequence < startFrame) continue; - if (endFrame !== undefined && read.sequence > endFrame) break; - - if (timestamps) { - // Convert each frame to the track's advertised timescale. - const ts = BigInt(Math.round(read.frame.timestamp.as(timescale))); - await hooks.guardGroup(group, stream.u62(zigzag(ts - prevTs))); - prevTs = ts; - } + for (;;) { + const read = await race([hooks.readGroupFrame(group), stream.closed]); + if (!read) { + // The group ended before the frame the subscriber asked to start + // at, so this publisher can't serve the range at all. FINning here + // would claim an empty group under that index; reset so it reads + // as the gap it is. + if (!reached) throw new Error(`group ended before frame ${startFrame}`); + break; + } - await hooks.guardGroup(group, stream.u53(read.frame.payload.byteLength)); - await hooks.guardGroup(group, stream.write(read.frame.payload)); - } finally { - read.complete(); + try { + // A group that ends exactly at the start is a valid, empty range. + if (read.sequence + 1 >= startFrame) reached = true; + // Frames below the requested start were excluded, and the receiver + // numbers what it gets from `startFrame`. + if (read.sequence < startFrame) continue; + if (endFrame !== undefined && read.sequence > endFrame) break; + + if (timestamps) { + // Convert each frame to the track's advertised timescale. + const ts = BigInt(Math.round(read.frame.timestamp.as(timescale))); + await hooks.guardGroup(group, stream.u62(zigzag(ts - prevTs))); + prevTs = ts; } - } - stream.close(); - group.close(); - } catch (err: unknown) { - const e = error(err); - stream.reset(e); - group.close(e); - } finally { - priority.remove(stream); + await hooks.guardGroup(group, stream.u53(read.frame.payload.byteLength)); + await hooks.guardGroup(group, stream.write(read.frame.payload)); + } finally { + read.complete(); + } } + + stream.close(); + group.close(); } catch (err: unknown) { const e = error(err); + stream.reset(e); group.close(e); + } finally { + priority.remove(stream); } } diff --git a/js/net/src/lite/subscribe.test.ts b/js/net/src/lite/subscribe.test.ts index 7f8f5e7fa5..afd4c7c0a6 100644 --- a/js/net/src/lite/subscribe.test.ts +++ b/js/net/src/lite/subscribe.test.ts @@ -152,10 +152,30 @@ test("SubscribeStart round-trips on draft-05", async () => { }); test("SubscribeEnd round-trips on draft-05", async () => { - const got = await responseRoundtrip(Version.DRAFT_05, { end: new SubscribeEnd(7) }); + // Type, length, group: no stream count before draft-07. + expect(await encode(Version.DRAFT_05, { end: new SubscribeEnd(7, 3) })).toEqual(new Uint8Array([1, 1, 7])); + const got = await responseRoundtrip(Version.DRAFT_05, { end: new SubscribeEnd(7, 3) }); expect("end" in got).toBe(true); if (!("end" in got)) throw new Error("expected end"); - expect(got.end.group).toBe(7); + expect([got.end.group, got.end.streams]).toEqual([7, 0]); +}); + +test("SubscribeEnd carries the stream count on draft-07", async () => { + expect(await encode(Version.DRAFT_07, { end: new SubscribeEnd(7, 3) })).toEqual(new Uint8Array([1, 2, 7, 3])); + const got = await responseRoundtrip(Version.DRAFT_07, { end: new SubscribeEnd(7, 3) }); + if (!("end" in got)) throw new Error("expected end"); + expect([got.end.group, got.end.streams]).toEqual([7, 3]); +}); + +test("SubscribeDrop is gone on draft-07", async () => { + const drop: SubscribeResponse = { drop: new SubscribeDrop({ start: 1, end: 3, error: 0 }) }; + await expect(encode(Version.DRAFT_07, drop)).rejects.toThrow(); + + // A draft-06 DROP is an unknown response type on draft-07. + const wire06 = await encode(Version.DRAFT_06, drop); + await expect(decodeSubscribeResponse(new Reader(undefined, wire06), Version.DRAFT_07)).rejects.toThrow( + "unknown subscribe response type: 2", + ); }); test("SubscribeDrop is type 0x2 on draft-05 and 0x1 on draft-04", async () => { diff --git a/js/net/src/lite/subscribe.ts b/js/net/src/lite/subscribe.ts index c5a5b37b78..3d1a1f2711 100644 --- a/js/net/src/lite/subscribe.ts +++ b/js/net/src/lite/subscribe.ts @@ -1,7 +1,7 @@ import * as Path from "../path.ts"; import type { Reader, Writer } from "../stream.ts"; import * as Message from "./message.ts"; -import { hasFrameBounds, hasGroupOrder, resolvesStart, Version } from "./version.ts"; +import { hasFrameBounds, hasGroupOrder, hasStreamCount, resolvesStart, Version } from "./version.ts"; /** * Encode the `Group Start` field shared by SUBSCRIBE and SUBSCRIBE_UPDATE. @@ -456,24 +456,35 @@ export class SubscribeEnd { /** The exclusive final group sequence: the first sequence that will never be produced. */ group: number; - constructor(group: number) { + /** + * The number of group streams the publisher opened for this subscription. + * Draft-07+ only; not on the wire before, where it decodes as 0. + */ + streams: number; + + constructor(group: number, streams = 0) { this.group = group; + this.streams = streams; } - async encode(w: Writer): Promise { + async encode(w: Writer, version: Version): Promise { return Message.encode(w, async (w) => { await w.u53(this.group); + if (hasStreamCount(version)) await w.u53(this.streams); }); } - static async decode(r: Reader): Promise { - return Message.decode(r, async (r) => new SubscribeEnd(await r.u53())); + static async decode(r: Reader, version: Version): Promise { + return Message.decode( + r, + async (r) => new SubscribeEnd(await r.u53(), hasStreamCount(version) ? await r.u53() : 0), + ); } } /// Indicates that one or more groups have been dropped. /// -/// Draft03+ only. +/// Draft-03 to Draft-06 only: Draft-07 counts group streams in SUBSCRIBE_END instead. export class SubscribeDrop { start: number; end: number; @@ -510,8 +521,9 @@ export class SubscribeDrop { * * The discriminator is version-dependent: * - Draft-03/04: `0x0` SUBSCRIBE_OK, `0x1` SUBSCRIBE_DROP. - * - Draft-05+: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END, `0x2` SUBSCRIBE_DROP + * - Draft-05/06: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END, `0x2` SUBSCRIBE_DROP * (SUBSCRIBE_OK was removed; acceptance is implicit). + * - Draft-07+: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END (SUBSCRIBE_DROP was removed). */ export type SubscribeResponse = | { ok: SubscribeOk } @@ -548,12 +560,12 @@ export async function encodeSubscribeResponse(w: Writer, resp: SubscribeResponse await resp.start.encode(w); } else if ("end" in resp) { await w.u53(0x1); - await resp.end.encode(w); - } else if ("drop" in resp) { + await resp.end.encode(w, version); + } else if ("drop" in resp && !hasStreamCount(version)) { await w.u53(0x2); await resp.drop.encode(w); } else { - throw new Error("SUBSCRIBE_OK not supported for this version"); + throw new Error("subscribe response not supported for this version"); } break; } @@ -582,8 +594,9 @@ export async function decodeSubscribeResponse(r: Reader, version: Version): Prom case 0x0: return { start: await SubscribeStart.decode(r) }; case 0x1: - return { end: await SubscribeEnd.decode(r) }; + return { end: await SubscribeEnd.decode(r, version) }; case 0x2: + if (hasStreamCount(version)) throw new Error(`unknown subscribe response type: ${typ}`); return { drop: await SubscribeDrop.decode(r) }; default: throw new Error(`unknown subscribe response type: ${typ}`); diff --git a/js/net/src/lite/version.ts b/js/net/src/lite/version.ts index 4fda174243..28d0470dd9 100644 --- a/js/net/src/lite/version.ts +++ b/js/net/src/lite/version.ts @@ -230,6 +230,25 @@ export function hasHidden(version: Version): boolean { } } +/** + * Whether SUBSCRIBE_END carries the subscription's group stream count, sent once every + * counted stream is open, in place of SUBSCRIBE_DROP. Added in lite-07. + */ +export function hasStreamCount(version: Version): boolean { + // Explicitly list older versions so future versions keep the lite-07+ behavior. + switch (version) { + case Version.DRAFT_01: + case Version.DRAFT_02: + case Version.DRAFT_03: + case Version.DRAFT_04: + case Version.DRAFT_05: + case Version.DRAFT_06: + return false; + default: + return true; + } +} + /// The WebTransport subprotocol identifier for moq-lite. /// Version negotiation still happens via SETUP when this is used. export const ALPN = "moql"; diff --git a/quest/m1/README.md b/quest/m1/README.md index 1f4d300698..5ef1b79dd9 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -19,8 +19,8 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Missing fetch group](/quest/m1/fetch-missing-group.md) - HTTP /fetch answers 404 and `moq fetch` fails cleanly for a group the track lacks - [libmoq hidden opt-in](/quest/m1/libmoq-hidden.md) - `moq_origin_announced` takes a `hidden` flag so C callers can list `.`-named broadcasts -- [lite-07 stream count](/quest/m1/lite-stream-count.md) - moq-lite-07 replaces SUBSCRIBE_DROP with a group-stream count in SUBSCRIBE_END, like moq-transport - [Announce compression](/quest/m1/announce-compression.md) - a lite-07 announce reuses the path head and hop-chain tail of a live announcement on its stream instead of resending them +- [lite-07 count settle](/quest/m1/lite-count-settle.md) - moq-lite-07 subscribers stop waiting for a subscription's tail once SUBSCRIBE_END's stream count is reached - [Session death error](/quest/m1/session-death-error.md) - a dying session ends its tracks with its own error in Rust and JS, never a clean end, `Dropped`, or `Cancel` - [JS bare FIN](/quest/m1/js-bare-fin.md) - a `@moq/net` subscriber aborts a track whose subscribe stream FINs before its declared end, like Rust - [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop` diff --git a/quest/m1/announce-compression.md b/quest/m1/announce-compression.md index c7c61169c1..cb0dc30bc7 100644 --- a/quest/m1/announce-compression.md +++ b/quest/m1/announce-compression.md @@ -102,7 +102,5 @@ Hops { ## Related -- [lite-07 stream count](/quest/m1/lite-stream-count.md) - the other lite-07 - wire change, landing in the same unreleased version - [Relay memory](/quest/m1/relay-memory.md) - route state per relay, which this leaves unchanged diff --git a/quest/m1/lite-count-settle.md b/quest/m1/lite-count-settle.md new file mode 100644 index 0000000000..0ff1f18a71 --- /dev/null +++ b/quest/m1/lite-count-settle.md @@ -0,0 +1,29 @@ +# [S] lite-07 subscribers settle on the stream count + +## Goal + +On moq-lite-07, Rust and JS subscribers stop waiting for a subscription's tail +once they have read the headers of as many group streams as SUBSCRIBE_END +counts, so a group the publisher skipped or never opened costs no grace. A +late stream below the end is still accepted within the grace, which stays for +a stream reset before its header arrived. lite-05 and -06 keep the DROP +accounting JS already has and Rust track tail adds. + +## Plan + +- Both publishers already send the count, and both subscribers decode it + (`lite::SubscribeEnd::streams` in Rust, `SubscribeEnd.streams` in JS) and + ignore it. +- JS track tail has landed, and its `Tail` already counts streams. Rust track + tail has landed (`rs/moq-net/src/tail.rs`), so lite-07's completion check + becomes "headers read >= Stream Count" after SUBSCRIBE_END, in place of + every sequence from start to end being covered. The Rust subscriber needs + the same count. +- Tests in both languages: a late stream after SUBSCRIBE_END, a skipped group + that settles without the grace, a reset stream, and a count of zero. Add + the Rust-JS case to the track tail interop test. + +## Related + +- [Track tail interop](/quest/m1/track-tail-interop.md) - the Rust-JS case this adds its count check to +- [Reliable stream reset](/quest/m1/quic/reliable-reset.md) - makes the count exact by keeping a reset stream's header diff --git a/quest/m1/lite-stream-count.md b/quest/m1/lite-stream-count.md deleted file mode 100644 index 053984fc56..0000000000 --- a/quest/m1/lite-stream-count.md +++ /dev/null @@ -1,43 +0,0 @@ -# [M] moq-lite-07 counts group streams instead of dropping groups - -## Goal - -On moq-lite-07, a subscriber knows a subscription has delivered everything -once it has seen as many group streams as the publisher opened, the way -moq-transport's PUBLISH_DONE Stream Count works. SUBSCRIBE_DROP is gone from -lite-07: a group the publisher skipped or never opened is simply not counted, -so nothing has to name it. Published versions (lite-01 to -06) keep decoding -SUBSCRIBE_DROP unchanged. - -Where reliable reset is negotiated, the count is exact and the subscriber -waits for nothing else. Where it is not (browsers today), a stream reset -before its header arrived is still invisible, so the track-tail grace stays. - -## Plan - -- Wire: SUBSCRIBE_END gains `Stream Count`, the number of group streams the - publisher opened for this subscription. The publisher sends it once every - group stream below the end has been opened (not finished), like - PUBLISH_DONE, so the boundary arrives slightly later than today. Remove - SUBSCRIBE_DROP and its type from lite-07 and reword the Subscribe Stream - section: the FIN follows once every counted stream has finished or been - reset. lite-07 is still work-in-progress (`moq-lite-07-wip`), so this changes it in place; update - `drafts/draft-lcurley-moq-lite.md` and its changelog. -- Rust and JS publishers count the streams they open per subscription and - send the count; a relay counts its own downstream streams, never forwarding - the upstream count. -- Subscribers on lite-07 stop waiting once the count is reached, accepting a - late stream below the end within the grace. On lite-05 and -06, the - DROP accounting already in moq-net (`rs/moq-net/src/tail.rs`) and `@moq/net` - (`js/net/src/tail.ts`) stays as it is. -- Tests in both languages: a late stream after SUBSCRIBE_END, a skipped group - that is never counted, a reset stream, and a count of zero. Add a Rust-JS - interop case. - -This lands before lite-07 is finalized. Published drafts keep SUBSCRIBE_DROP, -which moq-net and `@moq/net` already account. lite-07 replaces it with the -count in both. Rust still does not send SUBSCRIBE_DROP. - -## Related - -- [Reliable stream reset](/quest/m1/quic/reliable-reset.md) - makes the count exact by keeping a reset stream's header diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index a841eefa34..c0ec29e3c2 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3,7 +3,10 @@ use crate::{SessionError, announce, frame, group, origin, track}; use std::{ collections::HashMap, ops::Bound, - sync::Arc, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, task::{Context, Poll, ready}, time::Duration, }; @@ -1151,6 +1154,7 @@ impl SubscribeServe { track_priority_seen: msg.priority, version: self.shared.version, timescale, + opens: Default::default(), }; let run = TrackRun::new(sub, track, Bounds::from(&msg), track_priority_tx); @@ -2134,6 +2138,18 @@ struct Subscription { /// Negotiated timestamp scale for this track. `Some(_)` on lite-05+ after /// TRACK_INFO; used to validate per-frame timestamps before encoding. timescale: Option, + /// The group streams this subscription opened, shared by every group it serves. + opens: Arc, +} + +/// Counts a subscription's group streams for lite-07's SUBSCRIBE_END. +/// +/// A group is pending from the moment it is queued until its stream opens, or it gives +/// up first (expired, or the open failed) and is never counted. +#[derive(Default)] +struct Opens { + pending: AtomicU64, + opened: AtomicU64, } impl Subscription { @@ -2216,6 +2232,9 @@ struct TrackRun { emit_range: bool, start_sent: bool, end_sent: bool, + // Lite07+ sends SUBSCRIBE_END with the stream count instead of as soon as the + // boundary is known, once every group below it has opened its stream. + count_streams: bool, // Serve datagrams off this same subscriber, but only on lite-05+ over a // datagram-capable transport (qmux/WebSocket/TCP/UDS report size 0). No group // fallback: otherwise off. @@ -2237,6 +2256,7 @@ impl TrackRun { track.end_at(bounds.end_group.map_or(Bound::Unbounded, Bound::Included)); let emit_range = ctx.version.has_track_stream(); + let count_streams = ctx.version.has_stream_count(); let datagrams = ctx.version.has_datagrams() && ctx.session.max_datagram_size() > 0; Self { @@ -2248,6 +2268,7 @@ impl TrackRun { emit_range, start_sent: false, end_sent: false, + count_streams, datagrams, children: kio::Tasks::new(), } @@ -2300,7 +2321,7 @@ impl TrackRun { // group and, when enabled, the next best-effort datagram. Groups are polled // first so a datagram burst can't starve them; datagrams flow whenever no // group is ready (including while groups are parked above the cap). - let emit_boundary = self.emit_range && !self.end_sent; + let emit_boundary = self.emit_range && !self.end_sent && !self.count_streams; if let Poll::Ready(res) = poll_recv_next(&mut self.track, self.datagrams, emit_boundary, waiter) { match res? { Recv::Group(mut group) => { @@ -2352,7 +2373,22 @@ impl TrackRun { self.end_sent = true; stream .writer - .buffer(&lite::SubscribeResponse::End(lite::SubscribeEnd { group }))?; + .buffer(&lite::SubscribeResponse::End(lite::SubscribeEnd { group, streams: 0 }))?; + } + Recv::Finished if self.count_streams && !self.end_sent => { + // The count is final only once no served group is still waiting to + // open its stream. A group that gives up first is never counted, so + // the subscriber is not left waiting for it. The child's wake + // re-polls this loop. + if self.ctx.opens.pending.load(Ordering::Relaxed) > 0 { + return Poll::Pending; + } + let group = ready!(self.track.poll_finished(waiter))?; + let streams = self.ctx.opens.opened.load(Ordering::Relaxed); + self.end_sent = true; + stream + .writer + .buffer(&lite::SubscribeResponse::End(lite::SubscribeEnd { group, streams }))?; } Recv::Finished => return Poll::Ready(Ok(TrackEnd::Finished)), } @@ -2410,6 +2446,7 @@ impl GroupServe { priority: PriorityHandle, group: group::Consumer, ) -> Self { + ctx.opens.pending.fetch_add(1, Ordering::Relaxed); Self { ctx, priority, @@ -2421,6 +2458,14 @@ impl GroupServe { } } + /// Leave [`GroupState::Open`], counting the stream if it opened. + fn settle_open(&mut self, opened: bool) { + self.ctx.opens.pending.fetch_sub(1, Ordering::Relaxed); + if opened { + self.ctx.opens.opened.fetch_add(1, Ordering::Relaxed); + } + } + /// Serve the group, aborting the stream with the real reason (Old, Lagged, /// Evicted, ...) on failure so the subscriber can tell a truncated group from /// a routine cancel. Without this the Writer's Drop fallback would report @@ -2430,6 +2475,7 @@ impl GroupServe { match &mut self.state { GroupState::Open => { if self.group.poll_expired(waiter) { + self.settle_open(false); self.state = GroupState::Done; return Poll::Ready(Err(Error::Old)); } @@ -2437,10 +2483,12 @@ impl GroupServe { let stream = match ready!(self.ctx.session.poll_open_uni(&mut cx)) { Ok(stream) => stream, Err(err) => { + self.settle_open(false); self.state = GroupState::Done; return Poll::Ready(Err(Error::from_transport(err))); } }; + self.settle_open(true); let mut writer = Writer::new(stream, self.ctx.version); writer.set_priority(self.priority.send_order()); @@ -2768,6 +2816,7 @@ mod serve_group_test { track_priority_seen: 0, version: Version::Lite06, timescale: Some(crate::Timescale::default()), + opens: Default::default(), }; let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); @@ -2808,6 +2857,7 @@ mod serve_group_test { track_priority_seen: 0, version: Version::Lite06, timescale: Some(crate::Timescale::default()), + opens: Default::default(), }; let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); @@ -2852,6 +2902,7 @@ mod serve_group_test { track_priority_seen: 0, version: Version::Lite06, timescale: Some(crate::Timescale::default()), + opens: Default::default(), }; let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); @@ -2914,6 +2965,7 @@ mod serve_group_test { track_priority_seen: 0, version: Version::Lite06, timescale: Some(crate::Timescale::default()), + opens: Default::default(), }; let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); @@ -2983,6 +3035,7 @@ mod serve_group_test { track_priority_seen: 0, version: Version::Lite06, timescale: Some(crate::Timescale::default()), + opens: Default::default(), }; let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); @@ -3007,6 +3060,108 @@ mod serve_group_test { "rank 0 must reach the transport as send order 255: {priorities:?}", ); } + + /// A lite-07 subscription's run loop, from group 0, and the log of its subscribe stream. + fn lite07_run( + session: SinkSession, + track: track::Subscriber, + ) -> (TrackRun, Stream, Log) { + let log = Log::default(); + let stream = Stream { + writer: Writer::new(SinkSend::new(log.clone()), Version::Lite07), + reader: crate::coding::Reader::new(PendingRecv, Version::Lite07), + }; + let track_priority = kio::Producer::new(0u8); + let ctx = Subscription { + session, + id: 0, + track_name: "test".into(), + priority: PriorityQueue::default(), + track_priority: track_priority.consume(), + track_priority_seen: 0, + version: Version::Lite07, + timescale: Some(crate::Timescale::default()), + opens: Default::default(), + }; + let bounds = Bounds { + start_group: Some(0), + start_frame: 0, + end_group: None, + end_frame: None, + }; + (TrackRun::new(ctx, track, bounds, track_priority), stream, log) + } + + fn write_group(track: &mut track::Producer, sequence: u64, millis: u64) { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + group + .write_frame(Timestamp::from_millis(millis).unwrap(), b"x".as_slice()) + .unwrap(); + group.finish().unwrap(); + } + + /// SUBSCRIBE_END counts the group streams opened, not the groups below the end: a + /// group the track never produced has no stream and is not counted. + #[tokio::test] + async fn lite07_end_counts_the_streams_opened() { + let mut track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); + let subscriber = track.subscribe(None); + let (mut run, mut stream, log) = lite07_run(SinkSession::new(Log::default()), subscriber); + let mut run = std::pin::pin!(kio::wait(move |waiter| run.poll(&mut stream, waiter))); + + write_group(&mut track, 0, 0); + assert!(futures::poll!(run.as_mut()).is_pending()); + write_group(&mut track, 2, 2); + track.finish().unwrap(); + assert!(matches!(run.await.unwrap(), TrackEnd::Finished)); + + // SUBSCRIBE_START at 0, then SUBSCRIBE_END at 3 with 2 streams. + assert_eq!(*log.writes.lock().unwrap(), [0, 1, 0, 1, 2, 3, 2]); + } + + /// A track that ends without a group still ends the subscription, with no stream owed. + #[tokio::test] + async fn lite07_end_counts_zero_streams() { + let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); + let subscriber = track.subscribe(None); + track.finish().unwrap(); + + let (mut run, mut stream, log) = lite07_run(SinkSession::new(Log::default()), subscriber); + kio::wait(|waiter| run.poll(&mut stream, waiter)).await.unwrap(); + assert_eq!(*log.writes.lock().unwrap(), [1, 2, 0, 0]); + } + + /// The count is sent once every served group has opened its stream or given up, so a + /// group still waiting for stream credit holds SUBSCRIBE_END back, and one that expires + /// first is never counted. + #[tokio::test] + async fn lite07_end_waits_for_every_stream_to_open() { + tokio::time::pause(); + + let gate = kio::Producer::new(false); + let mut track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); + let subscriber = track.subscribe(None); + write_group(&mut track, 0, 0); + + let (mut run, mut stream, log) = lite07_run(SinkSession::gated_open_uni(gate.consume()), subscriber); + let mut run = std::pin::pin!(kio::wait(move |waiter| run.poll(&mut stream, waiter))); + assert!(futures::poll!(run.as_mut()).is_pending()); + + // Group 1 lands a second later, expiring group 0 before it ever opened. + tokio::time::advance(Duration::from_secs(1)).await; + write_group(&mut track, 1, 1000); + track.finish().unwrap(); + assert!(futures::poll!(run.as_mut()).is_pending(), "group 1 is still opening"); + assert_eq!(*log.writes.lock().unwrap(), [0, 1, 0], "only SUBSCRIBE_START so far"); + + let Ok(mut open) = gate.write() else { + panic!("transport gate closed"); + }; + *open = true; + drop(open); + run.await.unwrap(); + assert_eq!(*log.writes.lock().unwrap(), [0, 1, 0, 1, 2, 2, 1]); + } } #[cfg(test)] diff --git a/rs/moq-net/src/lite/subscribe.rs b/rs/moq-net/src/lite/subscribe.rs index 5de12e097f..4656e94521 100644 --- a/rs/moq-net/src/lite/subscribe.rs +++ b/rs/moq-net/src/lite/subscribe.rs @@ -322,6 +322,9 @@ impl Message for SubscribeStart { #[derive(Clone, Debug)] pub struct SubscribeEnd { pub group: u64, + /// The number of group streams the publisher opened for this subscription. + /// Lite07+ only; not on the wire before, where it decodes as 0. + pub streams: u64, } impl Message for SubscribeEnd { @@ -329,16 +332,23 @@ impl Message for SubscribeEnd { if !version.has_track_stream() { return Err(DecodeError::Version); } - Ok(Self { - group: u64::decode(r, version)?, - }) + let group = u64::decode(r, version)?; + let streams = match version.has_stream_count() { + true => u64::decode(r, version)?, + false => 0, + }; + Ok(Self { group, streams }) } fn encode_msg(&self, w: &mut W, version: Version) -> Result<(), EncodeError> { if !version.has_track_stream() { return Err(EncodeError::Version); } - self.group.encode(w, version) + self.group.encode(w, version)?; + if version.has_stream_count() { + self.streams.encode(w, version)?; + } + Ok(()) } } @@ -429,7 +439,7 @@ impl Message for SubscribeUpdate { /// The range `[start, end]` is inclusive on both ends. For example, /// `start = 5, end = 7` means groups 5, 6, and 7 were dropped. /// -/// Lite03+ only. +/// Lite03 to Lite06 only: Lite07 counts group streams in [`SubscribeEnd`] instead. #[derive(Clone, Debug)] pub struct SubscribeDrop { /// The first absolute group sequence in the dropped range. @@ -449,6 +459,7 @@ impl Message for SubscribeDrop { Version::Lite01 | Version::Lite02 => { return Err(DecodeError::Version); } + _ if version.has_stream_count() => return Err(DecodeError::Version), _ => {} } @@ -464,6 +475,7 @@ impl Message for SubscribeDrop { Version::Lite01 | Version::Lite02 => { return Err(EncodeError::Version); } + _ if version.has_stream_count() => return Err(EncodeError::Version), _ => {} } @@ -480,8 +492,9 @@ impl Message for SubscribeDrop { /// /// The discriminator is version-dependent: /// - Lite03/04: `0x0` SUBSCRIBE_OK, `0x1` SUBSCRIBE_DROP. -/// - Lite05+: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END, `0x2` SUBSCRIBE_DROP +/// - Lite05/06: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END, `0x2` SUBSCRIBE_DROP /// (SUBSCRIBE_OK was removed; acceptance is implicit). +/// - Lite07+: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END (SUBSCRIBE_DROP was removed). #[derive(Clone, Debug)] pub enum SubscribeResponse { Ok(SubscribeOk), @@ -525,8 +538,8 @@ impl Encode for SubscribeResponse { _ => match self { Self::Start(start) => encode_typed(w, 0, start, version)?, Self::End(end) => encode_typed(w, 1, end, version)?, - Self::Drop(drop) => encode_typed(w, 2, drop, version)?, - Self::Ok(_) => return Err(EncodeError::Version), + Self::Drop(drop) if !version.has_stream_count() => encode_typed(w, 2, drop, version)?, + Self::Drop(_) | Self::Ok(_) => return Err(EncodeError::Version), }, } @@ -551,7 +564,7 @@ impl Decode for SubscribeResponse { match typ { 0 => Ok(Self::Start(SubscribeStart::decode(buf, version)?)), 1 => Ok(Self::End(SubscribeEnd::decode(buf, version)?)), - 2 => Ok(Self::Drop(SubscribeDrop::decode(buf, version)?)), + 2 if !version.has_stream_count() => Ok(Self::Drop(SubscribeDrop::decode(buf, version)?)), _ => Err(DecodeError::InvalidMessage(typ)), } } @@ -577,16 +590,53 @@ mod test { #[test] fn subscribe_end_roundtrips_on_lite05() { - let resp = SubscribeResponse::End(SubscribeEnd { group: 7 }); + let resp = SubscribeResponse::End(SubscribeEnd { group: 7, streams: 3 }); let mut buf = Vec::new(); resp.encode(&mut buf, Version::Lite05).unwrap(); + // Type, length, group: no stream count before lite-07. + assert_eq!(buf, [1, 1, 7]); let mut slice = buf.as_slice(); match SubscribeResponse::decode(&mut slice, Version::Lite05).unwrap() { - SubscribeResponse::End(end) => assert_eq!(end.group, 7), + SubscribeResponse::End(end) => assert_eq!((end.group, end.streams), (7, 0)), + other => panic!("expected End, got {other:?}"), + } + } + + #[test] + fn subscribe_end_carries_the_stream_count_on_lite07() { + let resp = SubscribeResponse::End(SubscribeEnd { group: 7, streams: 3 }); + let mut buf = Vec::new(); + resp.encode(&mut buf, Version::Lite07).unwrap(); + assert_eq!(buf, [1, 2, 7, 3]); + let mut slice = buf.as_slice(); + match SubscribeResponse::decode(&mut slice, Version::Lite07).unwrap() { + SubscribeResponse::End(end) => assert_eq!((end.group, end.streams), (7, 3)), other => panic!("expected End, got {other:?}"), } } + #[test] + fn subscribe_drop_is_gone_on_lite07() { + let resp = SubscribeResponse::Drop(SubscribeDrop { + start: 1, + end: 3, + error: 0, + }); + let mut buf = Vec::new(); + assert!(matches!( + resp.encode(&mut buf, Version::Lite07), + Err(EncodeError::Version) + )); + + // A lite-06 DROP is an unknown response type on lite-07. + let mut buf = Vec::new(); + resp.encode(&mut buf, Version::Lite06).unwrap(); + assert!(matches!( + SubscribeResponse::decode(&mut buf.as_slice(), Version::Lite07), + Err(DecodeError::InvalidMessage(2)) + )); + } + #[test] fn subscribe_drop_is_type_2_on_lite05() { let resp = SubscribeResponse::Drop(SubscribeDrop { diff --git a/rs/moq-net/src/lite/version.rs b/rs/moq-net/src/lite/version.rs index dfd78d0c57..72e4ffa30b 100644 --- a/rs/moq-net/src/lite/version.rs +++ b/rs/moq-net/src/lite/version.rs @@ -18,13 +18,26 @@ pub enum Version { /// as `moq-lite-06` and preferred by the default version sets. Lite06, /// Lite-07. Adds the hidden opt-in to ANNOUNCE_REQUEST: without it, a route with - /// a `.`-prefixed segment below the requested prefix is left out. The wire format is - /// still work-in-progress, so it is advertised over ALPN as `moq-lite-07-wip` and - /// only when explicitly requested; the default version sets leave it out. + /// a `.`-prefixed segment below the requested prefix is left out. SUBSCRIBE_END + /// carries the number of group streams opened, replacing SUBSCRIBE_DROP. + /// The wire format is still work-in-progress, so it is advertised over ALPN as + /// `moq-lite-07-wip` and only when explicitly requested; the default version sets + /// leave it out. Lite07, } impl Version { + /// Whether SUBSCRIBE_END carries the subscription's group stream count, sent once + /// every counted stream is open, in place of SUBSCRIBE_DROP. Added in lite-07. + #[allow(clippy::match_like_matches_macro)] + pub fn has_stream_count(self) -> bool { + // Match form so future versions default forward (CLAUDE.md convention). + match self { + Self::Lite01 | Self::Lite02 | Self::Lite03 | Self::Lite04 | Self::Lite05 | Self::Lite06 => false, + _ => true, + } + } + /// Whether the version has lite-05's dedicated TRACK stream and related stream /// layout changes. /// From d30aaf9b268348c2def69ec1aaa94b1efe34bff5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 00:33:32 -0700 Subject: [PATCH 3/5] test(net): widen the subscribeEnd version parameter Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/publisher.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 125e1f223a..15b70bd1ab 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -86,7 +86,7 @@ test.each([Version.DRAFT_01, Version.DRAFT_03, Version.DRAFT_06])( // Delivers `sequences` in the given order, finishes the track, and returns the // SUBSCRIBE_END the publisher put on the wire. -async function subscribeEnd(sequences: number[], version = Version.DRAFT_05): Promise { +async function subscribeEnd(sequences: number[], version: Version = Version.DRAFT_05): Promise { const pair = createMockTransportPair(version === Version.DRAFT_07 ? ALPN_07_WIP : ALPN_05); const origin = new OriginProducer(); const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); From 5b3dc9b17acd03e2a14058109900466c5539e747 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 14:25:13 -0700 Subject: [PATCH 4/5] fix(net): type the lite-07 open wait The open promise resolves to a writer, so the set that SUBSCRIBE_END waits on cannot be Promise. Co-Authored-By: Grok 4.7 --- js/net/src/lite/publisher.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index c7e8ee2030..b740c43a61 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -750,7 +750,7 @@ export class Publisher { // delays it until they are released. const countStreams = hasStreamCount(this.version); let streams = 0; - const opening = new Set>(); + const opening = new Set>(); // The track's exclusive final boundary. A Rust subscriber feeds SUBSCRIBE_END // straight into finish_at, so it must name the track's boundary (which counts From a31e62556d35a5ad9a1257c60813806934cb2f32 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 14:33:37 -0700 Subject: [PATCH 5/5] fix(net): hold lite-07 SUBSCRIBE_END until produced groups open finishAt can name the end while groups below it are still being written. Sending the count on that idle path omits streams that open afterwards. Co-Authored-By: Grok 4.7 --- js/net/src/lite/publisher.test.ts | 45 +++++++++++++++++++++++++++++++ js/net/src/lite/publisher.ts | 7 ++--- 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 15b70bd1ab..bbaf0a2731 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -999,6 +999,51 @@ test("lite draft-07: subscribe end counts zero streams when no groups were produ expect([end.group, end.streams]).toEqual([0, 0]); }); +// finishAt names the end while groups below it are still being produced. The count +// cannot include a stream that has not opened, so SUBSCRIBE_END waits for them. +test("lite draft-07: subscribe end waits for groups below a declared finish", async () => { + const pair = createMockTransportPair(ALPN_07_WIP); + const origin = new OriginProducer(); + const publisher = new Publisher(pair.server, Version.DRAFT_07, randomHop(), origin.consume()); + const broadcast = publish(origin, Path.from("test")); + const track = broadcast.createTrack("video"); + + const client = await Stream.open(pair.client); + const server = await Stream.accept(pair.server); + if (!server) throw new Error("publisher never accepted the subscribe stream"); + void publisher.runSubscribe( + new Subscribe({ id: 0n, broadcast: Path.from("test"), track: "video", priority: 0 }), + server, + ); + + try { + const first = new GroupProducer(0); + first.writeString("hello"); + first.close(); + track.writeGroup(first); + track.finishAt(2); + + const start = await decodeSubscribeResponse(client.reader, Version.DRAFT_07); + expect("start" in start).toBe(true); + const pending = decodeSubscribeResponse(client.reader, Version.DRAFT_07); + const early = await Promise.race([pending, new Promise((resolve) => setTimeout(resolve, IDLE_MS))]); + expect(early).toBeUndefined(); + + const second = new GroupProducer(1); + second.writeString("hello"); + second.close(); + track.writeGroup(second); + track.close(); + + const resp = await pending; + if (!("end" in resp)) throw new Error("expected SUBSCRIBE_END"); + expect([resp.end.group, resp.end.streams]).toEqual([2, 2]); + } finally { + publisher.close(); + client.close(); + } +}); + // Serves one group with its stream open held until `open(ok)`, and returns the pending // SUBSCRIBE_END plus the call that lets the open succeed or fail. async function heldOpenEnd() { diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index b740c43a61..f35fce72be 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -828,9 +828,10 @@ export class Publisher { case "error": throw recv.error; case "idle": - // An end declared ahead of the live edge goes out as soon as it is - // known, while the remaining groups are still being produced. - if (!endSent && track.final() !== undefined) { + // Before lite-07, an end declared ahead of the live edge goes out as + // soon as it is known, while the remaining groups are still being + // produced. The lite-07 count is not final until those groups open. + if (!endSent && !countStreams && track.final() !== undefined) { if (!(await sendEnd())) return; continue; }