diff --git a/quest/m1/README.md b/quest/m1/README.md index afb1c5d677..92ae9a3733 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -126,7 +126,6 @@ tracks, so per-group stats stop costing cluster announces. - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide - [Replica resume](/quest/m1/replica-resume.md) - a lagging same-epoch replica resumes with every frame exactly once and in order, on lite-07 and on moq-transport - [Rust resume checks compression](/quest/m1/resume-compress.md) - a same-epoch route copy with a different `compress` is refused instead of corrupting framing -- [Rust resume backs single-route groups](/quest/m1/resume-handed-out.md) - a group first read from one route keeps the later copy that holds it, with a passthrough benchmark - [WebGPU on Safari](/quest/m1/webgpu-safari.md) - the WebGPU renderer is verified on Safari 26 for macOS and iOS - [Exporters follow a rendition to a new track](/quest/m1/export-rendition-rebind.md) - the fMP4, TS, MKV and FLV exporters rebuild a rendition whose resolved `track` or `broadcast` changes under the same ID - [HLS cursor failure](/quest/m1/hls-segments-failure.md) - the export segment cursor returns a failed timeline as an error after draining, so a truncated recording never looks complete diff --git a/quest/m1/resume-handed-out.md b/quest/m1/resume-handed-out.md deleted file mode 100644 index 449fe5269b..0000000000 --- a/quest/m1/resume-handed-out.md +++ /dev/null @@ -1,29 +0,0 @@ -# [S] Rust resume keeps a backup for a group read from a single route - -## Goal - -A Rust `resume::Reader` keeps a replaced copy that holds any group it handed -out and is still reading, including a group first read while only one copy -existed. Today such a group is never recorded, so the copy that later holds it -is dropped and the read waits out the reader's `max_delay` before ending with -`Error::Old`, even though the missing frames were available. - -## Plan - -Found by an external audit on 2026-10-22, confirmed on main. `hand_out` -(`rs/moq-net/src/model/resume.rs`) pushes onto `out` only when -`copies.len() > 1`, a guard from -[#6141](https://github.com/moq-dev/moq/pull/6141). Reproduction: read the -first frame of group 0 from A; swap to B, which holds all of group 0; B's -route fails, keeping its finished group; swap to C, which lacks group 0, and -read group 1; A's partial group 0 fails. B was dropped, so group 0 stalls. - -- Decided 2026-10-22 (maintainer): record every handed-out group, dropping the - guard, as [#6136](https://github.com/moq-dev/moq/pull/6136) does in JS. The - existing sweep keeps the list at the live group count. -- Decided 2026-10-22 (maintainer): add a passthrough benchmark in - `rs/moq-net/benches` that reads groups through a resumed track with a - single copy, so the per-group record's cost shows up. -- Regression test on a mocked clock for the reproduction above. - -Public API: behavior only. Wire: none. diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index 5decd12b2f..70b2bf3d1a 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -17,6 +17,9 @@ //! An equal-cost pool is swept the same way, over its members and the paths it //! already serves, and over those paths and the cursors watching the pool. //! +//! One-frame groups through a front with a single route are swept over readers and the +//! groups each holds open, which is what a reader records per group it hands out. +//! //! 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 @@ -1113,6 +1116,65 @@ fn bench_relay(c: &mut Criterion) { group.finish(); } +/// `(readers, open)` shapes for [`bench_passthrough`]: readers of one track, and the +/// groups each holds open at once. +const PASSTHROUGH: [(usize, usize); 4] = [(1, 1), (1, 64), (100, 1), (100, 64)]; + +/// One-frame groups read through a front that only ever had one route, so each handed-out +/// group's per-reader record (kept for a later route to back it) is a visible share of the +/// cost. Each reader holds `open` groups at once, the live count that record tracks. +fn bench_passthrough(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/passthrough"); + for (readers, open) in PASSTHROUGH { + group.throughput(Throughput::Elements((readers * open) as u64)); + group.bench_function(BenchmarkId::from_parameter(format!("{readers}r_{open}g")), |b| { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let (producer, driver) = origin::Producer::new(origin::Config::default()); + runtime.spawn(moq_net::time::run(driver)); + let broadcast = producer.publish("room/live", origin::Route::default()).unwrap(); + let source = broadcast.create_track("0", timed()).unwrap(); + let mut subscriptions = runtime.block_on(async { + let resolved = producer.consume().request_broadcast("room/live", None).await.unwrap(); + let track = resolved.track("0").unwrap(); + let mut subscriptions = Vec::with_capacity(readers); + for _ in 0..readers { + subscriptions.push(track.subscribe(None).await.unwrap()); + } + subscriptions + }); + let mut held: Vec> = (0..readers).map(|_| Vec::with_capacity(open)).collect(); + + b.iter_custom(|iterations| { + runtime.block_on(async { + let started = std::time::Instant::now(); + for _ in 0..iterations { + // Received as each is written: a reader behind the live edge skips ahead. + for _ in 0..open { + let mut group = source.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, b"frame".as_ref()).unwrap(); + group.finish().unwrap(); + for (subscription, held) in subscriptions.iter_mut().zip(&mut held) { + held.push(subscription.recv_group().await.unwrap().expect("group")); + } + } + for held in &mut held { + for group in held.iter_mut() { + group.read_frame().await.unwrap().expect("frame"); + } + held.clear(); + } + } + started.elapsed() + }) + }); + }); + } + group.finish(); +} + /// Re-poll stalled groups through a front, including their subscription budget. /// Sweep tracks and readers separately to expose unrelated table scans. fn bench_parked(c: &mut Criterion) { @@ -1525,6 +1587,7 @@ criterion_group!( bench_narrow_delivery, bench_narrow, bench_relay, + bench_passthrough, bench_parked, bench_copy_walk, bench_resume_overlap diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 9279f58117..5970ef21b3 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -833,7 +833,7 @@ struct Reader { groups: (u64, Cap), /// Whether groups were ever asked for; until then the reader keeps no replaced copy. groups_read: bool, - /// The groups handed out while copies overlap, by sequence, held while each is read. + /// The groups handed out, by sequence, held while each is read. out: Vec<(u64, Weak<()>)>, } @@ -1181,18 +1181,15 @@ impl Reader { let sequence = group.sequence; self.newest = self.newest.max(Some(sequence)); // Another copy holding the group, now or once it lands there, is where a failure - // continues it, so it stays while the group is read; see [`Copy::retains`]. Only - // while copies overlap: a passthrough read has none. - let out = (self.copies.len() > 1).then(|| { - // Sweep the groups no longer read once the list fills, so it tracks the live count. - if self.out.len() == self.out.capacity() { - self.out.retain(|(_, out)| out.strong_count() > 0); - self.out.reserve(self.out.len().max(1)); - } - let out = Arc::new(()); - self.out.push((sequence, Arc::downgrade(&out))); - out - }); + // continues it, so it stays while the group is read; see [`Copy::retains`]. Recorded + // even with one copy, since a later route's copy may come to hold it. + // Sweep the groups no longer read once the list fills, so it tracks the live count. + if self.out.len() == self.out.capacity() { + self.out.retain(|(_, out)| out.strong_count() > 0); + self.out.reserve(self.out.len().max(1)); + } + let out = Arc::new(()); + self.out.push((sequence, Arc::downgrade(&out))); let copy = &self.copies[index]; let handoff = group.handoff(); group.with_recover(Recover { @@ -1204,7 +1201,7 @@ impl Reader { copy: copy.track.clone(), handoff, lease: Some(copy.lease.clone()), - out, + out: Some(out), logical: None, fetch: None, }) @@ -2842,6 +2839,50 @@ mod test { ); } + /// A group first read while only one copy existed is still backed by a later copy that + /// holds it, so a failure continues from that copy instead of waiting out the budget. + #[test] + #[cfg(feature = "async")] + fn a_group_read_from_a_single_route_keeps_its_backup() { + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = subscribe(&logical, Duration::from_secs(10)); + + let mut open = a.create_group(group::Info { sequence: 0 }).unwrap(); + open.write_frame(ts(0), b"a".as_ref()).unwrap(); + let mut reading = recv(&mut sub); + assert_eq!(read(&mut reading), Some(b"a".to_vec())); + + // B holds the whole group, then its route fails, keeping it. + let b = copy(); + let mut whole = b.create_group(group::Info { sequence: 0 }).unwrap(); + whole.write_frame(ts(0), b"a".as_ref()).unwrap(); + whole.write_frame(ts(1), b"b".as_ref()).unwrap(); + whole.finish().unwrap(); + routes.serve(b.consume()); + assert!( + sub.poll_recv_group(&kio::Waiter::noop()).is_pending(), + "B replays only group 0" + ); + b.abort(Error::SessionClosed).unwrap(); + + // C lacks the group; reading a later one from it lets go of what nothing needs. + let c = copy(); + let mut later = c.create_group(group::Info { sequence: 1 }).unwrap(); + later.write_frame(ts(2), b"c".as_ref()).unwrap(); + later.finish().unwrap(); + routes.serve(c.consume()); + assert_eq!(recv(&mut sub).sequence, 1); + assert!(sub.poll_recv_group(&kio::Waiter::noop()).is_pending()); + + open.abort(Error::SessionClosed).unwrap(); + a.abort(Error::SessionClosed).unwrap(); + assert_eq!(read(&mut reading), Some(b"b".to_vec())); + assert_eq!(read(&mut reading), None); + } + #[test] #[cfg(feature = "async")] fn a_group_both_routes_deliver_is_handed_out_once() {