diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 1a412abba2f..20a2582dec3 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -259,7 +259,10 @@ untimed. A relay cancels its upstream subscription once nobody subscribes, but keeps its copy of the track for 30 seconds after the last reader leaves, so a returning reader or the next fetch finds it. That copy is not live meanwhile: readers get -nothing from it until the source answers again. +nothing from it until the source answers again, and its latest group is unknown +until then too, so a player starting at the latest waits for the answer instead +of starting from what the ended feed left. A reader's latest group stays within +the groups it limits itself to, and ends once the track does. The publisher may declare a retention window per track. Omission sets no limit; zero keeps only the live edge. Relays preserve that declaration, and may still diff --git a/js/net/MIGRATE.md b/js/net/MIGRATE.md index fbcb210ee2d..4d7076305f6 100644 --- a/js/net/MIGRATE.md +++ b/js/net/MIGRATE.md @@ -4,6 +4,14 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `largest()` reports within a reader's `setGroups` bounds + +`Track.Subscriber.largest()` and `Track.Ordered.largest()` report the track's largest only while it is within the bounds set with `setGroups` or `withGroups`, where they used to ignore the bounds. +A largest past the cap or below the floor reports nothing; the cache never stands in for it. +A cap moved below the largest stays pending. Moving the bounds wakes a pending read, which reports the largest once it is within them and newer than the last report. +Unbounded readers and `Track.Producer.largest()` are unchanged. +To read the track's edge from a capped reader, lift the cap with `setGroups({})` first. + ### Lite-07 accounts for groups with `SUBSCRIBE_DROP` Construct `SubscribeEnd(group)` without a stream count; remove reads of `.streams` and calls to `hasStreamCount()`. diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 1b7f06d4800..bf7ae262fe7 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -304,7 +304,10 @@ export interface ResumeSource { bufferedLargest(): Location | undefined; /** The serving copy's largest location now, and whether it can still advance. */ peekLargest(): { largest?: Location; ended: boolean }; - /** The serving copy's largest location once newer than `seen`, or undefined once the logical track ends. */ + /** + * The serving copy's largest location once within the reader's cursor and newer than `seen`, + * or undefined once the logical track ends. + */ largest(seen: Location | undefined): Promise; /** An independent subscriber to the serving copy. */ fork(options?: Subscription): Subscriber; diff --git a/js/net/src/resume.test.ts b/js/net/src/resume.test.ts index d09f886cf0c..6dd1be4a362 100644 --- a/js/net/src/resume.test.ts +++ b/js/net/src/resume.test.ts @@ -20,6 +20,17 @@ async function settle() { await new Promise((resolve) => setTimeout(resolve, 0)); } +/** Whether `promise` is still pending once queued work settles. */ +async function isPending(promise: Promise): Promise { + let settled = false; + void promise.then( + () => (settled = true), + () => (settled = true), + ); + await settle(); + return !settled; +} + /** A stand-in session announcing `path` under the epoch at `cost`, serving `source`. */ function serve(origin: Producer, source: BroadcastProducer, cost: bigint) { const handle = wireOf(origin).receive(path, Route.normalize({ epoch: EPOCH, cost })); @@ -1151,6 +1162,36 @@ test("a fetch still out on the old route moves to the new one", async () => { b.close(); }); +// A front's reader reports the serving copy's largest only within its bounds: the cap it sets +// and the floor its subscription names. +test("the largest honours the reader's bounds on the serving copy", async () => { + const front = new Front(); + const a = new BroadcastProducer(); + const ta = copy(a); + front.serve(a.consume()); + const capped = front.consumer + .track("seq") + .subscribe() + .withGroups({ end: { excluded: 3 } }); + const floored = front.consumer.track("seq").subscribe({ groups: { start: { included: 8 } } }); + for (let sequence = 0; sequence < 5; sequence++) ta.write(sequence); + + const next = capped.largest(); + expect(await isPending(next)).toBe(true); + capped.setGroups({}); + expect(await next).toEqual({ group: 4, frame: 0 }); + + const floor = floored.largest(); + expect(await isPending(floor)).toBe(true); + ta.write(8); + expect(await floor).toEqual({ group: 8, frame: 0 }); + + capped.close(); + floored.close(); + front.consumer.close(); + a.close(); +}); + test("the largest follows the serving copy across a swap without moving back, and ends with the route", async () => { const front = new Front(); const a = new BroadcastProducer(); diff --git a/js/net/src/resume.ts b/js/net/src/resume.ts index 2fc0b0fc443..97b9ffb386f 100644 --- a/js/net/src/resume.ts +++ b/js/net/src/resume.ts @@ -651,9 +651,12 @@ class Reader implements ResumeSource { const serving = this.#serving(); const view = serving && hooks.peekLargest(serving.sub); const largest = view?.largest; + const { start, end } = this.#shell.cursor.peek(); // A replacement copy can start below what an earlier one reported. if ( largest && + largest.group >= start && + (end === undefined || largest.group < end) && (!seen || largest.group > seen.group || (largest.group === seen.group && largest.frame > seen.frame)) ) return largest; @@ -666,7 +669,7 @@ class Reader implements ResumeSource { for (const dispose of disposes) dispose(); resolve(); }; - disposes.push(this.changed(wake), this.#shell.closed.changed(wake)); + disposes.push(this.changed(wake), this.#shell.closed.changed(wake), this.#shell.cursor.changed(wake)); if (serving) disposes.push(hooks.largestChanged(serving.sub, wake)); }); } diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 365dfba6664..e5929cce134 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -2027,6 +2027,92 @@ test("a datagram raises the largest", async () => { producer.close(); }); +/** Write untimed group `sequence` with `frames` frames, left open. */ +function framedGroup(producer: TrackProducer, sequence: number, frames: number): GroupProducer { + const group = new GroupProducer(sequence); + producer.writeGroup(group); + for (let i = 0; i < frames; i++) group.writeFrame({ payload: enc.encode("x") }); + return group; +} + +/** Whether `promise` is still pending once queued work settles. */ +async function isPending(promise: Promise): Promise { + let settled = false; + void promise.then( + () => (settled = true), + () => (settled = true), + ); + await settle(); + return !settled; +} + +// The largest reports only while it is within the bounds: a cap below it or a floor above it +// reports nothing, rather than a cached group standing in for it. +test("largest reports only within the bounds", async () => { + const producer = new TrackProducer("test").accept({}); + for (let sequence = 0; sequence < 6; sequence++) framedGroup(producer, sequence, 2); + + const capped = producer.subscribe().withGroups({ end: { excluded: 3 } }); + expect(await isPending(capped.largest())).toBe(true); + + // An inclusive cap reports the group it names. + const ordered = producer + .subscribe() + .ordered() + .withGroups({ end: { included: 5 } }); + expect(await ordered.largest()).toEqual({ group: 5, frame: 1 }); + + const floored = producer.subscribe().withGroups({ start: { included: 8 } }); + const next = floored.largest(); + expect(await isPending(next)).toBe(true); + framedGroup(producer, 8, 0); + expect(await next).toEqual({ group: 8, frame: 0 }); + producer.close(); +}); + +// Moving the bounds wakes a pending read, which reports the largest once it is within them; +// bounds that leave it where it was report nothing new. +test("largest follows the bounds as they move", async () => { + const producer = new TrackProducer("test").accept({}); + for (let sequence = 0; sequence < 6; sequence++) framedGroup(producer, sequence, 1); + + const reader = producer + .subscribe() + .ordered() + .withGroups({ end: { excluded: 3 } }); + let next = reader.largest(); + expect(await isPending(next)).toBe(true); + reader.setGroups({}); + expect(await next).toEqual({ group: 5, frame: 0 }); + + next = reader.largest(); + expect(await isPending(next)).toBe(true); + reader.setGroups({ end: { excluded: 2 } }); + expect(await isPending(next)).toBe(true); + + // Lifting it again leaves the last report standing. + reader.setGroups({}); + expect(await isPending(next)).toBe(true); + framedGroup(producer, 6, 0); + expect(await next).toEqual({ group: 6, frame: 0 }); + reader.setGroups({ start: { included: 3 } }); + expect(await isPending(reader.largest())).toBe(true); + producer.close(); +}); + +// Once the stream ended, widening the bounds over a largest it never reported does not bring +// it back. +test("largest stays ended when the bounds widen", async () => { + const producer = new TrackProducer("test").accept({}); + framedGroup(producer, 5, 1).close(); + producer.close(); + + const reader = producer.subscribe().withGroups({ end: { excluded: 3 } }); + expect(await reader.largest()).toBeUndefined(); + reader.setGroups({}); + expect(await reader.largest()).toBeUndefined(); +}); + // A reader that starts at the largest group while it is still empty is at the live edge: it // gets that group, which ends with nothing, then the next one. test("a live-edge reader starts on an empty largest group", async () => { @@ -2682,8 +2768,10 @@ test("a mid-group end survives eviction and a filtered late reader", async () => group.close(); producer.close(); await clock.advance(10_000 + CACHE_WINDOW_MS + 1); + expect(await producer.subscribe().largest()).toEqual({ group: 5, frame: 4 }); + // The cap excludes the only group, so nothing is within it. const reader = producer.subscribe().withGroups({ end: { excluded: 5 } }); - expect(await reader.largest()).toEqual({ group: 5, frame: 4 }); + expect(await reader.largest()).toBeUndefined(); expect(reader.end()).toEqual({ group: 5, frame: 5 }); reader.close(); } finally { diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 632df7bcfd4..c91229a6601 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -494,6 +494,11 @@ async function resolveInfo(state: TrackState): Promise { } } +// Whether `a` is past `b`. +function after(a: Location, b: Location): boolean { + return a.group > b.group || (a.group === b.group && a.frame > b.frame); +} + async function nextLargest( state: TrackState, seen: Location | undefined, @@ -501,10 +506,7 @@ async function nextLargest( ): Promise { for (;;) { const largest = state.largest.peek(); - if ( - largest && - (!seen || largest.group > seen.group || (largest.group === seen.group && largest.frame > seen.frame)) - ) { + if (largest && (!seen || after(largest, seen))) { commit(largest); return largest; } @@ -1100,12 +1102,7 @@ export class Producer { #advanceLargest(largest: Location | undefined): void { if (!largest) return; const current = this.#state.largest.peek(); - if ( - current && - (current.group > largest.group || (current.group === largest.group && current.frame >= largest.frame)) - ) { - return; - } + if (current && !after(largest, current)) return; this.#largestOpen = true; this.#state.largest.set(largest); for (const sink of this.#sinks) sink.largest.set(largest); @@ -1594,7 +1591,10 @@ export class Subscriber { #delivered = new Delivered(); // Whether a resumed reader has read groups, so a datagram read keeps copies they still need. #groupsRead = false; + // What {@link largest} last reported, and whether it ended: moving the bounds afterwards + // does not bring it back. #largestSeen?: Location; + #largestDone = false; #nextSequence = 0; #cursor = new Signal<{ start: number; end?: number }>({ start: 0 }); // A FETCH's own budget in place of the subscription's; unlike SUBSCRIBE's zero @@ -1951,18 +1951,39 @@ export class Subscriber { } /** - * The track's largest location, its newest group or datagram at the newest frame (0 while - * empty): the current one at first, then each newer one (skipping any in between), and - * `undefined` once the track ends. Seeded by the publisher's report, so it can name a frame - * before its group arrives. + * The largest location within {@link setGroups}: the current one at first, then each newer + * one (skipping any in between), and `undefined` once the track ends. + * + * The largest is the track's newest group or datagram at the newest frame (0 while empty), as + * the publisher reported it and its feed delivered since, so it can name a frame before its + * group arrives. A largest past the cap or below the floor reports nothing: the cache does not + * stand in for it, so a cap moved below the largest stays pending. Moving the bounds wakes a + * pending read, which reports the largest once it is within them and newer than the last report. */ async largest(): Promise { + if (this.#largestDone) return undefined; if (this.#resume) { const largest = await this.#resume.largest(this.#largestSeen); if (largest) this.#largestSeen = largest; + else this.#largestDone = true; return largest; } - return nextLargest(this.#state, this.#largestSeen, (largest) => (this.#largestSeen = largest)); + for (;;) { + const { start, end } = this.#cursor.peek(); + const largest = this.#state.largest.peek(); + const within = + largest && largest.group >= start && (end === undefined || largest.group < end) ? largest : undefined; + const seen = this.#largestSeen; + if (within && (!seen || after(within, seen))) { + this.#largestSeen = within; + return within; + } + if (this.#state.closed.peek() !== undefined || this.#state.largestEnded.peek()) { + this.#largestDone = true; + return undefined; + } + await Signal.race(this.#state.largest, this.#state.closed, this.#state.largestEnded, this.#cursor); + } } // The newest frame buffered for this subscriber, which a wire answer names as its edge: @@ -2014,6 +2035,7 @@ export class Subscriber { * Limit subsequent reads without rewinding read progress or changing the wire request. * An omitted start preserves the current floor; an omitted end removes the cap. * Raising the end makes unread buffered groups available again. + * {@link largest} reports within these bounds. */ setGroups(groups: Groups): void { const { start, end } = groupBounds(groups); @@ -2526,7 +2548,7 @@ export class Ordered { return this.#subscriber.subscription; } - /** The track's largest location; see {@link Subscriber.largest}. */ + /** The largest location within {@link setGroups}; see {@link Subscriber.largest}. */ largest(): Promise { return this.#subscriber.largest(); } diff --git a/quest/m1/README.md b/quest/m1/README.md index 192327a9510..9e072b805f8 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -122,6 +122,7 @@ lands before the cut. - [Gateway linger](/quest/m1/linger-gateways.md) - HLS, RTMP, SRT and RTC imports linger at a clean end, cut short by a newer epoch at the same path - [Exporters follow a rendition to a new track](/quest/m1/export-rendition-rebind.md) - the fMP4, TS, MKV and FLV exporters rebuild a rendition whose resolved `track` or `broadcast` changes under the same ID - [lite-06 max sequence](/quest/m1/lite06-max-sequence.md) - group sequence 2^62-1 reaches a lite-06 subscriber or is refused loudly, instead of vanishing +- [Track live signal](/quest/m1/track-live.md) - a reader can wait for its track's feed to be live, so a player skips old cached groups while a fresh upstream subscription starts - [C++ and OBS package proof](/quest/m1/cpp-package-check.md) - the installed artifacts pass before their first release diff --git a/quest/m1/public-model/serve.md b/quest/m1/public-model/serve.md index ea41b4811a5..7f30db46b6c 100644 --- a/quest/m1/public-model/serve.md +++ b/quest/m1/public-model/serve.md @@ -19,40 +19,72 @@ Rust items (`rs/moq-net/src/model/`), as they stood at b4eab0db1: - `track::Subscriber::poll_start` (lite publisher): public. A publisher reports the resolved start to its peer (SUBSCRIBE_START, SUBSCRIBE_OK). - `track::Subscriber::poll_live` (both publishers): remove the need through - the public `poll_largest` (#5725). It does not wait out an idle copy the - way `poll_live` does: it follows the source's reported largest, which can - name a frame the cache does not hold yet, so the publishers still answer - with the buffered edge. Decide here whether wire answers move to it, and + the public `poll_largest` (#5725). It now waits out an idle copy too + (#5924), but it follows the source's reported largest, which can name a + frame the cache does not hold yet, so the publishers still answer with the + buffered edge. A public live signal is + [its own quest](/quest/m1/track-live.md). Decide here whether wire answers move to it, and with them `latest_sequence` on `track::{Producer,Consumer,Subscriber}`, the private snapshot that replaced the public `latest()` (#5725). - **Cursor** (`track::Subscriber::start_at`, `raise_start_to`, `end_at`, `subscription::Cap`; both publishers). **Decision**: - - ✅ One public method assigns the floor and the cap in either direction, - renamed from `set_groups` to `set_bounds` (JS `setBounds`) on - `track::Subscriber` and `track::Ordered`, since it sets the reader's - bounds. Each reader already hands a sequence out once (#5446), so a - SUBSCRIBE_UPDATE that widens the start re-offers only unread groups. + - ✅ One public consuming builder assigns the floor and the cap in either + direction: `with_range(self, range) -> Self` on `track::Subscriber` and + `track::Ordered` (JS `withRange`), replacing `set_groups` and the + chaining `withGroups`. The maintainer decided "range" on 2026-10-19, + replacing the 2026-10-18 `limit` and the earlier `set_bounds`. Each + reader already hands a sequence out once (#5446), so a SUBSCRIBE_UPDATE + that widens the start re-offers only unread groups. `start_at`, `raise_start_to`, `end_at`, and `Cap` go. Precondition: lowering the floor re-offers unread cached groups in every mode. Today only the resume cursor does; the arrival-order (Plain) cursor walks a monotonic arrival index, so it must rescan below its old floor for groups it never handed out. Rewrite the doc ("without rewinding read - progress") to say what lowering does. JS's chaining form, - `withGroups` on `Track.Subscriber` and `Track.Ordered`, becomes - `withBounds`. + progress") to say what lowering does. + - Rebinding: a consuming builder fits readers bounded once at creation, + but these callers move a live reader's bounds through `&mut` today, so + the rename needs a way to rebind (take and replace the reader, or a + crate-private setter the builder wraps): + - `rs/moq-c/src/consume.rs` `RawReader::apply` re-applies the range on + every subscription update, counting a relative start from the track + `Consumer`'s unbounded largest. `set_groups` only raises the floor, so an + update cannot lower a floor an earlier one put above the live edge + (Codex on #5924). Making the floor assign fixes it; land that with a + moq-c test where a later relative update delivers the next group. + - The lite and IETF publishers move the cursor on SUBSCRIBE_UPDATE + (`start_at`, `end_at`; JS `hooks.replaceGroups` and the IETF + publisher's `setGroups({ start })`). + - `js/json` snapshot and `js/flate` `latest()` raise the floor of the same + reader on each read; `js/room` chat sets it after `largest()` resolves. + The snapshot consumers (`js/json/src/snapshot/consumer.ts` `#read`, + `js/flate/src/snapshot/consumer.ts` `latest()`) call + `setGroups({ start: { included: latest.group } })`, whose omitted end + removes the caller's cap. The rebinding must keep that cap; land it with + a regression test where a capped snapshot reader stays below its cap. + - `rs/moq-hls` export narrows a fetched group with `set_frames(..end)`. + - Readers that set a floor once right after `poll_largest` (moq-audio, + moq-video, moq-cli play, moq-json window, moq-mux) fit the builder. + - ✅ The range floor counts as a local bound (maintainer, 2026-10-19): + `largest()` reports within both the floor and the cap, so a reader whose + floor sits above the live edge reports nothing until a group reaches it + (#5924). + - ✅ `largest()` reports only upstream-confirmed state (maintainer, + 2026-10-19): the source's reported largest and what its live feed + delivered since, never a cache-derived stand-in. It waits while the copy + is idle, though cached groups are still served, and past the cap it + reports nothing rather than the newest cached group below it (#5924). - A second public `replace_groups`. -- **Bounds naming** (decided 2026-10-16 in #5717): ✅ the siblings follow the - rename, so every range setter says bounds. `group::Consumer::set_frames` - becomes `set_bounds`, and `Subscription::with_groups` becomes - `with_bounds`, with the JS mirrors (the `Subscription.groups` field - becomes `bounds`, and any group-level frame bounds setter matches). +- **Bounds naming** (decided 2026-10-16 in #5717, changed to "range" by the + maintainer on 2026-10-19): ✅ the siblings follow: + `Subscription::with_range` replaces `with_groups` as the single-range form + of `with_ranges`, and `group::Consumer::set_frames` becomes the consuming + `with_range`, with the JS mirrors. - `group::Consumer::start_at`, `end_at` (both publishers): remove the need; - the public `set_bounds` (today `set_frames`) and `skip_to` cover them. + the public `with_range` (today `set_frames`) and `skip_to` cover them. - `Subscription::{start, end, set_bounds, with_start, with_end}` (both publishers, IETF subscriber, from #5732): the covering span a wire without ranges carries. Derive it from the public `ranges` instead, or make it - public. Either way, rename the crate `set_bounds` so it doesn't clash with - the public `with_bounds`. + public. - `subscription::Position::final_sequence` (lite publisher): public, the one-line wire translation of an exclusive end that both languages already duplicate. `track::Consumer::final_sequence` (IETF publisher) goes; @@ -76,7 +108,7 @@ Rust items (`rs/moq-net/src/model/`), as they stood at b4eab0db1: JS items (`js/net/src/`): - `hooks.replaceGroups` (lite publisher): mirror the cursor decision, so - `Track.Subscriber.setBounds` assigns the floor and the hook goes. + `Track.Subscriber.withRange` assigns the floor and the hook goes. - `hooks.tryRecvGroup` with its `Recv` union, and `hooks.groupChanged` (lite publisher). The public `Subscriber.tryRecvGroup()` returns `undefined` for both a held boundary and a drained end, so a synchronous diff --git a/quest/m1/track-live.md b/quest/m1/track-live.md new file mode 100644 index 00000000000..f5c5532e5d9 --- /dev/null +++ b/quest/m1/track-live.md @@ -0,0 +1,31 @@ +# [M] A reader can tell when its track is live + +## Goal + +A track reader in moq-net and `@moq/net` can wait for, and check, whether its +track's feed is live: a local producer from the start, a relay's copy once its +upstream subscription answers. A player uses it to skip the old cached groups +a warm copy serves while it waits for a fresh upstream subscription, instead +of rendering them. + +## Plan + +`largest()` already waits for live internally (#5924), so it never reports +what an ended feed left; this quest makes the signal itself public. + +- Rust has the state: `TrackState::feed` (`Live`, `Idle`, `Answered`) and + `readable()` in `rs/moq-net/src/model/track.rs`, the crate-private + `resume` `poll_live`, and the crate-private `Producer::is_live`. Expose one + reader-side `poll_live`/`live()` (or `is_live`) on `track::Subscriber` and + `track::Ordered`, and decide whether the publishers' crate-private + `poll_live` folds into it (see + [Public serving](/quest/m1/public-model/serve.md)). +- JS has no equivalent: a track never goes idle, so its reader is live from + the start. Mirror the name, and the idle state if JS gains warm copies. +- Decide what a reader does with groups served before live: keep serving + them (today) and let the caller skip by the signal, or offer a mode that + starts at the live edge. + +## Related + +- [Public serving](/quest/m1/public-model/serve.md) - the publishers' `poll_live` and the `largest()` decision diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index c4abe0d0e2b..a77ad17ebe7 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -43,7 +43,8 @@ pub enum Start { /// Decode every cached group. #[default] Oldest, - /// Start at the newest cached group, including earlier groups for decoder warmup. + /// Start at the newest group, including earlier groups for decoder warmup. Opening + /// waits for the track's live edge, or its end. Latest, } @@ -193,10 +194,12 @@ impl Consumer { .await?; // Latest skips cached history unless decoder warmup needs it. Move the // local cursor, not Subscription::ranges: the latter is aggregated - // across subscribers and can ask the publisher to rewind for everyone. + // across subscribers and can ask the publisher to rewind for everyone. The live edge + // is known only once the feed is live, so wait for it rather than read the backlog + // of a warm copy still waiting on its source; the wait ends with the track. if options.start == Start::Latest && warmup.is_zero() - && let std::task::Poll::Ready(Some(live_edge)) = subscriber.poll_largest(&moq_net::kio::Waiter::noop()) + && let Some(live_edge) = subscriber.largest().await { subscriber.set_groups(live_edge.group..); } @@ -759,6 +762,119 @@ mod tests { (stamps, frames) } + /// An origin at `hop`, its driver spawned. + fn origin(hop: u64) -> moq_net::origin::Producer { + let config = moq_net::origin::Config::new(moq_net::Hop::new(hop).unwrap()); + let (producer, driver) = moq_net::origin::Producer::new(config); + tokio::spawn(moq_net::time::run(driver)); + producer + } + + /// A lite-07 session over memory carrying `from`'s broadcasts into `to`, open while held. + async fn link( + from: &moq_net::origin::Producer, + to: &moq_net::origin::Producer, + ) -> (moq_net::Session, moq_net::Session) { + use moq_net::transport::memory; + + let version: moq_net::Version = "moq-lite-07-wip".parse().unwrap(); + let mut config = memory::Config::default(); + config.protocol = Some(version.alpn().to_string()); + let (client_transport, server_transport) = memory::pair(config); + let client = moq_net::Client::new() + .with_versions(version.into()) + .with_publisher(from.consume()); + let server = moq_net::Server::new() + .with_versions(version.into()) + .with_subscriber(to.clone()); + tokio::join!( + async { + let (session, driver) = client + .connect_lite(moq_net::time::now(), client_transport) + .await + .unwrap(); + tokio::spawn(moq_net::time::run(driver)); + session + }, + async { + let (session, driver) = server + .accept_lite(moq_net::time::now(), server_transport) + .await + .unwrap(); + tokio::spawn(moq_net::time::run(driver)); + session + } + ) + } + + /// A relay copy whose upstream subscription went idle keeps the groups it cached. A + /// latest start waits for the copy to go live and joins at its live edge, rather than + /// replaying that backlog. + #[tokio::test(start_paused = true)] + async fn latest_skips_a_warm_copys_backlog() { + let publisher = origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let track = broadcast + .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) + .unwrap(); + broadcast.announce(Default::default()).unwrap(); + let mut demand = track.demand(); + let mut producer = moq_mux::container::Producer::new( + track, + moq_mux::catalog::hang::Container::Legacy(moq_mux::container::Kind::Audio), + ); + let mut write = |packet: u64| { + producer + .write(moq_mux::container::Frame { + timestamp: Timestamp::from_micros(packet * 20_000).unwrap(), + duration: None, + payload: vec![0; 960 * size_of::()].into(), + keyframe: true, + }) + .unwrap(); + producer.cut(None).unwrap(); + }; + for packet in 0..4 { + write(packet); + } + + let relay = origin(2); + let _link = link(&publisher, &relay).await; + let consumer = relay.consume(); + consumer.routed("bcast").await.unwrap(); + let remote = consumer.request_broadcast("bcast", None).await.unwrap(); + + // A first reader warms the relay's copy with every group. + { + let subscription = moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(10)); + let mut warm = remote.track("audio").unwrap().subscribe(subscription).await.unwrap(); + let mut groups = 0; + while groups < 4 { + warm.recv_group().await.unwrap().unwrap(); + groups += 1; + } + } + // The relay cancels the unread upstream subscription; its copy keeps the groups. + while demand.wanted_changed().await.unwrap().is_some() {} + write(4); + + let catalog = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Pcm, 48_000, 1); + let mut consumer = Consumer::new( + &remote, + &catalog, + "audio", + Options { + start: Start::Latest, + max_delay: Duration::from_secs(10), + ..Options::new() + }, + ) + .await + .unwrap(); + let frame = next_frame(&mut consumer).await.unwrap().unwrap(); + assert_eq!(frame.timestamp.as_micros(), 80_000, "the backlog was replayed"); + } + #[tokio::test(start_paused = true)] async fn warmup_trims_the_join_without_trimming_groups_or_declared_returns() { let (stamps, frames) = warmup_output(Start::Oldest, true).await; diff --git a/rs/moq-c/src/consume.rs b/rs/moq-c/src/consume.rs index 2597f60932a..1855647055f 100644 --- a/rs/moq-c/src/consume.rs +++ b/rs/moq-c/src/consume.rs @@ -111,6 +111,90 @@ struct RawTaskEntry { callback: OnStatus, } +/// A raw track's reader, with the track handle its subscription updates resolve against. +struct RawReader { + // A relative range counts from this track's largest: the reader's own reports only + // within the bounds the update is replacing. Kept unpolled, so each clone reports the + // current largest rather than only a newer one. + edge: moq_net::track::Consumer, + track: moq_net::track::Ordered, + // An update with a relative range, whose bounds wait for the edge to be live. + deferred: Option, +} + +impl RawReader { + fn new(edge: moq_net::track::Consumer, track: moq_net::track::Ordered) -> Self { + Self { + edge, + track, + deferred: None, + } + } + + /// Apply a subscription update: its wire preferences at once, and its local bounds once + /// any relative range has a live edge to count from. + fn apply(&mut self, subscription: Option, waiter: &kio::Waiter) { + let subscription = subscription.unwrap_or_default(); + // A closed track makes the update meaningless; the reader already sees the close. + let _ = self.track.update(subscription.clone()); + let relative = subscription + .ranges + .iter() + .any(|range| matches!(range.start, moq_net::track::Anchor::Relative(_))); + match relative { + true => { + self.deferred = Some(subscription); + self.poll_deferred(waiter); + } + false => { + self.deferred = None; + self.set_bounds(&subscription, None); + } + } + } + + /// Apply a deferred update's bounds once the edge is live, or the track ended. + fn poll_deferred(&mut self, waiter: &kio::Waiter) { + let Some(subscription) = &self.deferred else { + return; + }; + if let Poll::Ready(largest) = self.edge.clone().poll_largest(waiter) { + let subscription = subscription.clone(); + self.deferred = None; + self.set_bounds(&subscription, largest.map(|position| position.group)); + } + } + + fn set_bounds(&mut self, subscription: &moq_net::track::Subscription, latest: Option) { + // Reads start where the lowest range does. Without a floor of its own (the oldest + // range, the default) a raw reader joins at the latest group. + let start = subscription + .ranges + .iter() + .map(|range| match range.start { + moq_net::track::Anchor::Absolute(start) => Some(start.group), + moq_net::track::Anchor::Relative(u64::MAX) => latest, + moq_net::track::Anchor::Relative(back) => latest.map(|latest| latest.saturating_sub(back)), + }) + .min() + .flatten(); + // They stop past the last absolute end; an open or relative one follows the live edge. + let end = subscription + .ranges + .iter() + .map(|range| match range.end { + Some(moq_net::track::Anchor::Absolute(end)) => Some(end), + _ => None, + }) + .collect::>>() + .map(|ends| ends.into_iter().max().unwrap_or_default()); + self.track.set_groups(( + start.map_or(Bound::Unbounded, Bound::Included), + end.map_or(Bound::Unbounded, moq_net::track::Position::group_end), + )); + } +} + /// Outcome of polling a raw track source alongside its control channels. enum RawStep { /// A delivered value: the next group, or the next frame within a group. @@ -665,9 +749,12 @@ impl Consume { // `subscribe` blocks on SUBSCRIBE_OK, so run it inside the task. tokio::spawn(async move { let res = crate::ffi::run(driver.watch(async move { - let mut track = broadcast.track(&name)?.subscribe(subscription.clone()).await?.ordered(); - Self::apply_raw_subscription(&mut track, subscription); - Self::run_raw(on_frame, track, channel.1, updates).await + let edge = broadcast.track(&name)?; + let track = edge.subscribe(subscription.clone()).await?.ordered(); + let mut reader = RawReader::new(edge, track); + // The task's own polls re-poll a deferred update with their waiter. + reader.apply(subscription, &kio::Waiter::noop()); + Self::run_raw(on_frame, reader, channel.1, updates).await })) .await; @@ -682,45 +769,9 @@ impl Consume { Ok(id) } - fn apply_raw_subscription(track: &mut moq_net::track::Ordered, subscription: Option) { - let subscription = subscription.unwrap_or_default(); - // Reads start where the lowest range does. Without a floor of its own (the oldest - // range, the default) a raw reader joins at the latest group. - let latest = match track.poll_largest(&kio::Waiter::noop()) { - std::task::Poll::Ready(Some(position)) => Some(position.group), - _ => None, - }; - let start = subscription - .ranges - .iter() - .map(|range| match range.start { - moq_net::track::Anchor::Absolute(start) => Some(start.group), - moq_net::track::Anchor::Relative(u64::MAX) => latest, - moq_net::track::Anchor::Relative(back) => latest.map(|latest| latest.saturating_sub(back)), - }) - .min() - .flatten(); - // They stop past the last absolute end; an open or relative one follows the live edge. - let end = subscription - .ranges - .iter() - .map(|range| match range.end { - Some(moq_net::track::Anchor::Absolute(end)) => Some(end), - _ => None, - }) - .collect::>>() - .map(|ends| ends.into_iter().max().unwrap_or_default()); - track.set_groups(( - start.map_or(Bound::Unbounded, Bound::Included), - end.map_or(Bound::Unbounded, moq_net::track::Position::group_end), - )); - // A closed track makes the update meaningless; the reader already sees the close. - let _ = track.update(subscription); - } - async fn run_raw( callback: OnStatus, - mut track: moq_net::track::Ordered, + mut reader: RawReader, mut close: oneshot::Receiver<()>, mut updates: mpsc::UnboundedReceiver>, ) -> Result<(), Error> { @@ -734,10 +785,14 @@ impl Consume { loop { let mut group = match moq_net::kio::wait(|waiter| -> Poll, Error>> { - if Self::poll_raw_control(&mut close, &mut updates, &mut track, waiter) { + if Self::poll_raw_control(&mut close, &mut updates, &mut reader, waiter) { return Poll::Ready(Ok(RawStep::Stop)); } - match ready!(track.poll_next_group(waiter))? { + // A group read before deferred bounds apply could sit under the old floor. + if reader.deferred.is_some() { + return Poll::Pending; + } + match ready!(reader.track.poll_next_group(waiter))? { Some(group) => Poll::Ready(Ok(RawStep::Item(group))), None => Poll::Ready(Ok(RawStep::End)), } @@ -751,7 +806,7 @@ impl Consume { loop { let frame = match moq_net::kio::wait(|waiter| -> Poll, Error>> { - if Self::poll_raw_control(&mut close, &mut updates, &mut track, waiter) { + if Self::poll_raw_control(&mut close, &mut updates, &mut reader, waiter) { return Poll::Ready(Ok(RawStep::Stop)); } match ready!(group.poll_read_frame(waiter))? { @@ -775,17 +830,18 @@ impl Consume { } } - /// Poll the close and update channels, applying any subscription updates inline. + /// Poll the close and update channels, applying any subscription updates inline, and a + /// deferred update once its edge is live. /// /// Returns `true` when the task must stop: either the consumer was closed /// (`close` fired) or the update channel was dropped. The caller must then /// unwind rather than poll `close` again, since a completed oneshot panics if - /// re-polled. Borrows `track` only for the duration of the call so the caller + /// re-polled. Borrows `reader` only for the duration of the call so the caller /// can poll a track/group source afterwards. fn poll_raw_control( close: &mut oneshot::Receiver<()>, updates: &mut mpsc::UnboundedReceiver>, - track: &mut moq_net::track::Ordered, + reader: &mut RawReader, waiter: &moq_net::kio::Waiter, ) -> bool { let mut cx = waiter.context(); @@ -795,11 +851,13 @@ impl Consume { loop { match updates.poll_recv(&mut cx) { - Poll::Ready(Some(subscription)) => Self::apply_raw_subscription(track, subscription), + Poll::Ready(Some(subscription)) => reader.apply(subscription, waiter), Poll::Ready(None) => return true, - Poll::Pending => return false, + Poll::Pending => break, } } + reader.poll_deferred(waiter); + false } pub fn raw_track_update( diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 3364763c905..3dec15a63fc 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -1810,6 +1810,96 @@ fn raw_track_subscription_options_and_update() { assert_eq!(moq_origin_close(origin), 0); } +#[test] +fn raw_track_relative_update_counts_from_the_track_edge() { + let origin = id(moq_origin_create()); + let path = b"raw-track-relative"; + let broadcast = publish_broadcast(origin, path); + + let track_name = b"data"; + let track = id(unsafe { + moq_publish_track( + broadcast, + track_name.as_ptr() as *const c_char, + track_name.len(), + std::ptr::null(), + ) + }); + let publish = |payload: &[u8]| { + assert_eq!( + unsafe { moq_publish_track_frame(track, payload.as_ptr(), payload.len(), 0) }, + 0 + ); + }; + let anchor = |kind: moq_anchor_kind, group: u64| moq_anchor { + kind: kind as u32, + group, + frame: 0, + }; + let subscribe = |ranges: &[moq_range]| moq_subscription { + priority: 0, + max_delay_us: 60_000_000, + ranges: ranges.as_ptr(), + ranges_len: ranges.len(), + order: 0, + }; + let recv = |frame_cb: &Callback| { + let frame_id = id(frame_cb.recv()); + let mut frame = moq_frame { + payload: std::ptr::null(), + payload_size: 0, + timestamp_us: 0, + timestamp_present: false, + keyframe: false, + }; + assert_eq!(unsafe { moq_consume_track_frame(frame_id, &mut frame) }, 0); + let received = unsafe { std::slice::from_raw_parts(frame.payload, frame.payload_size) }.to_vec(); + assert_eq!(moq_consume_track_frame_free(frame_id), 0); + received + }; + + publish(b"zero"); + let consume = request_broadcast(origin, path); + + // Group 0 only: the reader's cap sits below where the track will be. + let first = [moq_range { + start: anchor(moq_anchor_kind::MOQ_ANCHOR_ABSOLUTE, 0), + end: anchor(moq_anchor_kind::MOQ_ANCHOR_ABSOLUTE, 1), + end_present: true, + }]; + let frame_cb = Callback::new(); + let consumer = id(unsafe { + moq_consume_track( + consume, + track_name.as_ptr() as *const c_char, + track_name.len(), + &subscribe(&first), + Some(channel_callback), + frame_cb.ptr, + ) + }); + assert_eq!(recv(&frame_cb), b"zero"); + + publish(b"one"); + publish(b"two"); + + // The latest group is the track's, group 2, not the newest one under the old cap. + let latest = [moq_range { + start: anchor(moq_anchor_kind::MOQ_ANCHOR_RELATIVE, 0), + end: anchor(moq_anchor_kind::MOQ_ANCHOR_ABSOLUTE, 0), + end_present: false, + }]; + assert_eq!(unsafe { moq_consume_track_update(consumer, &subscribe(&latest)) }, 0); + assert_eq!(recv(&frame_cb), b"two"); + + assert_eq!(moq_consume_track_cancel(consumer), 0); + assert_eq!(frame_cb.recv_terminal(), 0); + assert_eq!(moq_publish_track_finish(track), 0); + assert_eq!(moq_consume_close(consume), 0); + assert_eq!(moq_publish_close(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} + #[test] fn json_snapshot_publish_consume() { let origin = id(moq_origin_create()); diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index afa944e04dc..7aa884a844d 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -258,12 +258,11 @@ impl Media { .with_max_delay(max_delay), ) .await?; - // Start at the local live edge without asking the shared publisher - // subscription to rewind to a cached sequence. Only what is known now: a - // track with nothing yet starts wherever its first group lands. - if let std::task::Poll::Ready(Some(largest)) = - subscriber.poll_largest(&moq_net::kio::Waiter::noop()) - { + // Start at the live edge without asking the shared publisher + // subscription to rewind to a cached sequence. The edge is known only + // once the feed is live, so wait for it rather than read the backlog of + // a warm copy still waiting on its source; the wait ends with the track. + if let Some(largest) = subscriber.largest().await { subscriber.set_groups(largest.group..); } let format = catalog::hang::Container::try_from(config)?; diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 70d23af50da..7ad13c6f47a 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -850,8 +850,9 @@ enum SiState { Requesting(crate::Binding, String), /// Waiting for the subscription to resolve. Subscribing(kio::Pending), - /// The resolved subscription, reading snapshot groups. - Active(moq_net::track::Ordered), + /// The resolved subscription, reading snapshot groups. Boxed: a reader is far larger + /// than the other states. + Active(Box), /// The track ended or failed; the last complete snapshot keeps re-emitting, /// mirroring how a real mux repeats its tables between (absent) revisions. Done, @@ -937,7 +938,7 @@ impl SiTrack { } }; self.state = match resolved { - Ok(track) => SiState::Active(track.ordered()), + Ok(track) => SiState::Active(Box::new(track.ordered())), Err(err) => { tracing::warn!(%err, "SI subscription failed; carrying the last snapshot"); SiState::Done diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index d2611b9b980..bda4b7f1efc 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,15 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `largest()` reports only upstream state, within a reader's `set_groups` bounds + +`track::Subscriber::largest()` and `track::Ordered::largest()` (and their `poll_largest`) report the track's largest only while it is within the bounds set with `set_groups`, where they used to ignore the bounds. +A largest past the cap or below the floor reports nothing; the cache never stands in for it. +A cap moved below the largest stays pending. Moving the bounds wakes a pending read, which reports the largest once it is within them and newer than the last report. +Every `largest()`, `track::Producer` and `track::Consumer` included, now waits while a relay's copy is idle and reports once its new upstream subscription is live, rather than the largest an ended feed left. +Cached groups are still served meanwhile. +To read the track's edge from a capped reader, lift the cap with `set_groups(..)` first. + ### Lite-07 accounts for groups with `SUBSCRIBE_DROP` `lite::SubscribeEnd` now contains only `group`; remove `streams` from struct literals and stop calling `Version::has_stream_count()`. diff --git a/rs/moq-net/src/model/cache.rs b/rs/moq-net/src/model/cache.rs index 640fedfa7bc..40f9e5cf9d9 100644 --- a/rs/moq-net/src/model/cache.rs +++ b/rs/moq-net/src/model/cache.rs @@ -745,13 +745,18 @@ impl Track { } } - /// Poll for a largest location newer than `seen`, recording it there. `None` once the - /// track ended with nothing newer. - pub(super) fn poll_largest(&self, seen: &mut Option, waiter: &kio::Waiter) -> Poll> { + /// Poll for a largest location within `bounds` newer than `seen`, recording it there. + /// `None` once the track ended with nothing newer. + pub(super) fn poll_largest( + &self, + seen: &mut Option, + bounds: track::Bounds, + waiter: &kio::Waiter, + ) -> Poll> { let newer = |largest: &Largest| { largest .position - .filter(|position| seen.is_none_or(|seen| *position > seen)) + .filter(|position| bounds.contains(position.group) && seen.is_none_or(|seen| *position > seen)) }; let largest = ready!( self.largest @@ -767,13 +772,13 @@ impl Track { Poll::Ready(next) } - /// The largest frame newer than `seen` without waiting, recording it there. - pub(super) fn take_largest(&self, seen: &mut Option) -> Option { + /// The largest frame within `bounds` newer than `seen` without waiting, recording it there. + pub(super) fn take_largest(&self, seen: &mut Option, bounds: track::Bounds) -> Option { let next = self .largest .read() .position - .filter(|position| seen.is_none_or(|seen| *position > seen)); + .filter(|position| bounds.contains(position.group) && seen.is_none_or(|seen| *position > seen)); if next.is_some() { *seen = next; } diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index a201bcd1878..fd95ba87c44 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -142,10 +142,16 @@ impl Consumer { self.state.read().copy.clone() } - /// Poll for a largest location newer than `seen` on the serving copy, recording it there. - /// A copy ending is its route failing, so the logical track ends only with the route: - /// at once when the front ends it, or with the last copy once the front concluded. - pub(super) fn poll_largest(&self, seen: &mut Option, waiter: &kio::Waiter) -> Poll> { + /// Poll for a largest location within `bounds` newer than `seen` on the serving copy, + /// recording it there. A copy ending is its route failing, so the logical track ends + /// only with the route: at once when the front ends it, or with the last copy once the + /// front concluded. + pub(super) fn poll_largest( + &self, + seen: &mut Option, + bounds: track::Bounds, + waiter: &kio::Waiter, + ) -> Poll> { loop { let (generation, copy) = { let route = self.state.read(); @@ -153,7 +159,7 @@ impl Consumer { }; let end = self.end(); // A replacement copy can start below what an earlier one reported. - match copy.map(|copy| copy.poll_largest_after(seen, waiter)) { + match copy.map(|copy| copy.poll_largest_after(seen, bounds, waiter)) { Some(Poll::Ready(Some(largest))) => return Poll::Ready(Some(largest)), _ if matches!(end, Some(Some(_))) => return Poll::Ready(None), Some(Poll::Pending) if end.is_some() => return Poll::Pending, @@ -167,6 +173,9 @@ impl Consumer { /// Subscribe to the logical track with the reader's own preferences. pub(super) fn subscribe(&self, subscription: track::Preferences) -> Subscriber { let mirrored = subscription.read().clone(); + // An absolute floor holds from the start, as on a plain cursor; a relative one + // counts back from an edge, so the reported largest is never below it. + let floor = mirrored.floor(None).unwrap_or(0); let reader = Arc::new_cyclic(|this| { Mutex::new(Reader { this: this.clone(), @@ -177,7 +186,7 @@ impl Consumer { copies: Vec::new(), newest: None, datagram: None, - groups: (0, Cap::from(..)), + groups: (floor, Cap::from(..)), groups_read: false, }) }); @@ -448,10 +457,24 @@ impl Subscriber { self.reader().latest() } - /// Poll for a largest location newer than `seen`; see [`Consumer::poll_largest`]. - pub(super) fn poll_largest(&mut self, seen: &mut Option, waiter: &kio::Waiter) -> Poll> { + /// Poll for a largest location within `bounds` newer than `seen`; see [`Consumer::poll_largest`]. + pub(super) fn poll_largest( + &mut self, + seen: &mut Option, + bounds: track::Bounds, + waiter: &kio::Waiter, + ) -> Poll> { let route = self.reader().route.clone(); - route.poll_largest(seen, waiter) + route.poll_largest(seen, bounds, waiter) + } + + /// The local group filter, as [`track::Subscriber::poll_largest`] reports within it. + pub(super) fn bounds(&self) -> track::Bounds { + let (start, end) = self.reader().groups; + track::Bounds { + start, + end: end.exclusive(), + } } /// The groups the copies' drift budget skipped since the last call, counted against @@ -1658,6 +1681,48 @@ mod test { assert_eq!(sub.largest().now_or_never(), Some(None)); } + /// A front's reader reports the serving copy's largest only within its cap. + #[test] + #[cfg(feature = "async")] + fn largest_honours_the_cap_on_the_serving_copy() { + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = subscribe(&logical, HOUR); + for sequence in 0..5 { + write(&a, sequence); + } + sub.set_groups(..3); + assert!(sub.largest().now_or_never().is_none(), "the largest is past the cap"); + sub.set_groups(..); + assert_eq!(sub.largest().now_or_never(), Some(Some(Position::group(4)))); + } + + /// A front's reader created with a floor above the serving copy's edge reports nothing + /// below it, as a plain reader does. + #[test] + #[cfg(feature = "async")] + fn largest_honours_the_subscription_floor_on_the_serving_copy() { + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let subscription = Subscription::default().with_max_delay(HOUR).with_groups(8..); + let mut sub = logical + .consume() + .subscribe(subscription) + .now_or_never() + .unwrap() + .unwrap(); + for sequence in 0..5 { + write(&a, sequence); + } + assert!(sub.largest().now_or_never().is_none(), "the largest is below the floor"); + write(&a, 8); + assert_eq!(sub.largest().now_or_never(), Some(Some(Position::group(8)))); + } + /// A track reader waiting before any route serves is woken by the first one's frame. /// The front ending the track ends the stream, though the copy's producer stays open. #[test] diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index c8858d0b38c..e0eee93a9db 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -2150,20 +2150,52 @@ fn commit_abort(mut state: kio::Mut<'_, TrackState>, err: Error) { state.close(); } -/// Poll `cache` for a largest location newer than `seen`. `ended` polls the track for -/// [`TrackState::largest_settled`], or its channel closing without ending the cache (a -/// dropped request), either of which ends the stream. +/// A reader's local group bounds: the floor, and the exclusive cap (`None` for none). +#[derive(Clone, Copy, Debug, Default)] +pub(super) struct Bounds { + pub start: u64, + pub end: Option, +} + +impl Bounds { + /// Whether group `sequence` is within these bounds. + pub fn contains(&self, sequence: u64) -> bool { + sequence >= self.start && super::subscription::before_end(sequence, self.end) + } +} + +/// Poll `cache` for a largest location within `bounds` newer than `seen`, recording it there. +/// +/// The largest is what the source reported and what its feed delivered since, never what +/// the cache happens to hold, so `live` holds the stream until the track's feed is live: an +/// idle copy's largest is only what an ended feed left. Past the cap or below the floor it +/// reports nothing. `ended` polls the track for [`TrackState::largest_settled`], or its +/// channel closing without ending the cache (a dropped request), either of which ends the +/// stream. fn poll_largest( cache: &cache::Track, seen: &mut Option, + bounds: Bounds, waiter: &kio::Waiter, + live: impl FnOnce(&kio::Waiter) -> Poll<()>, ended: impl FnOnce(&kio::Waiter) -> Poll<()>, ) -> Poll> { - if let Poll::Ready(next) = cache.poll_largest(seen, waiter) { + ready!(live(waiter)); + if let Poll::Ready(next) = cache.poll_largest(seen, bounds, waiter) { return Poll::Ready(next); } ready!(ended(waiter)); - Poll::Ready(cache.take_largest(seen)) + Poll::Ready(cache.take_largest(seen, bounds)) +} + +/// Ready once the track's feed is live, or nothing more can move its largest. A declared +/// end that is not reached yet does not count: an idle copy's largest is still only what +/// its ended feed left. +fn largest_live(state: &TrackState) -> Poll<()> { + match state.feed == Feed::Live || state.sealed || state.abort.is_some() || state.largest_settled() { + true => Poll::Ready(()), + false => Poll::Pending, + } } /// Ready once the track's largest can no longer advance; see [`TrackState::largest_settled`]. @@ -2751,9 +2783,15 @@ impl Producer { /// Poll for the track's largest location: the current one, then each newer one, and `None` once the track ends. pub fn poll_largest(&mut self, waiter: &kio::Waiter) -> Poll> { let cache = self.state.read().cache.clone(); - poll_largest(&cache, &mut self.largest, waiter, |waiter| { - self.state.poll(waiter, |state| largest_ended(state)).map(|_| ()) - }) + let state = &self.state; + poll_largest( + &cache, + &mut self.largest, + Bounds::default(), + waiter, + |waiter| state.poll(waiter, |state| largest_live(state)).map(|_| ()), + |waiter| state.poll(waiter, |state| largest_ended(state)).map(|_| ()), + ) } /// The track's largest location, its newest group or datagram at the newest frame (0 while empty): the current one, then each newer one, and `None` once the track ends. @@ -2837,7 +2875,7 @@ impl Producer { // A producer-side (in-process) subscribe is not egress: stay untagged. stats: stats::Scope::default(), _stats_sub: stats::Subscription::default(), - largest: None, + largest: Reported::default(), } } @@ -3931,17 +3969,18 @@ impl Consumer { /// Poll for the track's largest location: the current one, then each newer one, and `None` once the track ends. pub fn poll_largest(&mut self, waiter: &kio::Waiter) -> Poll> { let mut seen = self.largest; - let next = self.poll_largest_after(&mut seen, waiter); + let next = self.poll_largest_after(&mut seen, Bounds::default(), waiter); self.largest = seen; next } - /// Poll for a largest frame newer than `seen`, recording it there. A front's logical - /// track follows the route serving it. Reads the instance served now without asking for - /// it, so a broadcast serving nothing under the name ends the stream. + /// Poll for a largest frame within `bounds` newer than `seen`, recording it there. A + /// front's logical track follows the route serving it. Reads the instance served now + /// without asking for it, so a broadcast serving nothing under the name ends the stream. pub(super) fn poll_largest_after( &self, seen: &mut Option, + bounds: Bounds, waiter: &kio::Waiter, ) -> Poll> { self.with_state(|state| { @@ -3953,10 +3992,15 @@ impl Consumer { (state.routes.clone(), state.cache.clone()) }; match routes { - Some(routes) => routes.poll_largest(seen, waiter), - None => poll_largest(&cache, seen, waiter, |waiter| { - state.poll(waiter, |state| largest_ended(state)).map(|_| ()) - }), + Some(routes) => routes.poll_largest(seen, bounds, waiter), + None => poll_largest( + &cache, + seen, + bounds, + waiter, + |waiter| state.poll(waiter, |state| largest_live(state)).map(|_| ()), + |waiter| state.poll(waiter, |state| largest_ended(state)).map(|_| ()), + ), } }) } @@ -4075,7 +4119,7 @@ impl Subscribing { delivered: Delivered::default(), stats: self.stats.clone(), _stats_sub: self.stats.subscribe(), - largest: None, + largest: Reported::default(), })) } @@ -4515,8 +4559,25 @@ pub struct Subscriber { // The subscription guard: bumps `subscriptions` (and the egress viewer refcount) // while held, closing them on drop. Empty (no-op) for an untagged track. _stats_sub: stats::Subscription, - // Last position this handle yielded through `largest`. - largest: Option, + largest: Reported, +} + +/// What a [`Subscriber`]'s `largest` stream last reported. +#[derive(Default)] +struct Reported { + position: Option, + // The stream ended: moving the bounds afterwards does not bring it back. + ended: bool, + // The parked poll, woken when the bounds move. Its registration dies with its waiter, + // so a dropped `largest()` future leaves nothing behind. + parked: kio::WaiterList, +} + +impl Reported { + /// The bounds moved: wake a parked poll, since the largest may now be within them. + fn bounds_moved(&mut self) { + self.parked.wake(); + } } /// Unread groups a logical reader can still take from a replaced copy. @@ -5548,6 +5609,7 @@ impl Subscriber { /// `2..=5` includes groups 2 through 5; `2..5` excludes group 5. An omitted /// start preserves the current floor, and an omitted end removes the cap. /// Groups above the end remain buffered and can be read after raising the cap. + /// [`Self::poll_largest`] reports within these bounds. /// This changes only local delivery; use [`Subscription::ranges`] and /// [`Self::update`] to change the requested groups. pub fn set_groups(&mut self, groups: impl RangeBounds) { @@ -5572,6 +5634,7 @@ impl Subscriber { // leave a finished group below the old floor unread. Inner::Resume(resume, _) => resume.start_at(sequence), } + self.largest.bounds_moved(); } /// Raise the read cursor's floor to `sequence`, keeping any higher floor already set. @@ -5581,6 +5644,7 @@ impl Subscriber { Inner::Plain(cursor) => cursor.min_sequence = cursor.min_sequence.max(sequence), Inner::Resume(resume, _) => resume.raise_start_to(sequence), } + self.largest.bounds_moved(); } /// Cap this subscriber's read cursor at `end`, or remove the cap with `..`. @@ -5605,6 +5669,18 @@ impl Subscriber { } Inner::Resume(resume, _) => resume.end_at(end.into()), } + self.largest.bounds_moved(); + } + + /// The local bounds `largest` reports within. + fn bounds(&self) -> Bounds { + match &self.inner { + Inner::Plain(cursor) => Bounds { + start: cursor.min_sequence, + end: cursor.end_sequence, + }, + Inner::Resume(resume, _) => resume.bounds(), + } } /// This subscriber's current preferences. @@ -5625,34 +5701,66 @@ impl Subscriber { channel.update(subscription) } - /// Poll for the track's largest location: the current one, then each newer one, and `None` once the track ends. + /// Poll for the track's largest location within [`Self::set_groups`]: the current one, then each newer one, and `None` once the track ends. + /// + /// The largest is the newest group or datagram at the newest frame (0 while empty), as + /// the source reported it and its live feed delivered since, so nothing reports until + /// the track's feed is live. A largest past the cap or below the floor reports nothing: + /// the cache does not stand in for it, so a cap moved below the largest stays pending. + /// Moving the bounds wakes a parked poll, which reports the largest once it is within + /// them and newer than the last report. pub fn poll_largest(&mut self, waiter: &kio::Waiter) -> Poll> { + if self.largest.ended { + return Poll::Ready(None); + } + let bounds = self.bounds(); + let poll = self.poll_largest_within(bounds, waiter); + // Only a parked poll waits on the bounds; a finished one is not woken by them. + match poll { + Poll::Pending => waiter.register(&mut self.largest.parked), + Poll::Ready(next) => { + self.largest.parked = kio::WaiterList::new(); + self.largest.ended = next.is_none(); + } + } + poll + } + + fn poll_largest_within(&mut self, bounds: Bounds, waiter: &kio::Waiter) -> Poll> { + let seen = &mut self.largest.position; match &mut self.inner { Inner::Plain(cursor) => { - let cache = cursor.state.read().cache.clone(); - poll_largest(&cache, &mut self.largest, waiter, |waiter| { - cursor.state.poll(waiter, |state| largest_ended(state)).map(|_| ()) - }) + let state = &cursor.state; + let cache = state.read().cache.clone(); + poll_largest( + &cache, + seen, + bounds, + waiter, + |waiter| state.poll(waiter, |state| largest_live(state)).map(|_| ()), + |waiter| state.poll(waiter, |state| largest_ended(state)).map(|_| ()), + ) } - Inner::Resume(resume, _) => resume.poll_largest(&mut self.largest, waiter), + Inner::Resume(resume, _) => resume.poll_largest(seen, bounds, waiter), } } - /// The track's largest location, its newest group or datagram at the newest frame (0 while empty): the current one, then each newer one, and `None` once the track ends. + /// The largest location within [`Self::set_groups`]; see [`Self::poll_largest`]. #[cfg(feature = "async")] pub async fn largest(&mut self) -> Option { kio::wait(|waiter| self.poll_largest(waiter)).await } - /// The largest location known now, without waiting or moving [`Self::poll_largest`] on: - /// past what the cache holds once the source reported a newer one. + /// The track's largest location known now, ignoring [`Self::set_groups`], without waiting + /// or moving [`Self::poll_largest`] on: past what the cache holds once the source + /// reported a newer one. #[cfg_attr(not(feature = "async"), allow(dead_code))] // private: capability gap pub(crate) fn peek_largest(&mut self) -> Option { let mut seen = None; match &mut self.inner { - Inner::Plain(cursor) => cursor.state.read().cache.take_largest(&mut seen), - Inner::Resume(resume, _) => match resume.poll_largest(&mut seen, &kio::Waiter::noop()) { + Inner::Plain(cursor) => cursor.state.read().cache.take_largest(&mut seen, Bounds::default()), + Inner::Resume(resume, _) => match resume.poll_largest(&mut seen, Bounds::default(), &kio::Waiter::noop()) { Poll::Ready(largest) => largest, Poll::Pending => None, }, @@ -5814,12 +5922,12 @@ impl Ordered { kio::wait(|waiter| self.poll_finished(waiter)).await } - /// Poll for the track's largest location: the current one, then each newer one, and `None` once the track ends. + /// Poll for the largest location within [`Self::set_groups`]; see [`Subscriber::poll_largest`]. pub fn poll_largest(&mut self, waiter: &kio::Waiter) -> Poll> { self.inner.poll_largest(waiter) } - /// The track's largest location, its newest group or datagram at the newest frame (0 while empty): the current one, then each newer one, and `None` once the track ends. + /// The largest location within [`Self::set_groups`]; see [`Subscriber::poll_largest`]. #[cfg(feature = "async")] pub async fn largest(&mut self) -> Option { kio::wait(|waiter| self.poll_largest(waiter)).await @@ -6376,6 +6484,223 @@ mod test { assert_eq!(subscriber.largest().await, Some(Position { group: 8, frame: 0 })); } + /// Wakes record that a parked poll was woken. + struct Flag(std::sync::atomic::AtomicBool); + + impl futures::task::ArcWake for Flag { + fn wake_by_ref(flag: &Arc) { + flag.0.store(true, std::sync::atomic::Ordering::SeqCst); + } + } + + impl Flag { + fn waiter() -> (Arc, kio::Waiter) { + let flag = Arc::new(Self(Default::default())); + let waiter = kio::Waiter::new(futures::task::waker(flag.clone())); + (flag, waiter) + } + + fn take(&self) -> bool { + self.0.swap(false, std::sync::atomic::Ordering::SeqCst) + } + } + + /// Write group `sequence` with `frames` frames, left open. + fn framed(producer: &Producer, sequence: u64, frames: usize) -> group::Producer { + let mut group = producer.create_group(group::Info { sequence }).unwrap(); + for _ in 0..frames { + group.write_frame(None, &b"x"[..]).unwrap(); + } + group + } + + /// The largest reports only while it is within the bounds: a cap below it or a floor + /// above it reports nothing, rather than a cached group standing in for it. + #[test] + fn largest_reports_only_within_the_bounds() { + let producer = track_producer("test", Info::default()); + let noop = kio::Waiter::noop(); + let _groups: Vec<_> = (0..6).map(|sequence| framed(&producer, sequence, 2)).collect(); + + let mut capped = producer.subscribe(replay()); + capped.set_groups(..3); + assert!(capped.poll_largest(&noop).is_pending(), "the largest is past the cap"); + + // An inclusive cap reports the group it names. + let mut ordered = producer.subscribe(replay()).ordered(); + ordered.set_groups(..=5); + assert_eq!( + ordered.poll_largest(&noop), + Poll::Ready(Some(Position { group: 5, frame: 1 })) + ); + + let mut floored = producer.subscribe(replay()); + floored.set_groups(8..); + assert!( + floored.poll_largest(&noop).is_pending(), + "the largest is below the floor" + ); + let _late = framed(&producer, 8, 0); + assert_eq!(floored.poll_largest(&noop), Poll::Ready(Some(Position::group(8)))); + } + + /// Moving the bounds wakes a parked reader, which reports the largest once it is within + /// them; bounds that leave it where it was report nothing new. + #[test] + fn largest_follows_the_bounds_as_they_move() { + let producer = track_producer("test", Info::default()); + let noop = kio::Waiter::noop(); + let _groups: Vec<_> = (0..6).map(|sequence| framed(&producer, sequence, 1)).collect(); + + let mut reader = producer.subscribe(replay()).ordered(); + reader.set_groups(..3); + let (flag, waiter) = Flag::waiter(); + assert!(reader.poll_largest(&waiter).is_pending()); + reader.set_groups(..); + assert!(flag.take(), "lifting the cap woke the reader"); + assert_eq!(reader.poll_largest(&noop), Poll::Ready(Some(Position::group(5)))); + + assert!(reader.poll_largest(&waiter).is_pending()); + reader.set_groups(..2); + assert!(flag.take(), "lowering the cap woke the reader"); + assert!( + reader.poll_largest(&noop).is_pending(), + "the largest is past the new cap" + ); + + reader.set_groups(..); + assert!( + reader.poll_largest(&noop).is_pending(), + "lifting it again leaves the last report standing" + ); + reader.set_groups(3..); + assert!( + reader.poll_largest(&noop).is_pending(), + "a floor below the largest leaves it where it was" + ); + } + + /// Once the stream ended, widening the bounds over a largest it never reported does not + /// bring it back. + #[test] + fn largest_stays_ended_when_the_bounds_widen() { + let producer = track_producer("test", Info::default()); + framed(&producer, 5, 1).finish().unwrap(); + producer.finish().unwrap(); + let noop = kio::Waiter::noop(); + + let mut reader = producer.subscribe(replay()); + reader.set_groups(..3); + assert_eq!(reader.poll_largest(&noop), Poll::Ready(None)); + reader.set_groups(..); + assert_eq!( + reader.poll_largest(&noop), + Poll::Ready(None), + "the stream already ended" + ); + } + + /// A dropped poll leaves no registration behind for a bounds move to wake. + #[test] + fn largest_forgets_a_dropped_poll() { + let producer = track_producer("test", Info::default()); + let mut reader = producer.subscribe(replay()); + let (flag, waiter) = Flag::waiter(); + assert!(reader.poll_largest(&waiter).is_pending()); + drop(waiter); + reader.set_groups(..); + assert!(!flag.take(), "the dropped poll was not woken"); + } + + /// A poll that reported is not parked, so moving the bounds afterwards does not wake it. + #[test] + fn largest_wakes_only_a_parked_poll_on_a_bounds_move() { + let producer = track_producer("test", Info::default()); + let _group = framed(&producer, 0, 1); + let mut reader = producer.subscribe(replay()); + let (flag, waiter) = Flag::waiter(); + assert_eq!(reader.poll_largest(&waiter), Poll::Ready(Some(Position::group(0)))); + reader.set_groups(..); + assert!(!flag.take(), "the finished poll was not woken"); + + assert!(reader.poll_largest(&waiter).is_pending()); + reader.set_groups(..); + assert!(flag.take(), "the parked poll was woken"); + } + + /// Bounds that move while the copy is idle keep the last report, so going live again + /// does not repeat it. + #[test] + fn largest_keeps_its_last_report_across_an_idle_move() { + let mut producer = track_producer("test", Info::default()); + let noop = kio::Waiter::noop(); + let _groups: Vec<_> = (0..3).map(|sequence| framed(&producer, sequence, 1)).collect(); + + let mut reader = producer.subscribe(replay()); + assert_eq!(reader.poll_largest(&noop), Poll::Ready(Some(Position::group(2)))); + producer.set_idle(); + reader.set_groups(..); + assert!(reader.poll_largest(&noop).is_pending(), "the copy is idle"); + + producer.set_live(Some(Position::group(2))); + assert!( + reader.poll_largest(&noop).is_pending(), + "the same largest is not reported again" + ); + let _next = framed(&producer, 3, 1); + assert_eq!(reader.poll_largest(&noop), Poll::Ready(Some(Position::group(3)))); + } + + /// An idle copy that declared an end it has not reached yet still waits for the live + /// feed, while one that finished reports its settled largest. + #[test] + fn largest_waits_for_an_idle_copy_with_a_future_end() { + let mut producer = track_producer("test", Info::default()); + let noop = kio::Waiter::noop(); + let _groups: Vec<_> = (0..3).map(|sequence| framed(&producer, sequence, 1)).collect(); + producer.finish_at(Position::group(10)).unwrap(); + producer.set_idle(); + + let mut reader = producer.subscribe(replay()); + assert!(reader.poll_largest(&noop).is_pending(), "the copy is idle"); + producer.set_live(Some(Position::group(5))); + let _answer = framed(&producer, 5, 1); + assert_eq!(reader.poll_largest(&noop), Poll::Ready(Some(Position::group(5)))); + + let mut finished = track_producer("test", Info::default()); + framed(&finished, 0, 1).finish().unwrap(); + finished.set_idle(); + finished.finish().unwrap(); + let mut reader = finished.subscribe(replay()); + assert_eq!(reader.poll_largest(&noop), Poll::Ready(Some(Position::group(0)))); + } + + /// An idle copy's largest is what an ended feed left, so nothing reports until the + /// route's answer shows, though the cache still holds groups. + #[test] + fn largest_waits_for_an_idle_copy_to_go_live() { + let mut producer = track_producer("test", Info::default()); + let noop = kio::Waiter::noop(); + let _groups: Vec<_> = (0..3).map(|sequence| framed(&producer, sequence, 1)).collect(); + producer.set_idle(); + + let mut reader = producer.subscribe(replay()); + let (flag, waiter) = Flag::waiter(); + assert!(reader.poll_largest(&waiter).is_pending(), "the copy is idle"); + + producer.set_live(Some(Position::group(5))); + producer.report_largest(Position::group(5)); + assert!( + reader.poll_largest(&waiter).is_pending(), + "the answer has not shown yet" + ); + flag.take(); + + let _answer = framed(&producer, 5, 1); + assert!(flag.take(), "the answer showing woke the reader"); + assert_eq!(reader.poll_largest(&noop), Poll::Ready(Some(Position::group(5)))); + } + #[test] #[cfg(feature = "async")] fn append_datagram_roundtrip() { diff --git a/rs/moq-room/src/chat.rs b/rs/moq-room/src/chat.rs index e80316df8e7..22ebb790c6c 100644 --- a/rs/moq-room/src/chat.rs +++ b/rs/moq-room/src/chat.rs @@ -128,7 +128,14 @@ impl Publisher { /// Reads changes to a participant's retained chat window. pub struct Subscriber { - consumer: moq_json::window::Consumer, + state: State, +} + +enum State { + /// Waiting for the newest window, so reading starts there rather than replaying older ones. + /// Taken only while moving to `Reading`. + Positioning(Option>), + Reading(Box>), } impl Subscriber { @@ -137,20 +144,30 @@ impl Subscriber { Ok(Self::new(broadcast.track(TRACK)?.subscribe(None).await?)) } - /// Read window changes from an existing subscription. - pub fn new(mut track: track::Subscriber) -> Self { - // Start at the newest window known now rather than replaying older ones. - if let Poll::Ready(Some(largest)) = track.poll_largest(&kio::Waiter::noop()) { - track.set_groups(largest.group..); - } + /// Read window changes from an existing subscription, starting at the newest window. + pub fn new(track: track::Subscriber) -> Self { Self { - consumer: moq_json::window::Consumer::new(track, Default::default()), + state: State::Positioning(Some(Box::new(track))), } } /// Poll the next window change, returning errors separately from clean completion. pub fn poll_recv(&mut self, waiter: &kio::Waiter) -> Poll, Error>> { - Poll::Ready(Ok(ready!(self.consumer.poll_next(waiter))?)) + if let State::Positioning(pending) = &mut self.state { + let track = pending.as_mut().expect("positioning holds the track"); + // The newest window is known only once the feed is live: a relay copy still waiting + // on its source reports nothing yet, so wait rather than replay its backlog. The + // wait ends with the track. + if let Some(largest) = ready!(track.poll_largest(waiter)) { + track.set_groups(largest.group..); + } + let track = pending.take().expect("positioning holds the track"); + self.state = State::Reading(Box::new(moq_json::window::Consumer::new(*track, Default::default()))); + } + let State::Reading(consumer) = &mut self.state else { + unreachable!("positioned above"); + }; + Poll::Ready(Ok(ready!(consumer.poll_next(waiter))?)) } /// Wait for the next window change, or None on clean completion. @@ -203,6 +220,103 @@ mod tests { assert!(matches!(subscriber.recv().await, Err(Error::Json(_)))); } + /// An origin at `hop`, its driver spawned. + fn origin(hop: u64) -> moq_net::origin::Producer { + let config = moq_net::origin::Config::new(moq_net::Hop::new(hop).unwrap()); + let (producer, driver) = moq_net::origin::Producer::new(config); + tokio::spawn(moq_net::time::run(driver)); + producer + } + + /// A lite-07 session over memory carrying `from`'s broadcasts into `to`, open while held. + async fn link( + from: &moq_net::origin::Producer, + to: &moq_net::origin::Producer, + ) -> (moq_net::Session, moq_net::Session) { + use moq_net::transport::memory; + + let version: moq_net::Version = "moq-lite-07-wip".parse().unwrap(); + let mut config = memory::Config::default(); + config.protocol = Some(version.alpn().to_string()); + let (client_transport, server_transport) = memory::pair(config); + let client = moq_net::Client::new() + .with_versions(version.into()) + .with_publisher(from.consume()); + let server = moq_net::Server::new() + .with_versions(version.into()) + .with_subscriber(to.clone()); + tokio::join!( + async { + let (session, driver) = client + .connect_lite(moq_net::time::now(), client_transport) + .await + .unwrap(); + tokio::spawn(moq_net::time::run(driver)); + session + }, + async { + let (session, driver) = server + .accept_lite(moq_net::time::now(), server_transport) + .await + .unwrap(); + tokio::spawn(moq_net::time::run(driver)); + session + } + ) + } + + /// A relay copy whose upstream subscription went idle keeps the windows it cached. A new + /// reader waits for the copy to go live and starts at its newest window, rather than + /// replaying a message that has since left it. + #[tokio::test(start_paused = true)] + async fn a_warm_copy_starts_at_the_newest_window() { + let publisher = origin(1); + let broadcast = publisher.create_broadcast("room").unwrap(); + let track = broadcast.create_track(TRACK, info()).unwrap(); + broadcast.announce(Default::default()).unwrap(); + let mut demand = track.demand(); + let mut chat = moq_json::window::Producer::::new( + track, + moq_json::window::ProducerConfig::default().with_op_ratio(0), + ); + let at = |millis| moq_net::Timestamp::from_millis(millis).unwrap(); + chat.push(moq_net::Timed::from(&"first".to_owned()).at(at(0))).unwrap(); + + let relay = origin(2); + let _link = link(&publisher, &relay).await; + let consumer = relay.consume(); + consumer.routed("room").await.unwrap(); + let remote = consumer.request_broadcast("room", None).await.unwrap(); + + // A first reader warms the relay's copy with the first window. + { + let mut warm = Subscriber::subscribe(&remote).await.unwrap(); + assert_eq!( + warm.recv().await.unwrap(), + Some(Event::Push { + index: 0, + value: "first".into() + }) + ); + } + // The relay cancels the unread upstream subscription; its copy keeps the window. + while demand.wanted_changed().await.unwrap().is_some() {} + chat.pop(moq_net::Timed::from(1).at(at(1))).unwrap(); + chat.push(moq_net::Timed::from(&"second".to_owned()).at(at(1))).unwrap(); + + // A delay budget long enough that the backlog is still worth delivering. + let subscription = track::Subscription::default().with_max_delay(Duration::from_secs(10)); + let mut late = Subscriber::new(remote.track(TRACK).unwrap().subscribe(subscription).await.unwrap()); + assert_eq!( + late.recv().await.unwrap(), + Some(Event::Push { + index: 1, + value: "second".into() + }), + "the warm copy's backlog was replayed" + ); + } + #[tokio::test] async fn propagates_track_failure() { let mut broadcast = broadcast::Info::new().produce(); diff --git a/rs/moq-video/src/decode/consumer.rs b/rs/moq-video/src/decode/consumer.rs index 85681fee175..e982a8ac425 100644 --- a/rs/moq-video/src/decode/consumer.rs +++ b/rs/moq-video/src/decode/consumer.rs @@ -39,7 +39,8 @@ pub enum Start { /// backlog at decode speed before reaching live media, which a viewer sees /// as playback jumping backwards and then sprinting to catch up. A /// rendition with a catalog `warmup` also decodes the group before it, for - /// the max-delay decoder (see [`Consumer`]). + /// the max-delay decoder (see [`Consumer`]). Opening waits for the track's + /// live edge, or its end. Latest, } @@ -287,8 +288,12 @@ impl Reader { // aggregated across every live subscriber, so naming a stale cached // sequence there asks the publisher to rewind the track for everyone // reading it. What a player wants is to skip what it already has. + // + // The live edge is known only once the feed is live: a warm copy still waiting on + // its source reports nothing yet, so wait for it rather than read the backlog. The + // wait ends with the track. if options.start == Start::Latest - && let std::task::Poll::Ready(Some(live_edge)) = subscriber.poll_largest(&moq_net::kio::Waiter::noop()) + && let Some(live_edge) = subscriber.largest().await { subscriber.set_groups(live_edge.group.saturating_sub(lookback)..); }