diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 0761ed4fcc..6e09b5a1b7 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -345,8 +345,11 @@ reader or the next fetch finds its cache. That copy is not live meanwhile: readers get nothing from its cache until the source answers again, since how stale it is cannot be told. lite-07 and moq-transport answer with their largest position, and older lite versions are asked from the head of the newest cached -group, so their first frame says where their feed is. A feed that has moved on -past everything cached leaves the cache below it to fetches. +group, so their first frame says where their feed is. Readers resume once the +cache shows the answered position: a received group stays hidden until its +first frame lands or its stream ends, since a bare group header would leave the +older cached group looking like the live edge. A feed that has moved on past +everything cached leaves the cache below it to fetches. The publisher may declare a retention window per track. Omission sets no limit; zero keeps only the live edge. The origin cache ceiling and cache pool may still diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index f39e3a9fc4..e461891a9b 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -32,7 +32,7 @@ Public API: none. Wire: none. ## Required -- [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - `just test media` late join stays within one GOP, or the regression is fixed +- [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - a late joiner shows video promptly and catches up to live, and the check asserts that under load - [Import catalog finish](/quest/m1/test-flakes-2/import-catalog-finish.md) - `moq-cli`'s subprocess EOF catalog-finish test holds up under load with event-based fixture coordination - [Relay restart rebind](/quest/m1/test-flakes-2/relay-restart-rebind.md) - the crash drill restarts on its original UDP address under concurrent load - [Impaired handshake](/quest/m1/test-flakes-2/impaired-handshake.md) - the impaired cluster drills' clients never time out while connecting diff --git a/quest/m1/test-flakes-2/media-late-join.md b/quest/m1/test-flakes-2/media-late-join.md index 1e5c92dd93..e22bb0d9d8 100644 --- a/quest/m1/test-flakes-2/media-late-join.md +++ b/quest/m1/test-flakes-2/media-late-join.md @@ -1,30 +1,54 @@ -# [S] Media late join within one GOP +# [S] Media late join catches up to live ## Goal -Late join holds its one-GOP budget under load in both `just test media` and -the interop browser lane, or the regression behind it is fixed. +A viewer joining a broadcast through a relay after a demand gap shows video +promptly and reaches the live edge within a bound, and the late-join check +asserts exactly that and passes under load. ## Plan -It failed once after [#4181](https://github.com/moq-dev/moq/pull/4181): -"joined at frame 111, 16 frames behind 127", against a budget of one GOP -(15). First decide whether 16 frames is a real regression (the player -joining at the previous GOP's keyframe) or an off-by-one in how the fixture -samples the live edge. Fix whichever it is; don't widen the budget without a -reason. - -The interop browser check "late join starts live" -(`test/interop/clients/js/media.ts`, `MAX_LATE_JOIN_LAG_FRAMES`, one GOP -compared with `<=`) fails the same way on `main`: 16 frames behind with a -15-frame GOP on 09-24, 09-25, 09-28, and in -[#4577](https://github.com/moq-dev/moq/pull/4577). It samples `live` just -before the page opens. Treat it as the same flake (decided 2026-10-01). - -Recorded in the 2026-10-05 audit: on the test-flakes-2 line, -[#4719](https://github.com/moq-dev/moq/pull/4719) replaced -`MAX_LATE_JOIN_LAG_FRAMES` in the interop browser check with a -published-keyframe check: the latecomer's presented timestamp must be at or -after the published track's newest keyframe. That fixes the browser lane's -sampling. The original 16-frame failure was never reproduced, so what remains -is the Rust `just test media` late join under load. +It failed once after [#4181](https://github.com/moq-dev/moq/pull/4181) ("16 +frames behind" a 15-frame GOP), and the interop browser check failed the +same way on `main`. `just test media` is that browser lane through a Rust +`moq-relay`; there is no separate Rust media test. +[#4719](https://github.com/moq-dev/moq/pull/4719) now compares the +latecomer against the published keyframe instead of the painted counter. + +It is real, not fixture sampling: the latecomer is handed a GOP from before +the demand gap. Reproduced by looping `bun media.ts --cases late-join` +against a local relay with debug logs. In the fixture, the old viewer +leaves, the publisher pauses (closing its open GOP and writing an empty +marker group), and the latecomer joins about a second later. + +[#4914](https://github.com/moq-dev/moq/pull/4914) fixed two relay windows, +each with a deterministic test: + +- A received group stays hidden from readers until its first frame lands or + its stream ends, and an idle copy goes live only once the cache shows the + route's answer (`rejoin_waits_for_the_answers_first_frame`). +- A rejoin's answer is judged against the groups that survived the leave, + not the idle snapshot, since the cancel resets the group in flight + (`an_answer_past_a_reset_group_skips_the_cache`). + +What remains (1 of 30 runs after both fixes): the latecomer's max delay rises +to its playout delay (about 145ms) right after it subscribes. When the +pre-gap GOP was short, the GOP before it still reaches within that budget of +the marker, so the relay hands it over by design. The player then presents +from that GOP's keyframe, not from its playout position. Relay log: groups +7, 8, 9 served in 1ms with a 145ms budget, and the player shows frames 360ms +behind the published keyframe. + +Decided 2026-10-06 by the maintainer: a joiner showing the reachable pre-gap +GOP at once and then fast-forwarding to live is the wanted behavior, so no +player or consumer change. The check is what's too strict: it asserts the +first presented frame is at or after the newest published keyframe. + +Remaining work: the late-join check (`just test media` and the interop +browser lane) asserts what a viewer cares about instead: video shows within a +start bound, and the player reaches the live edge within a catch-up bound. +Set both bounds from measurements, looped under synthetic CPU load (the +maintainer approved a bounded load generator such as `stress-ng` for these +runs), and keep the relay fixes' deterministic tests as the regression guard. + +Public API: none. Wire: none. diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index bd9ae5fe5d..03bed5d771 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -474,11 +474,13 @@ impl Subscriber { if let Ok(mut tail) = entry.tail.write() { tail.account(dg.sequence..dg.sequence.saturating_add(1), self.runtime.now()); } - // A datagram says where the live feed is as well as a group does. - if !entry.producer.is_live() { + // A datagram says where the live feed is as well as a group does. Inserted first, so + // the copy goes live with it already showing. + let live = entry.producer.is_live(); + entry.producer.insert_datagram(dg.sequence, timestamp, dg.payload)?; + if !live { entry.producer.set_live(Some(Position::group(dg.sequence))); } - entry.producer.insert_datagram(dg.sequence, timestamp, dg.payload)?; Ok(()) } @@ -792,6 +794,8 @@ enum GroupRecvState { group: crate::recv::Group, track: track::Producer, ingest: FrameIngest, + /// Whether readers see the group yet; see [`track::Producer::receive_group`]. + shown: bool, _reading: Reading, }, Done, @@ -819,19 +823,21 @@ impl GroupRecv { // The subscription's end waits until this stream is read. let reading = Reading::open(&entry.tail, Some(hdr.sequence), self.subscriber.runtime.now()); + let group_info = group::Info { sequence: hdr.sequence }; + // Stats (groups/frames/bytes) are counted in the model as the group + // is written, through the tagged `track::Producer`. Withheld from + // readers until its first frame lands. + let received = entry.producer.receive_group(group_info); // A route's first group says where its live feed is, when its answer did // not: even one the cache already holds, as an idle copy asked from its - // head gets back. + // head gets back. The copy goes live once the cache shows that group. if !entry.producer.is_live() { entry.producer.set_live(Some(Position { group: hdr.sequence, frame: hdr.frame_start, })); } - let group_info = group::Info { sequence: hdr.sequence }; - // Stats (groups/frames/bytes) are counted in the model as the group - // is written, through the tagged `track::Producer`. - let mut group = match entry.producer.create_group(group_info) { + let mut group = match received { Ok(group) => group, // The group is at or past the end the publisher declared, which no // later stream can repair. Before lite-05 only a local finish sets @@ -865,11 +871,16 @@ impl GroupRecv { group: crate::recv::Group::new(group), track, ingest: FrameIngest::new(&self.subscriber, timescale), + shown: false, _reading: reading, }; } GroupRecvState::Serve { - group, track, ingest, .. + group, + track, + ingest, + shown, + .. } => { // The track or group dying cancels the stream; the peer's own close // arrives through the ingest's reads. @@ -880,7 +891,13 @@ impl GroupRecv { if let Poll::Ready(err) = group.poll_closed(waiter) { break 'serve Err(err); } - match ingest.poll(&mut self.reader, group, waiter) { + let res = ingest.poll(&mut self.reader, group, waiter); + // Shown once its first frame opens, even mid-payload. + if !*shown && group.is_started() { + track.reveal_group(group); + *shown = true; + } + match res { Poll::Ready(res) => break 'serve res, Poll::Pending => return Poll::Pending, } @@ -890,27 +907,35 @@ impl GroupRecv { // short drops into an already-aborted group instead of reporting a loss. let GroupRecvState::Serve { group, + track, ingest: _ingest, + shown, _reading, - .. } = std::mem::replace(&mut self.state, GroupRecvState::Done) else { unreachable!() }; - match res { + // A group that ended without a frame shows once it settles. + let handle = (!shown).then(|| (*group).clone()); + let res = match res { Ok(()) => { let _ = group.finish(); + Ok(()) } Err(err @ (Error::Cancel | Error::Stream(crate::StreamError::Cancel))) => { let _ = group.abort(err); + Ok(()) } Err(err) => { tracing::debug!(%err, group = %group.sequence, "group error"); let _ = group.abort(err.clone()); - return Poll::Ready(Err(err)); + Err(err) } + }; + if let Some(handle) = handle { + track.reveal_group(&handle); } - return Poll::Ready(Ok(())); + return Poll::Ready(res); } GroupRecvState::Done => return Poll::Ready(Ok(())), } diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 0970d0da25..1ea02ec174 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -192,6 +192,11 @@ enum Timeline { } impl GroupState { + /// Whether the first frame opened, or the group ended without one. + fn started(&self) -> bool { + !matches!(self.timeline, Timeline::Empty) || self.fin.is_some() || self.abort.is_some() + } + /// Content still available to a reader of this group. /// /// Counts the in-flight frame at its declared size, like [`Self::content_range`]: a @@ -895,6 +900,16 @@ impl Producer { (state.committed > state.offset).then_some(state.committed) } + /// Whether `other` is a handle to this same group. + pub(crate) fn is_clone(&self, other: &Self) -> bool { + self.state.same_channel(&other.state) + } + + /// Whether the group opened its first frame, or ended without one. + pub(crate) fn is_started(&self) -> bool { + self.state.read().started() + } + /// Where the group starts in presentation time: its first frame's timestamp, or /// `None` while no frame has been opened or when that frame is untimed. /// @@ -998,7 +1013,7 @@ impl Producer { /// none has been opened. pub(crate) fn poll_started(&self, waiter: &kio::Waiter) -> Poll<()> { match self.state.poll(waiter, |state| { - if !matches!(state.timeline, Timeline::Empty) || state.fin.is_some() || state.abort.is_some() { + if state.started() { Poll::Ready(()) } else { Poll::Pending diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 017dc99d92..30bcb2a697 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -161,6 +161,20 @@ impl Info { } } +/// Whether a track's cache reflects its live feed; see [`TrackState::set_idle`]. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum Feed { + /// Readers take from the cache. + #[default] + Live, + /// The upstream subscription ended with the copy still held, so how stale the cache + /// is cannot be told until the route answers again. + Idle, + /// The route answered that its feed reaches group `from`, which the cache cannot show + /// yet: until it can, the newest group cached would read as the live edge. + Answered { from: u64 }, +} + #[derive(Default)] pub(crate) struct TrackState { // The publisher's properties, once known; always Some for Subscriber/Producer. @@ -263,8 +277,8 @@ pub(crate) struct TrackState { // Whether the cache reflects the live feed: readers get nothing from it while it does // not, since how stale it is cannot be told; fetches still do. A track is live from // creation, and only a session's copy goes idle: when its upstream subscription ends - // with the copy still held, until the route answers again. - live: bool, + // with the copy still held, until the route's answer shows in the cache. + feed: Feed, // Readers start at this group: the route's live feed went on past a gap after // everything cached, so nothing bounds how old the cache below it is. live_floor: Option, @@ -316,6 +330,10 @@ struct Slot { // subscription content. Fetch-only backfill stays cached but never anchors drift. visible: bool, + // A live group received from a route, withheld from readers until its first frame + // lands or its stream ends; see [`Producer::receive_group`]. + pending: bool, + // A fetched copy of a live `group` that the feed started past the frames a fetch // asked for. It runs to the end of the group too, so it is served in the live // group's place while it lasts; `group` stays the slot, still written by the feed, @@ -329,6 +347,11 @@ impl Slot { self.visible && !self.group.is_aborted() } + /// Whether readers see this slot as live content: [`Self::is_live`], and not withheld. + fn is_shown(&self) -> bool { + self.is_live() && !self.pending + } + /// The copy to hand readers: the fetched head while it lasts, else the live group. fn serving(&self) -> &group::Producer { match &self.head { @@ -463,7 +486,7 @@ impl TrackState { /// Whether readers may take from the cache: it reflects the live feed, or the track /// ended, which makes the cache the whole of it. fn readable(&self) -> bool { - self.live || self.sealed || self.final_sequence.is_some() || self.abort.is_some() + self.feed == Feed::Live || self.sealed || self.final_sequence.is_some() || self.abort.is_some() } fn poll_recv_group(&self, index: usize, min_sequence: u64) -> Poll>> { @@ -524,7 +547,9 @@ impl TrackState { self.datagrams.pop_front(); self.datagram_offset += 1; } + let sequence = datagram.sequence; self.datagrams.push_back(datagram); + self.shown(sequence); } /// Find the smallest-sequence cached group satisfying @@ -639,7 +664,7 @@ impl TrackState { .rev() .filter(|(seq, _)| super::subscription::before_end(**seq, cap)) .find_map(|(_, slot)| { - if !slot.visible || slot.group.is_aborted() { + if !slot.is_shown() { return None; } // The map is ordered by sequence, so the first stamped group from the @@ -669,7 +694,7 @@ impl TrackState { cap.map_or(std::ops::Bound::Unbounded, std::ops::Bound::Excluded), )) .rev() - .find(|(_, slot)| slot.visible && !slot.group.is_aborted()) + .find(|(_, slot)| slot.is_shown()) .map(|(sequence, _)| *sequence) } @@ -709,7 +734,7 @@ impl TrackState { .range(from..) .map(|(_, slot)| slot) .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap)) - .find(|slot| slot.visible && !slot.group.is_aborted()) + .find(|slot| slot.is_shown()) .map(|slot| &slot.group) } @@ -1002,7 +1027,6 @@ impl TrackState { fn spawn(broadcast: Arc) -> kio::Producer { let state = kio::Producer::new(Self { broadcast: broadcast.clone(), - live: true, ..Default::default() }); let cache = cache::Track::new(broadcast.pool.clone(), state.downgrade()); @@ -1032,8 +1056,9 @@ impl TrackState { /// Updates the live edge, demoting the previous latest into the eviction order; /// the current latest is never enqueued, which is what protects it from /// eviction. `visible` controls arrival-order delivery: publisher-produced - /// groups reach subscribers, fetched backfill is served by sequence only. - fn insert_group(&mut self, group: &group::Producer, visible: bool) { + /// groups reach subscribers, fetched backfill is served by sequence only. A + /// `pending` group joins arrival order only once [`Self::reveal`] shows it. + fn insert_group(&mut self, group: &group::Producer, visible: bool, pending: bool) { let sequence = group.sequence; self.next_stamp = self.next_stamp.wrapping_add(1); let stamp = self.next_stamp; @@ -1059,32 +1084,75 @@ impl TrackState { } self.max_sequence = Some(self.max_sequence.map_or(sequence, |max| max.max(sequence))); - let below = visible - .then(|| self.lookup.range(..sequence).rev().find(|(_, slot)| slot.is_live())) - .flatten() - .map(|(below, _)| *below); self.lookup.insert( sequence, Slot { group: group.clone(), stamp, visible, + pending, head: None, }, ); - if visible { + if visible && !pending { self.arrival.push_back((sequence, stamp)); - let oldest = self.lookup.first_key_value().map_or(sequence, |(oldest, _)| *oldest); - self.cache.wakes().landed(below, sequence, oldest); + self.landed(sequence); + } + } + + /// Show a [`Producer::receive_group`] group to readers, if `group` still holds its slot. + fn reveal(&mut self, group: &group::Producer) { + let sequence = group.sequence; + let Some(slot) = self.lookup.get_mut(&sequence) else { + return; + }; + if !slot.pending || !slot.group.is_clone(group) { + return; + } + slot.pending = false; + 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() { + return; + } + let latest = slot.group.latest(); + self.landed(sequence); + // Its frames presented while hidden, so a read woken by them may have judged the + // older edge and parked again: present them once more now that they count. + if let Some(latest) = latest { + self.cache.wakes().presented(latest); + } + } + + /// Content at `sequence` became servable: wake the reads it succeeds, and show an answer + /// waiting on it. + fn landed(&mut self, sequence: u64) { + let below = self + .lookup + .range(..sequence) + .rev() + .find(|(_, slot)| slot.is_shown()) + .map(|(below, _)| *below); + let oldest = self.lookup.first_key_value().map_or(sequence, |(oldest, _)| *oldest); + self.cache.wakes().landed(below, sequence, oldest); + self.shown(sequence); + } + + /// Content at `sequence` became readable: an answer waiting on it now shows. + fn shown(&mut self, sequence: u64) { + if let Feed::Answered { from } = self.feed + && sequence >= from + { + self.feed = Feed::Live; } } /// Admit a freshly-created group: settle eviction debt first (so the newcomer /// can never be a victim of the very write that created it), insert it, then /// expire idle groups. - fn commit_group(&mut self, group: &group::Producer, visible: bool) { + fn commit_group(&mut self, group: &group::Producer, pending: bool) { self.charge_debt(); - self.insert_group(group, visible); + self.insert_group(group, true, pending); self.evict_expired(); } @@ -1285,12 +1353,12 @@ impl TrackState { /// stale, so readers get nothing from it until the route answers again. Buffered /// datagrams go too, since a reader returning later must not be handed them. fn set_idle(&mut self) { - self.live = false; + self.feed = Feed::Idle; self.idle_newest = self .lookup .iter() .rev() - .find(|(_, slot)| slot.visible && !slot.group.is_aborted()) + .find(|(_, slot)| slot.is_shown()) .map(|(sequence, _)| *sequence); self.datagram_offset += self.datagrams.len(); self.datagrams.clear(); @@ -1300,23 +1368,47 @@ impl TrackState { /// is current up to there. A feed that went on past a gap after everything cached /// leaves the cache unjudgeable, since nothing bounds how far an old group reached, so /// readers skip it; whatever the feed delivers stays. + /// + /// Readers wait until the cache shows that position: before then its newest group would + /// read as the live edge, though the route already holds a newer one. fn set_live(&mut self, largest: Option) { - if self.live { + if self.feed != Feed::Idle { return; } - self.live = true; - let cached = self.idle_newest.take(); + // Judged by what survived: the cancel that idled the copy resets the group in flight, + // and an answer past it leaves a gap after the newest group still cached. + let cached = self.idle_newest.take().and_then(|newest| { + self.lookup + .range(..=newest) + .rev() + .find(|(_, slot)| slot.is_shown()) + .map(|(sequence, _)| *sequence) + }); self.live_floor = match (largest, cached) { (Some(largest), Some(cached)) if largest.group > cached.saturating_add(1) => Some(cached + 1), // The route has nothing, so whatever is cached is not its feed. (None, Some(cached)) => Some(cached.saturating_add(1)), _ => self.live_floor, }; + self.feed = match largest { + Some(largest) if !self.shows(largest.group) => Feed::Answered { from: largest.group }, + _ => Feed::Live, + }; + } + + /// Whether readers can already see content at or past `sequence`. + fn shows(&self, sequence: u64) -> bool { + self.lookup.range(sequence..).any(|(_, slot)| slot.is_shown()) + || self.datagrams.iter().any(|datagram| datagram.sequence >= sequence) } /// The newest live group the cache holds, and the number of frames it has so far. fn newest(&self) -> Option<(u64, u64)> { - let (sequence, slot) = self.lookup.iter().rev().find(|(_, slot)| slot.visible)?; + let (sequence, slot) = self + .lookup + .iter() + .rev() + .find(|(_, slot)| slot.visible && !slot.pending)?; Some((*sequence, slot.group.frame_count() as u64)) } @@ -1407,7 +1499,7 @@ impl TrackState { } // Invisible to arrival-order subscribers: fetched on demand, not produced // live by the publisher. - None => self.insert_group(&group, false), + None => self.insert_group(&group, false, false), } self.evict_expired(); Ok(group) @@ -1505,6 +1597,29 @@ impl Producer { /// Create a new group with the given sequence number. pub fn create_group(&self, group: group::Info) -> Result { + self.insert(group, false) + } + + /// Create a group a session is receiving from a route, withheld from readers until + /// [`Self::reveal_group`]. + /// + /// A publisher writes a group's first frame as it creates it, but a route's group + /// header lands before its first frame. Shown in between, it would leave the group + /// before it with no stamped successor, so a joiner at the live edge would be handed + /// that older group first. + pub(crate) fn receive_group(&self, group: group::Info) -> Result { + self.insert(group, true) + } + + /// Show a [`Self::receive_group`] group to readers: its first frame landed, or its + /// stream ended. A no-op once another incarnation took its sequence. + pub(crate) fn reveal_group(&self, group: &group::Producer) { + if let Ok(mut state) = self.modify() { + state.reveal(group); + } + } + + fn insert(&self, group: group::Info, pending: bool) -> Result { let mut state = self.modify()?; if let Some(fin) = state.final_sequence && group.sequence >= fin @@ -1517,7 +1632,7 @@ impl Producer { state.claim_sequence(group.sequence, 0)?; let group = group::Producer::new(group, track, state.cache.clone()).with_meter(self.stats.meter()); - state.commit_group(&group, true); + state.commit_group(&group, pending); Ok(group) } @@ -1539,7 +1654,7 @@ impl Producer { let group = group::Producer::new(group::Info { sequence }, track, state.cache.clone()).with_meter(self.stats.meter()); - state.commit_group(&group, true); + state.commit_group(&group, false); Ok(group) } @@ -1739,7 +1854,8 @@ impl Producer { } /// The route answered with its largest position, `None` for nothing yet; see - /// [`Self::set_idle`]. A no-op while live. + /// [`Self::set_idle`]. Readers resume once the cache shows that position. A no-op + /// unless idle. pub(crate) fn set_live(&mut self, largest: Option) { if let Ok(mut state) = self.modify() { state.set_live(largest); @@ -1748,14 +1864,14 @@ impl Producer { /// Whether readers may take from the cache; see [`Self::set_idle`]. pub(crate) fn is_live(&self) -> bool { - self.state.read().live + self.state.read().feed == Feed::Live } /// While idle, the newest group cached when the track went idle: a route asked from /// its head sends it again, and its first frame says whether the cache is current. pub(crate) fn idle_newest(&self) -> Option { let state = self.state.read(); - state.idle_newest.filter(|_| !state.live) + state.idle_newest.filter(|_| state.feed == Feed::Idle) } /// Declare the floor a subscription asked for while the serving session has yet to @@ -5680,6 +5796,35 @@ mod test { sequences } + /// A rejoined copy judges the route's answer by the groups that survived the leave. + /// + /// The cancel that idles a copy resets the group in flight. The route then answers past + /// it, so the newest group still cached is two behind: a gap, which the cached group + /// cannot be measured across. Judged by the group the copy held when it went idle, the + /// answer would read as contiguous, and the cached group would reach all the way to the + /// answer, fresh to any positive budget. + #[test] + fn an_answer_past_a_reset_group_skips_the_cache() { + let mut producer = track_producer("test", None); + append_at(&mut producer, 0); + let mut open = producer.append_group().unwrap(); + open.write_frame(Timestamp::from_millis(500).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + + producer.set_idle(); + open.abort(Error::Cancel).unwrap(); + + let mut answer = producer.receive_group(group::Info { sequence: 2 }).unwrap(); + producer.set_live(Some(Position::group(2))); + answer + .write_frame(Timestamp::from_millis(533).unwrap(), bytes::Bytes::from_static(b"")) + .unwrap(); + producer.reveal_group(&answer); + + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_millis(400))); + assert_eq!(drain(&mut subscriber), vec![2]); + } + #[test] fn real_time_skips_a_backlog_to_the_live_edge() { let mut producer = track_producer("test", None); @@ -6188,6 +6333,49 @@ mod test { assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); } + /// A received group's frames present while it is hidden, so a parked read they wake + /// still sees the older edge and parks again. Revealing the group has to wake it once + /// more, though the read sits below the group's nearest shown predecessor. + #[test] + fn revealing_a_presented_group_wakes_a_read_it_expires() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_millis(500))); + let mut head = producer.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head")) + .unwrap(); + append_at(&mut producer, 1000); // the successor bounds the head's reach + + let mut held = subscriber + .recv_group() + .now_or_never() + .unwrap() + .unwrap() + .expect("head group"); + assert_eq!(held.sequence, 0); + assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some()); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waker = futures::task::waker(Arc::new(FlagWake(woken.clone()))); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(held.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + + let mut received = producer.receive_group(group::Info { sequence: 2 }).unwrap(); + received + .write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + woken.store(false, Ordering::SeqCst); + assert!( + next.as_mut().poll(&mut cx).is_pending(), + "the hidden group moves no edge" + ); + + producer.reveal_group(&received); + assert!(woken.load(Ordering::SeqCst), "the reveal wakes the expired read"); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + /// The ordinary live case, at the default real-time budget: 2s GOPs produced one at /// a time and read as they arrive. The budget must take the live edge without /// shortening the group the reader is already on, so every frame of every group is diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index 3bddb07142..6357236a99 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -391,6 +391,77 @@ async fn rejoin_during_the_cancel_skips_the_cache() { } } +/// A rejoining reader is not handed the cached group while the route's next group has no +/// frame yet. +/// +/// The publisher moved on by one group while nobody subscribed, so the route answers the +/// rejoin with that group, right after the one the relay cached. Its header lands before its +/// first frame. Shown in between, it would leave the cached group without a stamped +/// successor, and a reader at the live edge would be handed that older group first. +#[moq_net_sim::test] +async fn rejoin_waits_for_the_answers_first_frame() { + // Lite01/02 carry no max age, so the publisher serves the relay's cached head again and + // that answer is what makes the cache current. + let unbudgeted = ["moq-lite-01", "moq-lite-02"]; + for version in Version::names().filter(|version| !unbudgeted.contains(version)) { + moq_net_sim::timeout(TEST_TIMEOUT, async { + let publisher = produce_origin(1); + let relay = produce_origin(2); + + let broadcast = publisher.create_broadcast("bench").unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(relay.clone()); + let pair = connect_mock(options).await; + + let consumer = relay.consume(); + consumer.routed("bench").await.unwrap(); + let remote = consumer.request_broadcast("bench", None).await.unwrap(); + let ts = |ms| Timestamp::from_millis(ms).unwrap(); + + // Held for fetches only: it keeps the relay's copy without subscribing. + let _held = remote.track("video").unwrap(); + + let mut group = track.append_group().unwrap(); + group.write_frame(ts(0), b"old".as_ref()).unwrap(); + group.finish().unwrap(); + let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap(); + let mut group = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(read_all(&mut group).await.unwrap(), [b"old".to_vec()], "{version}"); + drop((group, sub)); + track.demand().unused().await.unwrap(); + + let mut group = track.append_group().unwrap(); + group.write_frame(ts(500), b"new".as_ref()).unwrap(); + group.finish().unwrap(); + + pair.server_transport.split_unis(); + let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap(); + let early = moq_net_sim::timeout(Duration::from_secs(1), sub.recv_group()).await; + assert!( + early.is_err(), + "{version}: the rejoining reader got the cached group {:?} first", + early.map(|group| group.unwrap().unwrap().sequence) + ); + + pair.server_transport.release_split(); + let mut group = sub.recv_group().await.unwrap().unwrap(); + // A track that arrives untimed has no media age to skip the cached group by, so + // its reader starts at the newest group cached and reads on from there. + if sub.info().timescale.is_none() && group.sequence == 0 { + group = sub.recv_group().await.unwrap().unwrap(); + } + assert_eq!(group.sequence, 1, "{version}"); + assert_eq!(read_all(&mut group).await.unwrap(), [b"new".to_vec()], "{version}"); + }) + .await + .unwrap_or_else(|_| panic!("{version}: timed out")); + } +} + /// A reader leaving long after its last read leaves the relay holding the latest group. /// /// A catalog's latest group can sit unread past the cache's idle expiry. A downstream diff --git a/rs/moq-net/tests/support/mock.rs b/rs/moq-net/tests/support/mock.rs index 427e8df6ad..0b0a62c0eb 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/rs/moq-net/tests/support/mock.rs @@ -227,6 +227,11 @@ pub struct MockRecvStream { buf: Bytes, /// Whether we hit FIN or reset. done: bool, + /// Once released, delivers what follows this stream's first write; see + /// [`MockSession::split_unis`]. + split: Option>>, + /// Data chunks handed to the reader so far. + chunks: usize, /// Shared signal to notify the peer's send-side `poll_closed`. closed: Arc, park: kio::Park, @@ -237,6 +242,17 @@ impl MockRecvStream { /// Pop the next chunk, mapping queue closure to an implicit FIN. Once everything /// sent before a CONNECTION_CLOSE is read, an unfinished stream fails with it. fn poll_chunk(&mut self, cx: &mut Context<'_>) -> Poll> { + if let Some(split) = &self.split + && self.chunks > 0 + && split + .poll(self.park.hold(cx), |released| match **released { + true => Poll::Ready(()), + false => Poll::Pending, + }) + .is_pending() + { + return Poll::Pending; + } if self.flight.is_none() { let waiter = self.park.hold(cx); // Register on the close first so one racing the pop still wakes this poll. @@ -284,6 +300,7 @@ impl poll::RecvStream for MockRecvStream { match std::task::ready!(self.poll_chunk(cx)) { Some(StreamChunk::Data(data)) => { + self.chunks += 1; let n = dst.len().min(data.len()); dst[..n].copy_from_slice(&data[..n]); if n < data.len() { @@ -367,6 +384,8 @@ fn new_stream_pair(conn: &Arc) -> (MockSendStream, MockRecvStre landing: None, buf: Bytes::new(), done: false, + split: None, + chunks: 0, closed, park: kio::Park::default(), conn: conn.clone(), @@ -429,6 +448,9 @@ struct SessionSide { held: Mutex>>, /// Whether the peer has withheld uni stream credit, parking every open. withheld: Mutex, + /// While set, the uni streams this side opens hold back everything after their first + /// write until it is released; see [`MockSession::split_unis`]. + split: Mutex>>>, /// Whether the datagrams this side sends are lost. lossy: Mutex, /// Whether this side drops the FIN of the bidi streams it opens. @@ -506,7 +528,8 @@ impl poll::Session for MockSession { return Poll::Ready(Err(self.close_error())); } - let (mut our_send, peer_recv) = new_stream_pair(&self.side.conn); + let (mut our_send, mut peer_recv) = new_stream_pair(&self.side.conn); + peer_recv.split = self.side.split.lock().unwrap().clone(); if let Some(held) = self.side.held.lock().unwrap().as_mut() { our_send.ack_fin = true; @@ -638,6 +661,21 @@ impl MockSession { *self.side.withheld.lock().unwrap() = true; } + /// Deliver only the first write of each uni stream this side opens from now on, holding + /// the rest until [`Self::release_split`]. A group stream's first write is its header, + /// so the header lands before the group's first frame, as QUIC may split them across + /// packets. + pub fn split_unis(&self) { + *self.side.split.lock().unwrap() = Some(Arc::default()); + } + + /// Deliver what [`Self::split_unis`] held, and stop splitting. + pub fn release_split(&self) { + if let Some(split) = self.side.split.lock().unwrap().take() { + *split.lock() = true; + } + } + /// Lose every datagram this side sends from now on. pub fn lose_datagrams(&self) { *self.side.lossy.lock().unwrap() = true; @@ -687,6 +725,7 @@ pub fn create_mock_session_pair(protocol: Option<&'static str>) -> (MockSession, conn: conn.clone(), held: Mutex::default(), withheld: Mutex::default(), + split: Mutex::default(), lossy: Mutex::default(), withhold_bidi_fins: Arc::default(), }); @@ -702,6 +741,7 @@ pub fn create_mock_session_pair(protocol: Option<&'static str>) -> (MockSession, conn, held: Mutex::default(), withheld: Mutex::default(), + split: Mutex::default(), lossy: Mutex::default(), withhold_bidi_fins: Arc::default(), });