diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index c1440ba7a00..e5dd9fe4c81 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -499,25 +499,28 @@ An advertiser signals capacity through its route alone, withdrawing or re-pricin 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 each Range's start 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. +The publisher replies with a SUBSCRIBE_OK message once each Range's start is resolved, and a SUBSCRIBE_FINISH message once the Track's final group is known. +For a live track the publisher MAY withhold SUBSCRIBE_OK until the first matching group resolves the start, and withholds the SUBSCRIBE_DROP messages for groups it gives up meanwhile until it sends SUBSCRIBE_OK. +It MUST NOT withhold it past the point where no group can resolve it: once every Group in a subscription whose every Range has an `End` other than None has been given up, or once the Track has finished, the publisher sends SUBSCRIBE_OK and then the SUBSCRIBE_DROP messages it withheld. +A Track that finished before producing any Group is the one case with nothing to resolve: SUBSCRIBE_FINISH with a Group of 0 needs no SUBSCRIBE_OK. A requested start past the publisher's largest position is the exception: the publisher MUST send SUBSCRIBE_OK without waiting, so a subscriber resuming just past what it holds learns where the live feed is from the answer alone (a quiet track may not reach the start for a long time). +A relay knows that position only from its upstream's answer, never from its cache: a group it cached earlier says nothing about where the live feed is now. 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. 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 start is resolved, and SUBSCRIBE_END as soon as the exclusive final group is known (see [SUBSCRIBE_OK](#subscribe-ok) and [SUBSCRIBE_END](#subscribe-end)). -The publisher closes its side of the stream (FIN) after SUBSCRIBE_END, once every Group Stream has finished or been reset. -A subscription whose every Range has an `End` other than None is covered once every Group in its Ranges has been sent as a Group Stream or a datagram, or named by SUBSCRIBE_DROP. -The publisher MAY then close its side (FIN) without SUBSCRIBE_END, once every Group Stream has finished or been reset, rather than wait for a SUBSCRIBE_UPDATE to widen it. -That FIN ends the subscription, not the Track: the subscriber MUST NOT treat it as the Track's end, and opens a new Subscribe Stream to receive more. -It can arrive after the subscriber widened the Ranges in a SUBSCRIBE_UPDATE that the publisher never read. -After reading the publisher's FIN, the subscriber MUST close its side (FIN) once its tail accounting has settled: every received Group Stream has been read to its FIN, reset, or dropped, and every sequence in range has arrived, been named by SUBSCRIBE_DROP, or exhausted the subscriber's grace period. +The publisher sends SUBSCRIBE_OK once the start is resolved, and SUBSCRIBE_FINISH as soon as the exclusive final group is known (see [SUBSCRIBE_OK](#subscribe-ok) and [SUBSCRIBE_FINISH](#subscribe-finish)). +Neither closes the stream: the publisher keeps reading SUBSCRIBE_UPDATE messages, so an update can still widen the Ranges, including into an ended Track's history, and the publisher serves or drops what it newly covers. +The subscriber judges a Range complete from what it observed: the Groups it received, the SUBSCRIBE_DROP messages, the Range Starts in SUBSCRIBE_OK, and SUBSCRIBE_FINISH. +Drops and Groups name absolute sequences, so the subscriber judges them against its latest Ranges, and an update still in flight changes nothing it has counted. +The publisher never closes its side first: a subscription whose every Range has an `End` other than None and whose every Group has been delivered or dropped stays open for an update to widen it. + +The subscriber closes its side (FIN) once it wants nothing more from the subscription, and the publisher then closes its side (FIN), cancelling any Group Stream still open. +After SUBSCRIBE_FINISH, the subscriber SHOULD close its side once its tail accounting has settled: every received Group Stream has been read to its FIN, reset, or dropped, and every sequence in range has arrived, been named by SUBSCRIBE_DROP, or exhausted the subscriber's grace period. A publisher closing its session gracefully MUST wait for the subscriber's FIN or reset on every Subscribe Stream before closing the connection; a transport acknowledgement alone does not confirm the subscriber read the tail. -A subscriber MAY cancel by closing its side (FIN); the publisher then closes its side (FIN) without owing SUBSCRIBE_END. -Any other subscription stays open until SUBSCRIBE_END, and either endpoint MAY reset the stream at any time. +Either endpoint MAY reset the stream at any time. ### Fetch A subscriber opens a Fetch Stream (0x3) to request a single Group from a Track. @@ -1117,7 +1120,7 @@ Each sum MUST saturate at 2^62-1. A publisher sends an ANNOUNCE_END message to retract a previously started route, referencing its Announce ID. The id is retired and MUST NOT be referenced again. Retraction claims nothing about the content: a broadcast whose publisher stops advertising may remain readable by exact path, serving its stored groups over FETCH. -Retraction does not disturb subscriptions already in flight, which conclude normally with SUBSCRIBE_END. +Retraction does not disturb subscriptions already in flight, which conclude normally with SUBSCRIBE_FINISH. How a subscriber learns the path of stored content is out of band, e.g. an application catalog. ~~~ @@ -1495,7 +1498,7 @@ The name a TRACKS_ADD on the stream listed. ## SUBSCRIBE_OK {#subscribe-ok} A SUBSCRIBE_OK message confirms a subscription and resolves where each Range starts. -It is the first message the publisher sends on the Subscribe Stream, once those starts are known. +It is the first message the publisher sends on the Subscribe Stream, once those starts are known, except that SUBSCRIBE_FINISH MAY precede it. This is the trimmed-down counterpart of MoqTransport's SUBSCRIBE_OK: it retains the name and the role of the publisher's positive response, but carries only the publisher's largest position and where each Range starts (all other per-track properties live in [TRACK_INFO](#track-info)). @@ -1525,11 +1528,12 @@ The Range's Groups from this one are owed as described in [SUBSCRIBE_DROP](#subs A Range whose `Range Start` is its requested start Group begins at the requested frame, and otherwise at frame 0 (see [Positions](#positions)). The Ranges are those of the SUBSCRIBE unless the publisher applied a SUBSCRIBE_UPDATE before answering; a subscriber that sent none MUST treat a count other than its number of Ranges as a PROTOCOL_VIOLATION. -## SUBSCRIBE_END {#subscribe-end} -A SUBSCRIBE_END message is sent by the publisher to signal that no group at or after a given sequence will be produced. +## SUBSCRIBE_FINISH {#subscribe-finish} +A SUBSCRIBE_FINISH message is sent by the publisher to signal that no group at or after a given sequence will be produced. +It ends the Track, not the subscription: the Subscribe Stream stays open (see [Subscribe](#subscribe)). ~~~ -SUBSCRIBE_END Message { +SUBSCRIBE_FINISH Message { Type (i) = 0x1 Message Length (i) Group (i) @@ -1538,7 +1542,7 @@ SUBSCRIBE_END Message { ~~~ **Type**: -Set to 0x1 to indicate a SUBSCRIBE_END message. +Set to 0x1 to indicate a SUBSCRIBE_FINISH message. **Group**: The exclusive end of the range: the absolute sequence number of the first group that will never be delivered. @@ -1549,7 +1553,7 @@ The subscriber MUST NOT wait for any group at or after this sequence. 0 if the track ended cleanly, otherwise the stream error code it ended with plus 1. Groups below the end stay deliverable either way; a subscriber reaching the end surfaces the error instead of a clean end, and a relay forwards it with the end. -The publisher SHOULD send SUBSCRIBE_END as soon as the exclusive final group is known, even while lower groups are still in flight. +The publisher SHOULD send SUBSCRIBE_FINISH as soon as the exclusive final group is known, even while lower groups are still in flight. ## SUBSCRIBE_DROP {#subscribe-drop} @@ -1573,7 +1577,7 @@ A subscriber MUST end the readers of a dropped group it holds with that error, i For every sequence in the subscription's range, the publisher MUST send a Group Stream, send a datagram, or name it in SUBSCRIBE_DROP. This includes sequences it skipped and streams reset before their headers were delivered. A reliably delivered reset retaining the Group Stream header accounts for that group instead. -The publisher SHOULD send SUBSCRIBE_DROP as soon as it gives a sequence up, without waiting for SUBSCRIBE_END. +The publisher SHOULD send SUBSCRIBE_DROP as soon as it gives a sequence up, without waiting for SUBSCRIBE_FINISH. A datagram counts as sent even if it is lost; a subscriber SHOULD bound its wait for unaccounted sequences, for example by Subscriber Max Age. ## FETCH @@ -1895,14 +1899,15 @@ The `Message Length` describes the payload size on the wire. - A SUBSCRIBE_UPDATE that lowers a `Start` covers the groups below the old start that the publisher still holds, not only those that arrive later. A group already sent is not sent again. - Replaced SUBSCRIBE_OK `Group` with a `Range Start` per Range: the Group where that Range's delivery starts, including at a Relative or Time Anchor, from which its Groups are owed. - Added `Largest Group` and `Largest Frame` to SUBSCRIBE_OK: the publisher's largest position when it answers, which a subscriber takes as where the live feed is. A publisher MUST answer at once when the requested start is past it. -- Added `Error Code` to SUBSCRIBE_END: a track can end with an error while keeping the groups below its end. -- Restored SUBSCRIBE_DROP (type 0x2), replacing Stream Count on SUBSCRIBE_END. Every sequence is accounted for by a Group Stream, a datagram, or a drop; a publisher SHOULD send a drop as soon as it gives a sequence up. +- Renamed SUBSCRIBE_END to SUBSCRIBE_FINISH, matching TRACKS_FINISH: it ends the Track, not the subscription. +- Added `Error Code` to SUBSCRIBE_FINISH: a track can end with an error while keeping the groups below its end. +- Restored SUBSCRIBE_DROP (type 0x2), replacing Stream Count on SUBSCRIBE_FINISH. Every sequence is accounted for by a Group Stream, a datagram, or a drop; a publisher SHOULD send a drop as soon as it gives a sequence up. - SUBSCRIBE_DROP's Error Code is a stream error code, which ends the dropped groups' readers, including one still arriving. - Refuse a SUBSCRIBE_DROP whose Group Start is greater than Group End plus one on every version that carries it. -- SUBSCRIBE_END declares the final boundary while lower groups can still be in flight; the Subscribe Stream FIN follows once every Group Stream has finished or been reset. -- A publisher MAY FIN a subscription whose every Range has an End, once each Group in them is sent or dropped, without SUBSCRIBE_END. The FIN ends the subscription, not the Track. -- The subscriber FINs its Subscribe Stream after settling its tail; graceful session close waits for that FIN or reset. -- A subscriber MAY cancel by FIN on its Subscribe Stream, and the publisher then FINs without SUBSCRIBE_END. +- SUBSCRIBE_FINISH declares the final boundary while lower groups can still be in flight, and no longer closes the Subscribe Stream: the publisher keeps reading SUBSCRIBE_UPDATE, so an update can reach an ended Track's history. +- A covered subscription, whose every Range has an End and whose every Group was sent or dropped, also stays open for an update; the publisher never FINs first. +- A publisher that withheld SUBSCRIBE_OK sends it, then the drops it withheld, once no group can resolve the start: every Group of a bounded subscription was given up, or the Track finished. +- The subscriber FINs its Subscribe Stream once it wants nothing more, such as after settling its tail past SUBSCRIBE_FINISH, and the publisher FINs in reply; graceful session close waits for the subscriber's FIN or reset. ### Delivery {#changelog-07-delivery} diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index a31250c8ff7..d9cd59e170d 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -500,11 +500,11 @@ test("integration: lite-07 serves sparse and relative ranges", async () => { } }); -// A lite-07 publisher ends a bounded subscription once it covered the range, releasing its -// demand, and the subscriber keeps the copy: the read ends with its window, the same range reads -// again from the cache, and a wider one subscribes again for the rest. Mirrors the Rust -// `a_covered_bounded_subscription_ends_on_lite07`; reads end as Rust's do since moq-dev/moq#7545. -test("integration: lite-07 keeps a covered bounded subscription's copy", async () => { +// A lite-07 publisher keeps a covered bounded subscription open for an update: the read ends +// with its window, the same range reads again from the cache, and a wider one reaches upstream +// for the rest, past a group that never existed. The demand goes with the readers. Mirrors the +// Rust `a_covered_bounded_subscription_stays_open`. +test("integration: lite-07 keeps a covered bounded subscription open", async () => { const pair = createMockTransportPair(Lite.ALPN_07_WIP); const origin = new OriginProducer(); const [client, server] = await Promise.all([ @@ -532,19 +532,20 @@ test("integration: lite-07 keeps a covered bounded subscription's copy", async ( try { const reader = bounded(3n); expect(await read(reader, 2)).toEqual([1n, 2n]); - // The publisher's demand ends with the covered subscription, and the read with its window. - while (producer.subscription.peek() !== undefined) await producer.subscription.changed(); + // The read ends with its window, while the covered subscription stays open. expect(await reader.recvGroup()).toBeUndefined(); + expect(producer.subscription.peek()).toBeDefined(); const again = bounded(3n); expect(await read(again, 2)).toEqual([1n, 2n]); - expect(producer.subscription.peek()).toBeUndefined(); const wider = bounded(6n); expect(await read(wider, 4)).toEqual([1n, 2n, 4n, 5n]); reader.close(); again.close(); wider.close(); + // The demand goes with the readers. + while (producer.subscription.peek() !== undefined) await producer.subscription.changed(); } finally { remote.close(); broadcast.close(); diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index 9488b7159b1..7af5f8d8db9 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -32,6 +32,7 @@ import { ALPN_06, ALPN_07_WIP, hasTrackStream, + keepsSubscribeOpen, Version, versionName, } from "./version.ts"; @@ -538,8 +539,10 @@ async function servedRawSubscription(alpn: string = ALPN_07_WIP, version: Versio expect("start" in (await decodeSubscribeResponse(subscriber.reader, version))).toBe(true); producer.finish(); producer.close(); - // SUBSCRIBE_END, then the publisher's FIN. - await subscriber.reader.readAll(); + // SUBSCRIBE_END, then before lite-07 the publisher's FIN; lite-07 keeps the stream open. + if (keepsSubscribeOpen(version)) { + while (!("end" in (await decodeSubscribeResponse(subscriber.reader, version)))) {} + } else await subscriber.reader.readAll(); let closed = false; void pair.server.closed.then(() => { @@ -715,7 +718,8 @@ test("close waits for a request served while withdrawals are in flight", async ( expect("start" in (await decodeSubscribeResponse(subscriber.reader, version))).toBe(true); producer.finish(); producer.close(); - await subscriber.reader.readAll(); + // SUBSCRIBE_END; lite-07 keeps the stream open past it. + while (!("end" in (await decodeSubscribeResponse(subscriber.reader, version)))) {} // The withdrawal completes, but the subscription still waits for the subscriber's FIN. fin.release(); diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 1a311ecfec6..a26c771cf6e 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -30,7 +30,7 @@ import { SubscribeUpdate, } from "./subscribe.ts"; import { Track as TrackMessage } from "./track.ts"; -import { ALPN_05, ALPN_06, ALPN_07_WIP, Version } from "./version.ts"; +import { ALPN_05, ALPN_06, ALPN_07_WIP, keepsSubscribeOpen, Version } from "./version.ts"; function publish(origin: OriginProducer, path: Path.Valid) { const broadcastPublication = origin.createBroadcast(path); @@ -892,8 +892,9 @@ test("lite-07 SUBSCRIBE_START names each resolved range start", async () => { }); // A bounded window that resolves empty has no group to answer it, so the publisher answers it -// at once and ends the subscription as covered, as a local read of it ends at once. -test("lite-07 answers and ends a bounded window that resolves empty", async () => { +// at once. The stream stays open for an update until the subscriber FINs, and the publisher +// FINs in reply. +test("lite-07 answers a bounded window that resolves empty and FINs in reply", async () => { const version = Version.DRAFT_07; const sub = await servedSubscription({ version, @@ -905,6 +906,7 @@ test("lite-07 answers and ends a bounded window that resolves empty", async () = const resp = await decodeSubscribeResponse(sub.client.reader, version); if (!("start" in resp)) throw new Error("expected SUBSCRIBE_START"); expect(resp.start.starts).toEqual([undefined]); + sub.client.writer.close(); expect(await decodeSubscribeResponseMaybe(sub.client.reader, version)).toBeUndefined(); } finally { await sub.close(); @@ -1041,6 +1043,8 @@ for (const version of [Version.DRAFT_06, Version.DRAFT_07]) { const resp = await decodeSubscribeResponseMaybe(sub.client.reader, version); if (!resp) break; if ("drop" in resp) codes.push(resp.drop.error); + // Lite-07 keeps the stream open past SUBSCRIBE_END, after the sweep's drops. + if ("end" in resp && keepsSubscribeOpen(version)) break; } expect(codes).toEqual([version === Version.DRAFT_07 ? StreamCode.NotFound : 0]); } finally { @@ -2124,7 +2128,8 @@ test("blocked DROP responses stop opening later groups", async () => { track.finish(); track.close(); let dropped = 0; - for (;;) { + // Lite-07 keeps the stream open past SUBSCRIBE_END, so read until every group is named. + while (dropped < 72) { const response = await decodeSubscribeResponseMaybe(client.reader, Version.DRAFT_07); if (!response) break; if ("drop" in response) dropped += Number(response.drop.end - response.drop.start) + 1; diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 0b084e85f15..67a0dac77a7 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -66,6 +66,7 @@ import { hasRouteCost, hasSynced, hasTrackStream, + keepsSubscribeOpen, resolvesStart, Version, waitsForSubscriberFin, @@ -299,6 +300,11 @@ class SubscriptionControls { return this.#end === null ? { kind: "done" } : { kind: "error", error: this.#end }; } + /** Whether {@link take} has a control to answer with now. */ + pending(): boolean { + return this.#update !== undefined || this.#end !== undefined; + } + /** Calls `fn` once {@link take} may answer differently. */ changed(fn: () => void): Dispose { return this.#changed.changed(fn); @@ -392,6 +398,16 @@ function waitForSubscription( }); } +/** Resolves once the subscriber's next control (an update, its FIN, or a failure) arrives. */ +function waitForControl(controls: SubscriptionControls): Promise { + return new Promise((resolve) => { + const dispose = controls.changed(() => { + dispose(); + resolve(); + }); + }); +} + /** * The budget to serve a peer with, given what its wire could tell us. * @@ -1334,11 +1350,15 @@ export class Publisher { switch (control.kind) { case "done": // The subscriber left. Its queued groups are pointless now, which - // the finally below acts on since `finished` stays false. - return false; + // the finally below acts on since `finished` stays false. On lite-07 it + // also FINs once it read a finished track's tail, which ends the + // subscription as finished. + return finished; case "error": throw control.error; case "update": { + // An update may reach the finished track's history, so it is served again. + finished = false; const update = control.update; console.debug(`subscribe update: broadcast=${broadcast} track=${track.name}`); // `cursorFloor` pins an unfloored pre-06 SUBSCRIBE to the latest @@ -1446,21 +1466,29 @@ export class Publisher { if (!(await sendEnd())) return false; continue; } - // Lite-07 ends a subscription whose every range has an end once it accounted for - // each group in them, without SUBSCRIBE_END: the FIN ends the subscription, not the - // track. What lies below SUBSCRIBE_START was accounted for by it. - if (hasRanges(this.version) && served && startSequence !== undefined) { + // Lite-07 keeps a covered subscription open for an update. One none of whose groups + // can be served has no group to resolve its start: answer it now, as the Rust + // publisher does, and the loop then names what it held back. + if (keepsSubscribeOpen(this.version) && !startSent) { const end = hooks.servingEnd(track); - const covered = - end !== undefined && - hooks - .servingRuns(track, startSequence, end) - .every((run) => runGaps(served, run.start, run.end).length === 0); - if (covered) { - if (!(await drained())) return false; - if (pendingDrops.length > 0) continue; - finished = true; - return true; + const runs = end === undefined ? [] : hooks.servingRuns(track, 0n, end); + const given: Run[] = []; + for (const range of skipped) insertRun(given, range.start, range.end); + if ( + runs.length > 0 && + runs.every((run) => runGaps(given, run.start, run.end).length === 0) + ) { + startSent = true; + startSequence = runs[0].start; + const start = new SubscribeStart( + rangeStarts(this.version, track, startSequence), + largest, + ); + if ( + !(await controls.response(encodeSubscribeResponse(stream, { start }, this.version))) + ) + return false; + continue; } } await waitForSubscription(controls, track, drops); @@ -1482,6 +1510,17 @@ export class Publisher { (bounds.endGroup === undefined || bounds.endGroup + 1n >= boundary()) ) throw endError; + // Lite-07 sends no FIN to say nothing was served, so a track that ended before + // any group resolved the start still says where delivery starts; the loop then + // names what it held back, and the drops below account for the rest. + if (keepsSubscribeOpen(this.version) && !startSent && boundary() > 0n) { + startSent = true; + startSequence = hooks.servingRuns(track, 0n, boundary())[0]?.start ?? boundary(); + const start = new SubscribeStart(rangeStarts(this.version, track, startSequence), largest); + if (!(await controls.response(encodeSubscribeResponse(stream, { start }, this.version)))) + return false; + continue; + } // Name every sequence from the start to the end this subscription never // got. A cap below the end bounds what the subscriber is owed, as on its side. if (startSequence !== undefined) { @@ -1495,8 +1534,25 @@ export class Publisher { if (!(await sendEnd())) return false; continue; } - if (!(await drained())) return false; + if (!(await drained())) { + // A lite-07 subscriber FINs once it read the finished track's tail, or once + // it wants nothing more. Either way the reply FIN goes out now and the groups + // still waiting for a slot are cancelled, but an opened stream may be a final + // group whose FIN is still being acknowledged, so a draining close still + // waits for it; a reader that wants none of it resets it. + if (keepsSubscribeOpen(this.version)) + void this.owe(Promise.allSettled(groups).then(() => undefined)); + return false; + } if (pendingDrops.length > 0) continue; + // Lite-07 keeps the stream open past SUBSCRIBE_FINISH: an update may still ask + // for the ended track's history, and the subscriber FINs once it wants nothing + // more. + if (keepsSubscribeOpen(this.version)) { + finished = true; + if (!controls.pending()) await waitForControl(controls); + continue; + } finished = true; return true; } diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 02e38e65c11..3815b0de301 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -80,6 +80,7 @@ import { hasSynced, hasTrackStream, hasTracks, + keepsSubscribeOpen, resolvesStart, updateSupported, Version, @@ -147,6 +148,13 @@ interface SubscribeEntry { starts?: (bigint | undefined)[]; // Whether a SUBSCRIBE_UPDATE went out, which the publisher may have read before answering. updated?: boolean; + // Whether SUBSCRIBE_START arrived, which `start` alone cannot say when no range starts. + started?: boolean; + // Whether our side is FINned, which lite-07 does once the tail settles past SUBSCRIBE_FINISH. + fin?: boolean; + // How many SUBSCRIBE_UPDATEs went out, so a tail settling past SUBSCRIBE_END waits a fresh + // grace for what one sent meanwhile asks for. + sent?: number; // Set by the subscribe stream's FIN before lite-05, the track's only end there. finished?: boolean; // Settles once the subscribe stream's responses are read to its FIN or reset, which @@ -1223,8 +1231,8 @@ export class Subscriber { // A settled lite-05+ tail acknowledges completion with FIN, without cancelling // the publisher's already-finished receive half. - if (tailSettled && waitsForSubscriberFin(this.version)) stream.writer.close(); - else stream.close(); + if (tailSettled && waitsForSubscriberFin(this.version)) this.#fin(stream, entry); + else if (!entry.fin) stream.close(); console.debug(`subscribe close: id=${id} broadcast=${broadcast} track=${producer.name}`); // A publisher that covered the range before reading an update widening it sent fewer @@ -1787,6 +1795,9 @@ export class Subscriber { start === undefined || (lowest !== undefined && lowest <= start) ? lowest : start, undefined, ); + entry.started = true; + // A lite-07 tail waits for SUBSCRIBE_START to say where delivery starts. + entry.tail.touch(); hooks.startTrack(entry.track, { start: entry.start ?? null, requested: entry.requested }); hooks.reportedLargest(entry.track, resp.start.largest); // The groups the SUBSCRIBE asked for below it are not waited for, whatever the @@ -1811,6 +1822,9 @@ export class Subscriber { } console.warn(`invalid SUBSCRIBE_END: ${reason(error(err))}`); } + // Lite-07 keeps the stream open past SUBSCRIBE_FINISH: FIN once the tail settles, + // and the publisher's FIN in reply ends this loop. + if (keepsSubscribeOpen(this.version)) void this.#finishTail(stream, entry).catch(() => {}); } else if ("drop" in resp) { // The publisher will never send these groups, so they are accounted for without a // stream, and a reader of one still arriving ends with the drop's error now. @@ -1820,6 +1834,24 @@ export class Subscriber { } } + // Settle a lite-07 tail past SUBSCRIBE_END, then FIN. An update sent meanwhile asks for more, + // which waits a fresh grace rather than one that may have passed while a stream was read. + async #finishTail(stream: Stream, entry: SubscribeEntry): Promise { + for (;;) { + const sent = entry.sent ?? 0; + await this.#settleTail(entry); + if ((entry.sent ?? 0) === sent) break; + } + this.#fin(stream, entry); + } + + // FIN our side once, acknowledging a settled tail without cancelling the publisher's half. + #fin(stream: Stream, entry: SubscribeEntry): void { + if (entry.fin) return; + entry.fin = true; + stream.writer.close(); + } + // Wait for the group streams the publisher still owes once it has ended the subscription. // // Received headers and SUBSCRIBE_DROP account for the range. An unaccounted @@ -1861,8 +1893,18 @@ export class Subscriber { : undefined : min(entry.end, span.end ?? entry.end); if (end === undefined) return false; - // Without SUBSCRIBE_START the publisher served no group at all. - if (entry.start === undefined) return true; + // Without SUBSCRIBE_START the publisher served no group at all, though lite-07 still + // answers with one once the track ends, unless nothing was asked below the end. One + // that started no range owes nothing either, unless an update since asked from a floor + // below the end, which lite-07's open stream still serves. + const floor = rangesStart(ranges); + let first = entry.start; + if (first === undefined) { + if (!keepsSubscribeOpen(this.version)) return true; + if (!entry.started) return !(end > (floor ?? 0n)); + if (!entry.updated || floor === undefined || floor >= end) return true; + first = floor; + } // Owed from the floor the demand last asked for, which an update can move either // way, or where SUBSCRIBE_START resolved a live-edge one. The groups the SUBSCRIBE // asked for below its SUBSCRIBE_START were accounted for when it arrived. Only the @@ -1871,7 +1913,7 @@ export class Subscriber { // accounts for a skipped prefix and resolves a relative or time start. Otherwise a time // start resolved at the publisher, at the lowest start it named. const starts = !entry.updated && sameRanges(ranges, entry.asked) && entry.starts; - const start = starts ? entry.start : (rangesStart(ranges) ?? entry.start); + const start = starts ? first : (floor ?? first); const owed = starts ? ranges.flatMap((range, index) => { const at = starts[index]; @@ -1944,12 +1986,12 @@ export class Subscriber { if (ranges.length === 0) throw new Error(EMPTY_RANGE); const span = wireSpan(ranges); - // A lowered floor owes groups nobody asked for until now. + // Changed ranges owe groups nobody asked for until now: below a lowered floor, or + // between ranges a widening joins while keeping the floor. Each gap they reach waits + // a fresh grace, which only delays giving up on one still owed. const start = rangesStart(ranges); - if (start !== undefined) { - const floor = rangesStart(wireRanges(lastSent.ranges ?? [{}])) ?? entry.start; - entry.tail.demand(start, floor ?? UNBOUNDED); - } + if (!sameRanges(ranges, wireRanges(lastSent.ranges ?? [{}]))) + entry.tail.demand(start ?? entry.start ?? 0n, UNBOUNDED); // Round-trip the other Subscribe parameters so the publisher doesn't // interpret SUBSCRIBE_UPDATE as a reset of ordered/delay/etc. @@ -1975,6 +2017,9 @@ export class Subscriber { if (refresh !== undefined) sentWanted = refresh; entry.bounded ||= bounded(ranges); entry.updated = true; + entry.sent = (entry.sent ?? 0) + 1; + // A tail settling past SUBSCRIBE_END re-judges what the update now asks for. + entry.tail.touch(); lastSent = { ...current }; console.debug(`subscribe update: id=${id} broadcast=${broadcast} track=${track.name}`); } diff --git a/js/net/src/lite/version.ts b/js/net/src/lite/version.ts index 0ce6dac4db9..688c9956a2c 100644 --- a/js/net/src/lite/version.ts +++ b/js/net/src/lite/version.ts @@ -394,6 +394,16 @@ export function waitsForSubscriberFin(version: Version): boolean { return hasTrackStream(version); } +/** + * Whether the Subscribe Stream stays open past SUBSCRIBE_FINISH (SUBSCRIBE_END before lite-07) + * and past a covered bounded subscription, so a SUBSCRIBE_UPDATE can still reach the track's + * history. The subscriber FINs once it wants nothing more, and the publisher FINs in reply. + * Added in lite-07; before it the publisher FINs once the track ends. + */ +export function keepsSubscribeOpen(version: Version): boolean { + return hasRanges(version); +} + /** Whether the announce stream has ANNOUNCE_RESTART: another publisher instance replacing an * advertisement in place. Added in lite-07. Older versions send an ANNOUNCE_END then an * ANNOUNCE_START. */ diff --git a/quest/m1/lite07-finalize.md b/quest/m1/lite07-finalize.md index ed639960eb8..c32a0d6de3c 100644 --- a/quest/m1/lite07-finalize.md +++ b/quest/m1/lite07-finalize.md @@ -83,7 +83,7 @@ Decided 2026-10-27 (maintainer): of [Subscribe ranges](/quest/m1/subscribe-range only the children that change the lite-07 wire gate the cut: [Lite-07 ranges](/quest/m1/subscribe-ranges/lite.md), [JS ranges](/quest/m1/subscribe-ranges/js.md), and -[Remove FETCH from lite-07](/quest/m1/subscribe-ranges/lite07-fetch.md). The +[Lite-07 FETCH is single-group only](/quest/m1/subscribe-ranges/lite07-fetch.md). The fill path, fetch as a range, and the start bookkeeping change no wire. Decided 2026-10-25 (maintainer): addressing groups by timestamp gates the @@ -110,7 +110,7 @@ timestamp bound in [#6168](https://github.com/moq-dev/moq/pull/6168). - [Drop set](/quest/m1/drop-set.md) - lite-07's SUBSCRIBE_DROP carries `Max Age`, so a widening re-offers groups reported dropped - [Lite-07 ranges](/quest/m1/subscribe-ranges/lite.md) - a relay fills a lite-07 upstream through the subscription - [JS ranges](/quest/m1/subscribe-ranges/js.md) - JS speaks lite-07 ranges -- [Remove FETCH from lite-07](/quest/m1/subscribe-ranges/lite07-fetch.md) - lite FETCH is gone from lite-07, in Rust and JS +- [Lite-07 FETCH is single-group only](/quest/m1/subscribe-ranges/lite07-fetch.md) - lite-07 FETCH reads one group or a slice of one, and SUBSCRIBE ranges of whole groups - [Lite-07 by start](/quest/m1/time-addressing/lite07.md) - lite-07 carries group starts, previous-start links, GROUP_END, and time bounds instead of sequence numbers - [@moq/net by start](/quest/m1/time-addressing/js.md) - JS speaks the same lite-07 wire - [A job's exact route continues its claim's readers](/quest/m1/announce-takeover.md) - the job's route inherits the claim's instance, with no new wire message diff --git a/quest/m1/subscribe-ranges/README.md b/quest/m1/subscribe-ranges/README.md index b8bd2fcc77d..18c8813079b 100644 --- a/quest/m1/subscribe-ranges/README.md +++ b/quest/m1/subscribe-ranges/README.md @@ -2,14 +2,16 @@ ## Goal -A moq-lite-07 SUBSCRIBE replaces FETCH. One subscription asks for any number +A moq-lite-07 SUBSCRIBE replaces FETCH for anything wider than one group. One subscription asks for any number of group ranges, past or live, bounded by group or by time, in ascending or descending order, and the publisher serves their union from one cursor without sending a group twice. Every sequence in range either arrives or is named by a SUBSCRIBE_DROP, so nothing blocks on a cache miss. Relays fill misses upstream with ranges, not one request per group, including over moq-transport where group numbers may be -sparse. Lite FETCH is removed from lite-07. +sparse. Lite-07 FETCH is single-group only: one group, or a slice within one +group, which SUBSCRIBE cannot express (maintainer, 2026-11-03, replacing the +earlier removal). ## Plan @@ -26,8 +28,8 @@ relay bounds a sparse FETCH only if the model can request ranges: historical requests also obey that cap and publisher max-age. An omitted FETCH budget means publisher max-age alone; translation to a range must not accidentally give it the live SUBSCRIBE zero-delay default. SUBSCRIBE_UPDATE - replaces the list. Lite FETCH is removed from - lite-07; published versions keep it, answered through the same model. + replaces the list. Lite-07 FETCH is single-group only (decided 2026-11-03, + replacing its removal); published versions keep it, answered through the same model. - **Holes.** SUBSCRIBE_DROP (restored in lite-07) names every sequence in the requested ranges the publisher won't deliver: never existed, expired, or refused upstream. On lite-01..06 datagram groups are always dropped in a past range (#4982); on lite-07 ranges never cover datagrams ([Datagram tracks](/quest/m1/datagram-tracks/README.md)). The @@ -98,7 +100,7 @@ every sequence delivered or dropped. - [Lite-07 ranges](/quest/m1/subscribe-ranges/lite.md) - the lite-07 wire, Rust publisher and subscriber, and the draft - [JS ranges](/quest/m1/subscribe-ranges/js.md) - `@moq/net`'s on-demand request surface and `fetchGroup` as a range -- [Remove FETCH from lite-07](/quest/m1/subscribe-ranges/lite07-fetch.md) - FETCH removed from lite-07 in Rust, JS, and the draft +- [Lite-07 FETCH is single-group only](/quest/m1/subscribe-ranges/lite07-fetch.md) - lite-07 FETCH reads one group or a slice of one, and SUBSCRIBE ranges of whole groups, in Rust, JS, and the draft - [Delete the old start bookkeeping](/quest/m1/subscribe-ranges/start-bookkeeping.md) - what each lite cursor is owed and dropped lives in the model, and `TrackRun` only translates it per version ## Related diff --git a/quest/m1/subscribe-ranges/js.md b/quest/m1/subscribe-ranges/js.md index 1ed038aeeff..abffd7eb703 100644 --- a/quest/m1/subscribe-ranges/js.md +++ b/quest/m1/subscribe-ranges/js.md @@ -51,16 +51,22 @@ re-ports its ranges onto them: 2. A fresh branch, closing and deleting this one. 3. Park it. -Mirror lite-07 bounded completion as the Rust copy implements it. The target -is [Lite-07 ranges](/quest/m1/subscribe-ranges/lite.md)' "Ends are signaled" -design: a covered bounded subscription sends SUBSCRIBE_FINISH, keeps the -Subscribe Stream open, and never FINs it. Until SUBSCRIBE_FINISH ships, the -wire FINs a covered bounded subscription without SUBSCRIBE_END, and the JS -subscriber reopens demand it no longer covers on the same track, as the Rust -copy does. +Lite-07 bounded completion follows +[Lite-07 ranges](/quest/m1/subscribe-ranges/lite.md)' explicit end, which +[moq-dev/moq#7675](https://github.com/moq-dev/moq/pull/7675) brought to JS +alongside Rust: neither SUBSCRIBE_END nor a covered bounded subscription closes +the Subscribe Stream. The JS publisher keeps reading updates and answers a held +start with SUBSCRIBE_START and its held drops once no group can resolve it, and +the JS subscriber FINs once its tail settles past SUBSCRIBE_END, judging it only +after SUBSCRIBE_START. What remains: +- **The explicit end's leftovers.** Rename SUBSCRIBE_END to SUBSCRIBE_FINISH in + `js/net`, as Rust and the draft did, and delete the subscriber's covered-FIN + path (`#parkCovered`, `#demandCovered`, the `short` reopen), which no lite-07 + publisher triggers any more. + - **On-demand request surface.** Decided in the 2026-10-05 audit: the producer half of JavaScript FETCH folds in here, so `@moq/net`'s on-demand request surface takes range requests from the start @@ -99,4 +105,4 @@ and subscription updates. Datagrams move to their own track kind ([Datagram trac ## Related - [Lite-07 ranges](/quest/m1/subscribe-ranges/lite.md) - the Rust lite-07 range fill this mirrors -- [Remove FETCH from lite-07](/quest/m1/subscribe-ranges/lite07-fetch.md) - follows this and Lite-07 ranges +- [Lite-07 FETCH is single-group only](/quest/m1/subscribe-ranges/lite07-fetch.md) - follows this and Lite-07 ranges diff --git a/quest/m1/subscribe-ranges/lite.md b/quest/m1/subscribe-ranges/lite.md index fcce6cb6211..63e8fc9bdb4 100644 --- a/quest/m1/subscribe-ranges/lite.md +++ b/quest/m1/subscribe-ranges/lite.md @@ -1,13 +1,12 @@ -# [XL] Lite-07 ranges +# [L] Lite-07 ranges ## Goal -A relay fills a lite-07 upstream by subscribing to the missing ranges, and -recovers a dropped group with a one-group SUBSCRIBE. SUBSCRIBE_FINISH signals -a track's end and the Subscribe Stream stays open, so one-range reads back to -back on one lite-07 session complete. FETCH stays on lite-07 until -[Remove FETCH from lite-07](/quest/m1/subscribe-ranges/lite07-fetch.md) -removes it from both implementations. +Lite-07 ranges carry implicit Range IDs, so a range a SUBSCRIBE_UPDATE adds +learns where it started. A relay fills a lite-07 upstream by subscribing to the +missing ranges, and recovers a dropped group with a one-group SUBSCRIBE. Lite-07 +FETCH stays, for one group or a slice within one group +([Lite-07 FETCH is single-group only](/quest/m1/subscribe-ranges/lite07-fetch.md)). ## Plan @@ -15,8 +14,9 @@ Decided 2026-10-25 (maintainer, on [moq-dev/moq#6807](https://github.com/moq-dev/moq/pull/6807)): FETCH removal moves out of this quest (to its own quest since 2026-10-27), since JS still requests and serves wire FETCH on lite-07 (`track.Consumer.fetchGroup`), so a Rust-only removal would drop a current -cross-language capability. This quest keeps FETCH on lite-07 and makes the -range fill work without it. +cross-language capability. Since 2026-11-03 FETCH stays on lite-07 for one +group or a slice within one group, and this quest fills multi-group ranges +through the subscription. Decided 2026-11-01 (maintainer, on [moq-dev/moq#7487](https://github.com/moq-dev/moq/pull/7487)): this quest @@ -31,16 +31,80 @@ Fetch as a range finished in remaining lite-07 work here (maintainer, 2026-11-02): the explicit end, the back-to-back hang, and the one-group recovery below. +The explicit end landed in +[moq-dev/moq#7675](https://github.com/moq-dev/moq/pull/7675): SUBSCRIBE_END is +renamed SUBSCRIBE_FINISH, and on lite-07 neither it nor a covered bounded +subscription closes the Subscribe Stream. The publisher keeps reading updates +and FINs only in reply to the subscriber's FIN. Once no group can resolve a +held start (every group of a bounded subscription given up, or the track +finished), it sends SUBSCRIBE_OK and then the drops it held. The Rust copy +starts its tail at SUBSCRIBE_FINISH, waits for SUBSCRIBE_OK before judging it, +and FINs once it settles. That fixed back-to-back one-range reads +(`range_read::one_range_reads_back_to_back_each_complete`), and the Rust +copy's covered-FIN path (`SubStream::covered`, `ServeLoop::covered`) is gone. +Datagrams take no ranges, since they become their own track kind (maintainer, +2026-11-01; the datagram-tracks plan, +[moq-dev/moq#7575](https://github.com/moq-dev/moq/pull/7575)). The same PR +brought the JS lite-07 publisher and subscriber to the explicit end, so the two +implementations share the wire; JS still names it SUBSCRIBE_END in code. + +Decided 2026-11-03 (maintainer, on +[moq-dev/moq#7675](https://github.com/moq-dev/moq/pull/7675)), answering the +first attempt's open questions: + +1. **Land part 1:** ✅ land it now. Options: ✅ land now; keep a draft. +2. **Fill budget:** ✅ the live demand's budget. Groups that budget drops + count as officially dropped, so the one-group SUBSCRIBE recovery (no max + delay) brings them back, and the live cap never loosens. Options: ✅ live + demand's budget; loosest budget; per-range wire budget. +3. **Partial-group fills:** "we can keep FETCH then but for a single group (or + slice within a single group). SUBSCRIBE is for multiple groups and does not + support slicing within a group." A fill starting past frame 0, such as route + recovery's (`resume::Recover`), keeps its FETCH. See + [Lite-07 FETCH is single-group only](/quest/m1/subscribe-ranges/lite07-fetch.md). +4. **Datagram-only groups:** ✅ wait for datagram tracks + ([Datagram tracks](/quest/m1/datagram-tracks/README.md), planned in + [moq-dev/moq#7575](https://github.com/moq-dev/moq/pull/7575)); after them + ranges never carry datagrams. Options: ✅ wait for datagram tracks; + NOT_FETCHABLE drops now; keep FETCH for these. +5. **Range IDs:** ✅ implicit range IDs now, using + [moq-dev/moq#7207](https://github.com/moq-dev/moq/pull/7207)'s encoding, so + a range added by SUBSCRIBE_UPDATE learns where it started. Options: + absolute-only fills; ✅ implicit range IDs now. + +Range IDs are this quest's next step. They did not fit #7675 as a checkpoint +addition: the encoding changes SUBSCRIBE_UPDATE in the Rust and JS codecs, +publishers, and subscribers together (#7207 was 1,420 lines across both and +now conflicts with main), and it needs a start report #7207 never defined. +What remains is Range IDs, then the range fill and the one-group recovery. + +### Range IDs + +[moq-dev/moq#7207](https://github.com/moq-dev/moq/pull/7207) is the reference: +each Range implicitly takes the next Range ID on the stream, from 0, and +SUBSCRIBE_UPDATE edits the live Ranges rather than restating them, with +`Remove Count` Range IDs to retire and then the Ranges to add. A Range ID never +assigned or already retired is a PROTOCOL_VIOLATION, and the live Ranges must +stay non-empty and fit in one SUBSCRIBE. SUBSCRIBE_OK's Range Starts answer the +Ranges live when it is sent, in Range ID order. A Range an update adds after +SUBSCRIBE_OK has its start reported by SUBSCRIBE_STARTED, below. Rust and JS +change together, with draft text and codec, publisher, and subscriber tests. + +Decided 2026-11-03 (maintainer, on +[moq-dev/moq#7675](https://github.com/moq-dev/moq/pull/7675)): ✅ a new +**SUBSCRIBE_STARTED** message (type 0x3) carrying `Range ID` and +`Range Start`, sent when an added Range's start resolves. Options: ✅ new +SUBSCRIBE_STARTED; repeat SUBSCRIBE_OK. `Range Start` is encoded as in +SUBSCRIBE_OK (the Group plus one, 0 for none), and the subscriber owes the +added Range's groups from it as it does a SUBSCRIBE_OK Range Start. + ### Range fill through the subscription Lite-07 SUBSCRIBE and SUBSCRIBE_UPDATE carry the range list (`lite::Range`, typed anchors), and the publisher serves it from the model. Older wires map at the codec, and a merged request goes upstream as one span, from group 0 when -its starts mix (`wire_span`). Today a lite-07 publisher FINs a covered bounded -subscription without SUBSCRIBE_END, and the Rust copy settles its tail, keeps -the cached copy, and subscribes again only for demand the covered ranges leave -out. The explicit end below replaces that FIN; build the fill on it, not on -the FIN. +its starts mix (`wire_span`). A covered bounded subscription stays open, so a +fill joins the live subscription's ranges with a SUBSCRIBE_UPDATE. The first attempt ([moq-dev/moq#7487](https://github.com/moq-dev/moq/pull/7487), WIP commit @@ -55,8 +119,10 @@ requirement landed. Reuse its design: cache as its groups arrive. - A fill is judged only against the subscription that asked for it, not the track's arrivals ledger, which still holds a group an earlier subscription - cut. A SUBSCRIBE_DROP over its groups, a SUBSCRIBE_START past its range's - start, or a covered end refuses it, so a cut group is filled whole. + cut. A SUBSCRIBE_DROP over its groups, or a SUBSCRIBE_OK Range Start past + its range's start, refuses it, so a cut group is filled whole. The WIP's + covered-FIN refusal has no counterpart now: the publisher drops what it + gives up instead. - A run stays on the wire while a fill owes part of it, then until each of its streams still being read finishes, so narrowing never resets a group in flight. @@ -64,7 +130,7 @@ requirement landed. Reuse its design: carrying only fills sets neither, and canceling it does not idle the copy. If the demand leaves while fills are out, the subscription keeps the demand's ranges until they are answered. -- A NotFound drop or a covered end names the fill's groups dropped +- A NotFound drop names the fill's groups dropped (`Fill::drop_groups`), as a FETCH miss does; Old and NOT_FETCHABLE reject them, and any other error refuses only that request. @@ -88,7 +154,8 @@ whole) passed. These failed, and must pass on #7545's miss judging: backfill rule covers it: a relay leaves cached groups out of a bounded upstream range. On lite-07 the test counts SUBSCRIBE streams, not FETCH. -Open questions the attempt raised, each with the attempt's recommendation: +Open questions the attempt raised, now decided (2026-11-03, above), kept for +their detail: - **Datagram sequences.** A range fill of a sequence served only as a datagram never completes: datagrams resolve no SUBSCRIBE_START, so a FIN or end @@ -100,7 +167,9 @@ Open questions the attempt raised, each with the attempt's recommendation: wire change. The alternative keeps FETCH for a fill the first answer shows is a datagram, at a round trip and a second path. Since settled in part: datagrams become their own track kind, which takes - no ranges (maintainer, 2026-11-01; see "Ends are signaled" below). + no ranges (maintainer, 2026-11-01). Today a lite-07 publisher that served a + bounded range's only group as a datagram never answers it with SUBSCRIBE_OK + (`TrackRun::unservable`), so such a fill parks until its readers leave. - **Fills from partway through a group.** Route recovery (`resume::Recover`) fills from frame N, which a whole-group range cannot say. Recommended: keep FETCH on lite-07 for a fill whose first group starts past frame 0, as @@ -113,7 +182,11 @@ Open questions the attempt raised, each with the attempt's recommendation: a stale group 5). Recommended: settle the partial-group question first, since the failing fill is a recovery fill, and rerun the test. The alternatives fill under the demand's budget, refusing older groups as Old, - or add a per-range budget to the wire. + or add a per-range budget to the wire. Since the one-group recovery below + was decided (2026-11-02), filling under the demand's budget composes with + it: a group the live budget drops as OLD is officially dropped, so a + one-group SUBSCRIBE with no max delay recovers it, and the live cap never + loosens. Decided that way on 2026-11-03. - **A range added after SUBSCRIBE_OK has no reported start.** SUBSCRIBE_OK's Range Starts are positional and sent once, yet the subscriber judges a bounded range complete from them, on a stream that stays open for later @@ -122,8 +195,9 @@ Open questions the attempt raised, each with the attempt's recommendation: ([moq-dev/moq#7207](https://github.com/moq-dev/moq/pull/7207)), whose recommendation was to decide here whether ranges get implicit IDs so a later message can report a start. SUBSCRIBE_UPDATE keeps restating the whole list - either way: each range is a few bytes, and restating stays stateless. If IDs - are adopted, #7207 is the reference encoding. + either way: each range is a few bytes, and restating stays stateless. IDs + were adopted on 2026-11-03, with #7207's encoding, which edits rather than + restates. - `auth::lite_07_narrowing_resets_a_fetch_in_flight` read `Stream(Cancel)` instead of `Unauthorized`/`Unroutable` when the grant narrowed mid-group; the refusal must carry the grant's error. @@ -134,73 +208,6 @@ bounded ranges complete upstream. A hole on a live track counts as dropped once the feed gives up on it, past the arrival grace (#7545's bounded wait, `TrackState::miss_grace`). -### Ends are signaled, not inferred - -The maintainer, 2026-11-01, on -[moq-dev/moq#7545](https://github.com/moq-dev/moq/pull/7545): "we should have -a SUBSCRIBE_END message or something to signal the last group of a track? we -were using QUIC FIN to signal that but that prevents updates in the future, so -we should have a signal instead". Settled the same day: - -- No SUBSCRIBE_COVERED message: a subscriber judges a bounded range complete - from the groups it received, SUBSCRIBE_DROP, the Range Starts in - SUBSCRIBE_OK, and SUBSCRIBE_FINISH. Drops and groups name absolute - sequences, so the subscriber judges against its latest ranges and an update - in flight changes nothing it has counted; the FIN's race with an unread - update goes away. -- "Keep stream open": SUBSCRIBE_FINISH names the track's final group and no - longer FINs the Subscribe Stream, so an update can still reach an ended - track's history. A covered bounded subscription is not FINed either. -- ✅ Rename SUBSCRIBE_END to SUBSCRIBE_FINISH. It matches TRACKS_FINISH and - `track::Producer::finish`, and "END" would read as closing the stream, which - it no longer does. -- Before SUBSCRIBE_OK a lite-07 publisher holds back what it gave up - (`TrackRun::skipped`) and answers only once a group stream resolves the - start, so a bounded subscription none of whose groups it can serve gets - neither, and only the covered FIN ends it today. Once such a subscription is - covered, the publisher sends SUBSCRIBE_OK and then the drops it held. -- The subscriber FINs the stream once it wants nothing more, and the publisher - FINs in reply. -- Datagrams take no ranges, since they become their own track kind - ([Datagram tracks](/quest/m1/datagram-tracks/README.md), maintainer - 2026-11-01), so datagram fills need no handling here. - -The draft's exception for a start past the publisher's largest position should -also say that a relay knows that position only from its upstream's answer, -never from its cache (maintainer, 2026-11-01: "NEVER use the cache as the live -edge"). - -### Back-to-back one-range reads - -Acceptance test (maintainer, 2026-11-02, on -[moq-dev/moq#7625](https://github.com/moq-dev/moq/pull/7625)): on lite-07, -one-range reads of the same track through one session, groups 1 then 2, and -2, 1, 0, each complete, direct and through a relay. Today each hangs on the -later read (on 2, and on 0). Lite-05/06 and moq-transport pass. It also hangs -hls's `a_live_fmp4_import_serves_segments_over_lite07` once hls reads a -segment's groups that way. Reproduce with sequential `one_range` reads in -`range_read.rs`'s fixture. - -The cause is the publisher's end-of-subscription FIN on a covered bounded -subscription, traced with tracing on: - -- When the copy widens `1..2` to `1..3` with SUBSCRIBE_UPDATE while the - publisher FINs the covered `1..2` without reading it, the copy's tail owes - `1..3` (`SubStream::covered`, `updated` set) and waits its grace, which is - `Duration::MAX` with no max delay (`Subscription::budget`). Group 2 never - comes and nothing re-subscribes. -- When the FIN lands first (as on main before - [moq-dev/moq#7625](https://github.com/moq-dev/moq/pull/7625)), the tail - settles, `TrackServe::begin` finds the lingering demand equal to the covered - ranges and keeps `Sub::None`, and the second read's demand never wakes the - serve loop again, so the copy is left with no upstream subscription. - -No SUBSCRIBE_DROP is involved, so the one-group recovery below does not reach -it, and a fresh SUBSCRIBE while group 2 may still be in flight could deliver -it twice. Fix it with the explicit end above: the Subscribe Stream stays open -and the covered FIN goes, so the publisher reads the update and serves or -drops the group. - ### Recovering a dropped group Decided 2026-11-02 (maintainer, on @@ -211,19 +218,20 @@ marked DROP so we don't receive two copies of it". A group a reader still wants, once an existing subscription names it in SUBSCRIBE_DROP, is recovered by a new SUBSCRIBE covering only that group with no max delay, never on a guess, a timeout, or while the group may still arrive. This is lite-07's -FETCH replacement for a dropped group, so that -[Remove FETCH from lite-07](/quest/m1/subscribe-ranges/lite07-fetch.md) can -remove FETCH from Rust and JS together, as decided on 2026-10-25. Tests: a +replacement for a FETCH of a dropped group wider than one group's slice, and +with the fill budget decided 2026-11-03 it also recovers what the live +budget drops. Tests: a group still in flight on the existing subscription sends no second SUBSCRIBE, and a dropped group is recovered exactly once. ### Draft and checks Update `drafts/draft-lcurley-moq-lite.md` with each wire change here, -including that a subscriber fills missing groups through its ranges. Run +including that a subscriber fills missing groups through its ranges, and +recovers a dropped group with a one-group SUBSCRIBE. Run `just drafts check` and `just test interop --all`, and the `moq-net-tests` `restart::retired_fetch_lite07` and `moq-relay` suites. ## Related -- [Remove FETCH from lite-07](/quest/m1/subscribe-ranges/lite07-fetch.md) - removes FETCH from lite-07 in both implementations once this and JS ranges land +- [Lite-07 FETCH is single-group only](/quest/m1/subscribe-ranges/lite07-fetch.md) - lite-07 FETCH reads one group or a slice of one, once this and JS ranges land diff --git a/quest/m1/subscribe-ranges/lite07-fetch.md b/quest/m1/subscribe-ranges/lite07-fetch.md index 49f653de502..b7a24ed03c5 100644 --- a/quest/m1/subscribe-ranges/lite07-fetch.md +++ b/quest/m1/subscribe-ranges/lite07-fetch.md @@ -1,35 +1,44 @@ -# [M] Remove FETCH from lite-07 +# [M] Lite-07 FETCH is single-group only ## Goal -Lite-07 has no FETCH, in Rust, JS, and the draft. Rust and JS stop sending -and serving lite-07 FETCH, and read past groups through range subscriptions -instead. Published versions (lite-03..06) keep FETCH. +On lite-07, FETCH reads one group or a slice within one group, and SUBSCRIBE +reads ranges of whole groups and never slices within a group. Rust and JS read +any multi-group range through SUBSCRIBE and use FETCH only for a single group +or a partial one, and the draft says so. Published versions (lite-03..06) keep +FETCH as it is. ## Plan +Decided 2026-11-03 (maintainer, on +[moq-dev/moq#7675](https://github.com/moq-dev/moq/pull/7675)), replacing the +2026-10-25 plan to remove FETCH from lite-07: "we can keep FETCH then but for a +single group (or slice within a single group). SUBSCRIBE is for multiple groups +and does not support slicing within a group." Route recovery +(`resume::Recover`), which fills from frame N, keeps its FETCH +(`group::Fetch::frame_start`). + Decided 2026-10-25 (maintainer, on -[moq-dev/moq#6807](https://github.com/moq-dev/moq/pull/6807)): the removal -lands in Rust and JS together, once both fill through the subscription, since -JS still requests and serves wire FETCH on lite-07 -(`track.Consumer.fetchGroup`) and a Rust-only removal would drop a current -cross-language capability. - -Decided 2026-10-27 (maintainer): split out of -[JS ranges](/quest/m1/subscribe-ranges/js.md), so the rest of JS ranges -starts now instead of waiting on -[Lite-07 ranges](/quest/m1/subscribe-ranges/lite.md). This quest waits on -both. - -- `drafts/draft-lcurley-moq-lite.md` drops FETCH from lite-07, with its - changelog bullets (FETCH max age and epoch). Run `just drafts check`. -- Lite-07 has the only `NotFetchable` code; check whether it still has a - sender once FETCH is gone. +[moq-dev/moq#6807](https://github.com/moq-dev/moq/pull/6807)) and still +standing: Rust and JS change together, since both request and serve wire FETCH +on lite-07 (`track.Consumer.fetchGroup` in JS). + +- Lite-07 SUBSCRIBE and SUBSCRIBE_UPDATE anchors stop carrying a frame: an + Absolute anchor names a group, so a Range covers whole groups. This is a wire + change to `Range`, the Positions section, and the changelog bullets that say + SUBSCRIBE may start at any frame. A read starting partway through a group goes + through FETCH instead. +- Rust and JS send FETCH on lite-07 only for one group or a slice of one + group; a fill of more than one group goes through + [Lite-07 ranges](/quest/m1/subscribe-ranges/lite.md)' subscription fill. +- `drafts/draft-lcurley-moq-lite.md` states the split in its Fetch and + Subscribe sections. Run `just drafts check`. - Run `just test interop --all`. -Public API: none expected. Wire: lite-07 (unpublished) loses FETCH. +Public API: none expected. Wire: lite-07 (unpublished) SUBSCRIBE anchors lose +their frame. ## Required -- [Lite-07 ranges](/quest/m1/subscribe-ranges/lite.md) - Rust fills lite-07 through the subscription -- [JS ranges](/quest/m1/subscribe-ranges/js.md) - JS reads and serves ranges, so it no longer needs lite-07 FETCH +- [Lite-07 ranges](/quest/m1/subscribe-ranges/lite.md) - Rust fills multi-group ranges through the subscription +- [JS ranges](/quest/m1/subscribe-ranges/js.md) - JS reads and serves ranges, so its multi-group reads need no FETCH diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 3255184b462..effb803fda4 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -307,7 +307,7 @@ pub fn lite_wire(data: &[u8]) -> bool { 6 => roundtrip::, _>(rest, &mut Vec::new(), version), 7 => roundtrip::(rest, &mut Vec::new(), version), 8 => roundtrip::(rest, &mut Vec::new(), version), - 9 => roundtrip::(rest, &mut Vec::new(), version), + 9 => roundtrip::(rest, &mut Vec::new(), version), 10 => roundtrip::(rest, &mut Vec::new(), version), 11 => roundtrip::(rest, &mut Vec::new(), version), 12 => roundtrip::(rest, &mut Vec::new(), version), diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 865488d068f..ff5974a230b 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -2268,9 +2268,9 @@ impl Request for SubscribeServe { *self = Self::Run(Box::new(run)); } Self::Run(run) => { - // The live edge reached the boundary, so SUBSCRIBE_END was already sent (or - // the version predates the track stream), or a bounded lite-07 - // subscription is covered. Drain the in-flight group machines, then FIN. + // The live edge reached the boundary, so SUBSCRIBE_FINISH was already sent (or + // the version predates the track stream). Drain the in-flight group + // machines, then FIN. Lite-07 never gets here: the subscriber's FIN ends it. if ready!(run.poll_step(writer, waiter))?.is_continue() { return Poll::Ready(Ok(ControlFlow::Continue(()))); } @@ -3612,11 +3612,11 @@ fn buffer_zigzag_delta( enum Recv { Group(group::Consumer), Datagram(crate::model::datagram::Stored), - Boundary(lite::SubscribeEnd), + Boundary(lite::SubscribeFinish), Finished, } -/// The group boundary SUBSCRIBE_END carries. Lite cannot name a frame, so a mid-group end +/// The group boundary SUBSCRIBE_FINISH carries. Lite cannot name a frame, so a mid-group end /// rounds up past its group, whose own stream FIN says where it stops. fn final_sequence(end: track::Position) -> u64 { end.final_sequence().unwrap_or(u64::MAX) @@ -3637,7 +3637,7 @@ fn final_sequence(end: track::Position) -> u64 { /// /// When `emit_boundary` is set, a declared-but-not-yet-reached final sequence surfaces as /// [`Recv::Boundary`] in an idle moment (after groups and datagrams), so the caller can send -/// SUBSCRIBE_END as soon as the ending is known rather than waiting for the live edge to reach +/// SUBSCRIBE_FINISH as soon as the ending is known rather than waiting for the live edge to reach /// it. The caller clears `emit_boundary` after the first boundary so it fires once. /// /// When `end_error` is set, the boundary carries the error an errored end has, and reaching @@ -3685,7 +3685,7 @@ fn poll_recv_next( if !withheld { let error = error.filter(|_| end_error); return Poll::Ready(res.map(|end| { - Recv::Boundary(lite::SubscribeEnd { + Recv::Boundary(lite::SubscribeFinish { group: final_sequence(end), error: error.as_ref().map(StreamError::from), }) @@ -3877,8 +3877,8 @@ impl Bounds { Ok(bounds) } - /// Whether a subscription to these ranges over `version` ends once they are covered: - /// lite-07 with every range bounded. Earlier versions read a FIN as the track's end. + /// Whether a subscription to these ranges over `version` is answered once they are + /// covered, even with no group to resolve its start: lite-07 with every range bounded. fn ends(&self, version: Version) -> bool { version.has_ranges() && self.ranges.iter().all(|range| range.end.is_some()) } @@ -4105,7 +4105,7 @@ struct TrackRun { // peer was told is untimed, which ends the subscription. refused: Option, // Lite05+ resolves the range on the Subscribe Stream itself: SUBSCRIBE_START - // once the first group is known, SUBSCRIBE_END as soon as the track declares its + // once the first group is known, SUBSCRIBE_FINISH as soon as the track declares its // exclusive final sequence (which may be ahead of the live edge). emit_range: bool, // Where SUBSCRIBE_START resolved the feed, once sent. @@ -4115,8 +4115,8 @@ struct TrackRun { // Groups given up before the start resolved, which it must not name, with why. skipped: RangeMap, end_sent: bool, - // Lite-07 with every range bounded: the subscription ends once nothing in its ranges is - // left to deliver, rather than parking for an update to widen it. + // Lite-07 with every range bounded: once every group in its ranges is given up, it is + // answered without a group to resolve its start, rather than left unanswered. bounded: bool, // Lite05+ names every sequence below the end that this subscription never // got with SUBSCRIBE_DROP, so the subscriber settles without waiting out its grace @@ -4196,7 +4196,7 @@ impl TrackRun { /// the cap are held in the producer's cache until the subscriber raises the cap (or /// unsets it), then served in order. Only a peer FIN ends the subscription. This is /// what lets relays pause an upstream subscription across consumer churn without - /// tearing it down. Lite-07 ends one whose every range is bounded once it is covered. + /// tearing it down. Lite-07 keeps every subscription open until the subscriber FINs. fn update(&mut self, upd: lite::SubscribeUpdate) { // The update moves the cursor and the bounds the track side judges against, which // notifies nothing it parked on. @@ -4319,8 +4319,8 @@ impl TrackRun { } /// Serve one ready group or datagram (`Continue`), or `Break` once the live edge - /// reaches the track's boundary or a bounded lite-07 subscription is covered. Returning per item lets the caller watch the - /// requester while work stays ready. + /// reaches the track's boundary before lite-07, which then keeps the stream open for + /// updates. Returning per item lets the caller watch the requester while work stays ready. fn poll_step( &mut self, writer: &mut Writer, @@ -4489,14 +4489,34 @@ impl TrackRun { // even if trailing groups (below it) are still in flight, then // keep serving them until the live edge reaches the boundary. self.end_sent = true; - writer.buffer(&lite::SubscribeResponse::End(end))?; + writer.buffer(&lite::SubscribeResponse::Finish(end))?; } Recv::Finished => { if !self.children.is_empty() { return Poll::Pending; } - self.drop_unserved(writer)?; - return Poll::Ready(Ok(ControlFlow::Break(()))); + // Lite-07 sends no FIN to say nothing was served, so a track that ended + // before any group resolved the start still says where delivery starts, + // and the drops below then account for every group. + let mut answered = false; + if self.ctx.version.keeps_subscribe_open() + && self.emit_range && self.start.is_none() + && let Poll::Ready(Ok(fin)) = self.track.poll_finished(&kio::Waiter::noop()) + && final_sequence(fin) > 0 + { + self.resolve_held(final_sequence(fin), writer)?; + answered = true; + } + let dropped = self.drop_unserved(writer)? || answered; + if !self.ctx.version.keeps_subscribe_open() { + return Poll::Ready(Ok(ControlFlow::Break(()))); + } + // Lite-07 leaves the stream open past SUBSCRIBE_FINISH: an update may still + // ask for the ended track's history, and the subscriber FINs once it wants + // nothing more. Drops just buffered go out on the next pass. + if !dropped { + return Poll::Pending; + } } } return Poll::Ready(Ok(ControlFlow::Continue(()))); @@ -4506,28 +4526,24 @@ impl TrackRun { if self.poll_forward_drop(writer, waiter)? { return Poll::Ready(Ok(ControlFlow::Continue(()))); } - // The FIN ends the subscription, not the track, so no SUBSCRIBE_END precedes it. - if self.covered() { - return Poll::Ready(Ok(ControlFlow::Break(()))); + // A bounded subscription none of whose groups can be served has no group to resolve + // its start: answer it now, with the drops held for that start, so the subscriber + // learns every group's fate without a FIN. + if self.unservable() { + self.resolve_held(self.floor, writer)?; + return Poll::Ready(Ok(ControlFlow::Continue(()))); } Poll::Pending } - /// Whether a bounded subscription has nothing left to deliver: every group its ranges - /// name was served or dropped, and every stream it opened is done. - fn covered(&mut self) -> bool { - if !self.bounded || self.opening.is_some() || self.first.is_some() || !self.children.is_empty() { - return false; - } - match (self.start, &self.served) { - (Some(start), Some(served)) => self.track.covered(start, served), - // Before SUBSCRIBE_START the subscriber owes itself nothing (see - // `SubStream::owed`), so only a range given up whole is covered. Datagrams - // resolve no start, and a FIN would overtake them, so a range they served - // keeps the subscription open. - (None, _) => self.track.covered(self.floor, &self.skipped), - (Some(_), None) => false, - } + /// Whether a bounded subscription still unanswered has given up every group its ranges + /// name. Datagrams resolve no start, so a range they served stays unanswered. + fn unservable(&mut self) -> bool { + self.bounded + && self.start.is_none() + && self.first.is_none() + && self.opening.is_none() + && self.track.covered(self.floor, &self.skipped) } /// Lite-05+ names a sequence as soon as this cursor gives it up; whether one was taken. @@ -4546,7 +4562,9 @@ impl TrackRun { }; match self.start { None => self.skipped.insert(range, err), - Some(_) => self.drop_range(range, &err, writer)?, + Some(_) => { + self.drop_range(range, &err, writer)?; + } } Ok(true) } @@ -4604,49 +4622,57 @@ impl TrackRun { } /// Drop, with `err`, whatever part of `range` this subscription has not been served or - /// dropped yet. + /// dropped yet. Returns whether any drop was buffered. fn drop_range( &mut self, range: Range, err: &StreamError, writer: &mut Writer, - ) -> Result<(), Error> { + ) -> Result { let Some(served) = &mut self.served else { - return Ok(()); + return Ok(false); }; if range.is_empty() { - return Ok(()); + return Ok(false); } let gaps: Vec<_> = served.gaps(&range).collect(); - for gap in gaps { + for gap in &gaps { served.insert(gap.clone()); let drop = lite::SubscribeDrop::new(gap.start..=gap.end - 1, err, self.ctx.version); writer.buffer(&lite::SubscribeResponse::Drop(drop))?; } - Ok(()) + Ok(!gaps.is_empty()) } /// Name every sequence from the start to the end that this subscription never got. The /// track has ended, so none of them will be served now: groups it never produced (a /// skipped sequence), and groups the cursor passed over (stale, or missing their head). + /// + /// Returns whether any drop was buffered. fn drop_unserved( &mut self, writer: &mut Writer, - ) -> Result<(), Error> { + ) -> Result { let Some(start) = self.start else { // Without a SUBSCRIBE_START the subscriber owes itself no group (see - // `SubStream::owed`) and settles on the FIN alone, so there is nothing to drop. - return Ok(()); + // `SubStream::owed`), so there is nothing to drop. + return Ok(false); }; let Poll::Ready(Ok(fin)) = self.track.poll_finished(&kio::Waiter::noop()) else { - // A track closed without a final sequence sent no SUBSCRIBE_END, so the subscriber + // A track closed without a final sequence sent no SUBSCRIBE_FINISH, so the subscriber // owes itself no range either. - return Ok(()); + return Ok(false); }; // A cap below the end bounds what the subscriber is owed, as on its side. let fin = final_sequence(fin); + // Only the groups the ranges name: one between disjoint ranges is not owed, and + // naming it would keep a later update widening over it from delivering it. let end = self.track.subscription().end().map_or(fin, |end| end.min(fin)); - self.drop_range(start..end, &StreamError::NotFound, writer) + let mut dropped = false; + for run in self.track.runs_in_range(start, end) { + dropped |= self.drop_range(run, &StreamError::NotFound, writer)?; + } + Ok(dropped) } } @@ -4682,12 +4708,48 @@ impl TrackRun { { start = skipped.end.min(group.sequence); } + self.resolve(start, writer)?; + // SUBSCRIBE_START names where delivery starts. Groups already skipped between + // the requested floor and this group are not served. A later group at or above an + // explicit floor, still inside the subscriber's max age, is delivered, so the cursor + // stays at that floor. + // + // A pre-06 subscription that named no group is the exception: those drafts define an + // absent Group Start as the latest group, and the resolved start becomes the + // floor. Lite-06 encodes a floor of group 0 as 0, which decodes as no named floor; + // that 0 is group 0, so it is not pinned. Raised, not assigned: an update that + // landed while the group was held may already have raised it past. + if self.named.is_none() && !self.ctx.version.resolves_start() { + self.raise_floor(start); + } + self.serve(group); + Ok(()) + } + + /// Resolve a start no group resolved: the lowest range start the cursor resolves, or + /// `none` when no range starts anywhere. + fn resolve_held( + &mut self, + none: u64, + writer: &mut Writer, + ) -> Result<(), Error> { + let start = self.starts(self.floor).into_iter().flatten().min().unwrap_or(none); + self.resolve(start, writer) + } + + /// Send SUBSCRIBE_START resolving delivery at `start`, then name the groups given up + /// above it while it was held back. + fn resolve( + &mut self, + start: u64, + writer: &mut Writer, + ) -> Result<(), Error> { let skipped = std::mem::take(&mut self.skipped); self.start = Some(start); // Only groups: the subscriber derives the start frame from its own request - // (see `lite::SubscribeStart`). The track is live by now, since its first group - // was readable. A relay can hold just the first group of a backlog its source - // already answered past, so the source's largest says where the live feed is. + // (see `lite::SubscribeStart`). A relay can hold just the first group of a backlog + // its source already answered past, so the source's largest says where the live + // feed is. let live = match self.track.poll_live(&kio::Waiter::noop()) { Poll::Ready(largest) => largest, Poll::Pending => None, @@ -4702,20 +4764,6 @@ impl TrackRun { for (range, err) in skipped.iter() { self.drop_range(range.start.max(start)..range.end, err, writer)?; } - // SUBSCRIBE_START names where delivery starts. Groups already skipped between - // the requested floor and this group are not served. A later group at or above an - // explicit floor, still inside the subscriber's max age, is delivered, so the cursor - // stays at that floor. - // - // A pre-06 subscription that named no group is the exception: those drafts define an - // absent Group Start as the latest group, and the resolved start becomes the - // floor. Lite-06 encodes a floor of group 0 as 0, which decodes as no named floor; - // that 0 is group 0, so it is not pinned. Raised, not assigned: an update that - // landed while the group was held may already have raised it past. - if self.named.is_none() && !self.ctx.version.resolves_start() { - self.raise_floor(start); - } - self.serve(group); Ok(()) } @@ -6675,10 +6723,10 @@ mod serve_group_test { assert!(futures::poll!(run.as_mut()).is_pending()); write_group(&mut track, 2, 2); track.finish().unwrap(); - run.await.unwrap(); + assert!(futures::poll!(run.as_mut()).is_pending(), "the stream stays open"); // SUBSCRIBE_START with the range starting at 0, a DROP for the skipped group 1 - // with NOT_FOUND (0x33), which the end gives up on at once, and SUBSCRIBE_END at 3. + // with NOT_FOUND (0x33), which the end gives up on at once, and SUBSCRIBE_FINISH at 3. assert_eq!( *log.writes.lock().unwrap(), [0, 4, 1, 0, 1, 1, 2, 3, 1, 1, 0x33, 1, 2, 3, 0] @@ -7276,23 +7324,26 @@ mod serve_group_test { } } - /// A bounded lite-07 subscription ends once its source gives up every range, even with no - /// start resolved, and not while one range is still open. + /// A bounded lite-07 subscription is answered once its source gives up every range, even + /// with no group to resolve its start, and not while one range is still open. The stream + /// stays open for an update. #[kio::test] - async fn giving_up_every_range_before_the_start_ends_it() { + async fn giving_up_every_range_before_the_start_answers_it() { // Disjoint ranges, then one range given up a group at a time with alternating errors. for (mut bench, last) in [(CoverageBench::new(3, 2), 4), (CoverageBench::adjacent(4), 4)] { bench.wake(); bench.wake(); + assert!(bench.run.start.is_none(), "a range is still open"); bench.track.drop_groups(last..last + 1, StreamError::NotFound); loop { match bench.run.poll_step(&mut bench.writer, &kio::Waiter::noop()) { Poll::Ready(Ok(ControlFlow::Continue(()))) => continue, - Poll::Ready(Ok(ControlFlow::Break(()))) => break, - other => panic!("the subscription should end, got {other:?}"), + Poll::Pending => break, + other => panic!("the subscription stays open, got {other:?}"), } } - assert!(bench.run.start.is_none()); + assert_eq!(bench.run.start, Some(0), "answered from the floor"); + assert!(bench.run.skipped.is_empty(), "every held drop went out"); } } @@ -7429,8 +7480,9 @@ mod serve_group_test { track.finish_at(track::Position::group(6), None).unwrap(); drop(track); kio::wait(|waiter| run.poll(&mut writer, waiter)).await.unwrap(); - // The feed gives up on 2..5 once its grace passes, a miss nothing fills. - assert_eq!(drops(&mut writer, &log, version).await, [(2, 4), (1, 1)]); + // The feed gives up on 2..5 once its grace passes, a miss nothing fills. Group 1 lies + // below the floor the ranges name, so the end sweep owes it nothing. + assert_eq!(drops(&mut writer, &log, version).await, [(2, 4)]); } /// A raised floor sends no early drop: a later update may lower it again, and the @@ -7822,9 +7874,10 @@ mod serve_group_test { assert!(relay.started_at(6), "named a skipped group"); } - /// A track that ends without a group still ends the subscription, with no stream owed. + /// A track that ends without a group still finishes the subscription, with no stream + /// owed, and the stream stays open. #[kio::test] - async fn lite07_empty_track_ends_without_a_start() { + async fn lite07_empty_track_finishes_without_a_start() { let track = track::Producer::new( Arc::new(broadcast::Info::default()), "test", @@ -7834,7 +7887,11 @@ mod serve_group_test { track.finish().unwrap(); let (mut run, mut writer, log) = lite07_run(SinkSession::new(Log::default()), subscriber); - kio::wait(|waiter| run.poll(&mut writer, waiter)).await.unwrap(); + assert!( + kio::wait(|waiter| run.poll(&mut writer, waiter)) + .now_or_never() + .is_none() + ); assert_eq!(*log.writes.lock().unwrap(), [1, 2, 0, 0]); } @@ -7871,7 +7928,7 @@ mod serve_group_test { }; *open = true; drop(open); - run.await.unwrap(); + assert!(futures::poll!(run.as_mut()).is_pending(), "the stream stays open"); assert_eq!( *log.writes.lock().unwrap(), [0, 4, 1, 0, 1, 1, 2, 3, 0, 0, 0x34, 1, 2, 2, 0] diff --git a/rs/moq-net/src/lite/subscribe.rs b/rs/moq-net/src/lite/subscribe.rs index efabfce87ae..2872f075f44 100644 --- a/rs/moq-net/src/lite/subscribe.rs +++ b/rs/moq-net/src/lite/subscribe.rs @@ -461,7 +461,7 @@ fn encode_frame_bounds( /// Publisher's acknowledgement on the Subscribe Stream for drafts 01-04. /// /// Lite05+ replaced this with implicit acceptance plus -/// [`SubscribeStart`]/[`SubscribeEnd`]; the immutable timescale/cache moved +/// [`SubscribeStart`]/[`SubscribeFinish`]; the immutable timescale/cache moved /// to [`super::TrackInfo`]. #[derive(Clone, Debug)] pub struct SubscribeOk { @@ -593,18 +593,21 @@ impl<'a> Message<'a> for SubscribeStart { } } -/// Signals the exclusive end of a Lite05+ subscription. +/// SUBSCRIBE_FINISH (SUBSCRIBE_END before lite-07): the exclusive end of a Lite05+ +/// subscription's track. /// -/// No group at or after `group` will be produced. `0` means the track ended +/// No group at or after `group` will be produced. Before lite-07 the publisher FINs the +/// Subscribe Stream after it; lite-07 leaves the stream open for updates, so the +/// subscriber FINs once it wants nothing more. `0` means the track ended /// before producing any groups. #[derive(Clone, Debug)] -pub struct SubscribeEnd { +pub struct SubscribeFinish { pub group: u64, /// The error the track ended with, if it did not end cleanly. Lite-07 and later. pub error: Option, } -impl<'a> Message<'a> for SubscribeEnd { +impl<'a> Message<'a> for SubscribeFinish { const CONTROL: Option = Some(crate::stats::Message::Subscribe); fn decode_msg(r: &mut Decoder<'a>, version: Version) -> Result { if !version.has_track_stream() { @@ -796,13 +799,14 @@ impl<'a> Message<'a> 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+: `0x0` SUBSCRIBE_START (SUBSCRIBE_OK on lite-07), `0x1` SUBSCRIBE_FINISH +/// (SUBSCRIBE_END before lite-07), `0x2` SUBSCRIBE_DROP /// (SUBSCRIBE_OK was removed; acceptance is implicit). #[derive(Clone, Debug)] pub enum SubscribeResponse { Ok(SubscribeOk), Start(SubscribeStart), - End(SubscribeEnd), + Finish(SubscribeFinish), Drop(SubscribeDrop), } @@ -832,7 +836,7 @@ impl Encode for SubscribeResponse { // Lite05+: SUBSCRIBE_OK is gone; START/END/DROP carry the resolved range. _ => match self { Self::Start(start) => encode_typed(w, 0, start, version)?, - Self::End(end) => encode_typed(w, 1, end, version)?, + Self::Finish(end) => encode_typed(w, 1, end, version)?, Self::Drop(drop) => encode_typed(w, 2, drop, version)?, Self::Ok(_) => return Err(EncodeError::Version), }, @@ -858,7 +862,7 @@ impl<'a> Decode<'a, Version> for SubscribeResponse { let typ = buf.varint()?; match typ { 0 => Ok(Self::Start(SubscribeStart::decode(buf, version)?)), - 1 => Ok(Self::End(SubscribeEnd::decode(buf, version)?)), + 1 => Ok(Self::Finish(SubscribeFinish::decode(buf, version)?)), 2 => Ok(Self::Drop(SubscribeDrop::decode(buf, version)?)), _ => Err(DecodeError::InvalidMessage(typ)), } @@ -960,7 +964,7 @@ mod test { #[test] fn subscribe_end_roundtrips_on_lite05() { - let resp = SubscribeResponse::End(SubscribeEnd { group: 7, error: None }); + let resp = SubscribeResponse::Finish(SubscribeFinish { group: 7, error: None }); let mut buf = Vec::new(); resp.encode(&mut Encoder::new(&mut buf, Version::Lite05.into()), Version::Lite05) .unwrap(); @@ -968,7 +972,7 @@ mod test { assert_eq!(buf, [1, 1, 7]); let mut slice = buf.as_slice(); match crate::coding::decode_buf(&mut slice, Version::Lite05, |r, v| SubscribeResponse::decode(r, v)).unwrap() { - SubscribeResponse::End(end) => assert_eq!(end.group, 7), + SubscribeResponse::Finish(end) => assert_eq!(end.group, 7), other => panic!("expected End, got {other:?}"), } } @@ -976,7 +980,7 @@ mod test { #[test] fn subscribe_end_carries_its_error_on_lite07() { let encode = |error: Option| { - let resp = SubscribeResponse::End(SubscribeEnd { group: 7, error }); + let resp = SubscribeResponse::Finish(SubscribeFinish { group: 7, error }); let mut buf = Vec::new(); resp.encode(&mut Encoder::new(&mut buf, Version::Lite07.into()), Version::Lite07) .unwrap(); @@ -987,7 +991,7 @@ mod test { match crate::coding::decode_buf(&mut slice, Version::Lite07, |r, v| SubscribeResponse::decode(r, v)) .unwrap() { - SubscribeResponse::End(end) => (end.group, end.error), + SubscribeResponse::Finish(end) => (end.group, end.error), other => panic!("expected End, got {other:?}"), } }; @@ -1000,7 +1004,7 @@ mod test { assert_eq!(decode(&errored), (7, Some(StreamError::App(7)))); // An older version has nowhere to carry it. - let resp = SubscribeResponse::End(SubscribeEnd { + let resp = SubscribeResponse::Finish(SubscribeFinish { group: 7, error: Some(StreamError::App(7)), }); diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 787ac6fc9c1..90983d7a9e5 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -2458,11 +2458,11 @@ mod tests { ranges: vec![lite::Range::OLDEST], tail: tail.clone(), served: None, + started: false, starts: Vec::new(), end: None, wanted: None, unanswered: None, - bounded: false, updated: false, _working: crate::session::Work::default().start(), }) @@ -2499,62 +2499,6 @@ mod tests { assert!(fetches.held.is_empty(), "released once abandoned"); } - /// A lite-07 publisher that FINs with no SUBSCRIBE_START served nothing, which covers the - /// ranges the SUBSCRIBE asked for but not ones an update it may not have read changed. - #[kio::test] - async fn a_startless_fin_covers_only_the_subscribed_ranges() { - let tail = kio::Producer::new(Tail::default()); - let Sub::Active(mut active) = feed_sub(&tail).await else { - unreachable!("feed_sub is active") - }; - active.ranges = vec![lite::Range::span(Some(0), 0, Some(1), None)]; - assert_eq!(active.covered(), Some(Vec::new())); - - active.updated = true; - assert_eq!(active.covered(), None, "the update may be unread"); - - active.served = Some(0); - assert_eq!(active.covered(), Some(std::iter::once(0..2).collect())); - } - - /// A covered FIN owes a relative range only from its own SUBSCRIBE_START, not from the - /// lowest start of a disjoint range below it. - #[kio::test] - async fn a_covered_fin_owes_each_range_from_its_own_start() { - let tail = kio::Producer::new(Tail::default()); - let Sub::Active(mut active) = feed_sub(&tail).await else { - unreachable!("feed_sub is active") - }; - let end = |group| Some(lite::Anchor::Absolute(crate::track::Position { group, frame: 0 })); - active.ranges = vec![ - lite::Range::span(Some(0), 0, Some(0), None), - lite::Range { - start: Some(lite::Anchor::Relative(0)), - end: end(11), - }, - lite::Range { - start: Some(lite::Anchor::Relative(0)), - end: end(20), - }, - ]; - active.served = Some(0); - active.starts = vec![Some(0), Some(10), None]; - assert_eq!(active.covered(), Some(vec![0..1, 10..11, 20..20])); - - active.ranges[0] = lite::Range::span(Some(0), 0, Some(2), None); - active.starts[0] = Some(1); - assert_eq!(active.covered(), Some(vec![1..3, 10..11, 20..20])); - active.starts[0] = None; - assert_eq!(active.covered(), Some(vec![3..3, 10..11, 20..20])); - - active.updated = true; - assert_eq!( - active.covered(), - Some(vec![0..3, 0..11, 0..20]), - "an update leaves only the lowest" - ); - } - /// A bounded feed goes live at its answer alone only when its resolved ranges stop short /// of Largest. One that serves Largest waits for it, so a finite read judges an old /// cached group against that edge rather than against itself. @@ -2663,11 +2607,11 @@ mod tests { ranges: vec![lite::Range::span(Some(floor), 0, None, None)], tail: Default::default(), served: None, + started: false, starts: Vec::new(), end: None, wanted: None, unanswered: None, - bounded: false, updated: false, _working: crate::session::Work::default().start(), }); @@ -3103,6 +3047,10 @@ mod tests { /// The owed groups start at the floor the demand last asked for, which an update can move /// either way after SUBSCRIBE_START answered the SUBSCRIBE. #[kio::test] + #[expect( + clippy::single_range_in_vec_init, + reason = "the owed groups are a list of runs, often one" + )] async fn the_owed_floor_follows_the_demand() { let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()); let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), VERSION) @@ -3120,40 +3068,86 @@ mod tests { ranges: vec![lite::Range::span(Some(2), 0, None, None)], tail: Default::default(), served: None, + started: false, starts: Vec::new(), - end: Some(lite::SubscribeEnd { group: 10, error: None }), + end: Some(lite::SubscribeFinish { group: 10, error: None }), wanted: None, unanswered: None, - bounded: false, updated: false, _working: crate::session::Work::default().start(), }; assert_eq!( sub.owed(None), - Some(10..10), + Some(vec![]), "without SUBSCRIBE_START nothing was served" ); + // An update moves the floor, the ranges, and `updated` together, as + // `begin_subscription` does. + let update = |sub: &mut SubStream, start: Option, ranges: Vec| { + sub.start = start; + sub.ranges = ranges; + sub.updated = true; + }; + sub.started = true; + update(&mut sub, Some(1), vec![lite::Range::span(Some(1), 0, None, None)]); + assert_eq!( + sub.owed(None), + Some(vec![1..10]), + "an update asks from a floor even when SUBSCRIBE_START started no range" + ); + + sub.start = Some(2); + sub.ranges = vec![ + lite::Range::span(Some(2), 0, Some(3), None), + lite::Range::span(Some(6), 0, None, None), + ]; + assert_eq!( + sub.owed(None), + Some(vec![2..4, 6..10]), + "an update that keeps the floor but adds a range still owes it" + ); + + update(&mut sub, Some(2), vec![lite::Range::span(Some(2), 0, None, None)]); + sub.updated = false; sub.served = Some(4); + sub.starts = vec![Some(4)]; assert_eq!( sub.owed(None), - Some(2..10), + Some(vec![4..10]), "the groups below START are accounted for on arrival" ); - sub.start = Some(6); - assert_eq!(sub.owed(None), Some(6..10), "a raised floor owes nothing below it"); - sub.start = Some(1); + update(&mut sub, Some(6), vec![lite::Range::span(Some(6), 0, None, None)]); assert_eq!( sub.owed(None), - Some(1..10), + Some(vec![6..10]), + "a raised floor owes nothing below it" + ); + update(&mut sub, Some(1), vec![lite::Range::span(Some(1), 0, None, None)]); + assert_eq!( + sub.owed(None), + Some(vec![1..10]), "a lowered floor owes what it newly asked for" ); - sub.start = None; + update(&mut sub, None, vec![lite::Range::span(None, 0, None, None)]); assert_eq!( sub.owed(Some(8)), - Some(4..8), + Some(vec![4..8]), "a live-edge floor starts where START resolved it" ); + update( + &mut sub, + Some(1), + vec![ + lite::Range::span(Some(1), 0, Some(3), None), + lite::Range::span(Some(6), 0, None, None), + ], + ); + assert_eq!( + sub.owed(None), + Some(vec![1..4, 6..10]), + "nothing is owed between disjoint ranges" + ); } async fn receive_padding(version: Version, bytes: Vec) { @@ -3841,7 +3835,7 @@ mod tests { } /// An idle copy that resubscribes gets its newest cached group again. That stream only - /// duplicates the cache, but lite-07's SUBSCRIBE_END still counts it, so the tail must + /// duplicates the cache, but lite-07's SUBSCRIBE_FINISH still counts it, so the tail must /// too, or the subscription waits out the grace before it ends. #[kio::test] async fn a_duplicate_group_stream_still_counts_toward_the_end() { @@ -3902,12 +3896,12 @@ mod tests { assert!(matches!(res, Err(Error::Duplicate)), "{res:?}"); drop(group); - // SUBSCRIBE_END counted one stream: the tail settles without the clock moving. + // SUBSCRIBE_FINISH counted one stream: the tail settles without the clock moving. let mut settle = Settle::new(&clock, tail.consume()); assert!(settle.poll(&kio::Waiter::noop(), |tail| tail.streams() >= 1).is_ready()); } - /// A SUBSCRIBE_END below a group already received contradicts that group. lite-05 + /// A SUBSCRIBE_FINISH below a group already received contradicts that group. lite-05 /// specified an inclusive end, and `@moq/net` 0.1.3 to 0.1.9 sent one, so there it only /// costs the early boundary; later drafts abort the track. #[kio::test] @@ -3923,7 +3917,7 @@ mod tests { version, ) .unwrap(); - lite::SubscribeResponse::End(lite::SubscribeEnd { group: 2, error: None }) + lite::SubscribeResponse::Finish(lite::SubscribeFinish { group: 2, error: None }) .encode( &mut crate::coding::Encoder::new(&mut responses, version.into()), version, @@ -4038,7 +4032,7 @@ mod tests { }) .encode(&mut encoder, version) .unwrap(); - lite::SubscribeResponse::End(lite::SubscribeEnd { group: 0, error: None }) + lite::SubscribeResponse::Finish(lite::SubscribeFinish { group: 0, error: None }) .encode(&mut encoder, version) .unwrap(); session.push(&answer); @@ -4079,7 +4073,7 @@ mod tests { } /// Lite-07's SUBSCRIBE_START says where the track is, so a reader knows its largest - /// frame before any group arrives. The FIN with no SUBSCRIBE_END then aborts the track, + /// frame before any group arrives. The FIN with no SUBSCRIBE_FINISH then aborts the track, /// and the stream ends with it. #[kio::test] async fn subscribe_start_reports_the_largest() { @@ -4299,7 +4293,7 @@ mod tests { .unwrap(); } if clean { - lite::SubscribeResponse::End(lite::SubscribeEnd { group: 0, error: None }) + lite::SubscribeResponse::Finish(lite::SubscribeFinish { group: 0, error: None }) .encode( &mut crate::coding::Encoder::new(&mut responses, version.into()), version, @@ -4331,7 +4325,7 @@ mod tests { } } - /// SUBSCRIBE_START at group 0 and SUBSCRIBE_END at 1: the publisher ended the track + /// SUBSCRIBE_START at group 0 and SUBSCRIBE_FINISH at 1: the publisher ended the track /// owing group 0. fn owed_responses(version: Version) -> Vec { let mut responses = Vec::new(); @@ -4342,7 +4336,7 @@ mod tests { }) .encode(&mut encoder, version) .unwrap(); - lite::SubscribeResponse::End(lite::SubscribeEnd { group: 1, error: None }) + lite::SubscribeResponse::Finish(lite::SubscribeFinish { group: 1, error: None }) .encode(&mut encoder, version) .unwrap(); responses @@ -5610,7 +5604,7 @@ mod tests { /// A session that dies before a tail settles hands the track back with the session's error, /// even one the publisher ended, and the abort `TrackServeRun` commits then ends readers - /// cleanly only if the declared end settled: SUBSCRIBE_END on lite-05+, the FIN before it. + /// cleanly only if the declared end settled: SUBSCRIBE_FINISH on lite-05+, the FIN before it. #[kio::test] async fn internal_session_end_aborts_an_unsettled_tail() { use crate::transport::poll::Session as _; @@ -5657,11 +5651,10 @@ mod tests { if version.has_track_stream() { serving.serving.finish_at(Position::group(1), None).unwrap(); } - // The tail machine has not seen group 0 accounted for, so it never settles. + // Nothing says which groups are owed, so only the grace settles it, which has not passed. serving.mode = ServeMode::Tail { settle: Settle::new(&serve.subscriber.runtime, tail.consume()), - owed: Some(std::iter::once(0..1).collect()), - track: true, + requested_end: None, }; assert!(serving.poll(&serve, &kio::Waiter::noop()).is_pending()); } @@ -5729,7 +5722,7 @@ mod tests { send.reset(crate::StreamError::NotFound.to_code()); } else { let mut end = Vec::new(); - lite::SubscribeResponse::End(lite::SubscribeEnd { group: 0, error: None }) + lite::SubscribeResponse::Finish(lite::SubscribeFinish { group: 0, error: None }) .encode(&mut crate::coding::Encoder::new(&mut end, version.into()), version) .unwrap(); kio::wait(|waiter| send.poll_write(waiter, &end)).await.unwrap(); @@ -5953,53 +5946,6 @@ mod tests { assert_eq!(h.wire().len(), established, "no SUBSCRIBE_UPDATE inverting the range"); } - /// A lite-07 FIN on unchanged bounded ranges covers them even when a group the publisher - /// sent, such as a lost datagram, never arrived: subscribing again would replay the rest - /// and repeat for as long as it stays undeliverable. Ranges an update changed are asked - /// again, since the FIN may predate the update. - #[kio::test] - async fn a_covered_fin_short_of_a_lost_group_resubscribes_only_after_an_update() { - for updated in [false, true] { - let h = Harness::new(Version::Lite07); - let request = h._broadcast.reserve_track("lossy").unwrap(); - let mut serving = ServeLoop::new(&h.serve, request, track::Info::default()); - let track = h._broadcast.consume().track("lossy").unwrap(); - let _reader = track.subscribe(Subscription::default().with_range(0..2)).await.unwrap(); - assert!(serving.poll(&h.serve, &kio::Waiter::noop()).is_pending()); - let Sub::Active(active) = &mut serving.sub else { - panic!("the reader opens the upstream") - }; - assert!(active.bounded); - active.updated = updated; - active.served = Some(0); - // Group 0 arrived; group 1 was sent and lost. - let Ok(mut tail) = active.tail.write() else { - panic!("ledger closed") - }; - tail.account(0..1, h.serve.subscriber.runtime.now()); - drop(tail); - serving.mode = ServeMode::Tail { - settle: Settle::new(&h.serve.subscriber.runtime, active.tail.consume()), - owed: active.covered(), - track: false, - }; - assert!(serving.poll(&h.serve, &kio::Waiter::noop()).is_pending()); - assert!( - matches!(serving.mode, ServeMode::Tail { .. }), - "{updated}: waits out the grace" - ); - - kio::clock::sleep(Duration::from_secs(120)).await; - let _ = serving.poll(&h.serve, &kio::Waiter::noop()); - assert_eq!(serving.covered.is_some(), !updated, "{updated}"); - assert_eq!( - matches!(serving.sub, Sub::None), - !updated, - "{updated}: subscribes again" - ); - } - } - /// Concurrent joins over the refresh ratio send one SUBSCRIBE_UPDATE upstream, a new /// group past the largest sends the next, and every update restates the latest. #[kio::test] @@ -7514,8 +7460,7 @@ fn wire_ranges(subscription: &Subscription, timescale: Option) -> Res Ok((!ranges.is_empty()).then_some(ranges)) } -/// Whether every range has an end, so a lite-07 publisher ends the subscription once it -/// covers them. +/// Whether every range has an end. fn bounded(ranges: &[lite::Range]) -> bool { ranges.iter().all(|range| range.end.is_some()) } @@ -7568,43 +7513,31 @@ struct SubStream { tail: kio::Producer, /// The first group the publisher serves (SUBSCRIBE_START), once declared. served: Option, + /// Whether SUBSCRIBE_START arrived, which `served` alone cannot say when no range starts. + started: bool, /// Where SUBSCRIBE_START says each range's delivery starts, `None` for one with nothing /// to deliver. It matches `ranges` until an update changes them. starts: Vec>, - /// The track's exclusive end (SUBSCRIBE_END), once declared. - end: Option, + /// The track's exclusive end (SUBSCRIBE_FINISH), once declared. + end: Option, /// The refresh join's wanted group, restated in every SUBSCRIBE_UPDATE. wanted: Option, /// Until the publisher's first response, what the source must still hold: a SUBSCRIBE /// without an instance resolves whichever one wins the path when it arrives, so one /// still unanswered once another instance takes the path may be answered by it. unanswered: Option, - /// Whether this stream ever asked for bounded ranges alone, which a lite-07 publisher - /// FINs without SUBSCRIBE_END once it covers them. - bounded: bool, - /// Whether an update changed the ranges since the SUBSCRIBE, which a publisher may - /// FIN before reading. + /// Whether an update changed the ranges since the SUBSCRIBE, which the publisher may + /// have read before its SUBSCRIBE_START, so the starts need not match the SUBSCRIBE. updated: bool, /// Holds the session busy for as long as the subscription is open. _working: crate::session::Working, } impl SubStream { - /// The groups the publisher still owes once it has ended the subscription. - /// - /// `None` when nothing says which: drafts before SUBSCRIBE_END only have the FIN. - /// Without a SUBSCRIBE_START the publisher served no group at all. Otherwise they start - /// at the floor the demand last asked for, which an update can move either way, or - /// where SUBSCRIBE_START resolved a live-edge one. The groups the SUBSCRIBE asked for - /// below its SUBSCRIBE_START are accounted for as unavailable when it arrives. - fn owed(&self, requested_end: Option) -> Option> { - let end = self.end.as_ref()?.group; - let end = requested_end.map_or(end, |requested| requested.min(end)); - let start = match self.served { - Some(served) => self.start.unwrap_or(served), - None => end, - }; - Some(start..end) + /// Whether a lite-07 publisher still owes SUBSCRIBE_START once it ended the track: it + /// answers even when no group resolved the start, unless nothing was asked below the end. + fn awaits_start(&self) -> bool { + !self.started && self.end.as_ref().is_some_and(|end| end.group > self.start.unwrap_or(0)) } /// `largest`, when this bounded subscription's ranges, as SUBSCRIBE_START resolved them, @@ -7615,21 +7548,15 @@ impl SubStream { largest.filter(|largest| { bounded(&self.ranges) && self - .covered() + .spans() .is_none_or(|spans| !spans.iter().any(|span| span.contains(&largest.group))) }) } - /// The groups the publisher owed once it FINs a covered bounded lite-07 subscription - /// without SUBSCRIBE_END: each range from its absolute start, or else from where - /// SUBSCRIBE_START resolved delivery, up to its end. Groups below SUBSCRIBE_START were - /// accounted for when it arrived. - /// - /// Without a SUBSCRIBE_START the publisher served no group, which covers the ranges the - /// SUBSCRIBE asked for, so it is empty unless an update changed them since. `None` when - /// a range is open, or the ranges changed with nothing served: the publisher covered - /// what it asked before an update it may not have read. - fn covered(&self) -> Option>> { + /// Where each bounded range delivers, as SUBSCRIBE_START resolved it: from its absolute + /// start, or else from where SUBSCRIBE_START resolved delivery, up to its end. `None` when + /// a range is open, or the ranges changed with nothing served, so nothing says where. + fn spans(&self) -> Option>> { let Some(served) = self.served else { return (!self.updated).then(Vec::new); }; @@ -7642,8 +7569,7 @@ impl SubStream { return None; }; let start = match self.starts.get(index) { - // Each range's own start accounts for its skipped prefix, including - // an absolute one when another range has no absolute requested floor. + // Each range's own start accounts for its skipped prefix. Some(start) if !self.updated => start.unwrap_or(end.group), _ => match range.start { Some(lite::Anchor::Absolute(start)) => start.group, @@ -7654,6 +7580,45 @@ impl SubStream { }) .collect() } + + /// The groups the publisher still owes once it has ended the subscription, one run per + /// range: the publisher accounts for nothing between disjoint ranges. + /// + /// `None` when nothing says which: drafts before SUBSCRIBE_FINISH only have the FIN. + /// Without a SUBSCRIBE_START the publisher served no group at all, unless an update since + /// moved the floor the demand asks from. Each range is owed from where SUBSCRIBE_START + /// said it starts while the ranges are still the SUBSCRIBE's, which accounts for a skipped + /// prefix and resolves a relative or time start; after an update, from its absolute start, + /// or else from the first group served. The groups the SUBSCRIBE asked for below its + /// SUBSCRIBE_START are accounted for as unavailable when it arrives. + fn owed(&self, requested_end: Option) -> Option>> { + let end = self.end.as_ref()?.group; + let end = requested_end.map_or(end, |requested| requested.min(end)); + // A SUBSCRIBE_START that started no range owes nothing, unless an update since asked + // for more, which the open stream still serves. + if self.served.is_none() && !(self.started && self.updated) { + return Some(Vec::new()); + } + let fallback = self.served.or(self.start); + let owed = self.ranges.iter().enumerate().filter_map(|(index, range)| { + let start = match (self.updated, self.starts.get(index)) { + (false, Some(start)) => (*start)?, + _ => match range.start { + Some(lite::Anchor::Absolute(start)) => start.group, + // A start counted back from the latest group resolves where delivery did. + _ => fallback?, + }, + }; + // Our absolute ends name a group's first frame (see `wire_ranges`), and the rest + // go upstream open. + let stop = match range.end { + Some(lite::Anchor::Absolute(stop)) => stop.group.min(end), + _ => end, + }; + (start < stop).then_some(start..stop) + }); + Some(owed.collect()) + } } #[expect( @@ -8072,10 +8037,16 @@ impl TrackServe { let floor = active.start.or(active.served); start..floor.unwrap_or(u64::MAX) }); + // So do changed ranges that keep the floor, such as a widening that joins + // two: each gap they reach waits a fresh grace, which only delays giving up + // on one still owed. + let widened = (active.ranges != ranges) + .then(|| subscription.start().or(active.served).unwrap_or(0)..u64::MAX); if let Ok(mut tail) = active.tail.write() { tail.set_grace(producer.tail_grace(subscription.budget())); - if let Some(groups) = lowered { - tail.demand(groups, self.subscriber.runtime.now()); + let now = self.subscriber.runtime.now(); + for groups in lowered.into_iter().chain(widened) { + tail.demand(groups, now); } } active.start = subscription.start(); @@ -8347,7 +8318,6 @@ fn wire_max_delay(max_delay: Option) -> std::time::Duration /// Buffer the current SUBSCRIBE parameters. The caller flushes. fn buffer_update(active: &mut SubStream) -> Result<(), Error> { - active.bounded |= bounded(&active.ranges); active.stream.writer.buffer(&lite::SubscribeUpdate { wanted_group: active.wanted, priority: active.priority, @@ -8447,10 +8417,10 @@ impl Establish { requested: self.subscription.start(), tail: self.tail.clone(), served: None, + started: false, starts: Vec::new(), end: None, wanted: None, - bounded: bounded(&self.ranges), updated: false, _working: self.session.work().start(), } @@ -8538,9 +8508,12 @@ enum TrackRunState { _working: crate::session::Working, }, Serve(ServeLoop), - /// Preserve the lite07 completion FIN until the publisher acknowledges it. + /// Preserve the completion FIN until the publisher acknowledges it. Lite-07 also reads + /// on to the publisher's FIN in reply, so dropping our receive half never refuses it. Finish { - writer: crate::coding::Writer, + /// `None` once the publisher acknowledged our FIN. + writer: Option>, + reader: Option>, /// The subscription keeps the session busy until then. _working: crate::session::Working, }, @@ -8784,8 +8757,10 @@ impl kio::Task for TrackServeRun { if active.stream.writer.finish().is_ok() && finished && self.serve.subscriber.version.waits_for_subscriber_fin() { + let keeps_open = self.serve.subscriber.version.keeps_subscribe_open(); self.state = TrackRunState::Finish { - writer: active.stream.writer, + writer: Some(active.stream.writer), + reader: keeps_open.then_some(active.stream.reader), _working: active._working, }; continue; @@ -8794,8 +8769,38 @@ impl kio::Task for TrackServeRun { return Poll::Ready(()); } - TrackRunState::Finish { writer, .. } => { - let _ = ready!(writer.poll_close(waiter)); + TrackRunState::Finish { writer, reader, .. } => { + // Responses after the tail settled change nothing: read past them, yielding + // to the session's other tasks while a peer streams them. + let mut budget = kio::coop::Budget::new(32); + while let Some(stream) = reader + && let Poll::Ready(res) = stream.poll_decode_maybe::(waiter) + { + match res { + Ok(Some(_)) => {} + // The publisher's FIN in reply, or a reset: the stream is done. + Ok(None) => { + *reader = None; + break; + } + // Malformed bytes are a protocol violation here as on any Subscribe + // Stream, which closes the session; a short read at a FIN is not. + Err(err) => { + let _ = lite::error::control(&mut self.serve.subscriber.session.clone(), err); + *reader = None; + break; + } + } + ready!(budget.poll_yield(waiter)); + } + if let Some(stream) = writer + && stream.poll_close(waiter).is_ready() + { + *writer = None; + } + if writer.is_some() || reader.is_some() { + return Poll::Pending; + } self.state = TrackRunState::Done; } TrackRunState::Done => return Poll::Ready(()), @@ -8984,20 +8989,13 @@ struct ServeLoop { /// The TRACK stream that answered, until a SUBSCRIBE has a response or nobody holds /// the copy. held: Option, - /// The ranges of a bounded lite-07 subscription the publisher covered, while the - /// demand still asks for just them: subscribing again would only resend them. - covered: Option>, + /// The tail of a lite-07 subscription whose publisher sent SUBSCRIBE_FINISH: once it + /// settles the copy finishes and FINs, while demand and responses keep flowing meanwhile. + finishing: Option<(u64, Settle)>, /// A pipelined subscription whose group streams stay parked until the responses that /// beat the accept are read, as they would have been without the wait: a /// SUBSCRIBE_START says where the feed starts before its first group does. parked: Option, - /// Our side of the covered subscription, finished and held until the publisher reads - /// the FIN, since dropping it would reset it. The subscription keeps the session busy - /// until then. - closing: Option<( - crate::coding::Writer, - crate::session::Working, - )>, } #[expect( @@ -9010,16 +9008,14 @@ enum ServeMode { /// Driving an upstream SUBSCRIBE open. The demand arms wait meanwhile, /// exactly like the old inline await. Establish(Establish), - /// The upstream FIN'd, but QUIC does not order streams: keep the subscription - /// routable until every owed group is accounted for. The grace bounds an unaccounted - /// sequence, such as a lost datagram. + /// The publisher FINned once the track ended, as before lite-07, but QUIC does not order + /// streams: keep the subscription routable until every owed group is accounted for. The + /// grace bounds an unaccounted sequence, such as a lost datagram. Lite-07 keeps serving + /// instead, settling its tail in [`ServeLoop::finishing`]. Tail { settle: Settle, - /// The groups owed, `None` when nothing says which, so only the grace settles. - owed: Option>>, - /// Whether the FIN ended the track, rather than a covered bounded lite-07 - /// subscription, after which the copy stays for its readers. - track: bool, + /// The demand's end when the tail started, which bounds the groups owed. + requested_end: Option, }, /// The upstream subscription failed with `err`. Every group stream whose header arrived /// before it, parked on the accept or not, is read to its end first, as a group ends on @@ -9276,8 +9272,7 @@ impl ServeLoop { mode: ServeMode::Select, linger: crate::time::Deadline::new(&serve.subscriber.runtime), held: None, - covered: None, - closing: None, + finishing: None, parked: None, } } @@ -9313,29 +9308,185 @@ impl ServeLoop { } } - /// Apply a demand change, opening a fresh upstream subscription after a covered one - /// unless the demand still asks for exactly what it covered. + /// Apply a demand change: open, update, or cancel the upstream subscription. fn begin(&mut self, serve: &TrackServe, pref: Option) -> Result<(), Error> { - if let Some(covered) = &self.covered { - let ranges = match &pref { - Some(pref) => wire_ranges(pref, self.serving.info().timescale)?, - None => None, - }; - if ranges.as_ref() == Some(covered) { - return Ok(()); - } - // What the copy cached may go stale before the next subscription says where - // the live feed is. - self.covered = None; - self.serving.set_idle(); - } match serve.begin_subscription(&mut self.serving, &mut self.sub, pref, self.supports_update)? { Begin::Establish(est) => self.mode = ServeMode::Establish(est), Begin::None => {} } + // A finished track's tail waits a fresh grace for what an update newly asked for, + // rather than a deadline that may have passed while an earlier stream was read. + if let Some((id, _)) = &self.finishing + && let Sub::Active(active) = &self.sub + && active.id == *id + { + self.finishing = Some((active.id, Self::settle(serve, &self.serving, active))); + } + Ok(()) + } + + /// Apply one response read from the upstream subscription. + fn apply_response(&mut self, serve: &TrackServe, msg: lite::SubscribeResponse) -> Result<(), Error> { + let Sub::Active(active) = &mut self.sub else { + return Ok(()); + }; + match &msg { + // SUBSCRIBE_FINISH declares the track's exclusive final + // sequence, which may arrive while trailing groups are + // still in flight. Record it on this segment's producer so + // consumers learn the boundary early; the tail then waits for + // the groups below it. + lite::SubscribeResponse::Finish(end) => { + if active.end.is_some() { + tracing::warn!(track = %serve.name, group = end.group, "duplicate subscribe finish"); + return Err(Error::ProtocolViolation); + } + // Lower groups may still be on the wire, behind a higher + // one, so the end holds readers at a hole until it fills or + // its grace passes. finish_at rejects a boundary at or below a group + // already received. lite-05 specified an inclusive end, + // and `@moq/net` 0.1.3 to 0.1.9 sent one, so there it + // only costs the early boundary: warn, and let the FIN + // finish the track. Later drafts made it exclusive, so + // the publisher contradicted its own end. + let error = end.error.clone().map(Error::from); + if let Err(err) = self.serving.finish_at(Position::group(end.group), error) { + match serve.subscriber.version { + Version::Lite05 => { + tracing::warn!(track = %serve.name, group = end.group, %err, "invalid subscribe end") + } + _ => { + tracing::warn!(track = %serve.name, group = end.group, %err, "subscribe end below a received group"); + return Err(Error::ProtocolViolation); + } + } + } + active.end = Some(end.clone()); + // Lite-07 keeps the stream open past it, so its tail starts now rather + // than at a FIN, while demand and responses keep flowing. + if serve.subscriber.version.keeps_subscribe_open() { + self.finishing = Some((active.id, Self::settle(serve, &self.serving, active))); + } + } + // SUBSCRIBE_START names where each range's delivery starts, + // the lowest being the first group this feed serves: the + // publisher skipped everything below it (e.g. it could not + // serve the requested frame). Record it as a drop signal, + // so a reader waiting on a skipped group fails over + // instead of stalling on a live route. + lite::SubscribeResponse::Start(start) => { + // One resolved start per range the SUBSCRIBE carried, unless an + // update the publisher read first replaced them. + if serve.subscriber.version.has_ranges() && !active.updated && start.starts.len() != active.ranges.len() + { + tracing::warn!(track = %serve.name, starts = start.starts.len(), ranges = active.ranges.len(), "subscribe start resolves other ranges"); + return Err(Error::ProtocolViolation); + } + // No range starting anywhere leaves nothing owed. + let served = start.starts.iter().flatten().min().copied(); + active.served = served; + active.starts = start.starts.clone(); + // Where the live feed is, on versions whose answer says. + if serve.subscriber.version.has_largest() { + // A bounded feed that never delivers Largest goes live at the + // answer alone; one that does waits for it to show. + match active.never_serves(start.largest) { + Some(largest) => self.serving.set_live_ahead(largest, served), + None => self.serving.set_live(start.largest, served), + } + if let Some(largest) = start.largest { + self.serving.report_largest(largest); + } + } else if let Some(entry) = serve.subscriber.subscribes.lock().get_mut(&active.id) { + entry.answered(served, None); + } + // A START describes the demand the SUBSCRIBE carried. + // It applies only while the current start still matches + // that demand (updates get no fresh START, so an update + // that moved the start makes it stale, and one that + // moved back restores it); elsewhere the + // request-tracked floor stands rather than a guess. + if active.start == active.requested { + let _ = match (serve.subscriber.version.resolves_start(), served) { + (true, Some(served)) => self.serving.start_resolved(served), + _ => self.serving.start_at(served), + }; + } + active.started = true; + // The groups the SUBSCRIBE asked for below it are not + // waited for, whatever the demand asks later. One that + // still arrives is delivered, so they are not dropped. + if let Some(requested) = active.requested + && let Some(served) = served + && let Ok(mut tail) = active.tail.write() + { + tail.account(requested..served, serve.subscriber.runtime.now()); + } + } + // The publisher will never send these groups, so they + // are accounted for without a stream, and a reader of + // one still arriving ends with the drop's error now. + lite::SubscribeResponse::Drop(dropped) => { + if let Ok(mut tail) = active.tail.write() { + tail.account( + dropped.start..dropped.end.saturating_add(1), + serve.subscriber.runtime.now(), + ); + } + self.serving.drop_groups( + dropped.start..dropped.end.saturating_add(1), + dropped.error(serve.subscriber.version), + ); + } + // OK just resolves the range (the producer already orders + // groups). + lite::SubscribeResponse::Ok(_) => { + tracing::debug!(track = %serve.name, ?msg, "subscribe response") + } + } Ok(()) } + /// Start waiting out `active`'s tail once the publisher ended the track. The effective + /// max delay is the stopgap grace: the wrong clock (it bounds presentation-time drift), + /// but it is how long the subscriber was willing to wait for a late group anyway. + fn settle(serve: &TrackServe, serving: &track::Producer, active: &SubStream) -> Settle { + let budget = serving.subscription().map(|sub| sub.budget()); + if let Ok(mut tail) = active.tail.write() { + // Lite-05+ declared the end with SUBSCRIBE_FINISH instead. + if !serve.subscriber.version.has_track_stream() { + tail.end(); + } + tail.set_grace(serving.tail_grace(budget.unwrap_or_default())); + tail.expire(serve.subscriber.runtime.now()); + } + Settle::new(&serve.subscriber.runtime, active.tail.consume()) + } + + /// Whether the tail of a lite-07 subscription the publisher finished has settled, so the + /// copy can FIN it: every owed group accounted for, or the grace passed. + fn poll_finishing(&mut self, waiter: &kio::Waiter) -> Poll<()> { + let (Some((id, settle)), Sub::Active(active)) = (&mut self.finishing, &self.sub) else { + return Poll::Pending; + }; + // The tail belongs to the subscription that received SUBSCRIBE_FINISH, not one that + // replaced it after the demand left and came back. + if *id != active.id { + self.finishing = None; + return Poll::Pending; + } + // Nothing says which groups are owed until SUBSCRIBE_START says where delivery starts, + // so until then only the grace settles. + let owed = match active.awaits_start() { + true => None, + false => active.owed(self.serving.subscription().and_then(|sub| sub.end())), + }; + settle.poll(waiter, |tail| { + owed.as_ref() + .is_some_and(|owed| owed.iter().all(|owed| tail.covers(owed.clone()))) + }) + } + fn poll(&mut self, serve: &TrackServe, waiter: &kio::Waiter) -> Poll { // Immutable TRACK_INFO stays usable locally after withdrawal. Its wire hold // is only a bridge to SUBSCRIBE while this source is still advertised. @@ -9388,46 +9539,26 @@ impl ServeLoop { }; return Poll::Ready(ServeEnd::GiveBack(err)); } - ServeMode::Tail { settle, owed, track } => { + ServeMode::Tail { settle, requested_end } => { // The tail waits for groups, parked ones included. if let Some(id) = self.parked.take() { serve.accept_pipelined(id, &self.serving); } + let owed = match &self.sub { + Sub::Active(active) => active.owed(*requested_end), + Sub::None => None, + }; let _ = self.fetches.poll(waiter); self.feed_fetches .poll(&self.sub, serve.subscriber.runtime.now(), waiter); - let mut whole = false; if settle .poll(waiter, |tail| { - whole = owed - .as_ref() - .is_some_and(|owed| owed.iter().all(|owed| tail.covers(owed.clone()))); - whole + owed.as_ref() + .is_some_and(|owed| owed.iter().all(|owed| tail.covers(owed.clone()))) }) .is_ready() { - if *track { - return Poll::Ready(ServeEnd::Finished); - } - // Only the subscription ended: close our side as the publisher waits - // for, and keep the copy and what it cached for the readers. - self.mode = ServeMode::Select; - if let Sub::Active(active) = std::mem::replace(&mut self.sub, Sub::None) { - serve.subscriber.remove_subscribe(active.id); - let mut writer = active.stream.writer; - if writer.finish().is_ok() { - self.closing = Some((writer, active._working)); - } - // Unchanged ranges are what the publisher covered, even if a datagram - // went missing. Changed ones settled by the grace instead may have - // been covered before an update it never read. - self.covered = (whole || !active.updated).then_some(active.ranges); - } - // Whatever the covered ranges leave out of the demand is asked again. - if let Err(err) = self.begin(serve, self.serving.subscription()) { - return Poll::Ready(ServeEnd::GiveBack(err)); - } - continue; + return Poll::Ready(ServeEnd::Finished); } if self.fetches.is_empty() && crate::util::poll_unwanted(&mut self.demand, waiter).is_ready() { return Poll::Ready(ServeEnd::Idle); @@ -9440,7 +9571,7 @@ impl ServeLoop { if poll_unparked(&self.sub, serve, waiter).is_pending() { return Poll::Pending; } - if *track && !serve.subscriber.version.has_track_stream() { + if !serve.subscriber.version.has_track_stream() { let _ = self.serving.finish(); } return Poll::Ready(ServeEnd::GiveBack(crate::lite::error::from_transport(err))); @@ -9448,12 +9579,6 @@ impl ServeLoop { return Poll::Pending; } ServeMode::Select => { - if let Some((writer, _)) = &mut self.closing - && writer.poll_close(waiter).is_ready() - { - self.closing = None; - } - // A SUBSCRIBE still unanswered once another instance takes the path may have // resolved it: refuse it. A withdrawal alone leaves it to its answer. Checked // before reading the stream, so an answer that only arrived alongside the @@ -9548,6 +9673,11 @@ impl ServeLoop { self.feed_fetches .poll(&self.sub, serve.subscriber.runtime.now(), waiter); + // A lite-07 track the publisher finished ends once its tail settles. + if self.poll_finishing(waiter).is_ready() { + return Poll::Ready(ServeEnd::Finished); + } + // (3) Nobody holds this copy anymore. Its upstream subscription already // went with the last subscriber, so it lingers, cache and all, for a // reader or fetch that asks again soon, then is dropped. In-flight @@ -9581,141 +9711,18 @@ impl ServeLoop { active.unanswered = None; match res { Ok(Some(msg)) => { - match &msg { - // SUBSCRIBE_END declares the track's exclusive final - // sequence, which may arrive while trailing groups are - // still in flight. Record it on this segment's producer so - // consumers learn the boundary early; the later stream FIN - // then finds the track already finished. - lite::SubscribeResponse::End(end) => { - // Lower groups may still be on the wire, behind a higher - // one, so the end holds readers at a hole until it fills or - // its grace passes. finish_at rejects a boundary at or below a group - // already received. lite-05 specified an inclusive end, - // and `@moq/net` 0.1.3 to 0.1.9 sent one, so there it - // only costs the early boundary: warn, and let the FIN - // finish the track. Later drafts made it exclusive, so - // the publisher contradicted its own end. - let error = end.error.clone().map(Error::from); - if let Err(err) = self.serving.finish_at(Position::group(end.group), error) { - match serve.subscriber.version { - Version::Lite05 => { - tracing::warn!(track = %serve.name, group = end.group, %err, "invalid subscribe end") - } - _ => { - tracing::warn!(track = %serve.name, group = end.group, %err, "subscribe end below a received group"); - return Poll::Ready(ServeEnd::GiveBack(Error::ProtocolViolation)); - } - } - } - active.end = Some(end.clone()); - } - // SUBSCRIBE_START names where each range's delivery starts, - // the lowest being the first group this feed serves: the - // publisher skipped everything below it (e.g. it could not - // serve the requested frame). Record it as a drop signal, - // so a reader waiting on a skipped group fails over - // instead of stalling on a live route. - lite::SubscribeResponse::Start(start) => { - // One resolved start per range the SUBSCRIBE carried, unless an - // update the publisher read first replaced them. - if serve.subscriber.version.has_ranges() - && !active.updated && start.starts.len() != active.ranges.len() - { - tracing::warn!(track = %serve.name, starts = start.starts.len(), ranges = active.ranges.len(), "subscribe start resolves other ranges"); - return Poll::Ready(ServeEnd::GiveBack(Error::ProtocolViolation)); - } - // No range starting anywhere leaves nothing owed. - let served = start.starts.iter().flatten().min().copied(); - active.served = served; - active.starts = start.starts.clone(); - // Where the live feed is, on versions whose answer says. - if serve.subscriber.version.has_largest() { - // A bounded feed that never delivers Largest goes live at the - // answer alone; one that does waits for it to show. - match active.never_serves(start.largest) { - Some(largest) => self.serving.set_live_ahead(largest, served), - None => self.serving.set_live(start.largest, served), - } - if let Some(largest) = start.largest { - self.serving.report_largest(largest); - } - } else if let Some(entry) = - serve.subscriber.subscribes.lock().get_mut(&active.id) - { - entry.answered(served, None); - } - // A START describes the demand the SUBSCRIBE carried. - // It applies only while the current start still matches - // that demand (updates get no fresh START, so an update - // that moved the start makes it stale, and one that - // moved back restores it); elsewhere the - // request-tracked floor stands rather than a guess. - if active.start == active.requested { - let _ = match (serve.subscriber.version.resolves_start(), served) { - (true, Some(served)) => self.serving.start_resolved(served), - _ => self.serving.start_at(served), - }; - } - // The groups the SUBSCRIBE asked for below it are not - // waited for, whatever the demand asks later. One that - // still arrives is delivered, so they are not dropped. - if let Some(requested) = active.requested - && let Some(served) = served && let Ok(mut tail) = active.tail.write() - { - tail.account(requested..served, serve.subscriber.runtime.now()); - } - } - // The publisher will never send these groups, so they - // are accounted for without a stream, and a reader of - // one still arriving ends with the drop's error now. - lite::SubscribeResponse::Drop(dropped) => { - if let Ok(mut tail) = active.tail.write() { - tail.account( - dropped.start..dropped.end.saturating_add(1), - serve.subscriber.runtime.now(), - ); - } - self.serving.drop_groups( - dropped.start..dropped.end.saturating_add(1), - dropped.error(serve.subscriber.version), - ); - } - // OK just resolves the range (the producer already orders - // groups). - lite::SubscribeResponse::Ok(_) => { - tracing::debug!(track = %serve.name, ?msg, "subscribe response") - } + if let Err(err) = self.apply_response(serve, msg) { + return Poll::Ready(ServeEnd::GiveBack(err)); } // Each pass re-polls the whole loop, so a peer streaming // responses must not hold the session's other tasks. ready!(budget.poll_yield(waiter)); continue; } - // Lite-07 FINs a covered bounded subscription without SUBSCRIBE_END, - // which ends the subscription but not the track. The demand may have - // widened it since, in an update the publisher never read. - Ok(None) - if serve.subscriber.version.has_ranges() && active.end.is_none() && active.bounded => - { - tracing::debug!(broadcast = %serve.subscriber.log_path(&serve.path), track = %serve.name, "subscribe covered"); - let budget = self.serving.subscription().map(|sub| sub.budget()); - if let Ok(mut tail) = active.tail.write() { - // Lite-05+ declared the end with SUBSCRIBE_END instead. - if !serve.subscriber.version.has_track_stream() { - tail.end(); - } - tail.set_grace(self.serving.tail_grace(budget.unwrap_or_default())); - tail.expire(serve.subscriber.runtime.now()); - } - self.mode = ServeMode::Tail { - settle: Settle::new(&serve.subscriber.runtime, active.tail.consume()), - owed: active.covered(), - track: false, - }; - continue; - } Ok(None) => { + // The publisher ends the subscription only once it declared the + // track's end: with SUBSCRIBE_FINISH from lite-05, which lite-07 + // reads into the tail without a FIN. if serve.subscriber.version.has_track_stream() && active.end.is_none() { return Poll::Ready(ServeEnd::GiveBack(Error::ProtocolViolation)); } @@ -9723,29 +9730,12 @@ impl ServeLoop { // Upstream FIN'd the subscription: the publisher only FINs // once the track's final sequence is known and every group // stream finished, so the logical track is over for good - // (before lite-07, bounded demand alone never FINs; the - // publisher parks, since a cap can be raised). Those streams - // can still be in flight, so wait for the tail before finishing. - let subscription = self.serving.subscription(); - let requested_end = subscription.as_ref().and_then(|sub| sub.end()); - // The effective max delay is the stopgap grace: the wrong clock - // (it bounds presentation-time drift), but it is how long the - // subscriber was willing to wait for a late group anyway. - if let Ok(mut tail) = active.tail.write() { - // Lite-05+ declared the end with SUBSCRIBE_END instead. - if !serve.subscriber.version.has_track_stream() { - tail.end(); - } - tail.set_grace( - self.serving - .tail_grace(subscription.map(|sub| sub.budget()).unwrap_or_default()), - ); - tail.expire(serve.subscriber.runtime.now()); - } + // (bounded demand alone never FINs; the publisher parks, since + // a cap can be raised). Those streams can still be in flight, + // so wait for the tail before finishing. self.mode = ServeMode::Tail { - settle: Settle::new(&serve.subscriber.runtime, active.tail.consume()), - owed: active.owed(requested_end).map(|owed| vec![owed]), - track: true, + settle: Self::settle(serve, &self.serving, active), + requested_end: self.serving.subscription().and_then(|sub| sub.end()), }; continue; } diff --git a/rs/moq-net/src/lite/version.rs b/rs/moq-net/src/lite/version.rs index 4ee7c9687c6..05c9d92a788 100644 --- a/rs/moq-net/src/lite/version.rs +++ b/rs/moq-net/src/lite/version.rs @@ -235,12 +235,20 @@ impl Version { } /// Whether subscription completion waits for the subscriber's FIN, which it sends once - /// its tail accounting settles. SUBSCRIBE_END says which groups are owed, so that is + /// its tail accounting settles. SUBSCRIBE_FINISH says which groups are owed, so that is /// prompt from lite-05; before it, only the subscriber's grace settles the tail. pub(crate) fn waits_for_subscriber_fin(self) -> bool { self.has_track_stream() } + /// Whether the Subscribe Stream stays open past SUBSCRIBE_FINISH and past a covered + /// bounded subscription, so a SUBSCRIBE_UPDATE can still reach the track's history. + /// The subscriber FINs once it wants nothing more, and the publisher FINs in reply. + /// Added in lite-07; before it the publisher FINs once the track ends. + pub(crate) fn keeps_subscribe_open(self) -> bool { + self.has_ranges() + } + /// Whether ANNOUNCE_REQUEST carries the hidden opt-in. Added in lite-07; older /// requests decode as not opted in. pub fn has_hidden(self) -> bool { @@ -289,7 +297,7 @@ impl Version { ) } - /// Whether SUBSCRIBE_END carries the error the track ended with, so an errored end + /// Whether SUBSCRIBE_FINISH carries the error the track ended with, so an errored end /// keeps its history across a hop. Added in lite-07; earlier versions can only reset. pub(crate) fn has_end_error(self) -> bool { !matches!( diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index f87ab6dfa06..378a187e353 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -13013,7 +13013,7 @@ mod tests { /// The publisher finishes a track, then its broadcast. A subscription already in /// flight must conclude normally: the track's last group, then the end. moq-lite, /// ANNOUNCE_END: "Retraction does not disturb subscriptions already in flight, - /// which conclude normally with SUBSCRIBE_END." + /// which conclude normally with SUBSCRIBE_FINISH." /// /// The runtime is single-threaded and the publisher's whole ending has no await in /// it, so the outcome does not depend on timing. @@ -13059,7 +13059,7 @@ mod tests { /// last group and end are still on their way. The subscription already in flight /// must still conclude normally. moq-lite, ANNOUNCE_END: "Retraction does not /// disturb subscriptions already in flight, which conclude normally with - /// SUBSCRIBE_END." + /// SUBSCRIBE_FINISH." #[kio::test] #[cfg(feature = "async")] async fn a_retracted_route_concludes_in_flight_subscriptions() { diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 10e31cc604c..112d10ce6d6 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -880,6 +880,12 @@ impl Subscriber { self.reader().window().runs(from).next().map(|run| *run.start()) } + /// The groups the reader's ranges name from `from` on; see + /// [`track::Subscriber::runs_in_range`]. + pub(super) fn runs(&self, from: u64) -> Vec> { + self.reader().window().runs(from).collect() + } + /// Whether `accounted` holds every group the reader's bounded ranges name from `from` /// up to `cap`; see [`track::Subscriber::covered`]. pub(super) fn covered(&self, from: u64, cap: Option, accounted: &impl Accounted) -> bool { diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index d0db688ce30..4f848bb7986 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -8837,6 +8837,27 @@ impl Subscriber { } } + /// The groups from `from` up to the exclusive `end` that this subscriber's ranges and + /// [`Self::with_range`] bounds name, lowest first: the groups it is owed, without the + /// ones between disjoint ranges. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn runs_in_range(&mut self, from: u64, end: u64) -> Vec> { + let bounds = self.bounds(); + let from = from.max(bounds.start); + let end = bounds.end.map_or(end, |cap| cap.min(end)); + let runs = match &mut self.inner { + Inner::Plain(cursor) => { + cursor.sync(); + cursor.filter.window.runs(from).collect::>() + } + Inner::Resume(resume, _) => resume.runs(from), + }; + runs.into_iter() + .take_while(|run| *run.start() < end) + .map(|run| *run.start()..run.end().saturating_add(1).min(end)) + .collect() + } + /// Whether `accounted` holds every group from `from` on that this subscriber's ranges /// and [`Self::with_range`] bounds name, once every range has a resolved end: `false` /// while one stays open, counts back from a group not seen yet, or ends at a time not diff --git a/rs/moq-net/tests/integration/errored_end.rs b/rs/moq-net/tests/integration/errored_end.rs index a8f8fc5caf2..1f80e822cc3 100644 --- a/rs/moq-net/tests/integration/errored_end.rs +++ b/rs/moq-net/tests/integration/errored_end.rs @@ -131,7 +131,7 @@ async fn an_errored_end_keeps_its_history_across_a_hop() { false => assert!(err.is_some(), "{label} {what}: ended cleanly"), } } - // Lite-07 ends the subscription after SUBSCRIBE_END rather than resetting it. + // Lite-07 ends the subscription after SUBSCRIBE_FINISH rather than resetting it. let reset = session.server_transport.resets().contains(&(64 + 7)); assert_eq!( reset, diff --git a/rs/moq-net/tests/integration/finished_broadcast_mock.rs b/rs/moq-net/tests/integration/finished_broadcast_mock.rs index 7b963360410..2c80d58a072 100644 --- a/rs/moq-net/tests/integration/finished_broadcast_mock.rs +++ b/rs/moq-net/tests/integration/finished_broadcast_mock.rs @@ -7,7 +7,7 @@ //! The track is complete before the broadcast ends, and the session stays open, so //! the subscription must conclude normally. moq-lite, ANNOUNCE_END: "Retraction does //! not disturb subscriptions already in flight, which conclude normally with -//! SUBSCRIBE_END." Instead the track is aborted when the broadcast ends, and +//! SUBSCRIBE_FINISH." Instead the track is aborted when the broadcast ends, and //! whatever the session had not sent yet is gone. //! //! Deterministic: time is paused, the runtime is single-threaded, and the publisher diff --git a/rs/moq-net/tests/integration/range_read.rs b/rs/moq-net/tests/integration/range_read.rs index c1253847722..b2c8682be50 100644 --- a/rs/moq-net/tests/integration/range_read.rs +++ b/rs/moq-net/tests/integration/range_read.rs @@ -160,3 +160,44 @@ async fn a_range_across_the_live_edge_reads_up_to_its_end() { } assert!(failures.is_empty(), "failed: {failures:?}"); } + +async fn back_to_back(version: Version, relay: bool, groups: &[u64], hold: bool) { + let (remote, _track, _held) = fixture(version, relay).await; + let track = remote.track("video").unwrap(); + let mut readers = Vec::new(); + for &group in groups { + let mut sub = track.subscribe(one_range(group..group + 1)).await.unwrap(); + assert_eq!( + read(&mut sub).await, + [group], + "{version} relay={relay} groups={groups:?} hold={hold}" + ); + if hold { + readers.push(sub); + } + } +} + +/// One-range reads of one track through one session complete back to back, ascending or +/// descending, whether each earlier reader is still held or gone. On lite-07 the second +/// read widens the open subscription with an update, which the publisher reads, since a +/// covered subscription stays open. +#[kio::test] +async fn one_range_reads_back_to_back_each_complete() { + let mut failures = Vec::new(); + for version in support::every_version() { + for relay in [false, true] { + for groups in [&[1, 2][..], &[2, 1, 0][..]] { + for hold in [false, true] { + if kio::clock::sim::spawn(back_to_back(version, relay, groups, hold)) + .await + .is_err() + { + failures.push((version, relay, groups, hold)); + } + } + } + } + } + assert!(failures.is_empty(), "failed: {failures:?}"); +} diff --git a/rs/moq-net/tests/integration/session_close.rs b/rs/moq-net/tests/integration/session_close.rs index 087dbe604b2..21d677f6ef7 100644 --- a/rs/moq-net/tests/integration/session_close.rs +++ b/rs/moq-net/tests/integration/session_close.rs @@ -262,7 +262,7 @@ async fn close_times_out_waiting_for_announcement_acknowledgements() { } } -/// The versions whose SUBSCRIBE_END lets a subscriber settle its tail, and FIN, at once. +/// The versions whose SUBSCRIBE_FINISH lets a subscriber settle its tail, and FIN, at once. const SETTLED_TAIL: [&str; 3] = ["moq-lite-05", "moq-lite-06", "moq-lite-07-wip"]; /// A transport ACK is not proof the subscriber read the final group: a receiver may drop diff --git a/rs/moq-net/tests/integration/subscribe_ranges.rs b/rs/moq-net/tests/integration/subscribe_ranges.rs index ed5b2d73e9c..119e94be3f2 100644 --- a/rs/moq-net/tests/integration/subscribe_ranges.rs +++ b/rs/moq-net/tests/integration/subscribe_ranges.rs @@ -736,13 +736,13 @@ async fn unresolved_time_reader_never_changes_sibling_demand() { } } -/// A lite-07 publisher ends a bounded subscription once every group in range is -/// delivered or dropped, releasing its demand, and the subscriber keeps the copy: the -/// track does not end, the same range reads again from the cache, and a wider one -/// subscribes again for the rest, past a group that never existed. Older versions park -/// instead, since their subscriber reads a FIN as the track's end. +/// A covered bounded subscription stays open on every version: a lite-07 publisher keeps +/// the stream open for an update rather than FIN it, and older ones park, since their +/// subscriber reads a FIN as the track's end. The track does not end, the same range reads +/// again from the cache, and a wider one reaches upstream for the rest, past a group that +/// never existed. The demand goes once the readers do. #[kio::test] -async fn a_covered_bounded_subscription_ends_on_lite07() { +async fn a_covered_bounded_subscription_stays_open() { for version in LITE { let publisher = produce_origin(1); let broadcast = publisher.create_broadcast("bcast").unwrap(); @@ -776,34 +776,30 @@ async fn a_covered_bounded_subscription_ends_on_lite07() { assert_eq!(next(&mut reader).await, Some(2), "{version}"); let released = kio::clock::timeout(TIMEOUT, crate::support::unwanted(track.demand())).await; - if *version != "moq-lite-07-wip" { - assert!(released.is_err(), "{version}: the publisher parks"); - drop((reader, track, session, subscriber, broadcast, publisher)); - continue; - } - released.expect("the covered subscription ends").unwrap(); + assert!(released.is_err(), "{version}: the covered subscription stays open"); assert!( kio::clock::timeout(TIMEOUT, reader.finished()).await.is_err(), - "the track goes on" + "{version}: the track goes on" ); let mut again = subscribe(&remote, bounded(1..3)).await; - assert_eq!(next(&mut again).await, Some(1)); - assert_eq!(next(&mut again).await, Some(2)); - assert!(track.subscription().is_none(), "the cache answers the same range"); + assert_eq!(next(&mut again).await, Some(1), "{version}"); + assert_eq!(next(&mut again).await, Some(2), "{version}"); let mut wider = subscribe(&remote, bounded(1..6)).await; let mut got = Vec::new(); for _ in 0..4 { got.extend(next(&mut wider).await); } - assert_eq!(got, [1, 2, 4, 5]); - // Group 3 may still arrive on a live track, so it is dropped once the feed gives up - // on it, past the arrivals grace. + // Before lite-07 the wire carries no order, so the upstream serves newest first. + got.sort(); + assert_eq!(got, [1, 2, 4, 5], "{version}"); + + drop((reader, again, wider)); kio::clock::timeout(Duration::from_secs(60), crate::support::unwanted(track.demand())) .await - .expect("the wider subscription ends past the missing group") + .unwrap_or_else(|_| panic!("{version}: the demand leaves with the readers")) .unwrap(); - drop((reader, again, wider, track, session, subscriber, broadcast, publisher)); + drop((track, session, subscriber, broadcast, publisher)); } } diff --git a/rs/moq-net/tests/integration/subscription_end_integrity.rs b/rs/moq-net/tests/integration/subscription_end_integrity.rs index dcf562285b4..efc605ff202 100644 --- a/rs/moq-net/tests/integration/subscription_end_integrity.rs +++ b/rs/moq-net/tests/integration/subscription_end_integrity.rs @@ -10,7 +10,7 @@ //! `Ok(None)` only after that FIN; any other ending is an error. //! //! The publisher opens the final group and writes its first frame (the group stream -//! opens), finishes the track (SUBSCRIBE_END goes out), then writes the rest, finishes +//! opens), finishes the track (SUBSCRIBE_FINISH goes out), then writes the rest, finishes //! the group and disconnects with no await in between, so the final group's tail and //! its FIN never leave. The close lands before the subscriber reads on. The subscriber //! must then either see every frame or end with an error; it must not end `Ok(None)` @@ -125,7 +125,7 @@ async fn round(drop_session: bool) -> (Vec>, Option) { .expect("reader gone"); // The final group opens and its first frame is sent, then the track's end is - // declared: SUBSCRIBE_END reaches the subscriber with the group still open, as it + // declared: SUBSCRIBE_FINISH reaches the subscriber with the group still open, as it // does whenever a finish races a group still in flight. let mut tail = track.append_group().unwrap(); tail.write_frame(Timestamp::ZERO, TAIL[0]).unwrap(); diff --git a/rs/moq-net/tests/integration/track_tail.rs b/rs/moq-net/tests/integration/track_tail.rs index 1cc939befb3..d1bb4b2b7d9 100644 --- a/rs/moq-net/tests/integration/track_tail.rs +++ b/rs/moq-net/tests/integration/track_tail.rs @@ -29,7 +29,7 @@ const PAYLOAD: &[u8] = b"frame"; /// How long a subscriber waits for a group it cannot account for, with no max delay set. const GRACE: Duration = Duration::from_secs(1); -/// moq-lite drafts with and without SUBSCRIBE_END, and IETF drafts with a joining +/// moq-lite drafts with and without SUBSCRIBE_FINISH, and IETF drafts with a joining /// FETCH (18) and with subscription fills (22). const VERSIONS: &[&str] = &[ "moq-lite-03",