diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index d4885ac745b..8467f76058d 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -123,7 +123,6 @@ End+Start. The child quests own their focused regressions. ## Required -- [Resume duplicates](/quest/m0/broadcast-epoch/resume-duplicates.md) - a delayed group is delivered once even beyond the old 1024-entry history - [Catalog references](/quest/m0/broadcast-epoch/catalog-references.md) - players and exporters pin siblings consistently, with optional reference epochs and documented legacy best effort ## Related diff --git a/quest/m0/broadcast-epoch/resume-duplicates.md b/quest/m0/broadcast-epoch/resume-duplicates.md deleted file mode 100644 index 3fa1765f039..00000000000 --- a/quest/m0/broadcast-epoch/resume-duplicates.md +++ /dev/null @@ -1,38 +0,0 @@ -# [M] Resume never delivers a late group twice - -## Goal - -An unordered reader receives each group once across same-epoch route -changes, even when a duplicate arrives very late. A genuinely unseen late -group remains eligible within the reader's bounds and budget. - -## Plan - -Reproduced in the 2026-10-10 audit: read group 0 from A, switch to B, read -groups 1 through 1024, then deliver B's delayed copy of group 0. The reader -receives group 0 twice. `model/resume.rs` forgets delivered IDs at -`MAX_DELIVERED = 1024` without proving those groups can no longer arrive. - -Decided: this data-correctness fix gates m0. Preserve both duplicate -suppression and unordered delivery; advancing a floor past every late -group would hide the duplication by losing valid data. Bound bookkeeping -using the actual reader/copy lifetime, retained ranges, and terminal -dispositions rather than an arbitrary number of recently delivered IDs. -Choose the representation during implementation; do not retain one entry -per delivered group for the lifetime of an endless broadcast. - -Land the failing regression, a genuinely unseen late group beyond the old -window, repeated route flaps, and ordered-reader controls. Cover readers -that keep old groups open and change their group bounds, and prove finished -copies release their bookkeeping. Tests use mocked time and existing CI -lanes. Extend the resume benchmark over readers and retained/reordered -groups so memory and delivery costs are measured. - -Public API: duplicate suppression is corrected; no signature changes. -Wire: none. Update stale resume comments inline. JS handover must preserve -the corrected contract instead of copying the fixed-size history. - -## Related - -- [Resume reorder](/quest/m1/resume-reorder.md) - an unseen reordered group is not mistaken for a missing one -- [JS track handover](/quest/m1/js-group-handover.md) - the same delivery contract in JS diff --git a/quest/m0/parked-drain.md b/quest/m0/parked-drain.md index 8ce8682a52b..ad34f3f5601 100644 --- a/quest/m0/parked-drain.md +++ b/quest/m0/parked-drain.md @@ -39,7 +39,3 @@ against capped subscribers per track: the parked axis flattens, and a per-abort cost that grows with subscribers shows up as a slope. Public API: none. Wire: none. - -## Required - -- [Resume duplicates](/quest/m0/broadcast-epoch/resume-duplicates.md) - reshapes `track::Subscriber`'s delivery record in the same code diff --git a/quest/m1/js-group-handover.md b/quest/m1/js-group-handover.md index f833f4ada5e..2a9f0d67aae 100644 --- a/quest/m1/js-group-handover.md +++ b/quest/m1/js-group-handover.md @@ -29,6 +29,17 @@ missing frame and byte. It does not copy through the earlier single-writer pump or immediately cancel every old copy. Decided in the 2026-10-10 audit: mirror the current identity, reader, recovery and lifetime contract, including the late-duplicate correction, rather than porting the obsolete pump. +Decided 2026-10-13: every track reader, direct or resumed, hands out each group +sequence at most once, including a group aborted and re-created and one a +second route delivers, so no consumer ever dedupes itself. The record is a +per-reader interval set of delivered sequences on the track subscriber, shared +by direct and resumed reads, with no fixed-size history. Once a group is past +the reader's budget (max delay, clamped by max age), the reader stops expecting +every group below it and counts them as delivered, so only gaps within the +budget cost memory. A genuinely unseen late group within the budget is still +delivered. Group timestamps on a track never rewind; which late groups are +delivered after a rewind is unspecified, so the floor is judged by the +delivered group above a gap rather than the gap itself. Things to settle along the way: - Whether the request's `active` broadcast stays the same object across a swap, @@ -64,7 +75,6 @@ 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 -- [Resume duplicates](/quest/m0/broadcast-epoch/resume-duplicates.md) - the corrected once-only delivery contract for reordered copies ## Related diff --git a/quest/m1/resume-reorder.md b/quest/m1/resume-reorder.md index db77e0e8134..0f61117f1e5 100644 --- a/quest/m1/resume-reorder.md +++ b/quest/m1/resume-reorder.md @@ -34,5 +34,4 @@ Public API: behavior only. Wire: none. ## Related -- [Resume duplicates](/quest/m0/broadcast-epoch/resume-duplicates.md) - retain once-only delivery without excluding genuinely unseen late groups - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - explicit terminal dispositions end recovery without waiting for a newer group diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index b9d5b9d0492..e29c7873942 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -19,6 +19,8 @@ //! //! A route swap on one front is swept over its tracks and the copies each track //! still holds from earlier routes, and a front retiring over the fronts around it. +//! Readers draining a swapped track are swept over their count, the groups every +//! route delivered, and the routes each reader still holds a copy of. //! //! Run with `cargo bench -p moq-net --bench origin`. @@ -1125,6 +1127,129 @@ fn bench_copy_walk(c: &mut Criterion) { group.finish(); } +/// `(readers, retained, routes)` for [`bench_resume_overlap`]: readers of one track, +/// the groups every route holds, and the routes it swapped through. +const RESUME_OVERLAP: [(usize, usize, usize); 9] = [ + (1, 1, 2), + (1, 64, 2), + (1, 1024, 2), + (32, 64, 2), + (256, 64, 2), + (32, 1024, 2), + (1, 64, 8), + (1, 64, 32), + (32, 64, 8), +]; + +/// A track that swapped through its routes while its readers' local cap held every +/// group back, so each reader keeps a copy per route and every copy holds the same +/// groups, each replacement in reverse arrival order. +struct ResumeOverlap { + _producer: origin::Producer, + _generations: Vec<(broadcast::Producer, track::Producer)>, + readers: Vec, +} + +async fn resume_overlap(readers: usize, retained: usize, routes: usize) -> ResumeOverlap { + let (producer, driver) = origin::Producer::new(origin::Config::default()); + tokio::spawn(moq_net::time::run(driver)); + let epoch = Epoch::mint(); + let generation = |order: &mut dyn Iterator| { + let broadcast = producer.create_broadcast(COPY_PATH).unwrap(); + let track = broadcast.create_track("0", timed()).unwrap(); + for sequence in order { + let mut group = track + .create_group(moq_net::group::Info { + sequence: sequence as u64, + }) + .unwrap(); + group.write_frame(Timestamp::ZERO, b"f".as_ref()).unwrap(); + group.finish().unwrap(); + } + broadcast + .announce(origin::Route::default().with_epoch(epoch.clone())) + .unwrap(); + (broadcast, track) + }; + let mut generations = vec![generation(&mut (0..retained))]; + let resolved = tokio::time::timeout(COPY_WAIT, producer.consume().request_broadcast(COPY_PATH, None)) + .await + .expect("front did not resolve") + .expect("front refused"); + let track = resolved.track("0").unwrap(); + // A budget covering every group, so none is written off as late. + let subscription = track::Subscription::default().with_max_delay(Duration::from_secs(3600)); + let mut subscribers = Vec::with_capacity(readers); + for _ in 0..readers { + let mut subscriber = tokio::time::timeout(COPY_WAIT, track.subscribe(subscription.clone())) + .await + .expect("track did not splice") + .expect("subscribe failed"); + // Held back until the drain, so every route's copy is kept for its unread groups. + subscriber.set_groups(..0); + subscribers.push(subscriber); + } + let waiter = kio::Waiter::noop(); + for _ in 1..routes { + generations.push(generation(&mut (0..retained).rev())); + let (_, latest) = generations.last().unwrap(); + tokio::time::timeout(COPY_WAIT, latest.demand().used()) + .await + .expect("route swap did not splice") + .expect("replacement track closed"); + // Each reader follows the swap, subscribing to the new copy. + for subscriber in &mut subscribers { + assert!( + subscriber.poll_recv_group(&waiter).is_pending(), + "the cap holds every group" + ); + } + } + for subscriber in &mut subscribers { + subscriber.set_groups(..); + } + ResumeOverlap { + _producer: producer, + _generations: generations, + readers: subscribers, + } +} + +/// Readers draining a track whose routes all delivered every group: each group is +/// handed out once, and every other copy's duplicate is skipped against what the reader +/// handed out. The slopes over retained groups and routes are that skipping's cost. +fn bench_resume_overlap(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/resume_overlap"); + group.sample_size(10); + for (readers, retained, routes) in RESUME_OVERLAP { + group.throughput(Throughput::Elements((readers * retained) as u64)); + group.bench_function( + BenchmarkId::from_parameter(format!("{readers}r_{retained}g_{routes}c")), + |b| { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let waiter = kio::Waiter::noop(); + b.iter_batched_ref( + || runtime.block_on(resume_overlap(readers, retained, routes)), + |rig| { + for reader in &mut rig.readers { + let mut delivered = 0; + while let Poll::Ready(Ok(Some(_))) = reader.poll_recv_group(&waiter) { + delivered += 1; + } + assert_eq!(delivered, retained, "every group once"); + } + }, + BatchSize::PerIteration, + ); + }, + ); + } + group.finish(); +} + criterion_group!( benches, bench_announce, @@ -1145,6 +1270,7 @@ criterion_group!( bench_narrow, bench_relay, bench_parked, - bench_copy_walk + bench_copy_walk, + bench_resume_overlap ); criterion_main!(benches); diff --git a/rs/moq-net/benches/track.rs b/rs/moq-net/benches/track.rs index 32a1542489b..376993fe895 100644 --- a/rs/moq-net/benches/track.rs +++ b/rs/moq-net/benches/track.rs @@ -23,6 +23,17 @@ //! each append must wake only the reads it can expire, so waking the backlog shows up //! as a slope over it. //! +//! `track_sparse_read` hands out groups from a feed with every other sequence missing, +//! holding a fixed number of gaps the reader still remembers: each hand-out must cost the +//! same however many gaps it holds, so a walk over them shows up as a slope. +//! +//! `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_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. +//! //! Run with `cargo bench -p moq-net --bench track`. use std::hint::black_box; @@ -521,6 +532,204 @@ fn bench_parked_read(c: &mut Criterion) { group.finish(); } +/// Hand-outs from a sparse feed holding `gaps` remembered gaps, on an untimed track and on +/// a timed one whose budget never writes a gap off. Each iteration hands out the next group, +/// opening a gap at the top, and the lowest missing one, closing the gap at the bottom, so +/// the count stays fixed. Swept over the gaps so a per-hand-out cost that grows with them +/// shows up as a slope. +fn bench_sparse_read(c: &mut Criterion) { + let mut group = c.benchmark_group("track_sparse_read"); + group.throughput(Throughput::Elements(2)); + for (name, info) in [("untimed", track::Info::default()), ("timed", timed())] { + for gaps in [16u64, 256, 4096] { + group.bench_function(BenchmarkId::new(name, gaps), |b| { + b.iter_custom(|iterations| { + let broadcast = broadcast::Info::default().produce(); + let track = broadcast.create_track("bench", info.clone()).unwrap(); + let mut sub = + track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(3600))); + // Microseconds, so no run lasts long enough to cross the budget. + let timestamp = |sequence: u64| info.timescale.map(|_| Timestamp::from_micros(sequence).unwrap()); + let mut hand_out = |sequence: u64| { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + group + .write_frame(timestamp(sequence), Bytes::from_static(b"x")) + .unwrap(); + group.finish().unwrap(); + let Poll::Ready(Ok(Some(read))) = sub.poll_recv_group(&kio::Waiter::noop()) else { + panic!("the group is cached"); + }; + black_box(read); + }; + // Every odd sequence is missing until the bottom gap closes. + for gap in 0..gaps { + hand_out(gap * 2); + } + let start = Instant::now(); + for iteration in 0..iterations { + hand_out((gaps + iteration) * 2); + hand_out(iteration * 2 + 1); + } + start.elapsed() + }); + }); + } + } + group.finish(); +} + +/// A reader remembering `gaps` gaps, each below a handed-out group since aborted but still +/// cached, then a stamped survivor after them all. +struct GapFlush { + _broadcast: broadcast::Producer, + track: track::Producer, + sub: track::Subscriber, + gaps: u64, +} + +impl GapFlush { + fn new(gaps: u64) -> Self { + let broadcast = broadcast::Info::default().produce(); + let track = broadcast.create_track("bench", timed()).unwrap(); + let mut sub = track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(3600))); + let mut hand_out = |sequence: u64, millis: u64| { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + group + .write_frame(Timestamp::from_millis(millis).unwrap(), Bytes::from_static(b"x")) + .unwrap(); + let Poll::Ready(Ok(Some(read))) = sub.poll_recv_group(&kio::Waiter::noop()) else { + panic!("the group is cached"); + }; + black_box(read); + group + }; + // Every odd sequence is missing. + let handed: Vec<_> = (0..gaps).map(|gap| hand_out(gap * 2, gap)).collect(); + hand_out(gaps * 2, gaps).finish().unwrap(); + for group in handed { + group.abort(Error::Cancel).unwrap(); + } + Self { + _broadcast: broadcast, + track, + sub, + gaps, + } + } + + /// Hand out a group two hours on, which puts every remembered gap past the budget. + fn flush(&mut self) { + let mut group = self + .track + .create_group(moq_net::group::Info { + sequence: self.gaps * 2 + 1, + }) + .unwrap(); + group + .write_frame(Timestamp::from_millis(7_200_000).unwrap(), Bytes::from_static(b"x")) + .unwrap(); + let Poll::Ready(Ok(Some(read))) = self.sub.poll_recv_group(&kio::Waiter::noop()) else { + panic!("the group is cached"); + }; + black_box(read); + } +} + +/// One hand-out forgetting every remembered gap at once, each below a group that was +/// aborted but is still cached, swept over the gaps. The groups judged share one +/// successor past the aborted run, so finding it once per gap shows up as a slope. +fn bench_gap_flush(c: &mut Criterion) { + let mut group = c.benchmark_group("track_gap_flush"); + group.sample_size(10); + for gaps in [64u64, 512, 4096] { + group.throughput(Throughput::Elements(gaps)); + group.bench_function(BenchmarkId::from_parameter(gaps), |b| { + b.iter_batched_ref(|| GapFlush::new(gaps), GapFlush::flush, BatchSize::PerIteration); + }); + } + group.finish(); +} + +/// Readers on one track holding group 0, a run of groups that finished without a frame, +/// and the groups after it. +struct EmptyRun { + _broadcast: broadcast::Producer, + _track: track::Producer, + readers: Vec, + /// Groups each reader should be handed. + handed: usize, +} + +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 { + let broadcast = broadcast::Info::default().produce(); + let track = broadcast.create_track("bench", timed()).unwrap(); + let append = |millis: Option| { + let mut group = track.append_group().unwrap(); + if let Some(millis) = millis { + group + .write_frame(Timestamp::from_millis(millis).unwrap(), Bytes::from_static(b"x")) + .unwrap(); + } + group.finish().unwrap(); + }; + append(Some(0)); + for _ in 0..run { + append(None); + } + let after: &[u64] = if stale { &[50, 100] } else { &[5] }; + for millis in after { + append(Some(*millis)); + } + let budget = track::Subscription::default().with_max_delay(Duration::from_millis(10)); + 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, + } + } + + fn drain(&mut self) { + for reader in &mut self.readers { + let mut handed = 0; + while let Poll::Ready(Ok(Some(group))) = reader.poll_recv_group(&kio::Waiter::noop()) { + black_box(group); + handed += 1; + } + assert_eq!(handed, self.handed); + } + } +} + +/// 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`). +/// 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 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::drain, + BatchSize::PerIteration, + ); + }); + } + } + } + group.finish(); +} + criterion_group!( benches, bench_fanout, @@ -531,6 +740,9 @@ criterion_group!( bench_subscriber_churn_after_peak, bench_subscriber_join, bench_datagram_subscriber_join, - bench_parked_read + bench_parked_read, + bench_sparse_read, + bench_gap_flush, + bench_empty_run ); criterion_main!(benches); diff --git a/rs/moq-net/src/model/delivered.rs b/rs/moq-net/src/model/delivered.rs new file mode 100644 index 00000000000..07cbe2c7dab --- /dev/null +++ b/rs/moq-net/src/model/delivered.rs @@ -0,0 +1,133 @@ +//! The group sequences a reader handed out, so it never hands one out twice. + +use std::collections::BTreeMap; + +/// The group sequences a reader handed out, as runs of consecutive sequences. +/// +/// A reader stops expecting a group once it is past its budget, and counts everything +/// below it as handed out (see [`Self::forget_below`]). The run starting at 0 is that +/// floor, so only the gaps above it cost memory. +#[derive(Default)] +pub(crate) struct Delivered { + /// First to last sequence, inclusive, never adjacent to each other. + runs: BTreeMap, +} + +impl Delivered { + /// Whether `sequence` was handed out, or is below the floor. + pub fn contains(&self, sequence: u64) -> bool { + self.runs + .range(..=sequence) + .next_back() + .is_some_and(|(_, last)| sequence <= *last) + } + + /// Record `sequence` as handed out; `false` when it already was. + pub fn insert(&mut self, sequence: u64) -> bool { + if self.contains(sequence) { + return false; + } + let first = sequence + .checked_sub(1) + .and_then(|below| self.runs.range(..=below).next_back()) + .filter(|(_, last)| **last + 1 == sequence) + .map_or(sequence, |(first, _)| *first); + let last = sequence + .checked_add(1) + .and_then(|above| self.runs.remove(&above)) + .unwrap_or(sequence); + self.runs.insert(first, last); + true + } + + /// Count every sequence below `floor` as handed out. + pub fn forget_below(&mut self, floor: u64) { + let Some(below) = floor.checked_sub(1) else { + return; + }; + // Every run starting at or below the floor joins it, including one starting right at it. + let above = match floor.checked_add(1) { + Some(above) => self.runs.split_off(&above), + None => BTreeMap::new(), + }; + let last = self.runs.last_key_value().map_or(below, |(_, last)| below.max(*last)); + self.runs = above; + self.runs.insert(0, last); + } + + /// The lowest sequence handed out above a gap: once it is past the reader's budget, + /// so is the gap, since group timestamps on a track never rewind. + pub fn above_gap(&self) -> Option { + let mut runs = self.runs.keys(); + match runs.next()? { + 0 => runs.next().copied(), + first => Some(*first), + } + } + + /// How many runs it holds: one per gap, plus the floor. + #[cfg(test)] + pub fn runs(&self) -> usize { + self.runs.len() + } +} + +#[cfg(test)] +mod test { + use super::*; + + fn runs(delivered: &Delivered) -> Vec<(u64, u64)> { + delivered.runs.iter().map(|(first, last)| (*first, *last)).collect() + } + + #[test] + fn insert_joins_neighbouring_runs() { + let mut delivered = Delivered::default(); + assert!(delivered.insert(5)); + assert!(delivered.insert(7)); + assert_eq!(runs(&delivered), [(5, 5), (7, 7)]); + assert!(delivered.insert(6)); + assert_eq!(runs(&delivered), [(5, 7)]); + assert!(!delivered.insert(6)); + assert!(delivered.insert(4)); + assert!(delivered.insert(8)); + assert_eq!(runs(&delivered), [(4, 8)]); + assert!(delivered.contains(4) && delivered.contains(8)); + assert!(!delivered.contains(3) && !delivered.contains(9)); + } + + #[test] + fn the_extremes_are_sequences_too() { + let mut delivered = Delivered::default(); + assert!(delivered.insert(0)); + assert!(delivered.insert(u64::MAX)); + assert!(!delivered.insert(u64::MAX)); + assert_eq!(runs(&delivered), [(0, 0), (u64::MAX, u64::MAX)]); + delivered.forget_below(u64::MAX); + assert_eq!(runs(&delivered), [(0, u64::MAX)]); + } + + #[test] + fn forget_below_closes_the_gaps_under_the_floor() { + let mut delivered = Delivered::default(); + for sequence in [3, 5, 6, 9, 12] { + delivered.insert(sequence); + } + assert_eq!(delivered.above_gap(), Some(3)); + delivered.forget_below(0); + assert_eq!(delivered.runs(), 4); + + // A floor at a run's start takes the run along. + delivered.forget_below(5); + assert_eq!(runs(&delivered), [(0, 6), (9, 9), (12, 12)]); + assert_eq!(delivered.above_gap(), Some(9)); + assert!(delivered.contains(4), "below the floor counts as handed out"); + + // A floor inside a gap keeps the run above it. + delivered.forget_below(11); + assert_eq!(runs(&delivered), [(0, 10), (12, 12)]); + delivered.forget_below(13); + assert_eq!(runs(&delivered), [(0, 12)]); + assert_eq!(delivered.above_gap(), None); + } +} diff --git a/rs/moq-net/src/model/mod.rs b/rs/moq-net/src/model/mod.rs index 37faf7e32ef..a26b558c69f 100644 --- a/rs/moq-net/src/model/mod.rs +++ b/rs/moq-net/src/model/mod.rs @@ -19,6 +19,7 @@ mod front; mod bytes; pub(crate) mod clock; mod datagram; +mod delivered; mod expiry; mod requests; mod resume; diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 595f112f2d1..85e43c55d0e 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -13,19 +13,19 @@ //! its [`Recover`]: once its copy fails with its route, or stalls while a newer route //! serves, it continues from the serving route's copy of the same group at the same frame //! (and byte), found in that copy's cache or fetched from it. +//! +//! A group is handed out once however many copies deliver it: the reader skips any +//! sequence its [`track::Subscriber`] already handed out, the same record that keeps a +//! direct reader from taking a re-created group twice. -use std::collections::BTreeSet; use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak}; use std::task::{Poll, ready}; use crate::{Datagram, Error, Result, StreamError, group, stats, track}; +use super::delivered::Delivered; use super::subscription::{Cap, Position, Subscription, max_some}; -/// How many delivered group sequences a reader remembers, so a group two routes both -/// deliver is handed out once. -const MAX_DELIVERED: usize = 1024; - /// What the front says about a logical track: the copy to read now, and how it ended. #[derive(Default)] struct Route { @@ -150,9 +150,7 @@ impl Consumer { mirrored, generation: None, copies: Vec::new(), - delivered: BTreeSet::new(), newest: None, - ordered: None, datagram: None, groups: (0, Cap::from(..)), }) @@ -320,14 +318,38 @@ impl Subscriber { &self.subscription } - /// Poll for the next group: in arrival order, or in sequence order for `ordered`. - pub(crate) fn poll_group(&mut self, ordered: bool, waiter: &kio::Waiter) -> Poll>> { - self.reader().poll_group(ordered, waiter) + /// Poll for the next group not in `delivered`: in arrival order, or in sequence order + /// for `ordered`. + pub(crate) fn poll_group( + &mut self, + ordered: bool, + delivered: &Delivered, + waiter: &kio::Waiter, + ) -> Poll>> { + self.reader().poll_group(ordered, delivered, waiter) + } + + /// Poll for the next datagram in arrival order. `delivered` decides which replaced + /// copies still have a group to hand out. + pub(crate) fn poll_recv_datagram( + &mut self, + delivered: &Delivered, + waiter: &kio::Waiter, + ) -> Poll>> { + self.reader().poll_recv_datagram(delivered, waiter) } - /// Poll for the next datagram in arrival order. - pub(crate) fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll>> { - self.reader().poll_recv_datagram(waiter) + /// Forget the gaps in `delivered` below each group past the reader's budget, judged on + /// the newest copy that answered; see [`track::Subscriber::forget_past_budget`]. + pub(crate) fn forget_past_budget(&self, delivered: &mut Delivered) { + let reader = self.reader(); + let mut ready = reader.copies.iter().rev().filter_map(|copy| match ©.sub { + Sub::Ready(sub) => Some(sub), + Sub::Pending(_) => None, + }); + if let Some(sub) = ready.next() { + sub.forget_past_budget(delivered); + } } /// Poll for the track's declared exclusive end. @@ -413,12 +435,8 @@ struct Reader { generation: Option, /// Oldest first: the last one is the serving route's, when it is subscribed. copies: Vec, - /// The groups handed out, so a group two copies deliver is handed out once. - delivered: BTreeSet, /// The newest group handed out: where a new copy is asked to start. newest: Option, - /// Reading in sequence order: the last group handed out. - ordered: Option, /// The newest datagram handed out. datagram: Option, /// The local group filter; see [`track::Subscriber::set_groups`]. @@ -485,11 +503,9 @@ impl Reader { } /// Drop replaced copies only once neither a handed-out nor an unread group needs them. - fn let_go(&mut self) { + fn let_go(&mut self, delivered: &Delivered) { let serving = self.generation; let groups = self.groups; - let ordered = self.ordered; - let delivered = &self.delivered; self.copies.retain_mut(|copy| { if copy.needed(Some(copy.generation) == serving) { return true; @@ -505,35 +521,22 @@ impl Reader { // order and applies the current bounds and drift budget. Ignore the local // cap here, since unread groups become eligible again when it rises. sub.has_unread_group(&track::Unread { - start: groups.0.max(ordered.map_or(0, |last| last.saturating_add(1))), + start: groups.0, end: until.and_then(|until| until.map_or(Some(0), |last| last.checked_add(1))), delivered, }) }); } - fn deliverable(&self, copy: &Copy, sequence: u64) -> bool { - if copy + fn deliverable(copy: &Copy, sequence: u64, delivered: &Delivered) -> bool { + let replaced = copy .until - .is_some_and(|until| until.is_none_or(|until| sequence > until)) - { - return false; - } - match self.ordered { - Some(last) => sequence > last, - None => !self.delivered.contains(&sequence), - } + .is_some_and(|until| until.is_none_or(|until| sequence > until)); + !replaced && !delivered.contains(sequence) } - fn hand_out(&mut self, index: usize, group: group::Consumer, ordered: bool) -> group::Consumer { + fn hand_out(&mut self, index: usize, group: group::Consumer) -> group::Consumer { let sequence = group.sequence; - if ordered { - self.ordered = Some(sequence); - } - self.delivered.insert(sequence); - if self.delivered.len() > MAX_DELIVERED { - self.delivered.pop_first(); - } self.newest = self.newest.max(Some(sequence)); let copy = &self.copies[index]; group.with_recover(Recover { @@ -547,9 +550,14 @@ impl Reader { }) } - fn poll_group(&mut self, ordered: bool, waiter: &kio::Waiter) -> Poll>> { + fn poll_group( + &mut self, + ordered: bool, + delivered: &Delivered, + waiter: &kio::Waiter, + ) -> Poll>> { loop { - if let Some(res) = self.poll_group_once(ordered, waiter) { + if let Some(res) = self.poll_group_once(ordered, delivered, waiter) { return res; } } @@ -560,6 +568,7 @@ impl Reader { fn poll_group_once( &mut self, ordered: bool, + delivered: &Delivered, waiter: &kio::Waiter, ) -> Option>>> { self.sync(waiter); @@ -587,11 +596,10 @@ impl Reader { match res { Poll::Pending => break, Poll::Ready(Ok(Some(group))) => { - let copy = &self.copies[index]; - if !self.deliverable(copy, group.sequence) { + if !Self::deliverable(&self.copies[index], group.sequence, delivered) { continue; } - let group = self.hand_out(index, group, ordered); + let group = self.hand_out(index, group); return Some(Poll::Ready(Ok(Some(group)))); } // Drained: the track ended as the copy did, which is not cleanly when @@ -610,7 +618,7 @@ impl Reader { } } } - self.let_go(); + self.let_go(delivered); self.poll_end(waiter, |copy| copy.done.clone()) } @@ -653,17 +661,21 @@ impl Reader { } } - fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll>> { + fn poll_recv_datagram(&mut self, delivered: &Delivered, waiter: &kio::Waiter) -> Poll>> { loop { - if let Some(res) = self.poll_datagram_once(waiter) { + if let Some(res) = self.poll_datagram_once(delivered, waiter) { return res; } } } - fn poll_datagram_once(&mut self, waiter: &kio::Waiter) -> Option>>> { + fn poll_datagram_once( + &mut self, + delivered: &Delivered, + waiter: &kio::Waiter, + ) -> Option>>> { self.sync(waiter); - self.let_go(); + self.let_go(delivered); let groups = self.groups; for index in (0..self.copies.len()).rev() { loop { @@ -1093,7 +1105,6 @@ mod test { use futures::FutureExt; use super::*; - #[cfg(feature = "async")] use crate::Timestamp; use crate::broadcast; @@ -1138,7 +1149,6 @@ mod test { frame.map(|frame| frame.payload.to_vec()) } - #[cfg(feature = "async")] fn ts(ms: u64) -> Timestamp { Timestamp::from_millis(ms).unwrap() } @@ -1297,6 +1307,215 @@ mod test { assert!(sub.recv_group().now_or_never().is_none()); } + /// A budget that never writes off a group in these tests. + const HOUR: Duration = Duration::from_secs(3600); + + /// Write a finished one-frame group, presented at its sequence in milliseconds. + fn write(track: &track::Producer, sequence: u64) { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + group.write_frame(ts(sequence), b"x".as_ref()).unwrap(); + group.finish().unwrap(); + } + + /// Every group the reader hands out right now. + fn drain(sub: &mut track::Subscriber, ordered: bool) -> Vec { + let mut sequences = Vec::new(); + loop { + let res = match ordered { + true => sub.poll_next_group(&kio::Waiter::noop()), + false => sub.poll_recv_group(&kio::Waiter::noop()), + }; + let Poll::Ready(Ok(Some(group))) = res else { + return sequences; + }; + sequences.push(group.sequence); + } + } + + /// A copy delivering a group long after the reader moved past it does not hand it out + /// again, however many groups came between; one never handed out is still delivered. + #[test] + fn a_late_group_is_handed_out_once_however_late() { + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = subscribe(&logical, HOUR); + write(&a, 0); + assert_eq!(drain(&mut sub, false), [0]); + + let b = copy(); + routes.serve(b.consume()); + for sequence in 2..=2000 { + write(&b, sequence); + } + assert_eq!(drain(&mut sub, false), (2..=2000).collect::>()); + assert_eq!(sub.delivered_runs(), 2, "the gap at 1 is within the budget"); + + // The replacement's copies of groups 1 and 0 land last: 1 is new, 0 is not. + write(&b, 1); + write(&b, 0); + assert_eq!(drain(&mut sub, false), [1]); + assert_eq!(sub.delivered_runs(), 1); + } + + /// An ordered reader never goes back, late duplicate or not. + #[test] + fn an_ordered_reader_skips_a_late_group() { + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = subscribe(&logical, HOUR); + write(&a, 0); + assert_eq!(drain(&mut sub, true), [0]); + + let b = copy(); + routes.serve(b.consume()); + for sequence in 1..=2000 { + write(&b, sequence); + } + assert_eq!(drain(&mut sub, true), (1..=2000).collect::>()); + write(&b, 0); + assert_eq!(drain(&mut sub, true), Vec::::new()); + } + + /// A group the serving copy aborts and re-creates is handed out once, as it is to a + /// direct reader of that copy. + #[test] + fn a_recreated_group_is_handed_out_once() { + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = subscribe(&logical, HOUR); + write(&a, 0); + let open = a.create_group(group::Info { sequence: 1 }).unwrap(); + assert_eq!(drain(&mut sub, false), [0, 1]); + + open.abort(Error::Cancel).unwrap(); + write(&a, 1); + write(&a, 2); + assert_eq!(drain(&mut sub, false), [2]); + } + + /// Two routes taking turns, each also delivering the other's groups late and out of + /// order, hand out every group once. What the reader remembers does not grow with the + /// groups or the flaps. + #[test] + fn route_flaps_hand_out_every_group_once() { + let routes = Producer::new(); + let logical = logical(&routes); + let tracks = [copy(), copy()]; + routes.serve(tracks[0].consume()); + let mut sub = subscribe(&logical, HOUR); + let mut seen = Vec::new(); + for round in 0..16u64 { + let (serving, other) = (&tracks[round as usize % 2], &tracks[1 - round as usize % 2]); + let groups = round * 10..round * 10 + 10; + // The serving route brings half the round, the reader takes it, then the rest. + for sequence in groups.clone().take(5) { + write(serving, sequence); + } + seen.extend(drain(&mut sub, false)); + for sequence in groups.clone().skip(5) { + write(serving, sequence); + } + // The other route takes over and the replaced copy hands out the rest; the + // other's own copy of the round lands only afterwards, newest first. + routes.serve(other.consume()); + seen.extend(drain(&mut sub, false)); + for sequence in groups.clone().rev() { + write(other, sequence); + } + seen.extend(drain(&mut sub, false)); + assert_eq!(sub.delivered_runs(), 1, "round {round}"); + } + let mut sorted = seen.clone(); + sorted.sort_unstable(); + assert_eq!(sorted, (0..160).collect::>(), "handed out: {seen:?}"); + } + + /// A reader holding an old group open keeps its replaced copy, and moving its group + /// bounds neither repeats a group nor loses one. + #[test] + fn held_groups_and_moved_bounds_hand_out_each_group_once() { + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = subscribe(&logical, HOUR); + sub.end_at(Cap::from(..=0)); + let mut held = a.create_group(group::Info { sequence: 0 }).unwrap(); + held.write_frame(ts(0), b"a".as_ref()).unwrap(); + let Poll::Ready(Ok(Some(reading))) = sub.poll_recv_group(&kio::Waiter::noop()) else { + panic!("group 0 is handed out") + }; + // Past the cap: parked on the copy, not handed out. + write(&a, 1); + write(&a, 2); + assert_eq!(drain(&mut sub, false), Vec::::new()); + + let b = copy(); + routes.serve(b.consume()); + for sequence in [2, 0, 3, 1] { + write(&b, sequence); + } + assert_eq!(drain(&mut sub, false), Vec::::new()); + + sub.end_at(Cap::from(..)); + let mut seen = drain(&mut sub, false); + seen.sort_unstable(); + assert_eq!(seen, [1, 2, 3]); + sub.start_at(0); + assert_eq!(drain(&mut sub, false), Vec::::new()); + + // The replaced copy finishes with the held group. + held.finish().unwrap(); + a.finish().unwrap(); + drop(reading); + assert_eq!(drain(&mut sub, false), Vec::::new()); + write(&b, 4); + assert_eq!(drain(&mut sub, false), [4]); + assert_eq!(sub.delivered_runs(), 1); + } + + /// Across route changes, an endless track with groups missing remembers only the gaps + /// within the reader's budget: one past it is forgotten, one within it still delivered. + #[test] + fn a_resumed_reader_forgets_gaps_past_its_budget() { + let routes = Producer::new(); + let logical = logical(&routes); + let tracks = [copy(), copy()]; + routes.serve(tracks[0].consume()); + // Groups present 1ms apart, so the budget spans ten of them. + let mut sub = subscribe(&logical, Duration::from_millis(10)); + let mut most = 0; + for sequence in 0..3000u64 { + let track = &tracks[(sequence / 500) as usize % 2]; + // Every 500 groups the other route takes over. + if sequence % 500 == 0 { + routes.serve(track.consume()); + } + // Every third group never arrives. + if sequence % 3 == 1 { + continue; + } + write(track, sequence); + assert_eq!(drain(&mut sub, false), [sequence]); + most = most.max(sub.delivered_runs()); + } + // The floor and the four gaps a budget spans, twice over across a route change: the + // new copy starts at its first group, so the gaps just before it wait for that one + // to fall past the budget. + assert_eq!(most, 9); + + // 2998 is within the budget, 2500 far past it. + write(&tracks[1], 2998); + write(&tracks[1], 2500); + assert_eq!(drain(&mut sub, false), [2998]); + } + /// A replacement copy asked to start at the newest group handed out keeps that /// floor when the reader's floor widens: an older group its route cached was /// never owed, and handing it out now would be out of order. diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 23f4f576963..65f7c311090 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -16,6 +16,7 @@ use crate::{Error, Result, Timescale, Timestamp, coding}; use crate::{broadcast, cache, group, stats}; +use super::delivered::Delivered; use super::{Datagram, Requests}; use super::Cap; @@ -743,9 +744,13 @@ impl TrackState { } /// 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 + /// timestamp, so looking for one would scan the whole cache on every poll. fn drift_edge(&self, cap: Option) -> Edge { + let untimed = self.info.as_ref().is_some_and(|info| info.timescale.is_none()); Edge { - presentation: self.live_edge(cap), + presentation: if untimed { None } else { self.live_edge(cap) }, cap, } } @@ -813,17 +818,16 @@ 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. fn drifted(&self, sequence: u64, edge: &Edge, budget: Duration) -> bool { - // The edge was resolved under an earlier lock, so confirm it still names the - // same servable incarnation before it convicts a candidate. Failing safe - // (delivering) is right, since the next poll resolves a fresh edge. - let Some(live) = edge - .presentation + self.live_above(sequence, edge) + .is_some_and(|live| live.behind(self.reach(sequence, edge.cap), budget)) + } + + /// The edge that could convict group `sequence`: above it, and still holding the + /// servable incarnation it was resolved from. It was resolved under an earlier lock, + /// and failing safe (delivering) is right, since the next poll resolves a fresh edge. + fn live_above(&self, sequence: u64, edge: &Edge) -> Option { + edge.presentation .filter(|live| live.sequence > sequence && self.holds(live.sequence, live.stamp)) - else { - return false; - }; - self.reach(sequence, edge.cap) - .is_some_and(|reach| matches!(live.timestamp.checked_sub(reach), Ok(age) if Duration::from(age) >= budget)) } /// [`Self::drifted`] under `cap`, parking `waiter` on exactly what else can move the @@ -2313,6 +2317,7 @@ impl Producer { broadcast, info, inner: Inner::Plain(Cursor::new(self.state.consume(), subscription)), + delivered: Delivered::default(), // A producer-side (in-process) subscribe is not egress: stay untagged. stats: stats::Scope::default(), _stats_sub: stats::Subscription::default(), @@ -3379,6 +3384,7 @@ impl Subscribing { broadcast: self.broadcast.clone(), info, inner, + delivered: Delivered::default(), stats: self.stats.clone(), _stats_sub: self.stats.subscribe(), })) @@ -3720,6 +3726,9 @@ pub struct Subscriber { broadcast: Arc, info: Info, inner: Inner, + /// The groups handed out, so none is handed out twice: not when its sequence is + /// re-created, nor when another route delivers it too. + delivered: Delivered, // Egress stats scope, used to meter the groups this subscriber reads. Empty // (no-op) for an untagged track. stats: stats::Scope, @@ -3732,7 +3741,7 @@ pub struct Subscriber { pub(crate) struct Unread<'a> { pub start: u64, pub end: Option, - pub delivered: &'a std::collections::BTreeSet, + pub delivered: &'a Delivered, } /// How a [`Subscriber`] reads: its track's own cache, or a front's routes. @@ -3828,6 +3837,13 @@ struct PresentationEdge { timestamp: Timestamp, } +impl PresentationEdge { + /// Whether a group that can present until `reach` is a full `budget` behind this edge. + fn behind(&self, reach: Option, budget: Duration) -> bool { + reach.is_some_and(|reach| matches!(self.timestamp.checked_sub(reach), Ok(age) if Duration::from(age) >= budget)) + } +} + /// The read cursor behind a [`Subscriber`]. struct Cursor { state: kio::Consumer, @@ -3839,9 +3855,10 @@ struct Cursor { datagram_index: usize, /// Minimum sequence to return from any `recv` method. Set by `start_at`. min_sequence: u64, - /// One past the highest sequence returned by `next_group`. + /// One past the highest sequence returned by `next_group`, or `None` once it returned + /// `u64::MAX` and no group can follow. /// Used only by that method to skip late arrivals; does not affect `recv_group`. - next_sequence: u64, + next_sequence: Option, /// Exclusive upper sequence bound for `next_group` and `recv_group`, in the form /// [`Cap::exclusive`] produces. `None` means no cap. Set by `end_at`; /// can be raised, lowered, or unset at any time. Groups at or past the cap stay in @@ -3909,7 +3926,7 @@ impl Cursor { min_sequence, index: 0, datagram_index, - next_sequence: 0, + next_sequence: Some(0), end_sequence: None, parked: BTreeMap::new(), drift_cap: kio::Producer::new(None), @@ -3977,6 +3994,34 @@ impl Cursor { }) } + /// Forget the gaps in `delivered` below each group past this cursor's budget now, + /// whether this track holds it or not: one landing there would be written off. + /// + /// One edge for every group judged, so forgetting N gaps never resolves it N times. + fn forget_past_budget(&self, delivered: &mut Delivered) { + if delivered.above_gap().is_none() { + return; + } + let max_delay = self.subscription.read().max_delay; + let state = self.state.read(); + let budget = clamp_max_delay(max_delay, state.max_age_bound()); + 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); + } + if !live.behind(successor.and_then(|group| group.timestamp()), budget) { + break; + } + delivered.forget_below(oldest); + } + } + fn with_expiry(&self, group: group::Consumer) -> group::Consumer { let sequence = group.sequence; group.with_expiry(Arc::new(GroupExpiry { @@ -4096,7 +4141,10 @@ impl Cursor { /// The lowest servable group past the last one returned, walking off everything the /// drift budget has convicted on the way. fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll>> { - let mut floor = self.next_sequence.max(self.min_sequence); + let Some(next) = self.next_sequence else { + return Poll::Ready(Ok(None)); + }; + let mut floor = next.max(self.min_sequence); let end = self.end_sequence; // One scan for the whole poll, so walking a backlog off stays linear in its size. let drift = ready!(self.poll_drift(end, waiter))?; @@ -4129,9 +4177,9 @@ impl Cursor { break self.with_expiry(group); }; - self.next_sequence = group.sequence.saturating_add(1); + self.next_sequence = group.sequence.checked_add(1); // The delivery commits everything the seek stepped over to reach this group. - self.commit_seek_stale(self.next_sequence); + self.commit_seek_stale(self.next_sequence.unwrap_or(u64::MAX)); // Delivery is a cache access, same as the arrival-order path. group.cache_refresh(); Poll::Ready(Ok(Some(group))) @@ -4235,6 +4283,12 @@ impl Subscriber { /// [`Self::ordered`] if you only want groups whose sequence number is higher than any /// previously returned. /// + /// Once means once per sequence: a group the publisher aborts and re-creates, or that + /// a second route delivers too, is not handed out again. Once a group is past the + /// budget below, the subscriber stops expecting every group before it, so a gap that + /// old is never filled. That assumes group timestamps on a track never rewind; which + /// late groups are handed out after a rewind is unspecified. + /// /// Groups are semi-reliable, and the [`Subscription::max_delay`] budget is the other /// thing (alongside eviction and a moving start) that decides which of them arrive: /// one that has drifted further behind the live edge than the budget tolerates is @@ -4257,15 +4311,60 @@ impl Subscriber { /// `Poll::Ready(Err(e))` when the track has been aborted, or /// `Poll::Pending` when no group is available yet. pub fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll>> { + self.poll_group(false, waiter) + } + + /// The next group in arrival order, or in sequence order for `ordered`, skipping any + /// already handed out. + fn poll_group(&mut self, ordered: bool, waiter: &kio::Waiter) -> Poll>> { let meter = self.stats.meter(); let res = match &mut self.inner { - Inner::Plain(cursor) => cursor.poll_recv_group(waiter), - Inner::Resume(resume, _) => resume.poll_group(false, waiter), + Inner::Plain(cursor) => loop { + let res = match ordered { + true => cursor.poll_next_group(waiter), + false => cursor.poll_recv_group(waiter), + }; + match res { + Poll::Ready(Ok(Some(group))) if self.delivered.contains(group.sequence) => continue, + res => break res, + } + }, + Inner::Resume(resume, _) => resume.poll_group(ordered, &self.delivered, waiter), }; + if let Poll::Ready(Ok(Some(group))) = &res { + self.hand_out(group.sequence, ordered); + } self.count_stale(&meter); res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter)))) } + /// Record `sequence` as handed out. An ordered read never goes back, and any reader + /// stops expecting what is past its budget, so only the gaps within it cost memory. + fn hand_out(&mut self, sequence: u64, ordered: bool) { + if ordered { + self.delivered.forget_below(sequence); + } + self.delivered.insert(sequence); + let mut delivered = std::mem::take(&mut self.delivered); + self.forget_past_budget(&mut delivered); + self.delivered = delivered; + } + + /// Forget the gaps in `delivered` below each group past this subscriber's budget now: + /// one landing there would be written off. + pub(crate) fn forget_past_budget(&self, delivered: &mut Delivered) { + match &self.inner { + Inner::Plain(cursor) => cursor.forget_past_budget(delivered), + Inner::Resume(resume, _) => resume.forget_past_budget(delivered), + } + } + + /// How many runs of handed-out groups this subscriber remembers. + #[cfg(test)] + pub(crate) fn delivered_runs(&self) -> usize { + self.delivered.runs() + } + /// Whether an unread cached group needs this subscription, without moving its cursor. pub(crate) fn has_unread_group(&self, unread: &Unread<'_>) -> bool { let cursor = match &self.inner { @@ -4277,7 +4376,7 @@ impl Subscriber { let eligible = |sequence: u64| { sequence >= floor && super::subscription::before_end(sequence, unread.end) - && !unread.delivered.contains(&sequence) + && !unread.delivered.contains(sequence) }; cursor .parked @@ -4299,7 +4398,7 @@ impl Subscriber { /// Receive the next group in arrival order. /// - /// Every group is returned exactly once, in the order it landed on the wire, which may + /// Every group sequence is returned at most once, in the order it landed on the wire, which may /// be out of sequence due to network reordering or loss. Use [`Self::ordered`] if you /// only want groups whose sequence number is higher than any previously returned. /// See [`Self::poll_recv_group`] for how [`Self::set_groups`] applies. @@ -4324,7 +4423,7 @@ impl Subscriber { let meter = self.stats.meter(); let res = match &mut self.inner { Inner::Plain(cursor) => cursor.poll_recv_datagram(waiter), - Inner::Resume(resume, _) => resume.poll_recv_datagram(waiter), + Inner::Resume(resume, _) => resume.poll_recv_datagram(&self.delivered, waiter), }; // Unlike a group (metered lazily as its frames are read), a datagram is // delivered whole here, so count it as the single-frame group it stands in for. @@ -4347,13 +4446,7 @@ impl Subscriber { /// The sequence cursor behind [`Ordered`], which owns the only public door to it. pub(crate) fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll>> { - let meter = self.stats.meter(); - let res = match &mut self.inner { - Inner::Plain(cursor) => cursor.poll_next_group(waiter), - Inner::Resume(resume, _) => resume.poll_group(true, waiter), - }; - self.count_stale(&meter); - res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter)))) + self.poll_group(true, waiter) } /// Read this track's groups in sequence order instead of arrival order. @@ -7299,7 +7392,6 @@ mod test { ); } - #[cfg(feature = "async")] fn untimed_producer() -> Producer { track_producer("test", None) } @@ -9229,6 +9321,8 @@ mod test { consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence, u64::MAX ); + // Nothing can follow the maximum sequence. + assert!(consumer.next_group().now_or_never().unwrap().unwrap().is_none()); } #[test] @@ -10958,4 +11052,143 @@ mod test { assert!(pending.await.is_err()); drop(dynamic); } + + /// Write a finished one-frame group, presented at its sequence in milliseconds. + fn write_group(producer: &Producer, sequence: u64) { + let mut group = producer.create_group(group::Info { sequence }).unwrap(); + group + .write_frame(Timestamp::from_millis(sequence).unwrap(), &b"x"[..]) + .unwrap(); + group.finish().unwrap(); + } + + /// Every group the subscriber hands out right now, in arrival order. + fn drain_groups(subscriber: &mut Subscriber) -> Vec { + let mut sequences = Vec::new(); + while let Poll::Ready(Ok(Some(group))) = subscriber.poll_recv_group(&kio::Waiter::noop()) { + sequences.push(group.sequence); + } + sequences + } + + /// A sequence the publisher aborts and re-creates is handed out once. + #[test] + fn a_recreated_group_is_handed_out_once() { + let producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(replay()); + write_group(&producer, 0); + let open = producer.create_group(group::Info { sequence: 1 }).unwrap(); + assert_eq!(drain_groups(&mut subscriber), [0, 1]); + + open.abort(Error::Cancel).unwrap(); + write_group(&producer, 1); + write_group(&producer, 2); + assert_eq!(drain_groups(&mut subscriber), [2]); + } + + /// A gap is remembered until a group above it is past the budget, max delay or max + /// age alike. Then every group below that one counts as handed out. + #[test] + fn a_gap_is_forgotten_once_a_group_above_it_is_past_the_budget() { + let budget = Duration::from_millis(10); + let readers = [ + (Info::timed(), Subscription::default().with_max_delay(budget)), + ( + Info::timed().with_max_age(budget), + Subscription::default().with_max_delay(Duration::from_secs(3600)), + ), + ]; + for (info, subscription) in readers { + let producer = track_producer("test", info); + let mut subscriber = producer.subscribe(subscription); + // Group 3 never arrives. + for sequence in [0, 1, 2, 4, 5, 6] { + write_group(&producer, sequence); + } + assert_eq!(drain_groups(&mut subscriber), [0, 1, 2, 4, 5, 6]); + // Group 4 reaches until group 5 presents at 5ms, so it is past a 10ms budget once + // the edge presents at 15ms. + for sequence in 7..=14 { + write_group(&producer, sequence); + assert_eq!(drain_groups(&mut subscriber), [sequence]); + assert_eq!(subscriber.delivered_runs(), 2, "the gap at 3 is kept at {sequence}"); + } + write_group(&producer, 15); + assert_eq!(drain_groups(&mut subscriber), [15]); + assert_eq!(subscriber.delivered_runs(), 1, "the gap at 3 is forgotten"); + write_group(&producer, 3); + assert_eq!(drain_groups(&mut subscriber), Vec::::new()); + } + } + + /// An endless track with groups missing remembers only the gaps within the budget, and + /// still delivers an unseen group that arrives late within it. + #[test] + fn an_endless_track_with_gaps_stays_bounded() { + let producer = track_producer("test", Info::timed()); + // Groups present 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 { + // Every third group never arrives. + if sequence % 3 == 1 { + continue; + } + 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"); + + // 2998 is within the budget, 2500 far past it. + write_group(&producer, 2998); + write_group(&producer, 2500); + assert_eq!(drain_groups(&mut subscriber), [2998]); + } + + /// 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. + #[test] + fn gaps_below_aborted_groups_are_forgotten_together() { + let producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(1))); + let mut handed = Vec::new(); + for sequence in (0..20).step_by(2) { + let mut group = producer.create_group(group::Info { sequence }).unwrap(); + group + .write_frame(Timestamp::from_millis(sequence).unwrap(), &b"x"[..]) + .unwrap(); + handed.push(group); + } + assert_eq!(drain_groups(&mut subscriber).len(), 10); + for group in handed.drain(..5) { + group.abort(Error::Cancel).unwrap(); + } + // Groups 0 to 8 reach until group 10 at 10ms, which a group at 1.01s puts past the + // budget. Group 10 reaches until 12ms, so the gap at 9 is kept. + write_group(&producer, 1010); + assert_eq!(drain_groups(&mut subscriber), [1010]); + assert_eq!( + subscriber.delivered_runs(), + 7, + "the floor at 8, then 10 to 18 and 1010 apart" + ); + } + + /// An untimed track has nothing past the budget, so a sparse feed 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] + fn an_untimed_sparse_feed_keeps_its_gaps() { + let producer = untimed_producer(); + let mut subscriber = producer.subscribe(replay()); + for sequence in (0..2000u64).step_by(2) { + let mut group = producer.create_group(group::Info { sequence }).unwrap(); + group.write_frame(None, &b"x"[..]).unwrap(); + group.finish().unwrap(); + assert_eq!(drain_groups(&mut subscriber), [sequence]); + } + assert_eq!(subscriber.delivered_runs(), 1000); + } }