diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index d15e0c74a82..9e6d534668d 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -197,7 +197,7 @@ Each subscription carries the knobs that decide behavior under congestion: | --- | --- | | **Priority** (0..255) | Higher-priority tracks get bandwidth first. Audio above video, base layer above enhancement. | | **Order** | Which group to send first when several are pending. Newest first for live (the default), oldest first for catch-up. Needs moq-lite 07 or moq-transport; older moq-lite always sends newest first. | -| **Max delay** | How far a non-latest group may fall behind the live edge before it is skipped. Zero means "live edge only", and raising it is also what asks for history. | +| **Max delay** | How long a non-latest group stays worth delivering before it is skipped. Zero means "live edge only", and raising it is also what asks for history. | Order only decides what goes first, never what is skipped, so catching up oldest first needs a max delay that covers the backlog. When a relay serves subscribers of @@ -205,16 +205,19 @@ both orders from one upstream subscription, it asks upstream for newest first, a still sends each subscriber its groups in that subscriber's own order. Over moq-transport the upstream order is fixed by whichever subscriber arrived first. -Max delay is measured on the media timeline, not the wall clock, so a backlog -delivered as a burst is still late while a congestion stall never expires -anything on its own. Both ends apply it: the publisher skips a group rather -than sending it, and the subscriber skips it again as it reads, since the -publisher only ever sees the most tolerant budget across its subscribers. +Max delay starts counting when a group's successor arrives, on two clocks, and +whichever reaches it first skips the group: the wall clock since the successor +arrived at this hop, and how far the successor's first timestamp trails the +live edge. Media time catches a backlog delivered as a burst; the wall clock +catches a publisher whose timestamps stall and an untimed track. The newest +group has no successor, so it is never skipped. Both ends apply it: the +publisher skips a group rather than sending it, and the subscriber skips it +again as it reads, since the publisher only ever sees the most tolerant budget +across its subscribers. A track is timed or untimed: it declares a timescale or none, and a frame that -doesn't match is refused. No receiver fills in an arrival time. An untimed -group never falls past max delay, so a new subscriber starts at its latest -group unless it names a start. Tracks from moq-lite before 05, or from +doesn't match is refused. No receiver fills in an arrival time, so an untimed +track is judged by the wall clock alone. Tracks from moq-lite before 05, or from moq-transport without a `TIMESCALE` (every track on drafts 14-16), arrive untimed. moq-lite 05 and 06 cannot declare an untimed track, so there its frames go out stamped with their send time; the work-in-progress 07 carries it @@ -228,9 +231,9 @@ nothing from it until the source answers again. 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 evict sooner under their own cache ceiling. Media tracks explicitly use 30 -seconds so a segmented egress can still find its segments. Retention uses media -timestamps and always keeps the newest group. IETF carries the same idea as -`MAX_CACHE_DURATION`, measured in wall time, which is only an approximate match. +seconds so a segmented egress can still find its segments. Retention ages a +group by the same two clocks as max delay and always keeps the newest group. +IETF carries the same idea as `MAX_CACHE_DURATION`, measured in wall time alone. Put together, a conference might use: diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 4abaa4c7082..b2b4de52a09 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -43,7 +43,7 @@ broadcast.announce(); - **One send estimate per connection.** `Bandwidth.Allocator` divides it by track priority, max-min fair within a tier. An idle track claims nothing. Publishers reserve against it so their targets sum to the estimate. - **One track per name.** Concurrent subscriptions share one request and producer. `createTrack` on a name with a request still queued answers it. The application owns sequence numbers: a track that replaces an ended one, created or requested, throws on an append that would reuse a sequence the name already wrote, so it writes its first group with an explicit sequence. Only a new broadcast starts over. - **Timing is per track, with no default.** A track that declares a `timescale` carries a `timestamp` on every frame; one without is [untimed](/concept/moq-lite#subscriptions) and its frames carry none. Omitting it means untimed, as in Rust. Nothing stamps a frame for you: the publisher passes the timestamp, and the `Timed` type (`{ value, at }`) carries one into the `@moq/json` and `@moq/flate` producers. -- **Subscriber staleness is `maxDelay`.** It is media time. Passing the old `maxAge` key throws a `TypeError` naming `maxDelay`. Publisher retention is the separate `Track.Info.maxAge`. +- **Subscriber staleness is `maxDelay`.** It counts from a group's successor, by media time and by wall clock, whichever reaches it first. Passing the old `maxAge` key throws a `TypeError` naming `maxDelay`. Publisher retention is the separate `Track.Info.maxAge`. - **Hidden paths** stay out of discovery unless the announce request opts in. See [hidden broadcasts](/concept/moq-lite#hidden-broadcasts). - **Authorization is in band.** On moq-lite 07 (offered only when `moq-lite-07-wip` is listed in `webtransport.protocols`; the WebSocket fallback cannot offer it) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `auth.grant` watches what the relay lets this side publish and subscribe to, and `auth.add(token)` presents another token without reconnecting. Publishing outside the grant closes the connection with `SessionCode.Unauthorized`, naming the path. On moq-lite, a subscription the grant stops covering resets with `StreamCode.Unauthorized` and the session stays up. - **A graceful close waits.** `await connection.close()` withdraws announcements and gives finished tracks up to one second. `abort()` ends immediately. diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index bc04030018c..f06b7c73c60 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -27,6 +27,14 @@ These land with the next breaking release, not the 2026-09-23 train. now take effect on parsed and merged configs unless the flag or environment variable was actually given. TOML keys, C ABI, binding records, and wire formats are unchanged. +- **Max age and max delay count wall-clock time too.** A superseded group is + stale once its successor arrived a full budget ago, as well as by media time, + so an untimed track is no longer exempt: with the default zero `max_delay`, + an untimed subscriber starts at the latest group and skips each superseded + one. Give it a budget to replay or tolerate reordering. +- **`cache::Pool::gc` takes a waiter.** In Rust, `gc(now)` is + `gc(now, waiter)`, like `Driver::poll`: call it again when the waiter wakes, + since a parked read can arm an earlier deadline than the one returned. - **A QUIC load-balancer nonce needs its server id.** `listen.lb_nonce`, `--listen-quic-lb-nonce`, and `MOQ_LISTEN_QUIC_LB_NONCE` without `lb_id` stop startup. Setting `lb_id` (or `lb_nonce`) and `load_balancer` together also diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 8ac3a301bc4..7eb866b76de 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -632,19 +632,23 @@ A subscriber MUST support gaps and out-of-order delivery regardless. ## Expiration Expiration governs when an older group is dropped. -The publisher SHOULD reset Group Streams for non-latest groups whose age relative to the latest group reaches `Subscriber Max Age` (see [SUBSCRIBE](#subscribe), and the age definition below); the subscriber MAY also locally drop such groups. +The publisher SHOULD reset Group Streams for non-latest groups whose age reaches `Subscriber Max Age` (see [SUBSCRIBE](#subscribe), and the age definition below); the subscriber MAY also locally drop such groups. Expiration only removes the group from live delivery; the publisher MAY still retain it for FETCH or new subscriptions until its age exceeds `Publisher Max Age` (see [TRACK_INFO](#track-info)). It is not crucial to aggressively expire groups thanks to [prioritization](#prioritization), but a lower priority group still consumes RAM, bandwidth, and potentially flow control. It is RECOMMENDED that an application set conservative limits and only resort to expiration when data is absolutely no longer needed. -A group is never expired until a later group (by sequence number) has presented a frame. -A group of an untimed Track has no timestamps, so it is never expired. -Once one has, the group's **timestamp age** is the difference between the *newest* frame timestamp of the latest group that has at least one frame, and this group's **reach**, defined below; the group is expired once that age meets the relevant `Max Age`. -Timestamps are the only measure: they are consistent across relays and unaffected by buffering or jitter. -Wall-clock reclamation of idle content is the retention cache's own policy (`Publisher Max Age`, or any implementation-defined bound), not part of this rule. +A group is never expired until its **successor** has arrived: the next group by sequence number that has arrived at this hop and was not reset. +The latest group has no successor, so it is never expired. +A group is expired once either of two ages meets the relevant `Max Age`: -A group is expired only once it **provably cannot overlap the Max Age window**: every frame it could still present is older than the budget allows. +- Its **timestamp age** is the difference between the *newest* frame timestamp of the latest group that has at least one frame, and this group's **reach**, defined below. +- Its **arrival age** is the time elapsed since its successor arrived at this hop. + +A group of an untimed Track has no timestamps, so arrival age alone expires it. + +By timestamp age, a group is expired only once it **provably cannot overlap the Max Age window**: every frame it could still present is older than the budget allows. +Arrival age needs no such proof: it bounds how long a superseded group may still wait to be delivered. Being behind is not itself a reason to expire anything. [Prioritization](#prioritization) already transmits newer groups first, so an older group consumes only whatever capacity is left over; it therefore closes the gap faster than the live edge advances, and a receiver that is behind can converge without losing content. What cannot be recovered is a group with nothing left worth delivering, and that is what expiration removes. @@ -652,13 +656,14 @@ What cannot be recovered is a group with nothing left worth delivering, and that A group's **reach** is the first frame timestamp of the next group by sequence number, or unbounded when no later group has presented a frame. A group cannot present past where its successor begins, so this is the furthest it could still reach. Its own frames do not bound it: frame durations are not carried on the wire, so a group's last frame timestamp is where that frame *starts* presenting, not where the group ends. -A group whose successor has not presented a frame is therefore never expired, because nothing yet proves where it stops. +A group whose successor has not presented a frame therefore has no timestamp age, because nothing yet proves where it stops. The group itself needs no timestamp: a zero-frame group (a keep-alive or gap marker) is bounded by its successor's first frame the same way. Reach is an exclusive bound, so a timestamp age **equal** to Max Age already expires the group: the freshest frame it could still hold sits strictly below its reach, and is therefore strictly older than the budget. A Max Age of zero follows from the same rule without a special case. -Measuring in timestamps rather than arrival means a burst of groups delivered together still reads as its true age, so catching up never resets the clock. +Timestamp age means a burst of groups delivered together still reads as its true age, so catching up never resets the clock. +Arrival age bounds what timestamps cannot: a publisher whose timestamps stop advancing, a successor that never presents a frame, an untimed Track, and a run of empty groups. An expired group SHOULD be reset at the QUIC level to avoid consuming flow control. @@ -1118,7 +1123,7 @@ See the [Group Order](#group-order) section for more information. **Subscriber Max Age**: The subscriber's preference, in milliseconds, for how long a non-latest group may remain in flight before being considered stale and dropped from live delivery. -The publisher SHOULD reset (at the QUIC level) Group Streams for groups whose age relative to the latest group exceeds this duration. +The publisher SHOULD reset (at the QUIC level) Group Streams for groups whose age, as defined in [Expiration](#expiration), reaches this duration. Applies only to non-latest groups; the latest group is never dropped on staleness grounds. A value of `0` means the subscriber wants only the latest group in live delivery (older groups are immediately stale once a newer group arrives). This is a delivery-time preference, not a retention rule: the publisher MAY still hold these groups for FETCH or future subscriptions (see `Publisher Max Age` in [TRACK_INFO](#track-info)). @@ -1221,7 +1226,7 @@ The publisher's priority for this Track, represented as a u8, used only to resol See the [Prioritization](#prioritization) section for more information. **Publisher Max Age**: -The maximum age, in milliseconds, that the publisher caches a non-latest group past the arrival of a newer group. +The maximum age, in milliseconds, that the publisher caches a non-latest group, with age as defined in [Expiration](#expiration). Applies only to non-latest groups; the latest group is always retained. It is an upper bound on retention, the inverse of an HTTP `Cache-Control: max-age` guarantee: @@ -1568,7 +1573,7 @@ The `Message Length` describes the payload size on the wire. - Added `Epoch` to ANNOUNCE_START, SUBSCRIBE, TRACK, and FETCH: a UUIDv7 naming the publisher instance, or empty. A path and an Epoch name one Broadcast. A relay MAY move a subscription between routes with the same Epoch, continuing from the first frame the subscriber lacks instead of at a group boundary, and never between routes with different Epochs or none. The newest Epoch wins a path for new requests, while subscriptions to the older one stay until their route goes. Replaces the first-hop identity. - Added ANNOUNCE_RESTART (0x3): another publisher instance replaces a live advertisement, under a new Epoch or, without one, through another route, including a path beneath a prefix moving to another route. Older versions send ANNOUNCE_END then ANNOUNCE_START for it. - An untimed Track declares a `Timescale` of 0 in TRACK_INFO, and its FRAME messages and datagrams omit the timestamp. lite-05 and lite-06 publishers declare 1000 for it and send each payload's send time instead. -- A group of an untimed Track is never expired. +- Expiration also measures a group's arrival age, the time since its successor arrived at this hop, so a group of an untimed Track expires by arrival age alone. - An open Track Stream is interest in the Track until the subscriber closes it, which a subscriber does once its Subscribe Stream has a response. lite-05 and lite-06 implementations do the same. - Assigned `moq-lite-07-wip` as this draft's protocol identifier until it is finalized as `moq-lite-07`. diff --git a/drafts/draft-lcurley-moq-timestamp.md b/drafts/draft-lcurley-moq-timestamp.md index 3d95a1e87c6..a1aa9b5a9fc 100644 --- a/drafts/draft-lcurley-moq-timestamp.md +++ b/drafts/draft-lcurley-moq-timestamp.md @@ -119,8 +119,8 @@ A track is either timed or untimed for its whole lifetime; no track mixes timed A receiver MUST NOT substitute a time for an untimed object, such as its arrival time, and a relay MUST forward an untimed track untimed, declaring no TIMESCALE downstream. Arrival time differs at every hop and changes when a route fails over, so a substituted time would not be the same timeline downstream. -Age-based dropping ({{age}}) never drops an untimed object; a relay bounds how long it keeps one with its own cache policy. -A new subscription to an untimed track that names no start begins at the latest group, since no age says how far back to reach. +Age-based dropping ({{age}}) does not apply to an untimed object. +A relay MAY instead drop a group of an untimed track once a later group arrived at it longer ago than the threshold, measured on its own clock. ## Age-Based Dropping {#age} Given two objects on the same track, both with TIMESTAMP, a relay computes their relative age as the difference of their timestamps divided by the Timescale. @@ -162,10 +162,9 @@ An earlier version of this document requested its own code points (`0x915C0` and ## moq-timestamp-02 {:numbered="false"} -- Replaced the wall-clock arrival-time fallback with untimed tracks: a track without TIMESCALE has no media time, its objects are never dropped for age, and a relay forwards it untimed. +- Replaced the wall-clock arrival-time fallback with untimed tracks: a track without TIMESCALE has no media time, and a relay forwards it untimed. A relay MAY drop one of its groups once a later group arrived longer ago than the threshold. - On a track with TIMESCALE, a publisher MUST attach TIMESTAMP to every object with a payload, and a receiver treats one without it as malformed. - A receiver ignores an object-scope TIMESCALE, and no longer reads a bare TIMESTAMP on a track without TIMESCALE as microseconds. -- A new subscription to an untimed track that names no start begins at the latest group. # Acknowledgments {:numbered="false"} diff --git a/js/net/src/group.ts b/js/net/src/group.ts index 5121131b9c1..fda4e4f31a6 100644 --- a/js/net/src/group.ts +++ b/js/net/src/group.ts @@ -68,6 +68,10 @@ class GroupState { timestamp?: Timestamp; // The newest frame's timestamp: where a reader that has taken every frame sits. latest?: Timestamp; + // When the group joined the track this mirror belongs to, at this hop, in + // `performance.now()` milliseconds: the wall clock a superseded predecessor ages from. + // Per track, since one group can join several. + arrived = 0; constructor(sequence: number) { this.sequence = sequence; @@ -428,6 +432,10 @@ export class Consumer { makeConsumer = (state) => new Consumer(state); hooks.groupTimestamp = (group) => group.#state.timestamp; hooks.groupLatest = (group) => group.#state.latest; + hooks.groupArrived = (group) => group.#state.arrived; + hooks.arriveGroup = (group, at) => { + group.#state.arrived = at; + }; hooks.expireGroup = (group, expiry) => { group.#expiry = expiry; }; diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index 89b00c3bf26..a1cd3924c6e 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -8,7 +8,7 @@ import { createMockTransportPair } from "../mock.ts"; import * as Path from "../path.ts"; import { Reader, Stream } from "../stream.ts"; import { Tail } from "../tail.ts"; -import { Timescale, Timestamp } from "../time.ts"; +import { Milli, Timescale, Timestamp } from "../time.ts"; import type * as track from "../track.ts"; import { ControlStreamAdapter, NativeSession } from "./adapter.ts"; import type * as Cluster from "./cluster.ts"; @@ -960,15 +960,17 @@ function encodeObjects(deltas: number[]): Uint8Array { async function subscribeTrack({ version = VERSION, properties = {}, + subscription, }: { version?: IetfVersion; properties?: Properties; + subscription?: track.Subscription; } = {}): Promise<{ subscriber: Subscriber; track: track.Subscriber }> { const pair = createMockTransportPair(version === Version.DRAFT_16 ? ALPN.DRAFT_16 : ALPN.DRAFT_19); const session = new NativeSession(pair.server, version, true); const subscriber = new Subscriber({ session }); - const track = subscriber.consume(Path.from("room")).track("video").subscribe(); + const track = subscriber.consume(Path.from("room")).track("video").subscribe(subscription); const peer = await nextStream(pair.client); if (!peer) throw new Error("the subscriber never opened a subscribe stream"); @@ -1447,7 +1449,8 @@ test.each([ [VERSION, true, [3, 4]], [Version.DRAFT_16, true, [3, 4]], ])("draft %s with FIRST_OBJECT=%s: a header-only stream delivers %p", async (version, firstObject, expected) => { - const { subscriber, track } = await subscribeTrack({ version }); + // A budget, so the empty group outlives its successor's arrival. + const { subscriber, track } = await subscribeTrack({ version, subscription: { maxDelay: Milli(60_000) } }); await subscriber.handleGroup( new GroupMessage({ diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 25452369c4f..4b900082866 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -223,6 +223,10 @@ export const hooks: { /** Return a group's first timestamp, retained even after its first frame is read. */ groupTimestamp: (group: GroupConsumer) => Timestamp | undefined; groupLatest: (group: GroupConsumer) => Timestamp | undefined; + /** Return when a group joined its track at this hop, in `performance.now()` milliseconds. */ + groupArrived: (group: GroupConsumer) => number; + /** Record when a track's mirror of a group joined that track at this hop. */ + arriveGroup: (group: GroupConsumer, at: number) => void; /** Keep applying a subscription's drift policy after it hands a group out. */ expireGroup: ( group: GroupConsumer, @@ -277,12 +281,18 @@ export const hooks: { limitGroup: () => { throw new Error("group.ts not loaded"); }, + arriveGroup: () => { + throw new Error("group.ts not loaded"); + }, groupTimestamp: () => { throw new Error("group.ts not loaded"); }, groupLatest: () => { throw new Error("group.ts not loaded"); }, + groupArrived: () => { + throw new Error("group.ts not loaded"); + }, expireGroup: () => { throw new Error("group.ts not loaded"); }, diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 5821d85e86e..d5d03967f5c 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -557,16 +557,17 @@ test("the latency budget skips a buffered timeline on either cursor", async () = // is the unsafe direction. test("an unstamped immediate successor leaves reach unbounded", async () => { const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI, maxAge: Milli(5000) }); - const track = producer.subscribe(); + // Any budget, so only media time judges: the wall clock barely moves here. + const track = producer.subscribe({ maxDelay: Milli(500) }); producer.writeFrame({ payload: enc.encode("0"), timestamp: Timestamp.fromMillis(0) }); producer.appendGroup(); // seq 1 stalls before its first frame producer.writeFrame({ payload: enc.encode("2"), timestamp: Timestamp.fromMillis(10_000) }); + producer.writeFrame({ payload: enc.encode("3"), timestamp: Timestamp.fromMillis(11_000) }); - // Group 1's reach is group 2's start, a full edge behind: stale at zero budget. + // Group 1's reach is group 2's start, a second behind the edge: stale. // Group 0's reach is unknown until group 1 presents its first frame, so it is kept. - expect((await track.recvGroup())?.sequence).toBe(0); - expect((await track.recvGroup())?.sequence).toBe(2); + expect(drain(track)).toEqual([0, 2, 3]); }); // Groups can arrive out of sequence order; the immediate successor bounds reach even when it @@ -1027,19 +1028,149 @@ test("a drained group finishes cleanly after the live edge advances", async () = expect(group.done).toBe(true); }); -test("a stalled track retains cached groups past its media maxAge", async () => { +// A stalled track keeps its newest group and gives up the ones before it by the wall clock: +// maxAge measures how long ago each group's successor arrived as well as media time, so a +// publisher that stops writing cannot pin its superseded groups. +test("a stalled track keeps only its newest group", () => { const clock = mockMonotonicTime(10_000); - const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI, maxAge: Milli(100) }); + const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI, maxAge: Milli(1000) }); try { - for (const value of ["first", "second"]) { - producer.writeFrame({ payload: enc.encode(value), timestamp: Timestamp.fromMillis(0) }); + for (const millis of [0, 10, 20]) { + producer.writeFrame({ payload: enc.encode(`${millis}`), timestamp: Timestamp.fromMillis(millis) }); } - // Media time has not advanced, even though wall-clock idleness exceeded maxAge. - clock.set(10_200); - const replay = producer.subscribe({ maxDelay: Milli(100) }); - expect(replay.tryRecvGroup()?.sequence).toBe(0); - expect(replay.tryRecvGroup()?.sequence).toBe(1); - replay.close(); + expect(drain(producer.subscribe({ maxDelay: Milli(1000) }))).toEqual([0, 1, 2]); + + // Nothing is written again: media time stands still. + clock.set(12_000); + expect(drain(producer.subscribe({ maxDelay: Milli(1000) }))).toEqual([2]); + } finally { + producer.close(); + clock.restore(); + } +}); + +// A peer whose timestamps stop advancing never ages a group out on media time. The wall +// clock still does. +test("frozen timestamps still age out by wall clock", () => { + const clock = mockMonotonicTime(10_000); + const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI }); + const frame = () => producer.writeFrame({ payload: enc.encode("x"), timestamp: Timestamp.fromMillis(0) }); + try { + frame(); + clock.set(10_500); + frame(); + clock.set(11_300); + frame(); + + // Group 0's successor arrived 800ms ago, group 1's just now. + expect(drain(producer.subscribe({ maxDelay: Milli(1000) }))).toEqual([0, 1, 2]); + clock.set(11_500); + expect(drain(producer.subscribe({ maxDelay: Milli(1000) }))).toEqual([1, 2]); + clock.set(12_500); + expect(drain(producer.subscribe({ maxDelay: Milli(1000) }))).toEqual([2]); + } finally { + producer.close(); + clock.restore(); + } +}); + +// Arrival is per track: a group joining a second track later doesn't restart the first +// track's clock. +test("a group shared by two tracks ages each from its own arrival", () => { + const clock = mockMonotonicTime(10_000); + const first = new TrackProducer("first").accept({}); + const second = new TrackProducer("second").accept({}); + try { + first.writeFrame({ payload: enc.encode("0") }); + const shared = new GroupProducer(1); + shared.writeFrame({ payload: enc.encode("1") }); + shared.close(); + first.writeGroup(shared); + clock.set(11_000); + second.writeGroup(shared); + + // Group 0's successor joined the first track a full budget ago. + expect(drain(first.subscribe({ maxDelay: Milli(1000) }))).toEqual([1]); + } finally { + first.close(); + second.close(); + clock.restore(); + } +}); + +// Groups that arrive at once are judged by media time, which convicts them before the wall +// clock can. +test("a startup burst ages out by media time", () => { + const clock = mockMonotonicTime(10_000); + const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI }); + try { + for (let second = 0; second < 5; second++) { + producer.writeFrame({ payload: enc.encode("x"), timestamp: Timestamp.fromMillis(second * 1000) }); + } + // Group N reaches N + 1 seconds against the 4s edge: groups 0 and 1 are 3s and 2s + // behind, past 1.5s. Their successors arrived just now. + expect(drain(producer.subscribe({ maxDelay: Milli(1500) }))).toEqual([2, 3, 4]); + } finally { + producer.close(); + clock.restore(); + } +}); + +// A read parked on a superseded group is released by the wall clock alone: nothing is +// written after its successor arrives, so no write could wake it. +test("a parked read wakes at its wall deadline", async () => { + for (const timescale of [undefined, Timescale.MILLI]) { + const clock = mockCacheTime(10_000); + const producer = new TrackProducer("test").accept({ timescale }); + try { + const track = producer.subscribe({ maxDelay: Milli(1000) }); + const held = producer.appendGroup(); + held.writeFrame({ payload: enc.encode("x"), timestamp: timescale && Timestamp.fromMillis(0) }); + const group = track.tryRecvGroup(); + if (!group) throw new Error("missing group"); + expect(await group.readString()).toBe("x"); + + // The successor never presents a frame, so media time never bounds the held group. + producer.appendGroup(); + let woken = false; + const read = group.readFrame().then((frame) => { + woken = true; + return frame; + }); + await Promise.resolve(); + + clock.advance(10_900); + await Promise.resolve(); + expect(woken).toBe(false); + clock.advance(11_000); + expect(await read).toBeUndefined(); + } finally { + producer.close(); + clock.restore(); + } + } +}); + +// Empty groups are handed out like any other and retired by the wall clock like an untimed +// one, whether they finished without a frame or are still open with none. A publisher +// flooding a run of them cannot pin the group before the run either. +test("a run of empty groups ages out by wall clock", () => { + const RUN = 1000; + const clock = mockMonotonicTime(10_000); + const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI }); + try { + const track = producer.subscribe({ maxDelay: Milli(1000) }); + producer.writeFrame({ payload: enc.encode("x"), timestamp: Timestamp.fromMillis(0) }); + for (let sequence = 1; sequence <= RUN; sequence++) { + const group = producer.appendGroup(); + if (sequence % 2 === 0) group.close(); + } + producer.writeFrame({ payload: enc.encode("x"), timestamp: Timestamp.fromMillis(10) }); + expect(drain(producer.subscribe({ maxDelay: Milli(1000) }))).toHaveLength(RUN + 2); + + clock.set(11_000); + expect(drain(track)).toEqual([RUN + 1]); + expect(drain(producer.subscribe({ maxDelay: Milli(1000) }))).toEqual([RUN + 1]); } finally { producer.close(); clock.restore(); @@ -1176,9 +1307,10 @@ test("an idle live edge ages out once a newer group arrives", async () => { test("an abandoned open group ages out with no further write", async () => { const clock = mockCacheTime(10_000); - const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI, maxAge: Milli(30) }); + const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI }); try { - const track = producer.subscribe({ maxDelay: Milli(30) }); + // A budget past the cache window, so the eviction rather than the budget ends it. + const track = producer.subscribe({ maxDelay: Milli(2 * CACHE_WINDOW_MS) }); const stalled = producer.appendGroup(); stalled.writeFrame(textFrame("first")); producer.appendGroup(); @@ -2191,34 +2323,33 @@ test("a datagram whose timedness disagrees with its track is refused", async () expect(got?.timestamp).toBeUndefined(); }); -test("an untimed track starts at its latest group", () => { - const producer = untimedTrack(5); - - // Nothing on an untimed track is ever stale, so no budget can say where to start. - const live = producer.subscribe(); - expect(drain(live)).toEqual([4]); - const patient = producer.subscribe({ maxDelay: Milli(60_000) }); - expect(drain(patient)).toEqual([4]); - - producer.writeFrame({ payload: enc.encode("5") }); - expect(drain(live)).toEqual([5]); -}); +// The wall clock judges an untimed track, so a new subscriber starts at the oldest cached +// group it doesn't find stale, the way a timed one does: a group is stale once its +// successor arrived a full budget ago. +test("an untimed subscriber starts at the oldest fresh group", () => { + const clock = mockMonotonicTime(10_000); + const producer = new TrackProducer("test").accept({}); + try { + // Group N's successor arrives N + 1 seconds in. + for (let i = 0; i < 5; i++) { + producer.writeFrame({ payload: enc.encode(`${i}`) }); + clock.set(10_000 + (i + 1) * 1000); + } -test("an unfloored subscriber made before accept starts at the latest group once untimed", () => { - const producer = new TrackProducer("test"); - for (let i = 0; i < 3; i++) producer.appendGroup().close(); - const early = producer.subscribe(); - const floored = producer.subscribe({ groups: { start: { included: 0 } } }); - const named = producer.subscribe(); - named.setGroups({ start: { included: 1 } }); + // Four seconds after group 4 arrived, groups 0 to 2 were superseded 4, 3, and 2 + // seconds ago: past a 1.5s budget. Group 3 was superseded 1s ago. + expect(drain(producer.subscribe({ maxDelay: Milli(1500) }))).toEqual([3, 4]); + // No budget takes the newest group alone, which is never stale. + const live = producer.subscribe(); + expect(drain(live)).toEqual([4]); + expect(drain(producer.subscribe({ maxDelay: Milli(60_000) }))).toEqual([0, 1, 2, 3, 4]); - producer.accept({}); - // Settled at accept(), so groups written before the first read still arrive. - producer.writeFrame({ payload: enc.encode("3") }); - producer.writeFrame({ payload: enc.encode("4") }); - expect(drain(early)).toEqual([2, 3, 4]); - expect(drain(floored)).toEqual([0, 1, 2, 3, 4]); - expect(drain(named)).toEqual([1, 2, 3, 4]); + producer.writeFrame({ payload: enc.encode("5") }); + expect(drain(live)).toEqual([5]); + } finally { + producer.close(); + clock.restore(); + } }); test("an untimed subscriber without a start begins at the latest group under its end", () => { @@ -2243,7 +2374,7 @@ test("accept closes the track when something written before it disagrees", async test("a single-frame write that throws aborts its group", () => { const producer = new TrackProducer("test").accept({}); - const subscriber = producer.subscribe(); + const subscriber = producer.subscribe({ maxDelay: Milli(60_000) }); // A timestamped frame is refused by an untimed track. expect(() => producer.writeFrame({ payload: enc.encode("x"), timestamp: Timestamp.fromMillis(1) })).toThrow( TimestampMismatch, @@ -2257,15 +2388,15 @@ test("a single-frame write that throws aborts its group", () => { expect(drain(subscriber)).toEqual([1]); }); -test("an explicit start holds on an untimed track, and nothing there is stale", () => { +// An explicit start is a floor, not a request: the budget still skips what it finds stale +// above it, as on a timed track. +test("an explicit start floors an untimed track", () => { const producer = untimedTrack(5); - // A named start says how far back to reach (a resume names the group it lacks), and a - // zero budget convicts nothing that has no place in media time. - const resumed = producer.subscribe({ groups: { start: { included: 2 } } }); + const resumed = producer.subscribe({ maxDelay: Milli(60_000), groups: { start: { included: 2 } } }); expect(drain(resumed)).toEqual([2, 3, 4]); - const replay = producer.subscribe({ groups: { start: { included: 0 } } }); - expect(drain(replay)).toEqual([0, 1, 2, 3, 4]); + const live = producer.subscribe({ groups: { start: { included: 2 } } }); + expect(drain(live)).toEqual([4]); }); test("a mid-group end survives eviction and a filtered late reader", () => { diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 8d7c04e080d..71dfc907e5f 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -76,11 +76,12 @@ export interface Info { */ timescale?: Timescale | null; /** - * Publisher Max Age: how far behind the live edge a group may fall in media - * timestamps, in milliseconds, before it is stale. Reported in TRACK_INFO (Lite05+) - * so relays re-serve with the same bound. Omission sets no publisher limit; zero - * serves only the live edge. A congestion stall cannot age content out; idle cache - * eviction has a separate wall-clock window. This clamps each subscriber budget. + * Publisher Max Age: how long, in milliseconds, a superseded group stays fresh. It is + * stale once either clock reaches the budget: the wall-clock time since its successor + * arrived at this hop, or how far its successor's first timestamp trails the live edge. + * The newest group is never stale, and an untimed track is judged by the wall clock alone. + * Reported in TRACK_INFO (Lite05+) so relays re-serve with the same bound. Omission sets + * no publisher limit; zero serves only the live edge. This clamps each subscriber budget. * Rounded up to a whole millisecond by {@link infoDefaults}, which refuses a negative * or non-finite value and a result past `Number.MAX_SAFE_INTEGER`. */ @@ -142,8 +143,9 @@ export interface Subscription { /** Delivery priority relative to this session's other subscriptions (`0..=255`). Defaults to `0`. */ priority?: number; /** - * How far (milliseconds) a non-latest group may fall behind the live edge before it is - * skipped. Defaults to `0`. + * How long (milliseconds) a non-latest group stays worth delivering, by the same two + * clocks as {@link Info.maxAge}: it is skipped once its successor arrived this long ago + * or starts this far behind the live edge. Defaults to `0`, the live edge only, timed or not. * Rounded up to a whole millisecond, so a value derived from a measurement is never * shortened. A negative or non-finite value, or one past `Number.MAX_SAFE_INTEGER` * after rounding, is refused. @@ -153,10 +155,8 @@ export interface Subscription { * The lowest group the publisher may deliver (a floor), or omit for none. * * A floor, not a request: only {@link maxDelay} asks for data, and the floor bounds how - * far back it may reach. Omitting it and a floor of 0 differ only on an untimed track, - * where nothing is ever stale: omitted starts at the latest group, and an explicit floor - * replays from there. A floor above the live edge simply waits there (a resumed - * subscription naming where it left off). + * far back it may reach. Omitting it is a floor of 0. A floor above the live edge simply + * waits there (a resumed subscription naming where it left off). */ groups?: Groups; /** @@ -345,8 +345,7 @@ export class Consumer { * The cursor starts at the group the subscription named (its floor), or 0. * {@link Subscription.maxDelay} is what asks for data: delivery skips everything above * the floor that the budget convicts, so the default budget of zero delivers only the - * latest group and a larger one reaches back over what it can still use. An untimed - * track has nothing to convict, so without a floor it starts at its latest group. + * latest group and a larger one reaches back over what it can still use. */ subscribe(options?: Subscription): Subscriber { return this.#broadcast.subscribe(this.name, options); @@ -418,9 +417,6 @@ class TrackState { update: Signal; /** Resolved once the producer commits the immutable properties. */ info = new Signal(undefined); - // Set by an unfloored subscriber made before accept(), which calls it to settle its - // start the moment the track's timedness is known. - settle?: () => void; constructor(subscription?: Subscription) { this.update = new Signal(subscription === undefined ? undefined : subscriptionDefaults(subscription)); @@ -456,7 +452,8 @@ async function resolveInfo(state: TrackState): Promise { // A source group retained in the producer cache, with the mirror handed to each sink // so eviction can drop them together. -type CachedGroup = { group: GroupProducer; mirrors: Map }; +// `arrived` is when the group joined this track, which its mirrors age a predecessor from. +type CachedGroup = { group: GroupProducer; mirrors: Map; arrived: number }; let bindProducerSequence: (producer: Producer, sequence: TrackSequence) => void; @@ -656,10 +653,7 @@ export class Producer { } this.#state.info.set(resolved); // Propagate to any sink handed out before accept (the on-demand path). - for (const sink of this.#sinks) { - sink.info.set(resolved); - sink.settle?.(); - } + for (const sink of this.#sinks) sink.info.set(resolved); this.#updateSubscription(); return this; } @@ -670,8 +664,7 @@ export class Producer { * Its cursor starts at the group the subscription named (its floor), or 0. * {@link Subscription.maxDelay} is what asks for data: delivery skips everything above * the floor that the budget convicts, so the default budget of zero delivers only the - * latest group and a larger one reaches back over what it can still use. An untimed - * track has nothing to convict, so without a floor it starts at its latest group. + * latest group and a larger one reaches back over what it can still use. */ subscribe(options: Subscription = {}): Subscriber { const sink = new TrackState(options); @@ -758,6 +751,7 @@ export class Producer { // reference. Tracked on the entry so eviction can drop it from the sink. #mirror(entry: CachedGroup, sink: TrackState): void { const dst = entry.group.mirror(); + hooks.arriveGroup(dst, entry.arrived); entry.mirrors.set(sink, dst); timelineInsert(sink.timeline, dst); void dst.readable().then(() => sink.timelineChanged.update((revision) => revision + 1)); @@ -896,7 +890,7 @@ export class Producer { // Retain a source group and fan it out to every live sink. #publish(group: GroupProducer): void { - const entry: CachedGroup = { group, mirrors: new Map() }; + const entry: CachedGroup = { group, mirrors: new Map(), arrived: performance.now() }; this.#cache.push(entry); this.#cached.set(group.sequence, entry); this.#advance(group.sequence); @@ -1184,6 +1178,13 @@ export class Subscriber { // The group the frame-level helpers are currently draining, acquired through the // sequence cursor so frame reads and {@link Ordered.nextGroup} share one floor. #frameGroup?: GroupConsumer; + // The earliest wall deadline a held read armed, and its timer, which bumps `revision` + // when it comes due. See #wakeAt. Kept apart from the subscriber so a pending timer + // retains only this, never the subscriber or its track state, once no read waits. + #wall: { revision: Signal; at: number; timer?: ReturnType } = { + revision: new Signal(0), + at: Number.POSITIVE_INFINITY, + }; #drift(): { budget: number; @@ -1217,21 +1218,23 @@ export class Subscriber { }; } - // The furthest presentation time the group at `sequence` could still reach: where its - // immediate servable successor begins, or undefined when nothing proves where it - // stops. An upper bound, deliberately: a frame's duration is not on the wire, so a - // group's own last timestamp says where it starts presenting, not where it ends. - // Only the *immediate* successor counts: timestamps need not rise with sequence (a - // rewind reorders them), so a later stamped group proves nothing about where an - // unstamped successor will begin, and shrinking the bound is the unsafe direction. - // An unstamped successor leaves the reach unbounded until it presents a frame. - #reach(sequence: number, end?: number): number | undefined { + // The group after `sequence` that bounds it: the first servable one below `end`, + // stamped or not, empty or not. + // + // Its first timestamp is the furthest presentation time `sequence` could still reach. + // An upper bound, deliberately: a frame's duration is not on the wire, so a group's own + // last timestamp says where it starts presenting, not where it ends. Only the + // *immediate* successor counts: timestamps need not rise with sequence (a rewind + // reorders them), so a later stamped group proves nothing about where an unstamped + // successor will begin, and shrinking the bound is the unsafe direction. An unstamped + // successor leaves the reach unbounded until it presents a frame. + #successor(sequence: number, end?: number): GroupConsumer | undefined { const timeline = this.#state.timeline; for (let i = timelineIndex(timeline, sequence + 1); i < timeline.length; i++) { const successor = timeline[i]; if (end !== undefined && successor.sequence >= end) break; if (successor.closed.peek() instanceof Error) continue; - return hooks.groupTimestamp(successor)?.asMillis(); + return successor; } return undefined; } @@ -1250,8 +1253,13 @@ export class Subscriber { // its stamped successor the same way. The bound is exclusive, so the comparison is `>=`: // the freshest frame a group could still hold sits just below its reach, so an age equal // to the budget already puts every frame in it strictly past the budget. A zero budget - // falls out for free. Only timestamps drive expiry; wall-clock reclamation of idle - // content is the cache's own policy, not the budget's. + // falls out for free. + // + // The wall clock judges it too: a group is also stale once its successor arrived a full + // budget ago, so a peer whose timestamps stop advancing, an untimed track, or an empty + // successor that never presents cannot pin a superseded group. The successor is re-read, + // not latched, like the reach. The newest group has no successor, so neither clock ever + // convicts it. `wake` arms a timer for the wall deadline, for a held read that waits on it. #isStale( group: GroupConsumer, drift: { @@ -1259,10 +1267,17 @@ export class Subscriber { presentation?: { sequence: number; timestamp: Timestamp }; end?: number; }, + wake = false, ): boolean { if (this.#state.timeline[timelineIndex(this.#state.timeline, group.sequence)] !== group) return false; - const reach = this.#reach(group.sequence, drift.end); + const successor = this.#successor(group.sequence, drift.end); + if (!successor) return false; + const deadline = hooks.groupArrived(successor) + drift.budget; + if (performance.now() >= deadline) return true; + if (wake) this.#wakeAt(deadline); + + const reach = hooks.groupTimestamp(successor)?.asMillis(); return ( drift.presentation !== undefined && drift.presentation.sequence > group.sequence && @@ -1271,13 +1286,32 @@ export class Subscriber { ); } + // Bump {@link #wall} at `at`, unless a wakeup is already armed sooner. Each held read + // re-arms on the bump, so one timer covers every deadline this subscriber waits on. + #wakeAt(at: number): void { + const wall = this.#wall; + if (!Number.isFinite(at) || at >= wall.at) return; + clearTimeout(wall.timer); + // setTimeout fires at once past 2^31 ms, so a longer budget wakes early and re-arms. + const delay = Math.min(Math.max(0, at - performance.now()), 2 ** 31 - 1); + wall.timer = setTimeout(() => { + wall.timer = undefined; + wall.at = Number.POSITIVE_INFINITY; + wall.revision.update((revision) => revision + 1); + }, delay); + // A staleness wakeup is never a reason to hold a Node/Bun process open. + (wall.timer as unknown as { unref?: () => void }).unref?.(); + wall.at = at; + } + #guard(group: GroupConsumer): GroupConsumer { if (!this.#enforceLatency) return group; hooks.expireGroup(group, { - expired: () => this.#isStale(group, this.#drift()), + expired: () => this.#isStale(group, this.#drift(), true), changed: [ this.#state.groups, this.#state.timelineChanged, + this.#wall.revision, this.#state.update, this.#state.info, this.#cursor, @@ -1292,36 +1326,9 @@ export class Subscriber { this.#state = state; // The cursor's floor is the group the subscription named, or 0. A floor is the // only thing a start contributes; {@link Subscription.maxDelay} is what asks for - // data, and delivery skips everything above the floor that the budget convicts. - // Nothing on an untimed track is ever stale, so there the budget would replay the - // whole cache: an unfloored subscriber starts at the latest group instead. - const groups = state.update.peek()?.groups ?? {}; - const bounds = groupBounds(groups); - const start = groups.start === undefined ? this.#untimedStart(bounds.end) : undefined; - this.#cursor.set({ start: start ?? bounds.start }); - // Timedness is unknown until accept(), so an unfloored subscriber made before it - // settles its start then, against the groups that existed at that moment. - if (groups.start === undefined && !state.info.peek()) state.settle = () => this.#settleStart(); - } - - // Raise the floor of an unfloored subscriber made before accept(), once the track is - // known to be untimed. - #settleStart(): void { - this.#state.settle = undefined; - const start = this.#untimedStart(this.#cursor.peek().end); - if (start === undefined) return; - this.#cursor.update((cursor) => ({ ...cursor, start: Math.max(cursor.start, start) })); - } - - // The newest servable group below the `end` cap, when the track is known to be untimed. - #untimedStart(end: number | undefined): number | undefined { - const info = this.#state.info.peek(); - if (!info || info.timescale != null) return undefined; - const timeline = this.#state.timeline; - for (let i = (end === undefined ? timeline.length : timelineIndex(timeline, end)) - 1; i >= 0; i--) { - if (!(timeline[i].closed.peek() instanceof Error)) return timeline[i].sequence; - } - return undefined; + // data, and delivery skips everything above the floor that the budget convicts, + // timed or not. + this.#cursor.set({ start: groupBounds(state.update.peek()?.groups ?? {}).start }); } static { @@ -1461,7 +1468,6 @@ export class Subscriber { */ setGroups(groups: Groups): void { const { start, end } = groupBounds(groups); - if (groups.start !== undefined) this.#state.settle = undefined; this.#cursor.update((cursor) => ({ start: Math.max(cursor.start, start), end })); } @@ -1469,7 +1475,6 @@ export class Subscriber { // Rust `start_at`. Local readers stay monotonic; only the wire publisher lowers. #replaceGroups(groups: Groups): void { const { start, end } = groupBounds(groups); - if (groups.start !== undefined) this.#state.settle = undefined; this.#cursor.update((cursor) => ({ start: groups.start === undefined ? cursor.start : start, end, @@ -1490,6 +1495,8 @@ export class Subscriber { this.#frameGroup?.close(abort); this.#frameGroup = undefined; this.#state.timeline.length = 0; + clearTimeout(this.#wall.timer); + this.#wall.timer = undefined; } /** diff --git a/quest/m1/README.md b/quest/m1/README.md index a32c8fa112b..29c0bc66aff 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -71,7 +71,6 @@ above them. - [Finalize moq-lite-07](/quest/m1/lite07-finalize.md) - when the maintainer cuts it, lite-07 negotiates as `moq-lite-07` and the next release ships it - [Interop close logs](/quest/m1/interop-close-log.md) - logging fix landed; delete once the required Interop run is green - [moq-transport passthrough](/quest/m1/ietf-extension/README.md) - subgroups, object IDs, properties, end status, datagram objects, and unknown track properties survive moq-transport -> moq-relay (and a relay cluster) -> moq-transport; moxygen's moq-test passes nightly -- [One max_age meaning](/quest/m1/cache-max-age.md) - a superseded group goes stale on wall clock since its successor arrived or on media time, whichever is first, in Rust and js/net; fixes the untimed failover stall - [FETCH max-delay](/quest/m1/fetch-max-delay.md) - historical readers obey publisher max-age and an optional tighter reader budget, including while holding a group - [Draft-20 FETCH edge-case backport](/quest/m1/fetch-edge-cases.md) - backport the final-object and response-group fixes to a later 0.17 patch after main lands - [Subscribe-close backport](/quest/m1/subscribe-close-backport.md) - the same fix and the `race` loser fix on `release`, so closing a subscription during setup leaves no unhandled rejection in a later 0.17 patch diff --git a/quest/m1/cache-max-age.md b/quest/m1/cache-max-age.md deleted file mode 100644 index 97643b8c2b9..00000000000 --- a/quest/m1/cache-max-age.md +++ /dev/null @@ -1,132 +0,0 @@ -# [L] One max_age meaning for both languages' track caches - -## Goal - -Publisher retention (`track::Info::max_age`) and a subscriber's budget -(`Subscription::max_delay`) apply the same -staleness rule in Rust and js/net. A group other than the newest is stale once either -its wall-clock age since its successor arrived at this hop, or its media-time -age (the live edge minus its reach), reaches the budget. The newest group is -never stale. An untimed group has no media time, so the wall clock alone -judges it. - -That one rule also gives up a resumed group no route continues after a -failover, on untimed tracks too, and lets an untimed track's start replay -what isn't stale instead of jumping to the latest group. The pool's (Rust) or -cache window's (JS) wall-clock idle bound stays separate from both. - -## Plan - -Decided (2026-10-06, maintainer): - -- Stale on either clock. The wall term bounds a peer that lies about PTS, - which media time alone would never evict. The media term evicts earlier - than the wall can, as when a startup burst arrives all at once. Rejected: - stale only when both clocks agree, which keeps the lying-peer case. -- The wall clock starts when the group's successor arrives at this hop, the - wall-clock twin of media time's reach (bounded by the immediate - successor). A group with no successor never ages. -- "Successor" is the group `reach` uses: `first_servable` above it, so - publisher-produced (not fetched backfill), not aborted, and below the - subscriber's `set_groups` cap (retention has no cap). The clock is that - group's arrival, re-read like reach rather than latched. If N+2 arrives - before N+1, N's clock starts at N+2's arrival, and restarts at N+1's - arrival once N+1 shows up. If the successor aborts, the clock falls to the - next servable successor's arrival, or stops if there is none. -- Accepted: during a total upstream stall, a reader still on a superseded - group is skipped forward after its budget, where media time alone would let - it wait. That reader is its budget of real time behind live, and the newest - group is never evicted. -- Applies to both retention and subscriber budgets, which share `is_stale` - in `rs/moq-net/src/model/track.rs` today. -- Supersedes two earlier decisions: the untimed model's ([#4822](https://github.com/moq-dev/moq/pull/4822)) - 2026-10-01 rejection of `max_age` on - max(wall, pts), whose worry was a congestion stall (answered by starting the - clock at the successor), and the retired untimed-failover quest's no-clock - rule, which this rule replaces. -- Blocked readers need a deadline wake. A reader parked in `Recover::poll` or - a subscriber waiting on a superseded group wakes only through - `model::expiry::Wakes`, which knows media-time deadlines, landings, and the - successor's first frame or abort, so a lazily checked wall term never fires - for it. Decided 2026-10-08: one wake mechanism. Arm successor arrival + - budget in `TrackState::poll_drifted`, which both reader paths share, once - the successor is resolved and before the early return for an unstamped one, - so re-selecting the successor re-arms it. The deadline needs a wall-clock - driver beside the index's write-driven ones. -- The swept benchmark measures cost and picks between evaluating the wall - term on a timer and evaluating it lazily on access, for retention only. It - no longer decides the semantics. -- Ranked in the clock chain right after the untimed model, because that - model ships the failover regression below until this lands. -- Docs update inline: `track::Info::max_age` and `Subscription::max_delay` docs, - js/net's `Info.maxAge` doc, `doc/concept`, the relay config docs, the - Expiration section and Max Age field definitions in - `drafts/draft-lcurley-moq-lite.md` (which today exclude wall-clock - reclamation from the rule), and the untimed text in - `drafts/draft-lcurley-moq-timestamp.md` that the untimed model adds - ("never media-stale, starts at the latest group"). No new guide. - -Facts, and work carried over: - -- Today Rust's `max_age` is media time and is applied only when the track - writes: a group ages out when a later one starts (`is_stale` and the expiry - scans in `rs/moq-net/src/model/track.rs`). A track that stops writing keeps - groups past `max_age` until the pool's idle expiry (`Pool::gc`, driven by - the origin driver without a write) or byte pressure reclaims them. - `max_age_does_not_drive_wall_eviction` pins that behaviour and changes here. -- JS matches Rust's media-time rule since - [#4659](https://github.com/moq-dev/moq/pull/4659): `#reach` and the drift - check in `js/net/src/track.ts` mirror `track.rs`, and a private - `CACHE_WINDOW_MS` idle window drives `#prune`, as Rust's pool does. The - remaining JS work is the wall term, and rewriting `Info.maxAge`'s doc, - which says a congestion stall cannot age content out. - [Rust-backed @moq/net](/quest/m2/rs2ts/README.md) replaces the JS model later; - the maintainer chose to fix it by hand first. -- Bench in `rs/moq-net/benches/track.rs`, swept over tracks (1 to 10k) and - cached groups per track (1 to 1k): write-path cost, timer cost per driver - pass, and retained memory for idle tracks. A cost that grows with the table - should show as a slope. `js/net/bench/track.ts` covers the JS side. -- Failover (from the retired untimed-failover quest, found in #4822): - `Recover::poll` (`rs/moq-net/src/model/resume.rs`) gives up a group the - serving route can't continue once `poll_stale` reports drift past the - reader's budget. `drifted` needs a timed live edge and a timed successor, - so on an untimed track nothing convicts the group and its reader waits - until the track ends; the replaced copy's lease stays alive too. The wall - term convicts it. Settle what "nothing pending" means against the recovery - fetch (`a_pending_recovery_fetch_does_not_disable_expiry`), so a fetch - about to fill the group is still subject to its reader's deadline. Do not - make a pending fetch disable expiry. Decided in the 2026-10-10 audit: - FETCH also obeys publisher max-age and an optional reader max-delay; - omission uses the publisher's limit alone. This supersedes the old FETCH - exemption. [FETCH max-delay](/quest/m1/fetch-max-delay.md) owns that API, - request propagation and per-reader enforcement, building on this rule. -- Start resolution: the untimed model starts an untimed track at the latest - group (`TrackState::untimed_start`). Replace that special case with the - normal rule, replaying the cached groups that aren't stale. Until then a - reader rejoining an untimed track starts at a stale cached group and reads - on from there, so `rejoin_during_the_cancel_skips_the_cache` - (`rs/moq-net/tests/rejoin.rs`) checks only the versions whose tracks arrive - timed. Make it check every version again. - -Tests, with mocked time: - -- A stalled track keeps its newest group and drops superseded ones by the - wall rule. -- A track whose PTS stops advancing still evicts by wall clock. -- A startup burst evicts by media time. -- A model test in `resume.rs` fails over an untimed track with an abandoned - group, against a cold route and against one whose copy already caches - untimed groups past the resumed one, so a cursor that skips ahead can't - hide the stall. Nothing writes after the successor arrives, so only the - deadline wake can release the reader. It fails without the fix. -- An untimed subscriber starts at the oldest group that isn't stale. - -Public API: no signature change; `max_age` behaviour changes. Wire: no -encoding change; Max Age semantics in the lite draft change. - -## Related - -- [Model ranges](/quest/m1/subscribe-ranges/model.md) - an untimed `Absolute(0)..` range follows this rule: replay what is not stale -- [JS track handover](/quest/m1/js-group-handover.md) - mirrors the failover rule in JS -- [Rust-backed @moq/net](/quest/m2/rs2ts/README.md) - retires js/net's hand-written model -- [FETCH max-delay](/quest/m1/fetch-max-delay.md) - applies the shared rule to historical requests and reconciles native IETF retention clocks diff --git a/quest/m1/fetch-max-delay.md b/quest/m1/fetch-max-delay.md index 64ea0adf8c9..eb7aa312ca7 100644 --- a/quest/m1/fetch-max-delay.md +++ b/quest/m1/fetch-max-delay.md @@ -63,10 +63,6 @@ field if needed; published formats are unchanged. Update API, concept, CLI and matching draft documentation inline, including the changed FETCH exemption and the distinction between age and upstream wait budgets. -## Required - -- [One max_age meaning](/quest/m1/cache-max-age.md) - one expiry rule and deadline wake mechanism - ## Related - [IETF fill timeout](/quest/m1/ietf-fill-timeout.md) - upstream waiting bounded through the existing FETCH parameter diff --git a/quest/m1/js-group-handover.md b/quest/m1/js-group-handover.md index 2a9f0d67aae..774099424cb 100644 --- a/quest/m1/js-group-handover.md +++ b/quest/m1/js-group-handover.md @@ -57,8 +57,8 @@ Things to settle along the way: in Rust, so `drain.ts`'s publisher mints an epoch. - `js/watch` and `js/hang` consumers that re-subscribe on `active` changes. Check whether they still need to. -- Giving up a resumed group no route continues. Mirror Rust's rule from - [One max_age meaning](/quest/m1/cache-max-age.md): give it up once its +- Giving up a resumed group no route continues. Mirror Rust's rule, which + js/net's `#isStale` already applies to subscriptions: give it up once its wall-clock age since its successor arrived, or its media-time drift, reaches the reader's budget. @@ -72,10 +72,6 @@ mirrors their final shape instead of chasing it. Public API: likely a behavior change to `Origin.Requesting.active` and track subscriptions across a swap. Report it in the PR. -## Required - -- [One max_age meaning](/quest/m1/cache-max-age.md) - the rule that gives up a resumed group no route continues - ## Related - [Retired requests](https://github.com/moq-dev/moq/pull/5294) - old handles cannot open unpinned requests after an identity change diff --git a/quest/m1/publisher-linger.md b/quest/m1/publisher-linger.md index f605d206f03..889684c41f2 100644 --- a/quest/m1/publisher-linger.md +++ b/quest/m1/publisher-linger.md @@ -25,7 +25,3 @@ landing it. Test with a mocked clock: a subscriber that requests a finished track within max-age gets its media, and the publisher closes once max-age passes. - -## Related - -- [One max_age meaning](/quest/m1/cache-max-age.md) - what max-age means for a superseded group diff --git a/quest/m1/subscribe-ranges/model.md b/quest/m1/subscribe-ranges/model.md index 032bb1fa5ec..f28dc4881a7 100644 --- a/quest/m1/subscribe-ranges/model.md +++ b/quest/m1/subscribe-ranges/model.md @@ -24,8 +24,8 @@ Guidance: a `Relative(0)..` subscriber on a quiet track whose newest group is 3 (the live one still gets group 3); a buffering `Relative(k)..` merged with a range above the live edge; a frame-precise start merged with live; an - untimed `Absolute(0)..` (it follows - [One max_age meaning](/quest/m1/cache-max-age.md)); and the lite-05 relay + untimed `Absolute(0)..` (it replays what + the wall clock has not made stale); and the lite-05 relay case #5000 left out: a live subscriber already reading group 1, then one from `Absolute(0)`, and a fresh group 0 reaches only the second. The closed #5085 branch has reusable pieces (older-wire codec mappings, the JS fix diff --git a/quest/m2/pipeline-requests/subscribe.md b/quest/m2/pipeline-requests/subscribe.md index f8ecf1be3a1..6c6119bf6f7 100644 --- a/quest/m2/pipeline-requests/subscribe.md +++ b/quest/m2/pipeline-requests/subscribe.md @@ -64,7 +64,7 @@ Decisions: promise a clamped aggregate (`track::Producer::subscription`, `max_age_bounds_the_budget`). Rejected: a SUBSCRIBE_UPDATE once TRACK_INFO lands. Narrowed 2026-10-08 from review. This agrees with - [One max_age meaning](/quest/m1/cache-max-age.md): both budgets apply one + the shared staleness rule (`Subscription::max_delay`): both budgets apply one staleness rule, so a group is stale under the smaller budget exactly when it is stale under either. - Publishers send a track's TRACK_INFO at a higher stream priority than that @@ -94,5 +94,4 @@ blocks the Related quests below; whichever lands second rebases. ## Related - [Pipelined first FETCH](/quest/m2/pipeline-requests/fetch.md) - the same change for fetch-only readers -- [One max_age meaning](/quest/m1/cache-max-age.md) - the staleness rule that makes the wire clamp redundant - [Lite-07 ranges](/quest/m1/subscribe-ranges/lite.md) - replaces the SUBSCRIBE floor with ranges diff --git a/rs/moq-archive/src/reader/tests.rs b/rs/moq-archive/src/reader/tests.rs index 6fb0cc8b220..217d99f2e22 100644 --- a/rs/moq-archive/src/reader/tests.rs +++ b/rs/moq-archive/src/reader/tests.rs @@ -722,12 +722,12 @@ async fn evicted_timeline_groups_remain_fetchable() { let serve = tokio::spawn(reader.serve()); let expected = fetch(&broadcast, &timeline("video"), 0, 0).await.unwrap(); let now = moq_net::time::Instant::now(); - pool.gc(now); - pool.gc(now + Duration::from_secs(1)); + pool.gc(now, &kio::Waiter::noop()); + pool.gc(now + Duration::from_secs(1), &kio::Waiter::noop()); assert_eq!(fetch(&broadcast, &timeline("video"), 0, 0).await.unwrap(), expected); // Past the end of an evicted complete group is EOF, as it is for a cached one. - pool.gc(now + Duration::from_secs(2)); - pool.gc(now + Duration::from_secs(3)); + pool.gc(now + Duration::from_secs(2), &kio::Waiter::noop()); + pool.gc(now + Duration::from_secs(3), &kio::Waiter::noop()); assert!( fetch(&broadcast, &timeline("video"), 0, expected.len() as u64 + 1) .await @@ -756,8 +756,11 @@ async fn enrolled_timeline_fetch_survives_finish() { let expected = fetch(&broadcast, &timeline("audio"), 0, 0).await.unwrap(); reader.finish().unwrap(); let now = moq_net::time::Instant::now(); - broadcast.info().pool.gc(now); - broadcast.info().pool.gc(now + Duration::from_secs(31)); + broadcast.info().pool.gc(now, &kio::Waiter::noop()); + broadcast + .info() + .pool + .gc(now + Duration::from_secs(31), &kio::Waiter::noop()); assert_eq!(fetch(&broadcast, &timeline("audio"), 0, 0).await.unwrap(), expected); let missing = fetch(&broadcast, &timeline("audio"), 2, 0).await; assert!( @@ -796,13 +799,13 @@ async fn benchmark_timeline_fetch_fanout() { let serving = tokio::spawn(reader.serve()); let name = timeline("video-0"); let mut now = moq_net::time::Instant::now(); - broadcast.info().pool.gc(now); + broadcast.info().pool.gc(now, &kio::Waiter::noop()); let mut elapsed = Duration::ZERO; for _ in 0..20 { // One sweep dates recent reads; the next expires them. for _ in 0..2 { now += Duration::from_secs(31); - broadcast.info().pool.gc(now); + broadcast.info().pool.gc(now, &kio::Waiter::noop()); } let subscription = moq_net::track::Subscription::default() .with_start(moq_net::track::Position::group(0)) diff --git a/rs/moq-hls/src/export/archive_tests.rs b/rs/moq-hls/src/export/archive_tests.rs index cb070ee5821..4a0bcc9a8e6 100644 --- a/rs/moq-hls/src/export/archive_tests.rs +++ b/rs/moq-hls/src/export/archive_tests.rs @@ -675,8 +675,11 @@ async fn history_after_expiry(finished: bool) { }; // Expire the publisher cache before any HLS consumer can retain a copy. let now = moq_net::time::Instant::now(); - broadcast.info().pool.gc(now); - broadcast.info().pool.gc(now + Duration::from_secs(31)); + broadcast.info().pool.gc(now, &kio::Waiter::noop()); + broadcast + .info() + .pool + .gc(now + Duration::from_secs(31), &kio::Waiter::noop()); let subscription = moq_net::track::Subscription::default() .with_start(moq_net::track::Position::group(0)) .with_max_delay(Duration::from_millis((1_u64 << 62) - 1)); diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index f909e3ca088..243e5020b47 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -1488,12 +1488,17 @@ mod tests { let mut track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.video)); let mut consumer = Consumer::new(track.subscribe(None), Container::Legacy(crate::container::Kind::Data)); write_group(&mut track, 0, &[ts(0), ts(33_000)]); + // Read the first group before the second arrives: a superseded group is skipped + // for latency, which is not what this checks. + for expected in [0, 33_000] { + assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(expected)); + } write_group(&mut track, 1, &[ts(66_000), ts(50_000)]); track.finish().unwrap(); let frames = continuous_frames(&mut consumer).await; assert_eq!( frames.iter().map(|f| f.timestamp.as_micros()).collect::>(), - vec![0, 33_000, 66_000, 50_000] + vec![66_000, 50_000] ); } diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs index 0a9f43ef621..bf28aba5879 100644 --- a/rs/moq-net/benches/session.rs +++ b/rs/moq-net/benches/session.rs @@ -452,7 +452,7 @@ impl Room { fn tick(&mut self) { self.cache_now += Duration::from_secs(1); for pool in self.pools() { - pool.gc(self.cache_now); + pool.gc(self.cache_now, &kio::Waiter::noop()); } } diff --git a/rs/moq-net/benches/track.rs b/rs/moq-net/benches/track.rs index 376993fe895..ab167b475cc 100644 --- a/rs/moq-net/benches/track.rs +++ b/rs/moq-net/benches/track.rs @@ -30,9 +30,16 @@ //! `track_gap_flush` forgets every remembered gap in one hand-out, each below an aborted //! group still cached: the successor they share must be found once, not once per gap. //! +//! `track_wall` is the wall-clock half of staleness, swept over tracks sharing one driven +//! pool and the groups each caches: a collection pass with a parked read's deadline armed on +//! every track (`idle`, none due, and `due`, every one due at once), and re-polling one of +//! those parked reads (`repoll`). A cost that grows with the table instead of the touched +//! path shows up as a slope. +//! //! `track_empty_run` drains a run of groups that finished without a frame across readers, -//! with the groups after it past the budget or within it: each empty group is handed out -//! like any other, so a walk over the rest of the run per group shows up as a slope over it. +//! with the groups after it past the budget, within it, or the whole run retired by the +//! wall clock: each empty group is handed out like any other, so a walk over the rest of +//! the run per group shows up as a slope over it. //! //! Run with `cargo bench -p moq-net --bench track`. @@ -80,6 +87,15 @@ const WRITERS: [usize; 4] = [1, 2, 4, 8]; /// Tracks sharing one expiring pool: the table a due collection pass walks. const SWEPT: [usize; 4] = [1, 64, 1_024, 16_384]; +/// Tracks sharing one driven pool, for the wall-clock deadlines. +const AGED_TRACKS: [usize; 4] = [1, 64, 1_024, 10_000]; + +/// Groups cached per track behind its parked read. +const AGED_GROUPS: [usize; 3] = [1, 32, 1_024]; + +/// The budget a parked read's wall deadline is armed for, past any setup on the real clock. +const AGED_BUDGET: Duration = Duration::from_secs(3600); + /// Keeps the ownership chain alive around the track and its subscribers. struct Fanout { _broadcast: broadcast::Producer, @@ -199,7 +215,7 @@ impl Swept { .collect(); // The first call starts the pool's clock and runs its first pass. let now = Instant::now(); - pool.gc(now); + pool.gc(now, &kio::Waiter::noop()); Self { _broadcast: broadcast, _tracks: tracks, @@ -247,6 +263,100 @@ impl Churn { } } +/// Tracks sharing one driven pool, each caching `groups` groups behind a read parked on its +/// oldest, which the group after it superseded: one wall-clock deadline per track. +struct Aged { + _broadcast: broadcast::Producer, + _tracks: Vec<(track::Producer, track::Subscriber, moq_net::group::Producer)>, + reads: Vec<(moq_net::group::Consumer, kio::Waiter)>, + pool: cache::Pool, + now: Instant, +} + +impl Aged { + fn new(tracks: usize, groups: usize) -> Self { + let broadcast = broadcast::Info::default().produce(); + let pool = broadcast.info().pool.clone(); + // The driver's first collection, which arms wall-clock deadlines. + let now = Instant::now(); + pool.gc(now, &kio::Waiter::noop()); + let mut reads = Vec::with_capacity(tracks); + let tracks = (0..tracks) + .map(|i| { + let track = broadcast.create_track(format!("bench{i}"), None).unwrap(); + let mut sub = track.subscribe(track::Subscription::default().with_max_delay(AGED_BUDGET)); + let mut held = track.append_group().unwrap(); + held.write_frame(None, Bytes::from_static(b"x")).unwrap(); + for _ in 1..groups.max(2) { + let mut group = track.append_group().unwrap(); + group.write_frame(None, Bytes::from_static(b"x")).unwrap(); + group.finish().unwrap(); + } + let waiter = kio::Waiter::noop(); + let Poll::Ready(Ok(Some(mut read))) = sub.poll_recv_group(&waiter) else { + panic!("the held group is cached"); + }; + assert!(matches!(read.poll_read_frame(&waiter), Poll::Ready(Ok(Some(_))))); + assert!(read.poll_read_frame(&waiter).is_pending()); + reads.push((read, waiter)); + (track, sub, held) + }) + .collect(); + Self { + _broadcast: broadcast, + _tracks: tracks, + reads, + pool, + now, + } + } +} + +fn bench_wall(c: &mut Criterion) { + let mut group = c.benchmark_group("track_wall"); + for tracks in AGED_TRACKS { + for groups in AGED_GROUPS { + // Skip the largest table: setup alone would dominate the run. + if tracks * groups > 1_000_000 { + continue; + } + let id = format!("{tracks}x{groups}"); + // A collection with every deadline armed and none due: every origin poll. + group.throughput(Throughput::Elements(1)); + group.bench_function(BenchmarkId::new("idle", &id), |b| { + let aged = Aged::new(tracks, groups); + b.iter(|| black_box(aged.pool.gc(aged.now, &kio::Waiter::noop()))); + }); + // Re-polling one parked read re-arms its deadline. + group.bench_function(BenchmarkId::new("repoll", &id), |b| { + let mut aged = Aged::new(tracks, groups); + b.iter(|| { + let (read, waiter) = &mut aged.reads[0]; + assert!(black_box(read.poll_read_frame(waiter)).is_pending()); + }); + }); + // The collection that releases every parked read at once. Its cost is per track + // released, so the cached groups behind each read are not swept. + if groups != 1 { + continue; + } + group.throughput(Throughput::Elements(tracks as u64)); + group.bench_function(BenchmarkId::new("due", &id), |b| { + b.iter_batched( + || Aged::new(tracks, groups), + |aged| { + // Past every deadline, which setup armed after `now`. + black_box(aged.pool.gc(aged.now + 2 * AGED_BUDGET, &kio::Waiter::noop())); + aged + }, + BatchSize::PerIteration, + ); + }); + } + } + group.finish(); +} + fn bench_gc(c: &mut Criterion) { let mut group = c.benchmark_group("track_gc"); for tracks in SWEPT { @@ -254,7 +364,7 @@ fn bench_gc(c: &mut Criterion) { group.throughput(Throughput::Elements(1)); group.bench_with_input(BenchmarkId::new("idle", tracks), &tracks, |b, &tracks| { let swept = Swept::new(tracks); - b.iter(|| black_box(swept.pool.gc(swept.now))); + b.iter(|| black_box(swept.pool.gc(swept.now, &kio::Waiter::noop()))); }); // A due pass, stepping one sweep interval per call. group.throughput(Throughput::Elements(tracks as u64)); @@ -264,7 +374,7 @@ fn bench_gc(c: &mut Criterion) { let mut now = swept.now; b.iter(|| { now += interval; - black_box(swept.pool.gc(now)) + black_box(swept.pool.gc(now, &kio::Waiter::noop())) }); }); } @@ -650,6 +760,17 @@ fn bench_gap_flush(c: &mut Criterion) { group.finish(); } +/// How the groups after an [`EmptyRun`] judge it. +#[derive(Clone, Copy)] +enum Run { + /// Two groups an hour apart in media time follow it, writing off its last group. + Stale, + /// One group follows it, keeping all of it within the budget. + Fresh, + /// A zero budget: the wall clock already retired every superseded group. + Retired, +} + /// Readers on one track holding group 0, a run of groups that finished without a frame, /// and the groups after it. struct EmptyRun { @@ -661,10 +782,9 @@ struct EmptyRun { } impl EmptyRun { - /// `stale` stamps the groups after the run 50 ms and 100 ms in, so a 10 ms budget writes - /// off the run's last group, the only one they bound. Otherwise one group 5 ms in follows - /// the run and keeps all of it within the budget. - fn new(run: u64, readers: usize, stale: bool) -> Self { + /// Media-time budgets span an hour, so setup on the real clock never ages a group out on + /// the wall clock. + fn new(run: u64, readers: usize, mode: Run) -> Self { let broadcast = broadcast::Info::default().produce(); let track = broadcast.create_track("bench", timed()).unwrap(); let append = |millis: Option| { @@ -680,17 +800,26 @@ impl EmptyRun { for _ in 0..run { append(None); } - let after: &[u64] = if stale { &[50, 100] } else { &[5] }; + const HOUR: u64 = 3_600_000; + let after: &[u64] = match mode { + Run::Stale => &[2 * HOUR, 4 * HOUR], + Run::Fresh | Run::Retired => &[5], + }; for millis in after { append(Some(*millis)); } - let budget = track::Subscription::default().with_max_delay(Duration::from_millis(10)); + let (budget, handed) = match mode { + // The run's last group is written off, but two groups follow it. + Run::Stale => (Duration::from_millis(HOUR), run as usize + 2), + Run::Fresh => (Duration::from_millis(HOUR), run as usize + 2), + Run::Retired => (Duration::ZERO, 1), + }; + let budget = track::Subscription::default().with_max_delay(budget); Self { readers: (0..readers).map(|_| track.subscribe(budget.clone())).collect(), _broadcast: broadcast, _track: track, - // The run's last group is written off when stale, but two groups follow it. - handed: run as usize + 2, + handed, } } @@ -707,19 +836,20 @@ impl EmptyRun { } /// Readers draining a run of groups that finished without a frame, swept over the run and -/// the readers, with the groups after it past the budget (`stale`) or within it (`fresh`). +/// the readers, with the groups after it past the budget (`stale`), within it (`fresh`), or +/// the run retired by the wall clock (`retired`). /// A reader's cost must stay linear in the run, so a walk over the rest of the run per /// group shows up as a slope over it. fn bench_empty_run(c: &mut Criterion) { let mut group = c.benchmark_group("track_empty_run"); group.sample_size(10); - for (name, stale) in [("stale", true), ("fresh", false)] { + for (name, mode) in [("stale", Run::Stale), ("fresh", Run::Fresh), ("retired", Run::Retired)] { for readers in [1usize, 8, 64] { for run in [64u64, 512, 4096] { group.throughput(Throughput::Elements(run * readers as u64)); group.bench_function(BenchmarkId::new(format!("{name}/{readers}"), run), |b| { b.iter_batched_ref( - || EmptyRun::new(run, readers, stale), + || EmptyRun::new(run, readers, mode), EmptyRun::drain, BatchSize::PerIteration, ); @@ -743,6 +873,7 @@ criterion_group!( bench_parked_read, bench_sparse_read, bench_gap_flush, + bench_wall, bench_empty_run ); criterion_main!(benches); diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 55fbe18a07f..45c68ab4782 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -4665,6 +4665,10 @@ mod serve_group_test { }); { let mut run = std::pin::pin!(kio::wait(|waiter| run.poll(&mut writer, waiter))); + // Group 2's successor arrives with it: group 5 arrived a full budget before + // the update, so group 2 bounded by it would be stale on the wall clock. Group 4 + // is, and goes unserved. + write_group(&mut track, 4, 0); write_group(&mut track, 2, 0); assert!(futures::poll!(run.as_mut()).is_pending()); track.finish_at(track::Position::group(6)).unwrap(); diff --git a/rs/moq-net/src/model/cache.rs b/rs/moq-net/src/model/cache.rs index 3f92e4d9218..c838da3bb88 100644 --- a/rs/moq-net/src/model/cache.rs +++ b/rs/moq-net/src/model/cache.rs @@ -22,9 +22,9 @@ //! nobody has read or written for that long is reclaimed, no matter what retention //! its track advertises. Only a live track's latest group is exempt: once a track //! ends, a stale consumer can't pin any of it. Track retention -//! ([`max_age`](crate::track::Info::max_age)) is measured in media timestamps, so a -//! congestion stall can't age content out; the pool's expiry is the orthogonal -//! wall-clock bound that keeps unwatched content from pinning RAM. +//! ([`max_age`](crate::track::Info::max_age)) decides which groups readers skip as +//! stale; the pool's expiry is the orthogonal bound that frees the bytes of content +//! nobody touches. //! //! Expiry is driven by [`Pool::gc`], also called by each origin driver. //! Reads and writes clear the expiration timestamp without reading a clock. @@ -38,6 +38,7 @@ //! [`DEFAULT_EXPIRY`], while a relay creates one configured pool and shares it across //! every origin so the whole process caches into a single policy. +use std::collections::BTreeMap; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, OnceLock, PoisonError, Weak}; use std::time::Duration; @@ -104,7 +105,7 @@ impl Config { /// A cached group (other than a live track's latest) that nobody reads or writes /// for this long is reclaimed, surfacing to any remaining reader as /// [`Error::Old`](crate::Error::Old). This is independent of track retention: - /// [`max_age`](crate::track::Info::max_age) uses media timestamps, while this + /// [`max_age`](crate::track::Info::max_age) decides what readers skip, while this /// window keeps idle content from pinning memory. Reclaiming without a write behind /// it needs [`Pool::gc`] called periodically (origins do this automatically). /// The value is fixed when the @@ -155,6 +156,30 @@ struct Inner { // budget needs no registry, since a track that never writes never grows the pool. // Weak, because a track owns its account and the account must not outlive it. tracks: kio::Lock>>, + // The first instant a driver supplied, and the latest as micros past it, so reading + // the wall clock never runs behind the time the drivers keep; see [`Pool::now_wall`]. + // Unset until a driver collects, which is also what arms wall-clock deadlines. + wall_epoch: OnceLock, + wall_micros: AtomicU64, + // Tracks with reads parked on a wall-clock deadline, by their earliest one. + wall: kio::Lock, +} + +/// The pool-wide index of wall-clock deadlines that [`Pool::gc`] fires. +/// +/// One entry per track and earliest deadline: the reads themselves park in the track's +/// own `expiry::Wakes`, so arming one takes this lock only when it moves its track's +/// earliest deadline forward in time, never per read. +#[derive(Default)] +struct WallDeadlines { + // Keyed by the track account's address too, so tracks sharing an instant coexist. + due: BTreeMap<(crate::time::Instant, usize), Weak>, + // The drivers collecting this pool, woken when a deadline earlier than any they + // were told about is armed. + drivers: kio::WaiterList, + // Reads that met a deadline before any driver collected, woken by the first + // collection to arm it. Holds live waiters only, so an undriven pool stays bounded. + undriven: kio::WaiterList, } struct Clock { @@ -189,6 +214,9 @@ impl Pool { access_sum: AtomicU64::new(0), access_count: AtomicU64::new(0), tracks: kio::Lock::new(slab::Slab::new()), + wall_epoch: OnceLock::new(), + wall_micros: AtomicU64::new(0), + wall: kio::Lock::default(), }), } } @@ -263,38 +291,142 @@ impl Pool { /// Collect idle cache entries and return the next cleanup time. /// - /// Call after polling and at the returned deadline, including when idle. - /// Calls before that deadline only advance the pool's sampled clock. A due + /// Collect idle cache entries and return the next cleanup time, like + /// [`Driver::poll`](crate::time::Driver::poll). + /// + /// Call at the returned deadline, and again whenever `waiter` wakes, including when + /// idle. Calls before that deadline only advance the pool's sampled clock. A due /// pass visits every cached group, dating accesses since the last pass and /// reclaiming idle groups except each live track's latest. Delayed calls extend /// retention. Shared pools use the latest supplied instant. /// - /// `None` means both cache policies are disabled. After enabling a capacity + /// `None` means both cache policies are disabled and no read waits on a wall-clock + /// deadline. After enabling a capacity /// with [`Self::resize`], call this again to resume periodic clock sampling. - pub fn gc(&self, now: crate::time::Instant) -> Option { - self.advance(now, true) + /// + /// It also releases reads parked on a wall-clock deadline (a group's successor + /// arriving plus the reader's budget), so the returned deadline includes the earliest + /// of those, and `waiter` wakes when a read arms one earlier than it. Until a pool is + /// first collected, no deadline fires; that first collection wakes the reads waiting + /// on one so they arm it. + pub fn gc(&self, now: crate::time::Instant, waiter: &kio::Waiter) -> Option { + self.advance(now, true, Some(waiter)) + } + + fn advance( + &self, + now: crate::time::Instant, + sweep: bool, + driver: Option<&kio::Waiter>, + ) -> Option { + let first; + let next = { + // Shared origins must not run overlapping collection passes. + let mut clock = self.inner.clock.lock().unwrap_or_else(PoisonError::into_inner); + let clock = clock.get_or_insert(Clock { + epoch: now, + now, + sweep: None, + }); + let now = now.max(clock.now); + let tick = u64::try_from(now.duration_since(clock.epoch).as_millis() / u128::from(TICK_MS)) + .expect("cache clock overflow"); + self.inner.tick.store(tick, Ordering::Relaxed); + clock.now = now; + // Collection passes are serialized by the clock lock, so only one sees the first. + first = self.inner.wall_epoch.get().is_none(); + let epoch = *self.inner.wall_epoch.get_or_init(|| now); + let micros = u64::try_from(now.duration_since(epoch).as_micros()).unwrap_or(u64::MAX); + self.inner.wall_micros.fetch_max(micros, Ordering::Relaxed); + if self.sweep_interval().is_none() { + clock.sweep = None; + } else if sweep && clock.sweep.is_none_or(|at| at <= now) { + self.sweep(); + clock.sweep = self.sweep_interval().and_then(|interval| now.checked_add(interval)); + } + clock.sweep + }; + if first { + let mut undriven = self.inner.wall.lock().undriven.take(); + undriven.wake(); + } + let wall = self.release_due(self.now_wall(), driver); + next.into_iter().chain(wall).min() + } + + /// Wake every read parked on a wall-clock deadline at or before `now`, returning the + /// earliest left and registering `driver` to hear of an earlier one. + fn release_due(&self, now: crate::time::Instant, driver: Option<&kio::Waiter>) -> Option { + loop { + let due: Vec<_> = { + let mut wall = self.inner.wall.lock(); + let mut due = Vec::new(); + while let Some(entry) = wall.due.first_entry() + && entry.key().0 <= now + { + due.push(entry.remove()); + } + if due.is_empty() { + if let Some(driver) = driver { + driver.register(&mut wall.drivers); + } + return wall.due.first_key_value().map(|((at, _), _)| *at); + } + due + }; + // Outside the lock: a track re-arms its next deadline here. + for track in due.iter().filter_map(Weak::upgrade) { + if let Some(next) = track.wakes.release_wall(now) { + self.schedule(next, None, &track); + } + } + } } - fn advance(&self, now: crate::time::Instant, sweep: bool) -> Option { - // Shared origins must not run overlapping collection passes. - let mut clock = self.inner.clock.lock().unwrap_or_else(PoisonError::into_inner); - let clock = clock.get_or_insert(Clock { - epoch: now, - now, - sweep: None, - }); - let now = now.max(clock.now); - let tick = u64::try_from(now.duration_since(clock.epoch).as_millis() / u128::from(TICK_MS)) - .expect("cache clock overflow"); - self.inner.tick.store(tick, Ordering::Relaxed); - clock.now = now; - if self.sweep_interval().is_none() { - clock.sweep = None; - } else if sweep && clock.sweep.is_none_or(|at| at <= now) { - self.sweep(); - clock.sweep = self.sweep_interval().and_then(|interval| now.checked_add(interval)); + /// The instant wall-clock staleness is judged at: the model's clock, never behind the + /// latest instant a driver collected this pool at. + /// + /// Both, because the clock alone runs behind a driver that is supplied time some other + /// way (a paused test runtime), which would leave a read woken at a deadline judging + /// it not yet reached and parking on it again. + pub(crate) fn now_wall(&self) -> crate::time::Instant { + let wall = super::clock::wall(); + match self.inner.wall_epoch.get() { + Some(epoch) => { + let driven = *epoch + Duration::from_micros(self.inner.wall_micros.load(Ordering::Relaxed)); + wall.max(driven) + } + None => wall, + } + } + + /// Whether a driver collects this pool, so a wall-clock deadline armed in it fires. + /// If not, `waiter` wakes on the first collection, to arm its deadline then. + fn watch_driven(&self, waiter: &kio::Waiter) -> bool { + // Checked under the lock the first collection drains under, so none slips between. + let mut wall = self.inner.wall.lock(); + if self.inner.wall_epoch.get().is_some() { + return true; + } + waiter.register(&mut wall.undriven); + false + } + + /// Index `track`'s earliest wall-clock deadline `at` in place of `replaced`, waking the + /// drivers if it is now the pool's earliest. + fn schedule(&self, at: crate::time::Instant, replaced: Option, track: &Arc) { + let mut wall = self.inner.wall.lock(); + let address = Arc::as_ptr(track) as usize; + if let Some(replaced) = replaced { + wall.due.remove(&(replaced, address)); + } + let key = (at, address); + wall.due.insert(key, Arc::downgrade(track)); + if wall.due.first_key_value().is_some_and(|(first, _)| *first == key) { + let mut drivers = std::mem::take(&mut wall.drivers); + drop(wall); + drivers.wake(); } - clock.sweep } /// Move the pool's sampled clock forward by `duration`, dating accesses on either @@ -308,14 +440,14 @@ impl Pool { .unwrap_or_else(PoisonError::into_inner) .as_ref() .map(|clock| clock.now); - let now = now.unwrap_or_else(crate::time::Instant::now); + let now = now.unwrap_or_else(super::clock::wall); self.date(now); self.date(now + duration); } #[cfg(test)] fn date(&self, now: crate::time::Instant) { - self.advance(now, false); + self.advance(now, false, None); let tracks: Vec<_> = self .inner .tracks @@ -537,6 +669,17 @@ impl Track { &self.wakes } + /// Park `waiter` until the pool's wall clock reaches `at`. While no driver collects the + /// pool, nothing would release it, so `waiter` parks on the first collection instead. + pub(crate) fn watch_wall(self: &Arc, at: crate::time::Instant, waiter: &kio::Waiter) { + if !self.pool.watch_driven(waiter) { + return; + } + if let Some(replaced) = self.wakes.watch_wall(at, waiter) { + self.pool.schedule(at, replaced, self); + } + } + /// Charge a new group's fixed overhead, returning its [`Charge`]. pub(crate) fn charge(self: &Arc) -> Charge { self.pool.add(ENTRY_OVERHEAD); @@ -652,6 +795,11 @@ impl Drop for Track { if let Some(key) = self.sweep.get() { self.pool.unregister(*key); } + // Its deadline would otherwise linger in a shared pool until it came due. + if let Some(at) = self.wakes.wall_armed() { + let address = std::ptr::from_ref::(self) as usize; + self.pool.inner.wall.lock().due.remove(&(at, address)); + } } } @@ -855,6 +1003,68 @@ mod test { Pool::new(config) } + struct Flag(std::sync::atomic::AtomicBool); + + impl std::task::Wake for Flag { + fn wake(self: Arc) { + self.0.store(true, Ordering::SeqCst); + } + } + + fn flag() -> (Arc, kio::Waiter) { + let flag = Arc::new(Flag(Default::default())); + let waiter = kio::Waiter::new(std::task::Waker::from(flag.clone())); + (flag, waiter) + } + + /// A wall-clock deadline armed after the driver collected wakes it to collect again by + /// then, and that collection releases the read parked on it. Before any collection, + /// nothing would release it, so the read waits for the first one to arm it then. + #[test] + fn wall_deadlines_wake_the_driver_and_then_the_read() { + let pool = Pool::unbounded(); + let track = Track::new(pool.clone(), kio::Weak::new()); + let now = super::super::clock::wall(); + let at = now + Duration::from_secs(1); + let (read, reading) = flag(); + track.watch_wall(at, &reading); + + let (driver, driving) = flag(); + assert_eq!(pool.gc(now, &driving), None, "an undriven pool indexes nothing"); + assert!( + read.0.load(Ordering::SeqCst), + "the first collection never woke the read" + ); + read.0.store(false, Ordering::SeqCst); + track.watch_wall(at, &reading); + assert!( + driver.0.load(Ordering::SeqCst), + "the driver slept through an earlier deadline" + ); + assert_eq!(pool.gc(now, &driving), Some(at)); + + pool.gc(at - Duration::from_millis(1), &driving); + assert!(!read.0.load(Ordering::SeqCst), "released early"); + assert_eq!(pool.gc(at, &driving), None); + assert!(read.0.load(Ordering::SeqCst), "the read was never released"); + } + + /// A track indexes one wall-clock deadline, its earliest, and takes it along when dropped. + #[test] + fn a_dropped_track_leaves_the_wall_index() { + let pool = Pool::unbounded(); + let now = super::super::clock::wall(); + pool.gc(now, &kio::Waiter::noop()); + let track = Track::new(pool.clone(), kio::Weak::new()); + let (_read, reading) = flag(); + track.watch_wall(now + Duration::from_secs(2), &reading); + track.watch_wall(now + Duration::from_secs(1), &reading); + assert_eq!(pool.inner.wall.lock().due.len(), 1, "the later deadline lingers"); + + drop(track); + assert!(pool.inner.wall.lock().due.is_empty(), "the dropped track lingers"); + } + #[test] fn unbounded_never_accrues() { let pool = Pool::unbounded(); @@ -1104,21 +1314,24 @@ mod test { fn collecting_before_the_deadline_does_not_postpone_it() { let pool = Pool::new(Config::default().with_expiry(Duration::from_secs(2))); let now = crate::time::Instant::now(); - let deadline = pool.gc(now); - assert_eq!(pool.gc(now + Duration::from_millis(500)), deadline); + let deadline = pool.gc(now, &kio::Waiter::noop()); + assert_eq!( + pool.gc(now + Duration::from_millis(500), &kio::Waiter::noop()), + deadline + ); } #[test] fn bounded_pools_sample_recency_without_expiration() { let pool = Pool::unbounded(); let now = crate::time::Instant::now(); - assert_eq!(pool.gc(now), None); + assert_eq!(pool.gc(now, &kio::Waiter::noop()), None); pool.resize(1024); - assert_eq!(pool.gc(now), Some(now + DEFAULT_EXPIRY / 2)); - pool.gc(now + DEFAULT_EXPIRY); + assert_eq!(pool.gc(now, &kio::Waiter::noop()), Some(now + DEFAULT_EXPIRY / 2)); + pool.gc(now + DEFAULT_EXPIRY, &kio::Waiter::noop()); assert!(pool.now() > 0); pool.resize(None); - assert_eq!(pool.gc(now + DEFAULT_EXPIRY), None); + assert_eq!(pool.gc(now + DEFAULT_EXPIRY, &kio::Waiter::noop()), None); } #[test] diff --git a/rs/moq-net/src/model/clock.rs b/rs/moq-net/src/model/clock.rs index f1a1a3dc195..6159032e181 100644 --- a/rs/moq-net/src/model/clock.rs +++ b/rs/moq-net/src/model/clock.rs @@ -16,6 +16,31 @@ pub(crate) fn now() -> crate::time::Instant { crate::time::Instant::now() } +/// The instant a cache judges wall-clock staleness by, before the time its drivers +/// supplied is folded in; see `cache::Pool::now_wall`. +/// +/// The ambient runtime's clock where there is one, so a paused test runtime pauses +/// staleness along with the drivers it supplies time to. +#[cfg(not(test))] +pub(crate) fn wall() -> crate::time::Instant { + #[cfg(all(feature = "async", not(target_family = "wasm")))] + let now = web_async::time::Instant::now().into_std(); + #[cfg(not(all(feature = "async", not(target_family = "wasm"))))] + let now = now(); + now +} + +/// Unit tests freeze the wall clock, so only time a test supplies moves it: the +/// simulated clock inside `moq_net_sim`, or a pool stepped by hand outside it. +#[cfg(test)] +pub(crate) fn wall() -> crate::time::Instant { + static FROZEN: LazyLock = LazyLock::new(now); + match moq_net_sim::is_running() { + true => moq_net_sim::now(), + false => *FROZEN, + } +} + /// Where [`Timestamp`](crate::Timestamp)s count from: an instant, and the /// timestamp it maps to. /// diff --git a/rs/moq-net/src/model/expiry.rs b/rs/moq-net/src/model/expiry.rs index 9b8de407c2f..1b88e943faf 100644 --- a/rs/moq-net/src/model/expiry.rs +++ b/rs/moq-net/src/model/expiry.rs @@ -1,5 +1,8 @@ //! Wakes for group reads parked on their subscription's drift budget, and for groups held //! across a route switch until the serving copy leaves them behind. +//! +//! The budget is judged on two clocks (see `TrackState::drifted`): media time, which only a +//! frame write advances, and the wall clock, which the cache pool's driver advances. use std::collections::BTreeMap; use std::sync::atomic::{AtomicU64, Ordering}; @@ -16,6 +19,10 @@ use super::Timestamp; /// nobody else. The successor itself is watched per group (its first frame, its abort), /// except for a group landing between the read and it, which reads park on by sequence. /// +/// A parked read also goes stale once the wall clock reaches its successor's arrival plus +/// its budget. Nothing written moves that clock, so those reads park by instant and the +/// pool's driver releases them; see `cache::Track::watch_wall`. +/// /// A group held across a route switch parks in the serving copy's index the same way, /// although that copy need not hold it; see `track::Consumer::poll_stale`. /// @@ -44,6 +51,11 @@ struct Parked { held: BTreeMap, /// The `held` size that sweeps out the lists every reader has left. held_sweep: usize, + /// Reads by wall-clock deadline, released by the pool's driver. + wall: BTreeMap, + /// The earliest `wall` deadline the pool was asked to release, so arming a later one + /// skips the pool's lock. + wall_armed: Option, } /// The smallest `held` size worth sweeping. @@ -91,6 +103,45 @@ impl Wakes { self.earliest.fetch_min(deadline, Ordering::Relaxed); } + /// Park `waiter` until the wall clock reaches `at`. When `at` is earlier than any + /// deadline the pool already releases for this track, so the caller asks it to, returns + /// `Some` of the one it replaces. + pub(crate) fn watch_wall( + &self, + at: crate::time::Instant, + waiter: &kio::Waiter, + ) -> Option> { + let mut parked = self.parked.lock(); + parked.wall.entry(at).or_default().register(waiter); + if parked.wall_armed.is_some_and(|armed| armed <= at) { + return None; + } + Some(parked.wall_armed.replace(at)) + } + + /// The wall-clock deadline the pool releases for this track next. + pub(crate) fn wall_armed(&self) -> Option { + self.parked.lock().wall_armed + } + + /// The wall clock reached `now`: wake the reads due by then, returning the earliest + /// deadline left for the pool to release next. + pub(crate) fn release_wall(&self, now: crate::time::Instant) -> Option { + let mut woken = Vec::new(); + let next = { + let mut parked = self.parked.lock(); + while let Some(entry) = parked.wall.first_entry() + && *entry.key() <= now + { + woken.push(entry.remove()); + } + parked.wall_armed = parked.wall.first_key_value().map(|(at, _)| *at); + parked.wall_armed + }; + woken.iter_mut().for_each(kio::WaiterList::wake); + next + } + /// Park `waiter` until a group lands above `sequence`. pub(crate) fn watch_landing(&self, sequence: u64, waiter: &kio::Waiter) { self.parked @@ -149,7 +200,8 @@ impl Wakes { std::mem::take(&mut *parked) }; let lists = parked.deadlines.into_values().chain(parked.landings.into_values()); - for mut list in lists.chain(parked.held.into_values()) { + let lists = lists.chain(parked.held.into_values()).chain(parked.wall.into_values()); + for mut list in lists { list.wake(); } } diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index d4771532410..58043030d20 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2378,7 +2378,8 @@ pub struct Driver { // Shared by this origin's lifecycle tasks; advanced only when polled. timers: Clock, // The cache pool this origin's groups charge into, swept on a wall-clock - // cadence so its idle window binds a track whose publisher stopped writing. + // cadence so its idle window binds a track whose publisher stopped writing, and + // collected at its reads' wall-clock deadlines. pool: cache::Pool, } @@ -2404,7 +2405,7 @@ impl Driver { waiter.batch(|| { self.timers.advance(now); let result = self.state.poll(waiter); - let gc = self.pool.gc(now); + let gc = self.pool.gc(now, waiter); if result.is_ready() { return Err(Error::Closed); } diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index ec0e70f63e3..ccb25b0bbc6 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -1627,6 +1627,76 @@ mod test { assert!(matches!(reading.read_frame().now_or_never(), Some(Err(Error::Old)))); } + /// An untimed track has no media time to convict a group no route continues, so only + /// the wall clock gives it up. Nothing is written after the replacement's successor + /// arrives, so only the deadline wake can release the reader: against a cold route, and + /// against one whose copy already caches groups past the held one, so a cursor that + /// skips ahead cannot hide the stall. + #[test] + #[cfg(feature = "async")] + fn an_untimed_group_no_route_continues_is_given_up_by_wall_clock() { + let untimed = || { + track::Producer::new( + Arc::new(broadcast::Info::default()), + "video", + crate::track::Info::default(), + ) + }; + for warm in [false, true] { + let routes = Producer::new(); + let logical = track::Request::new(Arc::new(broadcast::Info::default()), "video") + .routes(routes.consume()) + .accept(crate::track::Info::default()); + let a = untimed(); + routes.serve(a.consume()); + let mut sub = subscribe(&logical, Duration::from_secs(1)); + let mut open = a.create_group(group::Info { sequence: 0 }).unwrap(); + open.write_frame(None, b"a".as_ref()).unwrap(); + let mut reading = recv(&mut sub); + assert_eq!(read(&mut reading), Some(b"a".to_vec())); + + let b = untimed(); + let pool = b.broadcast().pool.clone(); + // A driver collects the replacement's pool, so its wall deadlines fire. + pool.step(Duration::ZERO); + let write = |sequence| { + let mut group = b.create_group(group::Info { sequence }).unwrap(); + group.write_frame(None, b"b".as_ref()).unwrap(); + group.finish().unwrap(); + }; + if warm { + (1..=3).for_each(write); + } + routes.serve(b.consume()); + if !warm { + write(1); + } + + let wake = Arc::new(Wake::default()); + let waiter = kio::Waiter::new(wake.clone().into()); + assert!( + reading.poll_read_frame(&waiter).is_pending(), + "warm {warm}: not stale yet" + ); + wake.0.store(false, std::sync::atomic::Ordering::Relaxed); + assert!( + reading.poll_read_frame(&waiter).is_pending(), + "warm {warm}: not stale yet" + ); + + pool.step(Duration::from_secs(1)); + assert!( + wake.0.load(std::sync::atomic::Ordering::Relaxed), + "warm {warm}: the deadline never woke the reader" + ); + let result = reading.poll_read_frame(&waiter); + assert!( + matches!(result, Poll::Ready(Err(Error::Old))), + "warm {warm}: {result:?}" + ); + } + } + #[test] #[cfg(feature = "async")] fn datagrams_a_new_copy_replays_are_handed_out_once() { diff --git a/rs/moq-net/src/model/subscription.rs b/rs/moq-net/src/model/subscription.rs index 487a1fb7ddd..c851b90d451 100644 --- a/rs/moq-net/src/model/subscription.rs +++ b/rs/moq-net/src/model/subscription.rs @@ -15,10 +15,10 @@ use std::{ pub struct Subscription { /// Delivery priority. Higher values preempt lower ones when bandwidth is constrained. pub priority: u8, - /// How far a group may fall behind the live edge before this subscriber gives up on it. + /// How long a superseded group stays worth delivering to this subscriber. /// /// [`Duration::ZERO`] (the default) skips immediately: group 8 arriving means group 7 - /// is abandoned. A larger budget tolerates that much reordering before giving up. + /// is abandoned, timed or not. A larger budget tolerates that much reordering before giving up. /// This never *adds* delay, since the bound is only reached once newer data is /// already that far ahead. /// @@ -43,24 +43,29 @@ pub struct Subscription { /// /// # How delay is measured /// - /// In presentation time only. A group is measured by its *reach*, where its immediate - /// successor begins, against the newest frame of the latest group: it cannot present - /// past its successor, so once everything it could still hold falls outside the budget - /// it is provably useless. The candidate needs no timestamp of its own, so an empty or - /// stalled group is bounded by its stamped successor the same way. Wall-clock - /// reclamation of idle content is the cache's own policy, not this budget's. + /// By two clocks, and a group is stale once either reaches the budget. Both start at + /// its immediate servable successor, so the newest group is never stale. /// - /// An untimed track (pre-Lite05 moq-lite, or moq-transport without TIMESCALE) has no - /// media time, so none of its groups is ever stale, and a new subscriber with no - /// [`Self::start`] begins at its latest group instead of replaying the cache. + /// In presentation time, a group is measured by its *reach*, where its successor + /// begins, against the newest frame of the latest group: it cannot present past its + /// successor, so once everything it could still hold falls outside the budget it is + /// provably useless. The candidate needs no timestamp of its own, so an empty or + /// stalled group is bounded by its stamped successor the same way. + /// + /// In wall-clock time, a group is measured by how long ago its successor arrived at + /// this hop. That bounds what media time can't: a peer whose timestamps stop + /// advancing, a successor that never presents a frame, and an untimed track + /// (pre-Lite05 moq-lite, or moq-transport without TIMESCALE), which is judged by the + /// wall clock alone. A read waiting in a superseded group wakes at that deadline only + /// when something collects the track's cache pool: an origin's driver does, and a + /// standalone track needs [`Pool::gc`](crate::cache::Pool::gc) called, as its idle + /// expiry already does. Otherwise it is judged only when something else wakes the read. pub max_delay: Duration, /// The lowest [`Position`] the publisher may deliver, or `None` for no floor. /// /// A floor, not a request: only [`Self::max_delay`] asks for data, and the floor bounds - /// how far back it may reach. `None` and a floor of group 0 mean the same thing on a - /// timed track, since nothing sits below group 0. On an untimed track, which no budget - /// can measure, `None` starts at the latest group and a floor is honored as given. - /// Delivery starts at the oldest group at or above the floor that the budget still + /// how far back it may reach. `None` and a floor of group 0 mean the same thing, since + /// nothing sits below group 0. Delivery starts at the oldest group at or above the floor that the budget still /// considers fresh, so a floor above the live edge simply waits there (a resumed /// subscription naming where it left off). /// diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 65f7c311090..4f680fd5ffc 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -100,19 +100,21 @@ pub struct Info { /// (pre-Lite05 moq-lite, moq-transport drafts 14-16, or moq-transport without /// TIMESCALE) is untimed. pub timescale: Option, - /// How far behind the live edge a group may fall, in media timestamps, before it - /// is stale. The newest group is always retained. + /// How long a superseded group stays fresh. The newest group is never stale. + /// + /// A group is stale once either clock reaches the budget: the wall-clock time since + /// its successor arrived at this hop, or how far its successor's first timestamp + /// trails the live edge (see [`Subscription::max_delay`]). An untimed track is judged + /// by the wall clock alone, and a peer whose timestamps stop advancing cannot pin a + /// superseded group. /// /// A retention bound rather than a delivery one, the inverse of an HTTP /// `Cache-Control: max-age`. [`Subscription::max_delay`] is clamped to this, since a /// group can't be waited for longer than it's kept around. Reported in TRACK_INFO so /// relays re-serve with the same window. `None` (the default) sets no limit; /// the origin cache ceiling and pool still apply. `Some(Duration::ZERO)` keeps the live edge. - /// - /// Measured against timestamps rather than the wall clock, so a congestion stall - /// (timestamps stop advancing) can't age content out. Wall-clock reclamation of - /// content nobody is accessing belongs to the cache pool's own - /// [`expiry`](crate::cache::Pool::expiry) window, not to this budget. + /// Stale groups are skipped by readers; reclaiming their bytes is the cache pool's + /// [`expiry`](crate::cache::Pool::expiry) window. /// /// This is the `Publisher Max Age` on the wire, the publisher-side half of the /// budget [`Subscription::max_delay`] sets for a subscriber. @@ -378,6 +380,10 @@ struct Slot { // lands or its stream ends; see [`Producer::receive_group`]. pending: bool, + // When the group arrived here, withheld or not: the start of the wall clock for the + // group before it (see [`TrackState::arrival`]). Read on the pool's clock. + arrived: crate::time::Instant, + // A fetched copy of a live `group` that the feed started past the frames a fetch // asked for. It runs to the end of the group too, so it is served in the live // group's place while it lasts; `group` stays the slot, still written by the feed, @@ -691,8 +697,8 @@ impl TrackState { /// The live edge a subscription bounded at `cap` measures drift against: the /// highest-sequence group that has presented a frame, or `None` while nothing - /// stamped is servable (an unstamped track drives no expiry at all; the cache's - /// own wall-clock policy is what bounds it). + /// stamped is servable (an unstamped track drives no media-time expiry; the wall + /// clock alone judges it). /// /// One scan answers a whole poll. The highest-sequence edge in range is above /// every candidate below it and above none at or past it. Recomputing per candidate @@ -723,26 +729,6 @@ impl TrackState { }) } - /// Where a new reader with no explicit start begins on an untimed track: its newest - /// servable group below the exclusive `cap`. - /// - /// A drift budget resolves a timed track's start, but nothing is ever stale on an - /// untimed track, so the budget would replay the whole cache. `None` on a timed track, - /// or while nothing is servable yet. - fn untimed_start(&self, cap: Option) -> Option { - if self.info.as_ref()?.timescale.is_some() { - return None; - } - self.lookup - .range(( - std::ops::Bound::Unbounded, - cap.map_or(std::ops::Bound::Unbounded, std::ops::Bound::Excluded), - )) - .rev() - .find(|(_, slot)| slot.is_shown()) - .map(|(sequence, _)| *sequence) - } - /// This track's own edge under the exclusive `cap`, for measuring drift. /// /// None on an untimed track, where nothing is ever stale: its groups carry no @@ -774,17 +760,62 @@ impl TrackState { /// therefore leaves the reach unbounded until it presents its first frame, and so /// does having no servable successor below `cap` at all. fn reach(&self, sequence: u64, cap: Option) -> Option { - self.first_servable(sequence.saturating_add(1), cap)?.timestamp() + self.successor(sequence, cap)?.group.timestamp() } - /// The first servable group in `from..cap`, stamped or not. - fn first_servable(&self, from: u64, cap: Option) -> Option<&group::Producer> { + /// The group after `sequence` that bounds it: the first servable one below `cap`, + /// stamped or not, empty or not. + fn successor(&self, sequence: u64, cap: Option) -> Option<&Slot> { self.lookup - .range(from..) + .range(sequence.saturating_add(1)..) .map(|(_, slot)| slot) .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap)) .find(|slot| slot.is_shown()) - .map(|slot| &slot.group) + } + + /// The group after `sequence` that starts its wall clock: like [`Self::successor`], but + /// counting a group a route is still withholding. Its header arrived, so a relay ages the + /// group before it as the origin does, even if its first frame never comes. + fn arrival(&self, sequence: u64, cap: Option) -> Option<&Slot> { + self.lookup + .range(sequence.saturating_add(1)..) + .map(|(_, slot)| slot) + .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap)) + .find(|slot| slot.is_live()) + } + + /// The slot `select` picks, with `waiter` parked on its abort. + /// + /// An abort only closes the group, never touching the track, so one that landed before + /// registering re-selects, or the replacement would go unwatched. + fn watch_abort<'a>( + &'a self, + waiter: &kio::Waiter, + select: impl Fn(&'a Self) -> Option<&'a Slot>, + ) -> Option<&'a Slot> { + loop { + match select(self) { + Some(slot) if slot.group.poll_closed(waiter).is_ready() && slot.group.is_aborted() => continue, + slot => return slot, + } + } + } + + /// Whether `sequence`'s successor arrived a full `budget` ago. + /// + /// The wall-clock twin of [`Self::reach`], re-read the same way rather than latched: + /// a group that lands between the two restarts the clock, and an aborted successor + /// hands it to the next one. The clock is read only once there is a successor, + /// so judging the newest group, the common case, costs no clock read. + fn superseded(&self, sequence: u64, cap: Option, budget: Duration) -> bool { + self.arrival(sequence, cap) + .is_some_and(|successor| self.aged(successor, budget)) + } + + /// Whether a full `budget` passed since `successor` arrived. + fn aged(&self, successor: &Slot, budget: Duration) -> bool { + let now = self.cache.pool().now_wall(); + now.saturating_duration_since(successor.arrived) >= budget } /// Whether the group at `sequence` has drifted further behind `edge` than `budget` @@ -799,9 +830,13 @@ impl TrackState { /// /// A group's reach is bounded by its immediate successor (see [`Self::reach`]): it /// cannot present past where the next group begins. The candidate itself needs no - /// timestamp: an empty group is bounded by its stamped successor the same way. Only - /// timestamps drive expiry, so nothing on an untimed track is ever stale; wall-clock - /// reclamation of idle content is the cache's own policy, not the budget's. + /// timestamp: an empty group is bounded by its stamped successor the same way. + /// + /// The wall clock judges it too: a group is also stale once its successor arrived a + /// full `budget` ago (see [`Self::superseded`]), so a peer whose timestamps stop + /// advancing, an untimed track, or an empty successor that never presents cannot pin + /// a superseded group. Media time still convicts first when a burst arrives at once. + /// The newest group has no successor, so neither clock ever convicts it. /// /// The bound is exclusive, so the comparison is `>=` rather than `>`: the freshest frame /// a group could still hold sits just *below* its reach, so an age equal to the budget @@ -816,8 +851,11 @@ impl TrackState { } /// [`Self::is_stale`] for group `sequence`, whether this track holds it or not: its - /// successor here starts a full `budget` behind the edge. + /// successor here starts a full `budget` behind the edge, or arrived a full `budget` ago. fn drifted(&self, sequence: u64, edge: &Edge, budget: Duration) -> bool { + if self.superseded(sequence, edge.cap, budget) { + return true; + } self.live_above(sequence, edge) .is_some_and(|live| live.behind(self.reach(sequence, edge.cap), budget)) } @@ -831,28 +869,32 @@ impl TrackState { } /// [`Self::drifted`] under `cap`, parking `waiter` on exactly what else can move the - /// verdict: the successor's first frame or abort, the edge's abort, and the edge - /// crossing the deadline. Parking on every track change instead wakes each of N parked reads per append. + /// verdict: the successor's first frame or abort, the edge's abort, the edge crossing + /// the media deadline, and the wall clock crossing the successor's arrival plus + /// `budget`. Parking on every track change instead wakes each of N parked reads per append. /// /// The caller parks on a group landing above `sequence` first, in the way that fits /// whether this track holds it. fn poll_drifted(&self, sequence: u64, cap: Option, budget: Duration, waiter: &kio::Waiter) -> bool { + // Unbounded until a group lands above. Registered before judging, so a racing abort + // is observed or wakes us, and armed for whichever successor this pass selected, so + // re-selecting re-arms it. + let Some(arrival) = self.watch_abort(waiter, |state| state.arrival(sequence, cap)) else { + return false; + }; + if let Some(deadline) = arrival.arrived.checked_add(budget) { + self.cache.watch_wall(deadline, waiter); + } + if self.aged(arrival, budget) { + return true; + } let reach = loop { - // An abort can change reach even after the successor is stamped. - // Register before judging, so a racing abort is observed or wakes us. - // An abort only closes the group, never touching the track, so one that - // landed before registering must re-select or the replacement goes unwatched. - let successor = loop { - let successor = self.first_servable(sequence.saturating_add(1), cap); - match successor { - Some(group) if group.poll_closed(waiter).is_ready() && group.is_aborted() => continue, - successor => break successor, - } - }; - // Unbounded until a group lands above, or the successor presents its first frame. - let Some(successor) = successor else { + // An abort can change reach even after the successor is stamped. Unbounded + // until a group shows above, or the successor presents its first frame. + let Some(successor) = self.watch_abort(waiter, |state| state.successor(sequence, cap)) else { return false; }; + let successor = &successor.group; if let Some(reach) = successor.timestamp() { break reach; } @@ -1211,12 +1253,16 @@ impl TrackState { stamp, visible, pending, + arrived: self.cache.pool().now_wall(), head: None, }, ); if visible && !pending { self.arrival.push_back((sequence, stamp)); self.landed(sequence); + } else if visible { + // Withheld, but its arrival starts the wall clock of the reads it succeeds. + self.wake_landed(sequence); } } @@ -1248,6 +1294,12 @@ impl TrackState { /// Content at `sequence` became servable: wake the reads it succeeds, and show an answer /// waiting on it. fn landed(&mut self, sequence: u64) { + self.wake_landed(sequence); + self.shown(sequence); + } + + /// Wake the reads a group landing at `sequence` succeeds. + fn wake_landed(&self, sequence: u64) { let below = self .lookup .range(..sequence) @@ -1256,7 +1308,6 @@ impl TrackState { .map(|(below, _)| *below); let oldest = self.lookup.first_key_value().map_or(sequence, |(oldest, _)| *oldest); self.cache.wakes().landed(below, sequence, oldest); - self.shown(sequence); } /// Content at `sequence` became readable: an answer waiting on it now shows. @@ -2296,8 +2347,7 @@ impl Producer { /// The read cursor starts at the group the subscription named (its floor), or 0. /// [`Subscription::max_delay`] is what asks for data: delivery skips everything above /// the floor that the budget convicts, so the default budget of zero delivers only - /// the latest group and a larger one reaches back over what it can still use. An - /// untimed track has nothing to convict, so it starts at its latest group. + /// the latest group and a larger one reaches back over what it can still use. pub fn subscribe(&self, subscription: impl Into>) -> Subscriber { let preferences = subscription.into().unwrap_or_default(); @@ -3043,8 +3093,7 @@ impl Consumer { /// The read cursor starts at the group the subscription named (its floor), or 0. /// [`Subscription::max_delay`] is what asks for data: delivery skips everything above /// the floor that the budget convicts, so the default budget of zero delivers only - /// the latest group and a larger one reaches back over what it can still use. An - /// untimed track has nothing to convict, so it starts at its latest group. + /// the latest group and a larger one reaches back over what it can still use. pub fn subscribe(&self, subscription: impl Into>) -> kio::Pending { let subscription = kio::Producer::new(subscription.into().unwrap_or_default()); @@ -3891,18 +3940,12 @@ struct Cursor { impl Cursor { fn new(state: kio::Consumer, subscription: kio::Producer) -> Self { - // An explicit start says how far back to reach, so only an unfloored subscription - // jumps to an untimed track's latest group. + // The budget picks where to start, timed or not: delivery skips the cached groups + // it already judges stale, on whichever clock convicts them. let (min_sequence, datagram_index) = { let preferences = subscription.read(); let state = state.read(); - let min_sequence = match preferences.start { - Some(_) => floor_of(&preferences), - None => { - let cap = preferences.end.and_then(|end| Cap::from(end.group_end()).exclusive()); - state.untimed_start(cap).unwrap_or(0) - } - }; + let min_sequence = floor_of(&preferences); // Arrival order need not be timestamp order. Only skip the initial stale prefix; // subsequent delivery stays in arrival order, including late datagrams. let newest = state.datagrams.iter().filter_map(|datagram| datagram.timestamp).max(); @@ -4008,14 +4051,26 @@ impl Cursor { let edge = state.drift_edge(self.end_sequence); // Every group below the last successor found shares it, so a run of cached groups // that cannot serve is walked once per pass, not once per gap. - let mut successor: Option<&group::Producer> = None; - while let Some(oldest) = delivered.above_gap() - && let Some(live) = state.live_above(oldest, &edge) - { - if successor.is_none_or(|group| group.sequence <= oldest) { - successor = state.first_servable(oldest.saturating_add(1), edge.cap); + let mut arrival: Option<&Slot> = None; + let mut successor: Option<&Slot> = None; + while let Some(oldest) = delivered.above_gap() { + if arrival.is_none_or(|slot| slot.group.sequence <= oldest) { + arrival = state.arrival(oldest, edge.cap); + } + // Either clock, as `drifted` judges a group: without a successor, neither. + let Some(first) = arrival else { + break; + }; + if successor.is_none_or(|slot| slot.group.sequence <= oldest) { + successor = state.successor(oldest, edge.cap); } - if !live.behind(successor.and_then(|group| group.timestamp()), budget) { + let past = state.aged(first, budget) + || successor.is_some_and(|slot| { + state + .live_above(oldest, &edge) + .is_some_and(|live| live.behind(slot.group.timestamp(), budget)) + }); + if !past { break; } delivered.forget_below(oldest); @@ -5282,8 +5337,8 @@ mod test { #[cfg(feature = "async")] fn recv_datagram_leaves_the_ordered_cursor_alone() { let mut producer = track_producer("test", Info::timed()); - let mut datagrams = producer.subscribe(None); - let mut subscriber = producer.subscribe(None).ordered(); + let mut datagrams = producer.subscribe(replay()); + let mut subscriber = producer.subscribe(replay()).ordered(); let ts = Timestamp::from_millis(5).unwrap(); producer @@ -5936,16 +5991,19 @@ mod test { fn cache_gc_dates_activity_before_expiring_it() { let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1))); let now = crate::model::clock::now(); - assert_eq!(pool.gc(now), Some(now + Duration::from_millis(500))); + assert_eq!( + pool.gc(now, &kio::Waiter::noop()), + Some(now + Duration::from_millis(500)) + ); let producer = track_producer_pooled("test", pool.clone()); let mut group = producer.append_group().unwrap(); group.write_frame(Timestamp::ZERO, b"first".as_slice()).unwrap(); producer.append_group().unwrap(); // Activity written while cleanup was idle gets the new supplied time. let later = now + Duration::from_secs(60); - pool.gc(later); + pool.gc(later, &kio::Waiter::noop()); assert!(producer.state.read().lookup.contains_key(&0)); - pool.gc(later + Duration::from_secs(2)); + pool.gc(later + Duration::from_secs(2), &kio::Waiter::noop()); assert!(!producer.state.read().lookup.contains_key(&0)); } @@ -5957,12 +6015,12 @@ mod test { let now = crate::model::clock::now(); let groups: Vec<_> = (0..EVICT_SCAN * 3).map(|_| producer.append_group().unwrap()).collect(); producer.append_group().unwrap(); - pool.gc(now); + pool.gc(now, &kio::Waiter::noop()); // Keep more than a write scan's worth of leading entries fresh. for group in &groups[..EVICT_SCAN * 2] { group.cache_refresh(); } - pool.gc(now + expiry * 2); + pool.gc(now + expiry * 2, &kio::Waiter::noop()); let state = producer.state.read(); for sequence in 0..EVICT_SCAN * 2 { assert!(state.lookup.contains_key(&(sequence as u64)), "fresh front survives"); @@ -6032,7 +6090,7 @@ mod test { let producer = track_producer_expiring("test", cache::DEFAULT_EXPIRY); producer.append_group().unwrap().finish().unwrap(); let mut live = producer.append_group().unwrap(); - let mut consumer = producer.subscribe(None); + let mut consumer = producer.subscribe(replay()); assert_eq!(consumer.assert_group().sequence, 0); assert_eq!(consumer.assert_group().sequence, 1); @@ -6094,20 +6152,178 @@ mod test { assert!(expired, "a forwarded datagram runs expiry"); } - /// A track's `max_age` is a media-timestamp budget: it does not drive wall-clock - /// eviction, so a stall (no accesses, no timestamp progress) shorter than the - /// pool's LRU window can't age content out no matter how small the window is. + /// A stalled track keeps its newest group and gives up the ones before it by the wall + /// clock: `max_age` measures how long ago each group's successor arrived as well as + /// media time, so a publisher that stops writing cannot pin its superseded groups. #[test] - fn max_age_does_not_drive_wall_eviction() { - let producer = track_producer("test", Info::timed().with_max_age(Duration::from_secs(1))); - producer.append_group().unwrap(); // seq 0 + #[cfg(feature = "async")] + fn a_stalled_track_keeps_only_its_newest_group() { + let mut producer = track_producer("test", Info::timed().with_max_age(Duration::from_secs(1))); + for millis in [0, 10, 20] { + append_at(&mut producer, millis); + } + assert_eq!(drain(&mut producer.subscribe(replay())), vec![0, 1, 2]); - // Far past max_age in wall time, but inside the pool's LRU window. - elapse(&producer, Duration::from_secs(10)); - producer.append_group().unwrap(); // seq 1 + // Nothing is written again: media time stands still. + elapse(&producer, Duration::from_secs(2)); + assert_eq!( + drain(&mut producer.subscribe(replay())), + vec![2], + "the newest group is kept" + ); + assert_eq!( + live_groups(&producer.state.read()), + 3, + "reclaiming the bytes is the pool's job" + ); + } - let state = producer.state.read(); - assert_eq!(live_groups(&state), 2, "max_age is media time, not a wall clock"); + /// A peer whose timestamps stop advancing never ages a group out on media time. The wall + /// clock still does. + #[test] + #[cfg(feature = "async")] + fn frozen_timestamps_still_age_out_by_wall_clock() { + let mut producer = track_producer("test", Info::timed()); + let budget = Subscription::default().with_max_delay(Duration::from_secs(1)); + append_at(&mut producer, 0); + elapse(&producer, Duration::from_millis(500)); + append_at(&mut producer, 0); + elapse(&producer, Duration::from_millis(800)); + append_at(&mut producer, 0); + + // Group 0's successor arrived 800ms ago, group 1's just now. + assert_eq!(drain(&mut producer.subscribe(budget.clone())), vec![0, 1, 2]); + elapse(&producer, Duration::from_millis(200)); + assert_eq!(drain(&mut producer.subscribe(budget.clone())), vec![1, 2]); + elapse(&producer, Duration::from_secs(1)); + assert_eq!(drain(&mut producer.subscribe(budget)), vec![2]); + } + + /// Groups that arrive at once are judged by media time, which convicts them before the + /// wall clock can. + #[test] + #[cfg(feature = "async")] + fn a_startup_burst_ages_out_by_media_time() { + let mut producer = track_producer("test", Info::timed()); + for second in 0..5 { + append_at(&mut producer, second * 1000); + } + + // Group N reaches N + 1 seconds against the 4s edge: groups 0 and 1 are 3s and 2s + // behind, past 1.5s. Their successors arrived just now. + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_millis(1500))); + assert_eq!(drain(&mut subscriber), vec![2, 3, 4]); + } + + /// A read parked on a superseded group is released by the wall clock alone: nothing is + /// written after its successor arrives, so no write could wake it. That holds whether a + /// driver collected the pool before the read parked or only after. + #[test] + #[cfg(feature = "async")] + fn a_parked_read_wakes_at_its_wall_deadline() { + for (info, primed) in [Info::default(), Info::timed()] + .into_iter() + .flat_map(|info| [(info.clone(), true), (info, false)]) + { + let producer = track_producer("test", info.clone()); + let at = info.timescale.map(|_| Timestamp::ZERO); + if primed { + elapse(&producer, Duration::ZERO); + } + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(1))); + let mut held = producer.append_group().unwrap(); + held.write_frame(at, bytes::Bytes::from_static(b"x")).unwrap(); + let mut reading = subscriber.assert_group(); + assert!(reading.read_frame().now_or_never().unwrap().unwrap().is_some()); + + // The successor never presents a frame, so media time never bounds the held group. + let _successor = producer.append_group().unwrap(); + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone())))); + assert!(reading.poll_read_frame(&waiter).is_pending()); + if !primed { + elapse(&producer, Duration::ZERO); + assert!( + woken.swap(false, Ordering::SeqCst), + "{info:?}: the first collection never woke the read" + ); + assert!(reading.poll_read_frame(&waiter).is_pending()); + } + + elapse(&producer, Duration::from_millis(900)); + assert!( + !woken.load(Ordering::SeqCst), + "{info:?} primed={primed}: woken before the deadline" + ); + elapse(&producer, Duration::from_millis(100)); + assert!( + woken.load(Ordering::SeqCst), + "{info:?} primed={primed}: the wall deadline never fired" + ); + let result = reading.poll_read_frame(&waiter); + assert!(matches!(result, Poll::Ready(Ok(None))), "{info:?}: {result:?}"); + } + } + + /// A route's successor whose header landed but whose first frame never does still + /// starts the wall clock, so a relay ages the group before it as the origin does. + #[test] + #[cfg(feature = "async")] + fn a_withheld_successor_ages_its_predecessor() { + for info in [Info::default(), Info::timed()] { + let producer = track_producer("test", info.clone()); + let at = info.timescale.map(|_| Timestamp::ZERO); + elapse(&producer, Duration::ZERO); + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(1))); + let mut held = producer.append_group().unwrap(); + held.write_frame(at, bytes::Bytes::from_static(b"x")).unwrap(); + let mut reading = subscriber.assert_group(); + assert!(reading.read_frame().now_or_never().unwrap().unwrap().is_some()); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone())))); + assert!(reading.poll_read_frame(&waiter).is_pending()); + let _successor = producer.receive_group(group::Info { sequence: 1 }).unwrap(); + assert!( + woken.swap(false, Ordering::SeqCst), + "{info:?}: the header never woke the read" + ); + assert!(reading.poll_read_frame(&waiter).is_pending()); + + elapse(&producer, Duration::from_millis(900)); + assert!(!woken.load(Ordering::SeqCst), "{info:?}: woken before the deadline"); + elapse(&producer, Duration::from_millis(100)); + assert!(woken.load(Ordering::SeqCst), "{info:?}: the wall deadline never fired"); + let result = reading.poll_read_frame(&waiter); + assert!(matches!(result, Poll::Ready(Ok(None))), "{info:?}: {result:?}"); + } + } + + /// Empty groups are handed out like any other and retired by the wall clock like an + /// untimed one, whether they finished without a frame or are still open with none. A + /// publisher flooding a run of them cannot pin the group before the run either. + #[test] + #[cfg(feature = "async")] + fn a_run_of_empty_groups_ages_out_by_wall_clock() { + const RUN: u64 = 1000; + let mut producer = track_producer("test", Info::timed()); + let budget = Subscription::default().with_max_delay(Duration::from_secs(1)); + let mut subscriber = producer.subscribe(budget.clone()); + append_at(&mut producer, 0); + let mut open = Vec::new(); + for sequence in 1..=RUN { + let group = producer.append_group().unwrap(); + match sequence % 2 { + 0 => group.finish().unwrap(), + _ => open.push(group), + } + } + append_at(&mut producer, 10); + assert_eq!(drain(&mut producer.subscribe(budget.clone())).len() as u64, RUN + 2); + + elapse(&producer, Duration::from_secs(1)); + assert_eq!(drain(&mut subscriber), vec![RUN + 1], "the whole run is retired"); + assert_eq!(drain(&mut producer.subscribe(budget)), vec![RUN + 1]); } /// Disabling the pool's expiry keeps idle groups until byte pressure reclaims them. @@ -6605,7 +6821,7 @@ mod test { .unwrap(); append_at(&mut producer, 1000); append_at(&mut producer, 20_000); - let mut sub = producer.subscribe(None); + let mut sub = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(1))); let mut reading = sub.recv_group().await.unwrap().unwrap(); assert_eq!(reading.sequence, 0, "the rewound successor extends the reach"); assert!(reading.read_frame().await.unwrap().is_some()); @@ -7354,16 +7570,18 @@ mod test { #[cfg(feature = "async")] fn an_unstamped_immediate_successor_leaves_reach_unbounded() { let mut producer = track_producer("test", Info::timed()); - let mut subscriber = producer.subscribe(None); + // Any budget, so only media time judges: the wall clock stands still here. + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_millis(500))); append_at(&mut producer, 0); // seq 0 producer.append_group().unwrap(); // seq 1 stalls before its first frame append_at(&mut producer, 10_000); // seq 2 + append_at(&mut producer, 11_000); // seq 3 - // Group 1's reach is group 2's start, a full edge behind: stale at zero budget. + // Group 1's reach is group 2's start, a second behind the edge: stale. // Group 0's reach is unknown until group 1 presents its first frame, so it is // kept rather than convicted on a bound that could shrink the wrong way. - assert_eq!(drain(&mut subscriber), vec![0, 2]); + assert_eq!(drain(&mut subscriber), vec![0, 2, 3]); } /// Reach is the *immediate* successor's start, never a minimum across later groups. @@ -7423,41 +7641,47 @@ mod test { group.sequence } - /// Nothing on an untimed track is ever stale, so a budget can't resolve where a new - /// subscriber starts. It takes the latest group instead of replaying the cache. + /// The wall clock judges an untimed track, so a new subscriber starts at the oldest + /// cached group it doesn't find stale, the way a timed one does: a group is stale once + /// its successor arrived a full budget ago. #[test] #[cfg(feature = "async")] - fn an_untimed_track_starts_at_its_latest_group() { + fn an_untimed_subscriber_starts_at_the_oldest_fresh_group() { let mut producer = untimed_producer(); + // Group N's successor arrives N + 1 seconds in. for _ in 0..5 { append_untimed(&mut producer); + elapse(&producer, Duration::from_secs(1)); } + // Four seconds after group 4 arrived, groups 0 to 2 were superseded 4, 3, and 2 + // seconds ago: past a 1.5s budget. Group 3 was superseded 1s ago. + let mut patient = producer.subscribe(Subscription::default().with_max_delay(Duration::from_millis(1500))); + assert_eq!(drain(&mut patient), vec![3, 4]); + // No budget takes the newest group alone, which is never stale. let mut live = producer.subscribe(None); assert_eq!(drain(&mut live), vec![4]); - let mut patient = producer.subscribe(replay()); - assert_eq!( - drain(&mut patient), - vec![4], - "no budget reaches back on an untimed track" - ); + let mut replaying = producer.subscribe(replay()); + assert_eq!(drain(&mut replaying), vec![0, 1, 2, 3, 4]); append_untimed(&mut producer); assert_eq!(drain(&mut live), vec![5], "every later group is delivered"); } - /// An explicit start says how far back to reach, so it wins over the untimed jump to - /// the latest group (a resume after failover names the group it lacks). + /// An explicit start is a floor, not a request: the budget still skips what it finds + /// stale above it, as on a timed track. #[test] #[cfg(feature = "async")] - fn an_explicit_start_holds_on_an_untimed_track() { + fn an_explicit_start_floors_an_untimed_track() { let mut producer = untimed_producer(); for _ in 0..5 { append_untimed(&mut producer); } - let mut subscriber = producer.subscribe(Subscription::default().with_start(Position::group(2))); + let mut subscriber = producer.subscribe(replay().with_start(Position::group(2))); assert_eq!(drain(&mut subscriber), vec![2, 3, 4]); + let mut live = producer.subscribe(Subscription::default().with_start(Position::group(2))); + assert_eq!(drain(&mut live), vec![4]); } /// An unfloored untimed subscription with an end starts at the latest group below it, @@ -7600,11 +7824,14 @@ mod test { append_at(&mut producer, 0); // seq 0 producer.append_group().unwrap(); // seq 1 stalls before its first frame append_at(&mut producer, 10_000); // seq 2 + append_at(&mut producer, 11_000); // seq 3 - // Group 1 reaches 10s, a full edge behind: convicted. Group 0 is bounded only by - // group 1, which has yet to say where it begins. - let mut subscriber = producer.subscribe(None).ordered(); - assert_eq!(drain_ordered(&mut subscriber), vec![0, 2]); + // Group 1 reaches 10s, a second behind the edge: convicted. Group 0 is bounded + // only by group 1, which has yet to say where it begins, and arrived just now. + let mut subscriber = producer + .subscribe(Subscription::default().with_max_delay(Duration::from_millis(500))) + .ordered(); + assert_eq!(drain_ordered(&mut subscriber), vec![0, 2, 3]); } #[test] @@ -8118,7 +8345,7 @@ mod test { producer.create_group(group::Info { sequence: 2 }).unwrap(); producer.finish_at(Position::group(4)).unwrap(); - let mut ordered = producer.subscribe(None).ordered(); + let mut ordered = producer.subscribe(replay()).ordered(); drop(producer); let first = ordered.next_group().now_or_never().expect("group 0").unwrap().unwrap(); @@ -8363,7 +8590,7 @@ mod test { #[cfg(feature = "async")] async fn a_pending_tail_holds_readers_at_a_hole() { let mut producer = track_producer("test", None); - let mut arrival = producer.subscribe(None); + let mut arrival = producer.subscribe(replay()); // The wire declares where the feed starts. producer.start_at(0).unwrap(); producer.finish_at_pending(Position::group(3)).unwrap(); @@ -8381,7 +8608,7 @@ mod test { // A settled tail ends readers at the boundary, hole and all. let mut producer = track_producer("test", None); - let mut arrival = producer.subscribe(None); + let mut arrival = producer.subscribe(replay()); producer.start_at(0).unwrap(); producer.finish_at_pending(Position::group(2)).unwrap(); let _high = producer.create_group(group::Info { sequence: 1 }).unwrap(); @@ -8392,7 +8619,7 @@ mod test { // So does the last producer going. let mut producer = track_producer("test", None); - let mut arrival = producer.subscribe(None); + let mut arrival = producer.subscribe(replay()); producer.start_at(0).unwrap(); producer.finish_at_pending(Position::group(2)).unwrap(); let _high = producer.create_group(group::Info { sequence: 1 }).unwrap(); @@ -8449,7 +8676,7 @@ mod test { #[cfg(feature = "async")] async fn an_end_after_the_highest_group_holds_the_reader() { let mut producer = track_producer("test", None); - let mut arrival = producer.subscribe(None); + let mut arrival = producer.subscribe(replay()); producer.start_at(0).unwrap(); let _high = producer.create_group(group::Info { sequence: 2 }).unwrap(); assert_eq!(arrival.assert_group().sequence, 2); @@ -8479,7 +8706,7 @@ mod test { #[cfg(feature = "async")] fn abort_before_the_end_settles_wins() { let mut producer = track_producer("test", Info::timed()); - let mut consumer = producer.subscribe(None); + let mut consumer = producer.subscribe(replay()); let head = producer.create_group(group::Info { sequence: 0 }).unwrap(); head.finish().unwrap(); @@ -8546,8 +8773,8 @@ mod test { #[cfg(feature = "async")] fn abort_keeps_finished_groups_for_a_slow_reader() { let producer = track_producer("test", Info::timed()); - let mut arrival = producer.subscribe(None); - let mut ordered = producer.subscribe(None).ordered(); + let mut arrival = producer.subscribe(replay()); + let mut ordered = producer.subscribe(replay()).ordered(); for sequence in 0..2 { producer @@ -8633,7 +8860,7 @@ mod test { #[cfg(feature = "async")] fn abort_after_the_end_settles_ends_clean() { let mut producer = track_producer("test", Info::timed()); - let mut consumer = producer.subscribe(None); + let mut consumer = producer.subscribe(replay()); for sequence in 0..2 { let group = producer.create_group(group::Info { sequence }).unwrap(); @@ -8654,7 +8881,7 @@ mod test { #[cfg(feature = "async")] fn abort_after_the_end_settles_ends_clean_for_an_ordered_reader() { let mut producer = track_producer("test", Info::timed()); - let mut consumer = producer.subscribe(None).ordered(); + let mut consumer = producer.subscribe(replay()).ordered(); for sequence in 0..2 { let group = producer.create_group(group::Info { sequence }).unwrap(); @@ -9229,7 +9456,7 @@ mod test { fn empty_local_cap_holds_while_another_subscriber_requests_everything() { let producer = track_producer("test", Info::timed()); let mut everything = producer.subscribe(replay()); - let mut empty = producer.subscribe(Subscription::default().with_end(Position::group(0))); + let mut empty = producer.subscribe(replay().with_end(Position::group(0))); empty.set_groups((Bound::Unbounded, Position::group(0).group_end())); for s in 0..3 { @@ -11146,6 +11373,91 @@ mod test { assert_eq!(drain_groups(&mut subscriber), [2998]); } + /// A group of one untimed frame at `sequence`. + fn write_untimed_group(producer: &Producer, sequence: u64) { + let mut group = producer.create_group(group::Info { sequence }).unwrap(); + group.write_frame(None, &b"x"[..]).unwrap(); + group.finish().unwrap(); + } + + /// The wall clock forgets the gaps on an untimed track, which has no media time to + /// measure them by. + #[test] + fn an_endless_untimed_track_with_gaps_stays_bounded() { + let producer = track_producer("test", None); + // Groups arrive 1ms apart, so the budget spans ten of them. + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_millis(10))); + let mut most = 0; + for sequence in 0..3000u64 { + elapse(&producer, Duration::from_millis(1)); + // Every third group never arrives. + if sequence % 3 == 1 { + continue; + } + write_untimed_group(&producer, sequence); + assert_eq!(drain_groups(&mut subscriber), [sequence]); + most = most.max(subscriber.delivered_runs()); + } + assert_eq!(most, 5, "the floor and the gaps within the budget"); + + // 2998 is within the budget, 2500 far past it. + write_untimed_group(&producer, 2998); + write_untimed_group(&producer, 2500); + assert_eq!(drain_groups(&mut subscriber), [2998]); + } + + /// An empty successor gives the lowest group above a gap no reach, so media time never + /// forgets that gap or any after it. The wall clock still does. + #[test] + fn an_empty_successor_still_lets_the_gaps_go() { + let producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_millis(10))); + let mut most = 0; + for sequence in 0..3000u64 { + elapse(&producer, Duration::from_millis(1)); + if sequence % 3 == 1 { + continue; + } + // Group 2 sits above the first gap, and its successor never presents a frame. + match sequence { + 3 => producer + .create_group(group::Info { sequence }) + .unwrap() + .finish() + .unwrap(), + _ => write_group(&producer, sequence), + } + assert_eq!(drain_groups(&mut subscriber), [sequence]); + most = most.max(subscriber.delivered_runs()); + } + assert_eq!(most, 5, "the floor and the gaps within the budget"); + } + + /// A gap within the budget below a run of frameless groups is forgotten once the group + /// above it is superseded for a full budget, so the run stops being rescanned. + #[test] + fn a_gap_below_a_frameless_run_lets_the_floor_advance() { + const RUN: u64 = 100; + let producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(1))); + write_group(&producer, 0); + // Group 1 never arrives. + let open: Vec<_> = (2..2 + RUN) + .map(|sequence| producer.create_group(group::Info { sequence }).unwrap()) + .collect(); + assert_eq!( + drain_groups(&mut subscriber), + [0].into_iter().chain(2..2 + RUN).collect::>() + ); + assert_eq!(subscriber.delivered_runs(), 2, "the floor and the run above the gap"); + + elapse(&producer, Duration::from_secs(1)); + write_group(&producer, 2 + RUN); + assert_eq!(drain_groups(&mut subscriber), [2 + RUN]); + assert_eq!(subscriber.delivered_runs(), 1, "the gap is forgotten"); + drop(open); + } + /// Gaps below groups aborted since, but still cached, share the successor past them, so /// one hand-out that puts them all past the budget forgets every one. The gap below a /// group still within it is kept. `track_gap_flush` measures the cost. @@ -11176,7 +11488,7 @@ mod test { ); } - /// An untimed track has nothing past the budget, so a sparse feed keeps every gap + /// An untimed sparse feed whose groups all arrive within the budget keeps every gap /// and still hands out every group. The hand-out cost stays flat over the retained /// gaps; `track_sparse_read` measures it. #[test] diff --git a/rs/moq-net/tests/fetch_from_feed.rs b/rs/moq-net/tests/fetch_from_feed.rs index 207c1351ca4..46605da88df 100644 --- a/rs/moq-net/tests/fetch_from_feed.rs +++ b/rs/moq-net/tests/fetch_from_feed.rs @@ -149,8 +149,10 @@ async fn an_unfed_fetch_is_a_miss() { assert!(matches!(err, moq_net::Error::NotFound), "{version}: no feed: {err}"); relay.track.write_frame(None, &b"1"[..]).unwrap(); - relay.track.write_frame(None, &b"2"[..]).unwrap(); - // Unfloored on an untimed track, the feed starts at the newest group. + // The publisher skips group 2, so the feed goes past it. + let mut skipped = relay.track.create_group(moq_net::group::Info { sequence: 3 }).unwrap(); + skipped.write_frame(None, &b"3"[..]).unwrap(); + skipped.finish().unwrap(); let mut subscriber = moq_net_sim::timeout(TIMEOUT, relay.remote.track("video").unwrap().subscribe(None)) .await .expect("subscribe timeout") @@ -161,8 +163,8 @@ async fn an_unfed_fetch_is_a_miss() { .unwrap() .expect("a group") .sequence; - assert_eq!(first, 2, "{version}"); - let err = fetch_err(&relay, 1).await; + assert_eq!(first, 3, "{version}"); + let err = fetch_err(&relay, 2).await; assert!(matches!(err, moq_net::Error::NotFound), "{version}: passed: {err}"); } } diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index d5c5e7f2ead..5e644ad9cac 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -378,16 +378,7 @@ async fn rejoin_during_the_cancel_skips_the_cache() { let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap(); let group = sub.recv_group().await.unwrap().unwrap(); - // A track that arrives untimed has no media age to skip the cached group by, so - // its reader starts at the newest group cached when it subscribed and reads on - // from there. - if sub.info().timescale.is_none() { - let mut sequence = group.sequence; - while sequence < 3 { - sequence = sub.recv_group().await.unwrap().unwrap().sequence; - } - return; - } + // A track that arrives untimed skips the cached group by the wall clock instead. assert_eq!( group.sequence, 3, "{version}: the rejoining reader got the stale cache first"