diff --git a/quest/m1/README.md b/quest/m1/README.md index 5be7d6e1dcc..7f1743eaedb 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -102,7 +102,6 @@ lands before the cut. - [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 - [A widened max age re-offers cached groups](/quest/m1/resolved-demand.md) - a warm relay serves a reader with a wider max age the cached groups a cold relay would, with no wire change - [Drop set](/quest/m1/drop-set.md) - each cursor judges missing groups against its own budget, so a narrow reader drops promptly and a widening re-offers groups already reported dropped on lite-07 -- [SUBSCRIBE_OK names where delivery starts](/quest/m1/subscribe-ok-start.md) - a warm relay names the same start group as a cold one instead of groups it skipped as stale - [Resumed close](/quest/m1/resume-closed.md) - a resumed JS track keeps its declared end across copies, and its `closed` and `largest()` settle on a clean end - [Resumed in-flight groups](/quest/m1/resume-in-flight.md) - a replaced route copy stays while a group below its bound is in flight or it backs a handed-out group late, and no longer than the reader's budget - [Recovered group readiness](/quest/m1/resume-readable.md) - a recovered JS group's `readable()` ignores frames its replacement replays diff --git a/quest/m1/archive/tracks.md b/quest/m1/archive/tracks.md index 56de3a9c8d9..1c0fabad466 100644 --- a/quest/m1/archive/tracks.md +++ b/quest/m1/archive/tracks.md @@ -207,5 +207,4 @@ answer it. Wire: a new lite-07-wip stream. ## Related - [SUBSCRIBE_TRACKS](/quest/m1/archive/subscribe-tracks.md) - moq-transport answers and learns listings through this hook -- [SUBSCRIBE_OK names where delivery starts](/quest/m1/subscribe-ok-start.md) - a warm relay answers like a cold one, found while building SUBSCRIBE_ALL - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - the subscription shape each per-track SUBSCRIBE uses diff --git a/quest/m1/subscribe-ok-start.md b/quest/m1/subscribe-ok-start.md deleted file mode 100644 index fcb998fb77e..00000000000 --- a/quest/m1/subscribe-ok-start.md +++ /dev/null @@ -1,34 +0,0 @@ -# [S] SUBSCRIBE_OK names where delivery starts - -## Goal - -SUBSCRIBE_OK's `Group` is the first group the publisher actually delivers, -as the lite draft defines it, so a warm relay and a cold one answer the same -subscription the same way. Today a publisher whose read cursor gives up -cached groups as stale before the start resolves still names the cache's -oldest group, claiming groups it never sends. Only a group skipped for a -missing head is excluded now. - -## Plan - -Found while building SUBSCRIBE_ALL -([moq-dev/moq#5931](https://github.com/moq-dev/moq/pull/5931), commit -`f074d09e7`), which fixed it in `TrackRun` (`rs/moq-net/src/lite/publisher.rs`) -and covered it with `the_start_names_past_the_stale_groups_it_skips`. That -PR is being replaced, so port the fix and its test on their own: record -every range the cursor gives up before the start resolves (a missing head, -stale, or dropped upstream), and resolve the start past them. Check whether -the moq-transport publisher's SUBSCRIBE_OK and the JS publisher have the same -gap, and fix them here if so. - -Add a relay test on mock time: the same SUBSCRIBE through a cold relay and a -warm one whose cache holds groups older than the subscriber's max age gets -the same `Group`. - -Public API: none. Wire: no encoding change; SUBSCRIBE_OK's `Group` changes -behavior on every lite version with a resolved start (lite-05 on) to match -its definition. - -## Related - -- [TRACKS](/quest/m1/archive/tracks.md) - the line of work that surfaced it diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs index 59a2793a370..bda41066c2c 100644 --- a/rs/moq-net/benches/session.rs +++ b/rs/moq-net/benches/session.rs @@ -107,6 +107,8 @@ struct Shape { /// Whole groups each watched broadcast writes per round, so a serve loop finds a /// backlog ready: the group rate a round stands in for. burst: usize, + /// Cached groups per track before a join, whose older groups the live budget skips. + history: usize, /// Bytes queued before a mock stream applies backpressure. window: usize, /// Read each group via a fresh FETCH rather than a live subscription. @@ -126,6 +128,7 @@ impl Shape { group_frames: FRAMES, paced: false, burst: 1, + history: 1, window: 1024 * 1024, fetch: false, }; @@ -180,6 +183,11 @@ impl Shape { String::new() }, if self.fetch { "/fetch" } else { "" }, + if self.history > 1 { + format!("/history={}", self.history) + } else { + String::new() + }, ) } } @@ -753,7 +761,15 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator) let cluster = Cluster::new(version, *shape).await; let payload = Bytes::from(vec![0; shape.frame]); for track in &cluster.tracks { - write_group(track, &payload, shape.group_frames); + for sequence in 0..shape.history { + let mut group = track.append_group().unwrap(); + for _ in 0..shape.group_frames { + group + .write_frame(Timestamp::from_millis(sequence as u64).unwrap(), payload.clone()) + .unwrap(); + } + group.finish().unwrap(); + } } cluster }) @@ -1255,6 +1271,20 @@ fn session(c: &mut Criterion) { }) }), ); + // A live join skips a cached prefix. Sweep publishers against that prefix so + // resolving one start cannot hide a scan of unrelated publishers or group IDs. + join( + c, + "stale", + [1, 8, 32].into_iter().flat_map(|publishers| { + [1, 64, 1024].map(move |history| Shape { + publishers, + history, + ..base + }) + }), + ); + join(c, "relays", [1, 2, 8].map(|relays| Shape { relays, ..base })); churn(c, "sessions", [1, 16, 128].map(|sessions| Churn { sessions, held: 16 })); diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index ce8a6b85f0c..4199d925eb7 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -4,7 +4,7 @@ use std::sync::atomic::AtomicU64; use crate::{SessionError, announce, frame, group, origin, track}; use rangemap::RangeSet; use std::{ - collections::{BTreeSet, HashMap}, + collections::HashMap, ops::{Bound, ControlFlow, Range}, sync::{ Arc, Mutex, @@ -3586,8 +3586,8 @@ struct TrackRun { start: Option, // The first servable group, held until the source resolves where its feed starts. first: Option, - // Groups skipped for a missing head before the start resolved, which it must not name. - skipped: BTreeSet, + // Groups given up before the start resolved, which it must not name. + skipped: RangeSet, end_sent: bool, // Lite05+ names every sequence below the end that this subscription never // got with SUBSCRIBE_DROP, so the subscriber settles without waiting out its grace @@ -3623,7 +3623,7 @@ impl TrackRun { emit_range, start: None, first: None, - skipped: BTreeSet::new(), + skipped: RangeSet::new(), end_sent: false, served, datagrams, @@ -3804,7 +3804,10 @@ impl TrackRun { // only be reset; the next servable group resolves the start. tracing::debug!(subscribe = self.ctx.id, track = %self.ctx.track_name, sequence = group.sequence, "skipping group with a missing head"); if self.emit_range && self.start.is_none() { - self.skipped.insert(group.sequence); + // At u64::MAX no later group can resolve a start past this one. + if let Some(end) = group.sequence.checked_add(1) { + self.skipped.insert(group.sequence..end); + } } else if self.start.is_some() && let Some(end) = group.sequence.checked_add(1) { @@ -3844,21 +3847,24 @@ impl TrackRun { Poll::Pending } - /// Lite-05+ names a sequence as soon as this cursor gives it up; whether one went out. + /// Lite-05+ names a sequence as soon as this cursor gives it up; whether one was taken. /// Before SUBSCRIBE_START there is no owed range yet, so the eventual start accounts - /// for it instead. + /// for it instead, including ranges skipped while finding the first servable group. fn poll_forward_drop( &mut self, writer: &mut Writer, waiter: &kio::Waiter, ) -> Result { - if self.start.is_none() || self.served.is_none() { + if !self.emit_range { return Ok(false); } let Poll::Ready(range) = self.track.poll_drop(waiter) else { return Ok(false); }; - self.drop_range(range, writer)?; + match self.start { + None => self.skipped.insert(range), + Some(_) => self.drop_range(range, writer)?, + } Ok(true) } @@ -3935,12 +3941,12 @@ impl TrackRun { writer: &mut Writer, ) -> Result<(), Error> { let mut start = source.map_or(group.sequence, |source| source.min(group.sequence)); - // A skipped group is never served, so it cannot be where the feed starts. This stops - // at the held group at the latest, since it was not skipped. - while self.skipped.contains(&start) { - start += 1; + // A group given up cannot be where the feed starts. This stops at the held group + // at the latest, since it was not skipped. + if let Some(skipped) = self.skipped.get(&start) { + start = skipped.end.min(group.sequence); } - self.skipped.clear(); + let skipped = std::mem::take(&mut self.skipped); self.start = Some(start); // Only the group: the subscriber derives the start frame from its own request // (see `lite::SubscribeStart`). The track is live by now, since its first group @@ -3955,6 +3961,11 @@ impl TrackRun { group: start, largest, }))?; + // The subscriber now owes itself everything from the start, so a range given up + // above it is named as it would have been after the start. + for range in skipped.iter() { + self.drop_range(range.start.max(start)..range.end, writer)?; + } // SUBSCRIBE_START names where delivery starts. Groups already skipped between // the requested floor and this group are not served. A later group at or above an // explicit floor, still inside the subscriber's max age, is delivered, so the cursor @@ -5479,6 +5490,219 @@ mod serve_group_test { assert_eq!(drops(&mut writer, &log, version).await, [(1, 1)]); } + /// A missing head at the model's largest sequence has no successor to name. + #[moq_clock::test] + async fn a_missing_head_at_the_maximum_sequence_does_not_panic() { + let track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + let subscriber = track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(60))); + let mut group = track.create_group(group::Info { sequence: u64::MAX }).unwrap(); + group.start_at(1).unwrap(); + group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap(); + group.finish().unwrap(); + let (mut run, mut writer, _) = lite_run(Version::Lite07, SinkSession::new(Log::default()), subscriber); + assert!( + kio::wait(|waiter| run.poll(&mut writer, waiter)) + .now_or_never() + .is_none() + ); + } + + /// A long-budget reader warms the relay without changing a live reader's resolved start. + #[moq_clock::test] + async fn cold_and_warm_relays_resolve_the_same_start() { + fn drive(mut driver: D) { + drop(moq_clock::sim::spawn(moq_clock::drive(move |now, waiter| { + crate::time::Driver::poll(&mut driver, now, waiter) + }))); + } + + fn origin(hop: u64) -> origin::Producer { + let (origin, driver) = origin::Producer::new(origin::Config::new(crate::Hop::new(hop).unwrap())); + drive(driver); + origin + } + + async fn connect( + version: Version, + publisher: &origin::Producer, + subscriber: &origin::Producer, + ) -> (crate::Session, crate::Session) { + let version = crate::Version::Lite(version); + let config = crate::transport::memory::Config { + protocol: Some(version.alpn().into()), + ..Default::default() + }; + let (client_transport, server_transport) = crate::transport::memory::pair(config); + let client = crate::Client::new() + .with_versions(version.into()) + .with_subscriber(subscriber.clone()); + let server = crate::Server::new() + .with_versions(version.into()) + .with_publisher(publisher); + futures::join!( + async { + let (session, driver) = client.connect(moq_clock::now(), client_transport).await.unwrap(); + drive(driver); + session + }, + async { + let (session, driver) = server.accept(moq_clock::now(), server_transport).await.unwrap(); + drive(driver); + session + } + ) + } + + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + let mut starts = Vec::new(); + for warm in [false, true] { + let publisher = origin(1); + let broadcast = publisher.create_broadcast("room").unwrap(); + let mut track = broadcast.create_track("test", crate::track::Info::timed()).unwrap(); + track.start_at(0).unwrap(); + broadcast.announce(Default::default()).unwrap(); + let relay = origin(2); + let _upstream = connect(version, &publisher, &relay).await; + let remote = relay.consume().routed_broadcast("room").await.unwrap(); + let mut keeper = match warm { + true => Some( + remote + .track("test") + .unwrap() + .subscribe( + track::Subscription::default() + .with_groups(0..) + .with_max_delay(Duration::from_secs(60)), + ) + .await + .unwrap(), + ), + false => None, + }; + for sequence in 0..3 { + write_group(&mut track, sequence, sequence * 10); + if let Some(keeper) = &mut keeper { + assert_eq!(keeper.recv_group().await.unwrap().unwrap().sequence, sequence); + } + } + moq_clock::sim::advance(Duration::from_secs(1)).await; + let subscriber = origin(3); + let _downstream = connect(version, &relay, &subscriber).await; + let remote = subscriber.consume().routed_broadcast("room").await.unwrap(); + let mut live = remote + .track("test") + .unwrap() + .subscribe(track::Subscription::default().with_groups(0..)) + .await + .unwrap(); + assert_eq!(live.recv_group().await.unwrap().unwrap().sequence, 2); + starts.push(kio::wait(|waiter| live.poll_start(waiter)).await); + } + assert_eq!( + starts, + [Some(2), Some(2)], + "{version}: cold and warm relay starts differ" + ); + } + } + + /// A relay's retained stale groups must not change the start it answers downstream. + #[moq_clock::test] + async fn the_start_names_past_the_stale_groups_it_skips() { + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + let mut starts = Vec::new(); + for warm in [false, true] { + let mut track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + track.start_at(if warm { 0 } else { 2 }).unwrap(); + for sequence in (if warm { 0 } else { 2 })..3 { + write_group(&mut track, sequence, sequence * 10); + } + moq_clock::sim::advance(Duration::from_secs(1)).await; + let subscriber = track.subscribe(track::Subscription::default()); + let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); + assert!( + kio::wait(|waiter| run.poll(&mut writer, waiter)) + .now_or_never() + .is_none() + ); + kio::wait(|waiter| writer.poll_flush(&mut waiter.context())) + .await + .unwrap(); + let writes = log.writes.lock().unwrap().clone(); + let (msg, _) = lite::SubscribeResponse::decode_slice(&writes, version).unwrap(); + let lite::SubscribeResponse::Start(start) = msg else { + panic!("expected SUBSCRIBE_START, got {msg:?}"); + }; + starts.push(start.group); + } + assert_eq!(starts, [2, 2], "{version}: cold and warm relay starts differ"); + } + } + + /// Upstream drops received before the start exclude the entire range, even on lite-07. + #[moq_clock::test] + async fn the_start_names_past_the_upstream_dropped_range() { + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + let mut track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + track.start_at(0).unwrap(); + let subscriber = track.subscribe(track::Subscription::default()); + let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); + track.drop_groups(0..1_000_000); + write_group(&mut track, 1_000_000, 0); + assert!( + kio::wait(|waiter| run.poll(&mut writer, waiter)) + .now_or_never() + .is_none() + ); + kio::wait(|waiter| writer.poll_flush(&mut waiter.context())) + .await + .unwrap(); + let writes = log.writes.lock().unwrap().clone(); + let (msg, _) = lite::SubscribeResponse::decode_slice(&writes, version).unwrap(); + let lite::SubscribeResponse::Start(start) = msg else { + panic!("expected SUBSCRIBE_START, got {msg:?}"); + }; + assert_eq!(start.group, 1_000_000, "{version}: named a dropped group"); + } + } + + /// A range given up before the start resolves, but above it, is still named promptly. + #[moq_clock::test] + async fn a_range_skipped_above_the_start_is_dropped() { + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + let mut track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + track.start_at(0).unwrap(); + let subscriber = track.subscribe(track::Subscription::default()); + let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); + // The newer group beats the older ones, so the start resolves at the source's 0. + track.drop_groups(2..4); + write_group(&mut track, 5, 0); + assert!( + kio::wait(|waiter| run.poll(&mut writer, waiter)) + .now_or_never() + .is_none() + ); + assert_eq!(run.start, Some(0), "{version}"); + assert_eq!(drops(&mut writer, &log, version).await, [(2, 3)], "{version}"); + } + } + /// A SUBSCRIBE_UPDATE that lowers the start below SUBSCRIBE_START asks for those groups /// too, so the drops at the end count from there. #[moq_clock::test]