diff --git a/quest/m1/README.md b/quest/m1/README.md index 9162652ca96..cae2ad1a76c 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -107,7 +107,6 @@ explicit exception to the public model audit. - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide - [T-STD TS export](/quest/m1/tstd/README.md) - `moq export ts` is a proper remux that passes the supported T-STD buffer models, starting with a fixed `--delay` - [Draft 14-16 updates](/quest/m1/ietf-legacy-updates.md) - Rust and JS apply and answer moq-transport 14-16 request updates without ending or leaking the request -- [Tail arrivals](/quest/m1/tail-arrivals.md) - a track judges its pending tail from recorded arrivals, not a cache scan, and errors a reader whose tail ends short - [Resume reorder](/quest/m1/resume-reorder.md) - a group whose stream header arrives after its successor's still resumes on failover; only a dropped or long-gone group fails - [Refuse rewinds](/quest/m1/refuse-rewinds.md) - a timed track refuses a group whose first frame is earlier than the previous group's, in Rust and js/net, on every hop - [JS pending tail](/quest/m1/js-pending-tail.md) - a JS reader holds for a track's pending tail like Rust diff --git a/quest/m1/js-pending-tail.md b/quest/m1/js-pending-tail.md index 2c0b0fc7603..17e3704fcf7 100644 --- a/quest/m1/js-pending-tail.md +++ b/quest/m1/js-pending-tail.md @@ -12,16 +12,16 @@ Mirror the Rust pending-tail hold from #4225 (`rs/moq-net/src/model/track.rs`) and test it against the Rust behavior with mocked time. Split from [JS session parity](https://github.com/moq-dev/moq/pull/5214) on 2026-10-08 so the caps work stays ready. -Mirror the arrival record from [Tail arrivals](/quest/m1/tail-arrivals.md) -rather than #4225's cache scan (decided 2026-10-10), so JS ports the final -design once. +Mirror the Rust arrival record rather than #4225's cache scan (decided +2026-10-10): `TrackState::arrivals` in `rs/moq-net/src/model/track.rs`, a +`Ledger` (`rs/moq-net/src/model/ledger.rs`) of the live groups and datagrams +that arrived, which eviction never touches. A gap between two arrivals folds +once it is a subscription's grace old, the same rule as the session's +`tail::Tail`. A reader whose pending tail still owes a group when the track +aborts gets the abort error, not a clean end. Public API: none. Wire: none. -## Required - -- [Tail arrivals](/quest/m1/tail-arrivals.md) - Rust judges tail holes from recorded arrivals and errors a short tail - ## Related - [JS session parity](https://github.com/moq-dev/moq/pull/5214) - per-session caps, the other half diff --git a/quest/m1/public-model/README.md b/quest/m1/public-model/README.md index 127ded010d5..403e0e53186 100644 --- a/quest/m1/public-model/README.md +++ b/quest/m1/public-model/README.md @@ -38,8 +38,7 @@ Rules every child follows: growing a public shape it deletes: fetch and per-group requests ([Model ranges](/quest/m1/subscribe-ranges/model.md)), largest and liveness ([Live largest location](/quest/m1/largest.md)), announce - interest ([Rust lazy discovery](/quest/m1/lazy-discovery/rust.md)), and - the received tail ([Tail arrivals](/quest/m1/tail-arrivals.md)). + interest ([Rust lazy discovery](/quest/m1/lazy-discovery/rust.md)). Inventory method, to rerun before each child starts (counts drift). Rust: replace every `pub(crate)` under `rs/moq-net/src/model/` with diff --git a/quest/m1/public-model/ingest.md b/quest/m1/public-model/ingest.md index 7c499ba14a5..596ff101884 100644 --- a/quest/m1/public-model/ingest.md +++ b/quest/m1/public-model/ingest.md @@ -108,13 +108,17 @@ Rust items (`rs/moq-net/src/model/`), as they stood at b4eab0db1: - **Received end** (`track::Producer::finish_at_pending`, `set_tail_pending`; both subscribers). Without it, a custom subscriber over unordered streams ends readers at a hole. **Decision**: - - Public after [Tail arrivals](/quest/m1/tail-arrivals.md), as one call - that returns a handle the session settles when its tail is in. + - Public after Tail arrivals, as one call that returns a handle the + session settles when its tail is in. - Keep it private. - ✅ Public after Tail arrivals, with the shape left to that quest. Once the model records arrivals per track bounded by the subscription's grace, it may own the grace timer itself, so a plain `finish_at` on a received track could be enough and no settle handle is needed. + - Tail arrivals landed the record (`TrackState::arrivals`) without a + timer: the model folds a gap only on a write, so the session still + settles the tail with `set_tail_pending(false)`. The shape is this + quest's to settle. - **Session end** (`track::Producer::close`, `abort_session`; both subscribers): public. Fold `abort_session` into the public `abort` (abort open groups too) unless the implementer finds a reader-visible difference, @@ -181,7 +185,6 @@ cover the behavior. ## Required - [Narrow the model before it goes public](/quest/m1/public-model/narrow.md) - its visibility rewrite and marker lint guard every model API this child edits -- [Tail arrivals](/quest/m1/tail-arrivals.md) - settles the received-end semantics before they become public ## Related diff --git a/quest/m1/resume-reorder.md b/quest/m1/resume-reorder.md index 0f61117f1e5..4603af6efd4 100644 --- a/quest/m1/resume-reorder.md +++ b/quest/m1/resume-reorder.md @@ -15,9 +15,9 @@ answers that the feed went past `seq` as soon as any newer group arrived beat N's. Resume (`rs/moq-net/src/model/resume.rs`, the refusal verdict) then fails an in-flight group it could have continued. -- Decided: build on [Tail arrivals](/quest/m1/tail-arrivals.md), the per-track - record of live arrivals in the model, and make `poll_group`'s "passed" verdict - read it. A gap stays open for the subscription's grace, as `tail::Tail` does +- Decided: build on the per-track record of live arrivals in the model + (`TrackState::arrivals`, a `model::ledger::Ledger`), and make + `poll_group`'s "passed" verdict read it. A gap stays open for the subscription's grace, as `tail::Tail` does in the lite subscriber today. - Decided: #5209's held fetches move off their local `Tail` read onto this verdict, so one rule remains. @@ -28,10 +28,6 @@ fails an in-flight group it could have continued. Public API: behavior only. Wire: none. -## Required - -- [Tail arrivals](/quest/m1/tail-arrivals.md) - the per-track arrival record this verdict reads - ## Related - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - explicit terminal dispositions end recovery without waiting for a newer group diff --git a/quest/m1/subscribe-drop-early.md b/quest/m1/subscribe-drop-early.md index a35fe915930..6cd4f65aaa3 100644 --- a/quest/m1/subscribe-drop-early.md +++ b/quest/m1/subscribe-drop-early.md @@ -69,4 +69,3 @@ Tests, with mock time: ## Related - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - the parent: lite-07, resets before the header, readers -- [Tail arrivals](/quest/m1/tail-arrivals.md) - per-track live dispositions survive cache eviction diff --git a/quest/m1/subscribe-drop.md b/quest/m1/subscribe-drop.md index 9c448855053..f67a1275b4b 100644 --- a/quest/m1/subscribe-drop.md +++ b/quest/m1/subscribe-drop.md @@ -102,6 +102,5 @@ tail interop could land first. ## Related -- [Tail arrivals](/quest/m1/tail-arrivals.md) - per-track live dispositions survive cache eviction - [Resume reorder](/quest/m1/resume-reorder.md) - late stream headers are not proof that a group was dropped - [FETCH max-delay](/quest/m1/fetch-max-delay.md) - age expiry and DROP independently settle pending recovery diff --git a/quest/m1/tail-arrivals.md b/quest/m1/tail-arrivals.md deleted file mode 100644 index 3d1e794ca5d..00000000000 --- a/quest/m1/tail-arrivals.md +++ /dev/null @@ -1,26 +0,0 @@ -# [M] A track's pending tail counts arrivals, not its cache - -## Goal - -A received track decides which groups are still missing below a declared end -from a per-track record of live groups that arrived, which eviction never -touches. An end check stops scanning the cache, and a reader whose track -ends short because a never-arrived group was aborted gets the abort error -instead of a clean end. A reader that got every group still ends cleanly. - -## Plan - -PR #4225 judges holes from the cached range (`finish_at_pending` / -`set_tail_pending` in `rs/moq-net/src/model/track.rs`), so each end check is -O(groups), and an evicted group, a datagram, or an abort over a group that -never arrived all look alike. Decided 2026-10-10: record live arrivals per -track (fetched copies never count, as in #4225), bounded by the subscription's -grace like `tail::Tail`, and error the reader on an aborted hole. -Open: whether an arrival that ages past grace before the end is declared still -counts as arrived or becomes a hole. - -Add a benchmark swept over tracks and groups held at the tail, per AGENTS.md's -fan-out rule. Mirror the record in JS through -[JS pending tail](/quest/m1/js-pending-tail.md), which waits for this. - -Public API: behavior only (a short tail errors). Wire: none. diff --git a/quest/m2/ordered-owed-wait.md b/quest/m2/ordered-owed-wait.md index fe2f00b44f0..d7a1798ce80 100644 --- a/quest/m2/ordered-owed-wait.md +++ b/quest/m2/ordered-owed-wait.md @@ -22,8 +22,8 @@ sequence until the budget elapsed. The model cannot tell "not yet" from ✅ Decided 2026-10-14 (maintainer): - Wait only on sequences the track still owes, read from the dispositions - that [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) and - [Tail arrivals](/quest/m1/tail-arrivals.md) record. Not chosen: an opt-in + that [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) and the track's + arrival record (`TrackState::arrivals`) keep. Not chosen: an opt-in waiting reader (new public API no consumer needs, and opted-in readers hit the same holes); fixing #5475 forward (discontinuity markers still stall, and unlimited-budget readers still hang on a never-produced group). @@ -71,7 +71,6 @@ Public API: behavior only. Wire: none. ## Required - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - every stream group in a subscription arrives or is named dropped, including publisher skips -- [Tail arrivals](/quest/m1/tail-arrivals.md) - per-track live dispositions survive cache eviction ## Related diff --git a/rs/moq-net/Cargo.toml b/rs/moq-net/Cargo.toml index 10c304767ba..d10165815db 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -126,6 +126,12 @@ name = "track" harness = false required-features = ["async"] +# Declares a pending tail through the hidden `fuzz` module. +[[bench]] +name = "tail" +harness = false +required-features = ["fuzz"] + [[bench]] name = "priority" harness = false diff --git a/rs/moq-net/benches/tail.rs b/rs/moq-net/benches/tail.rs new file mode 100644 index 00000000000..4c76c9325be --- /dev/null +++ b/rs/moq-net/benches/tail.rs @@ -0,0 +1,79 @@ +//! The end check of a received track whose declared end still waits on its tail. +//! +//! `track_pending_tail` holds a reader at a hole below a pending end on every track, each +//! track caching `groups` groups above the hole, and re-polls one of those readers. The +//! check is judged by the track's record of arrivals, so a cost that grows with the groups +//! cached or the tracks held shows up as a slope over either axis. +//! +//! Run with `cargo bench -p moq-net --features fuzz --bench tail`. + +use std::{hint::black_box, task::Poll, time::Duration}; + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use moq_net::{broadcast, group, kio, track}; + +/// Tracks each holding a reader at the tail. +const TRACKS: [usize; 3] = [1, 32, 1024]; + +/// Groups each track caches above the hole. +const GROUPS: [usize; 3] = [1, 32, 1024]; + +/// Readers held at a hole below a pending end, one per track. +struct Held { + _broadcast: broadcast::Producer, + _tracks: Vec, + readers: Vec, +} + +impl Held { + fn new(tracks: usize, groups: usize) -> Self { + let broadcast = broadcast::Info::default().produce(); + let waiter = kio::Waiter::noop(); + let mut readers = Vec::with_capacity(tracks); + let tracks = (0..tracks) + .map(|i| { + let mut track = broadcast.create_track(format!("bench{i}"), None).unwrap(); + // A budget wide enough that no cached group is skipped as stale. + let mut reader = + track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(3600))); + track.start_at(0).unwrap(); + let end = track::Position::group(groups as u64 + 1); + moq_net::fuzz::finish_at_pending(&mut track, end).unwrap(); + // Group 0 never arrives: the hole that holds the reader. + for sequence in 1..=groups as u64 { + track.create_group(group::Info { sequence }).unwrap().finish().unwrap(); + } + for _ in 0..groups { + assert!(matches!(reader.poll_recv_group(&waiter), Poll::Ready(Ok(Some(_))))); + } + assert!(reader.poll_recv_group(&waiter).is_pending(), "held at the hole"); + readers.push(reader); + track + }) + .collect(); + Self { + _broadcast: broadcast, + _tracks: tracks, + readers, + } + } +} + +fn bench_pending_tail(c: &mut Criterion) { + let mut group = c.benchmark_group("track_pending_tail"); + group.throughput(Throughput::Elements(1)); + for tracks in TRACKS { + for groups in GROUPS { + let id = format!("{tracks}x{groups}"); + group.bench_function(BenchmarkId::new("repoll", &id), |b| { + let mut held = Held::new(tracks, groups); + let waiter = kio::Waiter::noop(); + b.iter(|| assert!(black_box(held.readers[0].poll_recv_group(&waiter)).is_pending())); + }); + } + } + group.finish(); +} + +criterion_group!(benches, bench_pending_tail); +criterion_main!(benches); diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 4eb964978a7..ced17bfc30a 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -58,6 +58,12 @@ pub fn object_group(frames: usize, properties: bool) -> (broadcast::Producer, tr (broadcast, track, group) } +/// Declare a received track's end while the wire may still owe groups below it, as a +/// subscriber does on SUBSCRIBE_END, for the pending-tail benchmark. +pub fn finish_at_pending(track: &mut track::Producer, end: track::Position) -> crate::Result<()> { + track.finish_at_pending(end) +} + /// Every target, keyed by the name of its `fuzz_targets/.rs` shim. pub const TARGETS: &[(&str, Target)] = &[ ("lite", lite_wire), diff --git a/rs/moq-net/src/model/ledger.rs b/rs/moq-net/src/model/ledger.rs new file mode 100644 index 00000000000..f505110d989 --- /dev/null +++ b/rs/moq-net/src/model/ledger.rs @@ -0,0 +1,303 @@ +//! The sequences a feed has accounted for, and how long each gap between them is waited for. +//! +//! A subscription's tail (the session's `tail::Tail`) and a track's record of live arrivals +//! both keep one, so they give up on a missing group by the same rule. + +use std::{ops::Range, time::Duration}; + +use crate::time::Instant; + +/// How long a subscriber waits for a group stream it cannot account for once the +/// publisher has ended the subscription. +/// +/// Bounds the wait on IETF, and on moq-lite when the subscription has no max delay to bound +/// it with. Matches `@moq/net`, so a reader cannot tell which side it is talking to. +pub(crate) const GRACE: Duration = Duration::from_secs(1); + +/// The grace for a subscription with `max_delay`: how long the subscriber was willing to +/// wait for a late group anyway, or [`GRACE`] without one. +pub(crate) fn grace(max_delay: Duration) -> Duration { + match max_delay.is_zero() { + true => GRACE, + false => max_delay, + } +} + +/// A run of accounted sequences. +#[derive(Debug, Clone, PartialEq, Eq)] +struct Run { + groups: Range, + /// When the gap below this run opened, which is when its missing groups became late. + since: Instant, +} + +/// The sequences accounted for, and how late each gap between them is. +#[derive(Debug)] +pub(crate) struct Ledger { + // Disjoint, sorted, non-adjacent runs of accounted sequences. A gap older than the grace + // can no longer be waited for, so it folds into the runs around it, which bounds this by + // the gaps opened within the grace rather than every gap ever seen. A fold is final: a + // later, longer grace or a lowered floor cannot reopen it. + runs: Vec, + grace: Duration, +} + +impl Default for Ledger { + fn default() -> Self { + Self::new(GRACE) + } +} + +impl Ledger { + /// A ledger that gives up on a missing group after `grace`. + pub fn new(grace: Duration) -> Self { + Self { + runs: Vec::new(), + grace, + } + } + + /// Change the grace, for a subscription whose max delay changed. + pub fn set_grace(&mut self, grace: Duration) { + self.grace = grace; + } + + /// Record `groups` as accounted for. Gaps fold only on [`Self::expire`]. + pub fn account(&mut self, groups: Range, now: Instant) { + if groups.is_empty() { + return; + } + // Every run the insert overlaps or touches merges with it into one. + let first = self.runs.partition_point(|run| run.groups.end < groups.start); + let last = self.runs.partition_point(|run| run.groups.start <= groups.end); + let merged = match &self.runs[first..last] { + [] => Run { + groups, + // Splitting a gap leaves both halves as late as it was. A run past every + // other one opens a new gap. + since: self.runs.get(first).map_or(now, |above| above.since), + }, + [head, .., tail] => Run { + groups: head.groups.start.min(groups.start)..tail.groups.end.max(groups.end), + since: head.since, + }, + [only] => Run { + groups: only.groups.start.min(groups.start)..only.groups.end.max(groups.end), + since: only.since, + }, + }; + self.runs.splice(first..last, [merged]); + } + + /// Fold every gap older than the grace into the runs around it. + pub fn expire(&mut self, now: Instant) { + let grace = self.grace; + self.runs.dedup_by(|run, below| { + let expired = now.duration_since(run.since) >= grace; + if expired { + below.groups.end = run.groups.end; + } + expired + }); + } + + /// Whether a gap separates two runs, so [`Self::expire`] has anything to fold. + pub fn has_gaps(&self) -> bool { + self.runs.len() > 1 + } + + /// The lowest sequence accounted for. + pub fn first(&self) -> Option { + self.runs.first().map(|run| run.groups.start) + } + + /// Whether every group in `groups` is accounted for. + pub fn covers(&self, groups: Range) -> bool { + // Runs merge on insert, so one run covers the span or none does. + groups.is_empty() + || self + .runs + .get(self.runs.partition_point(|run| run.groups.end <= groups.start)) + .is_some_and(|run| run.groups.start <= groups.start && groups.end <= run.groups.end) + } +} + +// What only a session's tail reads. +#[cfg_attr(not(feature = "async"), allow(dead_code))] +impl Ledger { + /// How long a gap is waited for once a later group opened it. + pub fn grace(&self) -> Duration { + self.grace + } + + /// Restart the age of every gap reaching into `groups`, which the demand newly asks for. + pub fn demand(&mut self, groups: Range, now: Instant) { + if groups.is_empty() { + return; + } + let mut below = 0; + for run in &mut self.runs { + if below < groups.end && groups.start < run.groups.start { + run.since = now; + } + below = run.groups.end; + } + } + + /// When the ledger gives up on a missing `group`: once the grace has passed since a later + /// group opened the gap below it. `None` while it is accounted for, or while no later + /// group has arrived, since nothing says it is late yet. + pub fn folds_at(&self, group: u64) -> Option { + let above = self + .runs + .get(self.runs.partition_point(|run| run.groups.end <= group))?; + match above.groups.start <= group { + true => None, + false => above.since.checked_add(self.grace), + } + } + + /// Gaps whose age bounds a held fetch, including the gap before the first arrival. + pub fn folds(&self) -> impl Iterator, Instant)> + '_ { + let mut below = 0; + self.runs.iter().filter_map(move |run| { + let gap = below..run.groups.start; + below = run.groups.end; + if gap.is_empty() { + return None; + } + Some((gap, run.since.checked_add(self.grace)?)) + }) + } + + /// The runs of `groups` not accounted for, in order. + pub fn gaps(&self, groups: Range) -> Vec> { + let mut gaps = Vec::new(); + let mut next = groups.start; + for run in &self.runs { + if next >= groups.end { + break; + } + if run.groups.start > next { + gaps.push(next..run.groups.start.min(groups.end)); + } + next = next.max(run.groups.end); + } + if next < groups.end { + gaps.push(next..groups.end); + } + gaps + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Account `groups` and fold what aged out, as each arrival does. + fn arrive(ledger: &mut Ledger, groups: Range, now: Instant) { + ledger.account(groups, now); + ledger.expire(now); + } + + fn groups(ledger: &Ledger) -> Vec> { + ledger.runs.iter().map(|run| run.groups.clone()).collect() + } + + #[test] + fn runs_merge_across_gaps() { + let now = Instant::now(); + let mut ledger = Ledger::default(); + arrive(&mut ledger, 0..1, now); + arrive(&mut ledger, 2..3, now); + assert!(!ledger.covers(0..3), "group 1 is missing"); + assert_eq!(groups(&ledger), vec![0..1, 2..3]); + + arrive(&mut ledger, 1..2, now); + assert!(ledger.covers(0..3) && ledger.runs.len() == 1, "adjacent runs merge"); + + arrive(&mut ledger, 5..7, now); + arrive(&mut ledger, 9..10, now); + arrive(&mut ledger, 4..9, now); + assert_eq!(groups(&ledger), vec![0..3, 4..10], "one insert swallows several runs"); + assert!(ledger.covers(4..10)); + assert!(!ledger.covers(2..5)); + assert!(ledger.covers(7..7), "an empty range is always covered"); + } + + /// A gap older than the grace folds away, so a lossy track keeps only the gaps it can + /// still wait for. + #[test] + fn a_gap_past_the_grace_folds_away() { + let start = Instant::now(); + let mut ledger = Ledger::new(GRACE); + arrive(&mut ledger, 0..1, start); + arrive(&mut ledger, 2..3, start); + arrive(&mut ledger, 4..5, start + GRACE / 2); + assert_eq!(groups(&ledger), vec![0..1, 2..3, 4..5]); + + // Splitting the younger gap keeps it as late as it was. + arrive(&mut ledger, 6..7, start + GRACE / 2); + arrive(&mut ledger, 8..9, start + GRACE); + assert_eq!( + groups(&ledger), + vec![0..3, 4..5, 6..7, 8..9], + "only the oldest gap expired" + ); + + ledger.expire(start + GRACE * 2); + assert_eq!(groups(&ledger), vec![0..9]); + assert!(ledger.covers(0..9)); + } + + /// A missing group is given up on a grace after a later one opened its gap, and not + /// before anything later has arrived. + #[test] + fn a_missing_group_folds_a_grace_after_its_gap_opened() { + let start = Instant::now(); + let mut ledger = Ledger::new(GRACE); + arrive(&mut ledger, 0..1, start); + assert_eq!(ledger.folds_at(0), None, "accounted for"); + assert_eq!(ledger.folds_at(1), None, "nothing later arrived"); + + arrive(&mut ledger, 2..3, start + GRACE / 2); + assert_eq!(ledger.folds_at(1), Some(start + GRACE / 2 + GRACE)); + assert_eq!(ledger.folds_at(3), None); + } + + /// Splitting a gap leaves both halves with its age. + #[test] + fn a_split_gap_keeps_its_age() { + let start = Instant::now(); + let mut ledger = Ledger::new(GRACE); + arrive(&mut ledger, 0..1, start); + arrive(&mut ledger, 9..10, start); + arrive(&mut ledger, 5..6, start + GRACE / 2); + assert_eq!(groups(&ledger), vec![0..1, 5..6, 9..10]); + + ledger.expire(start + GRACE); + assert_eq!(groups(&ledger), vec![0..10], "both halves opened at the start"); + } + + /// A gap the demand newly asks for is late only from then, however old the run above it. + #[test] + fn a_lowered_floor_restarts_the_gap_age() { + let start = Instant::now(); + let mut ledger = Ledger::new(GRACE); + arrive(&mut ledger, 3..4, start); + arrive(&mut ledger, 9..10, start); + + let lowered = start + GRACE * 2; + ledger.demand(1..3, lowered); + arrive(&mut ledger, 1..2, lowered); + assert_eq!( + groups(&ledger), + vec![1..2, 3..10], + "only the gap the floor reached restarts" + ); + assert!(!ledger.covers(1..4)); + + ledger.expire(lowered + GRACE); + assert!(ledger.covers(1..10)); + } +} diff --git a/rs/moq-net/src/model/mod.rs b/rs/moq-net/src/model/mod.rs index 7f5c8ae3843..eba4e141de1 100644 --- a/rs/moq-net/src/model/mod.rs +++ b/rs/moq-net/src/model/mod.rs @@ -22,6 +22,7 @@ pub(crate) mod clock; pub(crate) mod datagram; mod delivered; mod expiry; +pub(crate) mod ledger; pub(crate) mod object; mod requests; mod resume; diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index d0335618cff..2bb986af93e 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -17,6 +17,7 @@ use crate::{Error, Result, Timescale, Timestamp, coding}; use crate::{broadcast, cache, group, stats}; use super::delivered::Delivered; +use super::ledger::{self, Ledger}; use super::{Datagram, Requests}; use super::Cap; @@ -307,6 +308,16 @@ pub(crate) struct TrackState { // last producer to go, rather than end at the boundary with groups still to arrive. tail_pending: bool, + // The sequences the live feed delivered, as a group or a datagram: what a pending tail + // is judged by. Fetched copies never count, and eviction never touches it. A gap folds + // away a subscription's grace after a later arrival opened it, as the session's tail + // does, so this holds only the gaps still worth waiting for. + arrivals: Ledger, + + // Cached groups a route is still withholding (see `Slot::pending`), so an end check + // need not scan the cache for one. + withheld: usize, + // No producer remains (aborted, sealed, or dropped), so nothing protects the live // edge any longer: the pool's idle expiry reclaims it like every other group, and a // stale consumer cannot pin the cache. @@ -601,6 +612,8 @@ impl TrackState { } let sequence = datagram.sequence; self.datagrams.push_back(datagram); + let now = self.cache.pool().now_wall(); + self.record_arrival(sequence, now); self.shown(sequence); } @@ -1070,7 +1083,7 @@ impl TrackState { // Already aborted: the frames are gone, reclaim the slot so a // later fetch can serve the sequence again. if slot.is_aborted() { - self.lookup.remove(&sequence); + self.remove_slot(sequence); continue; } if self.protects(sequence) || !slot.is_expired(&scan) { @@ -1085,7 +1098,7 @@ impl TrackState { // Take the group out of the cache and abort it, so any consumer // still reading surfaces `Error::Old` instead of blocking forever // on a frame that will never arrive. - self.lookup.remove(&sequence).unwrap().abort(Error::Old); + self.remove_slot(sequence).unwrap().abort(Error::Old); } } @@ -1148,6 +1161,33 @@ impl TrackState { fn drop_open_groups(&mut self) { self.lookup .retain(|_, slot| slot.copies().any(group::Producer::is_finished)); + self.withheld = self.lookup.values().filter(|slot| slot.pending).count(); + } + + /// Take `sequence` out of the cache, keeping the withheld count in step. + fn remove_slot(&mut self, sequence: u64) -> Option { + let slot = self.lookup.remove(&sequence)?; + self.withheld -= usize::from(slot.pending); + Some(slot) + } + + /// Record that the live feed delivered `sequence`, then fold the gaps that aged out. + fn record_arrival(&mut self, sequence: u64, now: crate::time::Instant) { + self.arrivals.account(sequence..sequence.saturating_add(1), now); + self.fold_arrivals(now); + } + + /// Fold every gap in the arrivals a subscription's grace stopped waiting for: the grace + /// the session's tail uses, from the max delay the track's subscribers ask for. + fn fold_arrivals(&mut self, now: crate::time::Instant) { + // Only a gap folds, so an in-order feed never reads the subscriptions. + if !self.arrivals.has_gaps() { + return; + } + let max_delay = snapshot_subscription(&self.subscriptions, self.max_age_bound()) + .map_or(Duration::ZERO, |subscription| subscription.max_delay); + self.arrivals.set_grace(ledger::grace(max_delay)); + self.arrivals.expire(now); } /// Attach the publisher's immutable metadata without replacing it with local cache policy. @@ -1209,7 +1249,7 @@ impl TrackState { if self.covering_group(sequence, frame_start).is_some() { return Err(Error::Duplicate); } - self.lookup.remove(&sequence); + self.remove_slot(sequence); Ok(()) } @@ -1246,17 +1286,23 @@ impl TrackState { } self.advance(sequence); - self.lookup.insert( + let now = self.cache.pool().now_wall(); + if visible { + self.record_arrival(sequence, now); + } + self.withheld += usize::from(pending); + let replaced = self.lookup.insert( sequence, Slot { group: group.clone(), stamp, visible, pending, - arrived: self.cache.pool().now_wall(), + arrived: now, head: None, }, ); + self.withheld -= replaced.map_or(0, |slot| usize::from(slot.pending)); self.cache.wakes().arrived(sequence..=sequence); if visible && !pending { self.arrival.push_back((sequence, stamp)); @@ -1278,6 +1324,7 @@ impl TrackState { return; } slot.pending = false; + self.withheld -= 1; self.arrival.push_back((sequence, slot.stamp)); // A group reset before its first frame shows nothing of the route's feed. if slot.group.is_aborted() { @@ -1411,7 +1458,7 @@ impl TrackState { } if slot.is_aborted() { // Aborted upstream: the frames are already gone, reclaim the slot. - self.lookup.remove(&sequence); + self.remove_slot(sequence); continue; } if self.protects(sequence) { @@ -1438,7 +1485,7 @@ impl TrackState { self.debt -= size; paid = paid.saturating_add(size); - self.lookup.remove(&sequence).unwrap().abort(Error::Evicted); + self.remove_slot(sequence).unwrap().abort(Error::Evicted); } } @@ -1529,7 +1576,7 @@ impl TrackState { let reached = self.sealed || self .final_sequence() - .is_some_and(|fin| self.edge() >= fin && !self.withholds(fin) && !self.awaits_tail(fin)); + .is_some_and(|fin| self.edge() >= fin && !self.withholds() && !self.awaits_tail(fin)); reached && (self.abort.is_none() || self.settled) } @@ -1538,44 +1585,48 @@ impl TrackState { self.max_sequence.map_or(0, |max| max.saturating_add(1)) } - /// Whether a group below `fin` is still withheld from readers until its first frame - /// lands (see [`Producer::receive_group`]): the end is not reached before it shows, or - /// readers would end without it. - fn withholds(&self, fin: u64) -> bool { - self.lookup.range(..fin).any(|(_, slot)| slot.pending) + /// Whether a cached group is still withheld from readers until its first frame lands + /// (see [`Producer::receive_group`]): the end is not reached before it shows, or + /// readers would end without it. Every cached group sits below a declared end. + fn withholds(&self) -> bool { + debug_assert_eq!(self.withheld, self.lookup.values().filter(|slot| slot.pending).count()); + self.withheld > 0 } - /// Whether a group below `fin` the wire subscription still owes has yet to arrive: - /// a sequence from where the feed starts up to the end with nothing cached. Only - /// such a hole holds readers, so a tail with nothing missing ends them at once. - /// - /// Judged by the cache, so an evicted group or a datagram also reads as a hole until - /// the session settles the tail. That only delays the end: an abort stops the wait. + /// Whether a group below `fin` the wire subscription still owes has yet to arrive. + /// Only such a hole holds readers, so a tail with nothing missing ends them at once. fn awaits_tail(&self, fin: u64) -> bool { // Nothing arrives after an abort, so a hole left then is final. - if !self.tail_pending || self.abort.is_some() { + self.abort.is_none() && self.owes(fin) + } + + /// Whether the pending tail misses a sequence from where the feed starts up to `fin`: + /// one no live group or datagram delivered, in a gap younger than the grace. + /// + /// Judged by the arrivals, not the cache, so an evicted group or a datagram counts + /// as delivered, and the check costs the same however much is cached. + fn owes(&self, fin: u64) -> bool { + if !self.tail_pending { return false; } // Fetched backfill never reaches a reader in arrival order, so only the live feed's - // groups count. Without a declared start, the lowest one stands in for it. + // arrivals count. Without a declared start, the first one stands in for it. let floor = match self.start_sequence { Some(start) => start, None if self.start_pending => return true, - None => match self.lookup.iter().find(|(_, slot)| slot.visible) { - Some((&first, _)) => first, + None => match self.arrivals.first() { + Some(first) => first, None => return fin > 0, }, - } - .min(fin); - let arrived = self.lookup.range(floor..fin).filter(|(_, slot)| slot.visible).count(); - (arrived as u64) < fin - floor + }; + !self.arrivals.covers(floor.min(fin)..fin) } - /// Whether the declared end is reached and every cached group below it finished, - /// so nothing the end promised is still in flight. + /// Whether the declared end is reached, the pending tail owes nothing below it, and + /// every cached group below it finished, so nothing the end promised is still in flight. /// - /// Decided as an abort lands, which ends any wait for the tail: a hole the cache - /// cannot tell from an evicted group does not turn a delivered track into a failure. + /// Decided as an abort lands, which ends any wait for the tail: a group the tail still + /// owed was cut off by the abort, so readers get the error rather than a clean end. fn is_settled(&self) -> bool { let Some(fin) = self.final_sequence() else { return false; @@ -1583,6 +1634,7 @@ impl TrackState { let reached = self.sealed || self.edge() >= fin; reached && (self.abort.is_none() || self.settled) + && !self.owes(fin) && self.lookup.range(..fin).all(|(_, slot)| slot.group.is_finished()) } @@ -1789,7 +1841,10 @@ fn commit_abort(mut state: kio::Mut<'_, TrackState>, err: Error) { // Snapshot the frame boundary before the open group it may sit in goes away: an // abort is exactly when a replacement route asks where to resume. state.resume = state.resume_position(); - // Decided before the open groups go: the groups below the end are the evidence. + // Decided before the open groups go: the groups below the end are the evidence. A gap + // past the grace was lost before the abort rather than cut off by it. + let now = state.cache.pool().now_wall(); + state.fold_arrivals(now); state.settled = state.is_settled(); state.abort = Some(err); // Keep what finished for consumers still draining: they get it, then the abort (or @@ -8620,11 +8675,11 @@ mod test { ); } - /// An abort ends the wait for a pending tail. Whatever reached the end finished, so - /// readers end cleanly: the cache cannot tell a hole from a group it already evicted. + /// An abort ends the wait for a pending tail. A group the tail still owed was cut off, + /// so readers get the abort rather than a clean end without it. #[moq_net_sim::test] #[cfg(feature = "async")] - async fn an_abort_ends_a_pending_tail_cleanly() { + async fn an_abort_errors_a_pending_tail_short_of_a_group() { let mut producer = track_producer("test", None); let mut arrival = producer.subscribe(None); producer.start_at(0).unwrap(); @@ -8635,6 +8690,104 @@ mod test { assert!(arrival.recv_group().now_or_never().is_none(), "held at the hole"); producer.abort(Error::Dropped).unwrap(); + assert!(matches!( + arrival.recv_group().now_or_never().unwrap(), + Err(Error::Dropped) + )); + } + + /// A reader that got every group ends cleanly, even when the track aborts before the + /// reader polls past the last one. + #[moq_net_sim::test] + #[cfg(feature = "async")] + async fn an_abort_after_a_full_tail_ends_cleanly() { + let mut producer = track_producer("test", None); + let mut arrival = producer.subscribe(replay()); + producer.start_at(0).unwrap(); + producer.finish_at_pending(Position::group(2)).unwrap(); + for sequence in [1, 0] { + producer + .create_group(group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + } + producer.clone().abort(Error::Dropped).unwrap(); + assert_eq!(arrival.assert_group().sequence, 1); + assert_eq!(arrival.assert_group().sequence, 0); + assert!(arrival.recv_group().now_or_never().unwrap().unwrap().is_none()); + } + + /// A gap the grace already gave up on was lost before the abort, not cut off by it, so + /// readers end cleanly. Within the grace, the same abort errors them. + #[moq_net_sim::test] + #[cfg(feature = "async")] + async fn an_abort_past_the_grace_ends_a_pending_tail_cleanly() { + for (waited, ended) in [(ledger::GRACE / 2, false), (ledger::GRACE, true)] { + let mut producer = track_producer("test", None); + let mut arrival = producer.subscribe(None); + producer.start_at(0).unwrap(); + producer.finish_at_pending(Position::group(3)).unwrap(); + for sequence in [0, 2] { + producer + .create_group(group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + assert_eq!(arrival.assert_group().sequence, sequence); + } + assert!(arrival.recv_group().now_or_never().is_none(), "held at the hole"); + + elapse(&producer, waited); + producer.abort(Error::Dropped).unwrap(); + let res = arrival.recv_group().now_or_never().unwrap(); + assert_eq!(matches!(res, Ok(None)), ended, "waited {waited:?}"); + } + } + + /// An evicted group still arrived, so the pending tail counts it: readers end at the + /// declared end rather than wait for the session to settle the tail. + #[test] + #[cfg(feature = "async")] + fn an_evicted_group_still_fills_a_pending_tail() { + let mut producer = track_producer("test", None); + let mut arrival = producer.subscribe(replay()); + producer.start_at(0).unwrap(); + producer.finish_at_pending(Position::group(3)).unwrap(); + for sequence in [0, 1] { + producer + .create_group(group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + assert_eq!(arrival.assert_group().sequence, sequence); + } + elapse(&producer, cache::DEFAULT_EXPIRY + Duration::from_secs(1)); + producer + .create_group(group::Info { sequence: 2 }) + .unwrap() + .finish() + .unwrap(); + assert!(!producer.state.read().lookup.contains_key(&0), "group 0 was evicted"); + assert_eq!(arrival.assert_group().sequence, 2); + assert!(arrival.recv_group().now_or_never().unwrap().unwrap().is_none()); + } + + /// A datagram is a live arrival too, so it fills its sequence in a pending tail. + #[test] + #[cfg(feature = "async")] + fn a_datagram_fills_a_pending_tail() { + let mut producer = track_producer("test", None); + let mut arrival = producer.subscribe(replay()); + producer.start_at(0).unwrap(); + producer.finish_at_pending(Position::group(2)).unwrap(); + producer.insert_datagram(0, None, b"x".as_slice()).unwrap(); + producer + .create_group(group::Info { sequence: 1 }) + .unwrap() + .finish() + .unwrap(); + assert_eq!(arrival.assert_group().sequence, 1); assert!(arrival.recv_group().now_or_never().unwrap().unwrap().is_none()); } diff --git a/rs/moq-net/src/tail.rs b/rs/moq-net/src/tail.rs index f6ac0c58eb5..f00964cd7d7 100644 --- a/rs/moq-net/src/tail.rs +++ b/rs/moq-net/src/tail.rs @@ -12,41 +12,16 @@ use std::{ops::Range, task::Poll, time::Duration}; -use crate::time::{Deadline, Instant}; - -/// How long a subscriber waits for a group stream it cannot account for once the -/// publisher has ended the subscription. -/// -/// Bounds the wait on IETF, and on moq-lite when the subscription has no max delay to bound -/// it with. Matches `@moq/net`, so a reader cannot tell which side it is talking to. -pub(crate) const GRACE: Duration = Duration::from_secs(1); - -/// The grace for a subscription with `max_delay`: how long the subscriber was willing to -/// wait for a late group anyway, or [`GRACE`] without one. -pub(crate) fn grace(max_delay: Duration) -> Duration { - match max_delay.is_zero() { - true => GRACE, - false => max_delay, - } -} - -/// A run of accounted sequences. -#[derive(Debug, Clone, PartialEq, Eq)] -struct Run { - groups: Range, - /// When the gap below this run opened, which is when its missing groups became late. - since: Instant, -} +pub(crate) use crate::model::ledger::{GRACE, grace}; +use crate::{ + model::ledger::Ledger, + time::{Deadline, Instant}, +}; /// The data streams a subscription has received, and the groups they account for. #[derive(Debug)] pub(crate) struct Tail { - // Disjoint, sorted, non-adjacent runs of accounted sequences. A gap older than the grace - // can no longer be waited for, so it folds into the runs around it, which bounds this by - // the gaps opened within the grace rather than every gap in the subscription. A fold is - // final: a later, longer grace or a lowered floor cannot reopen it. - runs: Vec, - grace: Duration, + ledger: Ledger, streams: u64, // Streams whose header arrived and that are still being read. active: u64, @@ -62,13 +37,7 @@ struct Waiting { impl Default for Tail { fn default() -> Self { - Self { - runs: Vec::new(), - grace: GRACE, - streams: 0, - active: 0, - waiting: Default::default(), - } + Self::new(GRACE) } } @@ -76,14 +45,16 @@ impl Tail { /// A tail that gives up on a missing group after `grace`. pub fn new(grace: Duration) -> Self { Self { - grace, - ..Self::default() + ledger: Ledger::new(grace), + streams: 0, + active: 0, + waiting: Default::default(), } } /// Change the grace, for a subscription whose max delay changed. pub fn set_grace(&mut self, grace: Duration) { - self.grace = grace; + self.ledger.set_grace(grace); } /// Record groups as accounted for without a stream: dropped, or sent as a datagram. @@ -91,83 +62,34 @@ impl Tail { if groups.is_empty() { return; } - self.wake_groups(groups.clone()); - // Every run the insert overlaps or touches merges with it into one. - let first = self.runs.partition_point(|run| run.groups.end < groups.start); - let last = self.runs.partition_point(|run| run.groups.start <= groups.end); - let merged = match &self.runs[first..last] { - [] => Run { - groups, - // Splitting a gap leaves both halves as late as it was. A run past every - // other one opens a new gap. - since: self.runs.get(first).map_or(now, |above| above.since), - }, - [head, .., tail] => Run { - groups: head.groups.start.min(groups.start)..tail.groups.end.max(groups.end), - since: head.since, - }, - [only] => Run { - groups: only.groups.start.min(groups.start)..only.groups.end.max(groups.end), - since: only.since, - }, - }; - self.runs.splice(first..last, [merged]); + self.ledger.account(groups, now); self.expire(now); } /// Restart the age of every gap reaching into `groups`, which the demand newly asks for. pub fn demand(&mut self, groups: Range, now: Instant) { - if groups.is_empty() { - return; - } - let mut below = 0; - for run in &mut self.runs { - if below < groups.end && groups.start < run.groups.start { - run.since = now; - } - below = run.groups.end; - } + self.ledger.demand(groups, now); } - /// Fold every gap older than the grace into the runs around it. + /// Fold every gap older than the grace into the runs around it, waking fetches held on them. pub fn expire(&mut self, now: Instant) { - let grace = self.grace; - for (gap, at) in self.folds() { + for (gap, at) in self.ledger.folds() { if at <= now { self.wake_groups(gap); } } - self.runs.dedup_by(|run, below| { - let expired = now.duration_since(run.since) >= grace; - if expired { - below.groups.end = run.groups.end; - } - expired - }); + self.ledger.expire(now); } - /// Whether every group in `groups` is accounted for. + /// See [`Ledger::covers`]. pub fn covers(&self, groups: Range) -> bool { - // Runs merge on insert, so one run covers the span or none does. - groups.is_empty() - || self - .runs - .get(self.runs.partition_point(|run| run.groups.end <= groups.start)) - .is_some_and(|run| run.groups.start <= groups.start && groups.end <= run.groups.end) + self.ledger.covers(groups) } - /// When the ledger gives up on a missing `group`: once the grace has passed since a later - /// group opened the gap below it. `None` while it is accounted for, or while no later - /// group has arrived, since nothing says it is late yet. + /// See [`Ledger::folds_at`]. pub fn folds_at(&self, group: u64) -> Option { - let above = self - .runs - .get(self.runs.partition_point(|run| run.groups.end <= group))?; - match above.groups.start <= group { - true => None, - false => above.since.checked_add(self.grace), - } + self.ledger.folds_at(group) } /// Register only for this group's accounting, including an explicit drop. @@ -191,36 +113,14 @@ impl Tail { woken.iter_mut().for_each(kio::WaiterList::wake); } - /// Gaps whose age bounds a held fetch, including the gap before the first arrival. + /// See [`Ledger::folds`]. pub fn folds(&self) -> impl Iterator, Instant)> + '_ { - let mut below = 0; - self.runs.iter().filter_map(move |run| { - let gap = below..run.groups.start; - below = run.groups.end; - if gap.is_empty() { - return None; - } - Some((gap, run.since.checked_add(self.grace)?)) - }) + self.ledger.folds() } - /// The runs of `groups` not accounted for, in order. + /// See [`Ledger::gaps`]. pub fn gaps(&self, groups: Range) -> Vec> { - let mut gaps = Vec::new(); - let mut next = groups.start; - for run in &self.runs { - if next >= groups.end { - break; - } - if run.groups.start > next { - gaps.push(next..run.groups.start.min(groups.end)); - } - next = next.max(run.groups.end); - } - if next < groups.end { - gaps.push(next..groups.end); - } - gaps + self.ledger.gaps(groups) } /// Data streams received, whether they finished or were reset. @@ -297,7 +197,7 @@ pub(crate) struct Settle { impl Settle { /// Start waiting on `tail`, giving up on missing streams after its grace. pub fn new(runtime: &crate::time::Clock, tail: kio::Consumer) -> Self { - let grace = tail.read().grace; + let grace = tail.read().ledger.grace(); Self { tail, grace: Deadline::after(runtime, grace), @@ -321,107 +221,6 @@ impl Settle { mod tests { use super::*; - fn groups(tail: &Tail) -> Vec> { - tail.runs.iter().map(|run| run.groups.clone()).collect() - } - - #[test] - fn runs_merge_across_gaps() { - let now = Instant::now(); - let mut tail = Tail::default(); - tail.account(0..1, now); - tail.account(2..3, now); - assert!(!tail.covers(0..3), "group 1 is missing"); - assert_eq!(groups(&tail), vec![0..1, 2..3]); - - tail.account(1..2, now); - assert!(tail.covers(0..3) && tail.runs.len() == 1, "adjacent runs merge"); - - tail.account(5..7, now); - tail.account(9..10, now); - tail.account(4..9, now); - assert_eq!(groups(&tail), vec![0..3, 4..10], "one insert swallows several runs"); - assert!(tail.covers(4..10)); - assert!(!tail.covers(2..5)); - assert!(tail.covers(7..7), "an empty range is always covered"); - } - - /// A gap older than the grace folds away, so a lossy track keeps only the gaps it can - /// still wait for. - #[test] - fn a_gap_past_the_grace_folds_away() { - let start = Instant::now(); - let mut tail = Tail::new(GRACE); - tail.account(0..1, start); - tail.account(2..3, start); - tail.account(4..5, start + GRACE / 2); - assert_eq!(groups(&tail), vec![0..1, 2..3, 4..5]); - - // Splitting the younger gap keeps it as late as it was. - tail.account(6..7, start + GRACE / 2); - tail.account(8..9, start + GRACE); - assert_eq!( - groups(&tail), - vec![0..3, 4..5, 6..7, 8..9], - "only the oldest gap expired" - ); - - tail.expire(start + GRACE * 2); - assert_eq!(groups(&tail), vec![0..9]); - assert!(tail.covers(0..9)); - } - - /// A missing group is given up on a grace after a later one opened its gap, and not - /// before anything later has arrived. - #[test] - fn a_missing_group_folds_a_grace_after_its_gap_opened() { - let start = Instant::now(); - let mut tail = Tail::new(GRACE); - tail.account(0..1, start); - assert_eq!(tail.folds_at(0), None, "accounted for"); - assert_eq!(tail.folds_at(1), None, "nothing later arrived"); - - tail.account(2..3, start + GRACE / 2); - assert_eq!(tail.folds_at(1), Some(start + GRACE / 2 + GRACE)); - assert_eq!(tail.folds_at(3), None); - } - - /// Splitting a gap leaves both halves with its age. - #[test] - fn a_split_gap_keeps_its_age() { - let start = Instant::now(); - let mut tail = Tail::new(GRACE); - tail.account(0..1, start); - tail.account(9..10, start); - tail.account(5..6, start + GRACE / 2); - assert_eq!(groups(&tail), vec![0..1, 5..6, 9..10]); - - tail.expire(start + GRACE); - assert_eq!(groups(&tail), vec![0..10], "both halves opened at the start"); - } - - /// A gap the demand newly asks for is late only from then, however old the run above it. - #[test] - fn a_lowered_floor_restarts_the_gap_age() { - let start = Instant::now(); - let mut tail = Tail::new(GRACE); - tail.account(3..4, start); - tail.account(9..10, start); - - let lowered = start + GRACE * 2; - tail.demand(1..3, lowered); - tail.account(1..2, lowered); - assert_eq!( - groups(&tail), - vec![1..2, 3..10], - "only the gap the floor reached restarts" - ); - assert!(!tail.covers(1..4)); - - tail.expire(lowered + GRACE); - assert!(tail.covers(1..10)); - } - /// The grace gives up on streams that never arrived, never on one still being read: /// its group, or the END_OF_TRACK it carries, still belongs to the track. #[moq_net_sim::test]