diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 42d8cab6ba2..79e83781678 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -57,6 +57,19 @@ a group at or past `SUBSCRIBE_END`, or a `SUBSCRIBE_END` below a group already received. moq-lite 05 specified an inclusive end, so there it only drops that group or the early boundary. +## Group reads across failover + +Rust origin readers can keep reading an in-flight group after its source fails. +A group reader waiting on a replacement copy subscribes to it from the frame it +needs, even when the caller is not polling for the next group, and holds that +subscription until the group ends. If that copy refuses the subscription, the +read ends with an error. The copy's declared start says whether the group is +still coming, so a newer group overtaking it on the wire does not end the wait. +A copy that declares no start (a local track) and has advanced past the group +ends the read with an error. This also applies when waiting for the group's +completion; no FETCH is issued. A replacement that later drops the group it is +serving still needs SUBSCRIBE\_DROP support to resolve that wait. + ## Discovery A session can ask for announcements matching a path prefix. The peer replies diff --git a/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md b/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md index 886144d2d0e..21eff93ccd7 100644 --- a/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md +++ b/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md @@ -89,5 +89,5 @@ the closed producer's cache or terminal state. ## Related -- [Resume latest](/quest/m1/resume-latest.md) - edits `resume.rs` and its takeover tests (#4491) +- [#4491](https://github.com/moq-dev/moq/pull/4491) - edits `resume.rs` and its takeover tests - [Parked reads wake](/quest/m1/parked-read-wakes.md) - edits the same `resume.rs` wakeups diff --git a/quest/m1/README.md b/quest/m1/README.md index 928f1394a27..44c0d2038c2 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -40,7 +40,6 @@ QUIC studies there on that rule. - [web-transport-moq 1.3.3 ships noq#24](/quest/m1/web-transport-moq-release.md) - the release that carries the close-capsule fix - [WebTransport close upstream](/quest/m1/wt-close-upstream.md) - web-transport-moq delivers the close capsule itself, and moq-tokio's `CLOSE_LINGER` is deleted - [Browser close code](/quest/m1/browser-close-code.md) - a playwright case proves the page reads a relay's close code and reason, on every web-transport backend -- [Resumed groups](/quest/m1/resume-latest.md) - a half-delivered group ends once the new copy is past it, so a group-only reader never parks after a mid-group failover - [Truncated groups fail finished()](/quest/m1/finished-truncated.md) - a spliced group whose continuation can never arrive fails `finished()` instead of reporting a clean end at the seam - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - every stream group in a lite subscription arrives or is dropped by name, and lite-07 drops its stream count for it - [Late lower groups](/quest/m1/lite-late-lower-group.md) - a moq-lite subscriber with a floor receives a group created below the first served one, as moq-transport does diff --git a/quest/m1/finished-truncated.md b/quest/m1/finished-truncated.md index 9e052dd2fe3..dec6d1f64be 100644 --- a/quest/m1/finished-truncated.md +++ b/quest/m1/finished-truncated.md @@ -34,4 +34,4 @@ Public API: behavior only (an error where `Ok` was returned). Wire: none. ## Related -- [Resumed groups](/quest/m1/resume-latest.md) - same `poll_finished` subscription path +- [#4491](https://github.com/moq-dev/moq/pull/4491) - the same `poll_finished` subscription path diff --git a/quest/m1/parked-read-wakes.md b/quest/m1/parked-read-wakes.md index 3e403f8b261..77b75479a24 100644 --- a/quest/m1/parked-read-wakes.md +++ b/quest/m1/parked-read-wakes.md @@ -22,12 +22,9 @@ Follow-ups from [#4484](https://github.com/moq-dev/moq/pull/4484), which made Settled: one quest, since both are the same wakeup invariant in `rs/moq-net/src/model/track.rs` and `resume.rs`. -Land after [resume latest](/quest/m1/resume-latest.md) (#4491), which edits -the same `resume.rs` wakeups, and rebase onto it. - Public API: none. Wire: none. ## Related -- [Resume latest](/quest/m1/resume-latest.md) - edits the same `resume.rs` wakeups; lands first (#4491) +- [#4491](https://github.com/moq-dev/moq/pull/4491) - a resumed group holds demand on its replacement copy in the same `resume.rs` wakeups - [#2991](/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md) - extends the `resume.rs` takeover tests diff --git a/quest/m1/resume-latest.md b/quest/m1/resume-latest.md deleted file mode 100644 index 3949bfec69e..00000000000 --- a/quest/m1/resume-latest.md +++ /dev/null @@ -1,75 +0,0 @@ -# [S] A resumed group ends once the new copy is past it - -## Goal - -A reader that awaits only the in-flight group's `read_frame` never parks forever -after its source fails mid-group. Once the track fails over to a new copy, the -half-delivered group either completes from that copy or ends with an error, on -every lite version and for local and remote sources alike. The one exception -is a copy that later drops the group it is serving; that waits on -[SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md). - -Two reports, one mechanism: - -- #4392: a publisher aborts a track mid-group over a session. On lite-03/04 the - far subscriber's in-flight `read_frame` stays pending forever; on lite-05/06 - it errors. -- #4365: a publisher aborts a track mid-group and re-creates it under the same - name in one origin. The origin reader parks in `read_frame` while a direct - reader gets the abort. - -## Plan - -Why it hangs: `track_ended` in `rs/moq-net/src/model/front.rs` treats a copy -that dies after delivering as failover and splices a new copy. The resumed -`resume::Group` then waits in `track.poll_peek_group` -(`rs/moq-net/src/model/resume.rs`) on that copy, but never asks it for -anything. Only `resume::Subscriber::apply`, driven by `recv_group`, subscribes -to a new copy. On lite-03/04 the re-query accepts the track without asking the -peer (no TRACK stream), so no SUBSCRIBE goes out and the refusal never comes -back. An in-process copy declares no start, so its peek answers "not yet" -forever. The doc comment on `resume::Group` promises that a group reader alone -picks up the continuation; that promise is what breaks. - -Decided: - -- While a resumed group waits on a copy, it holds a subscription on that copy - with `start = latest`, the same shape as a fresh subscriber. A refusal ends - the group with that error. -- If the copy's latest group is past the resumed group, the group ends with an - error at once. A same-name re-create is a continuation, so group 2 of the old - track ends and group 3 of the new one arrives. No special case for local - broadcasts. -- If the resumed group is the copy's latest, it reads the rest from that - subscription. A copy that later drops it says so with SUBSCRIBE_DROP once - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) lands; until then that case - waits as today. -- No FETCH, not even where the version supports it. Fetching older groups from - the new copy may come later as its own quest. -- Don't abort the old group eagerly on failover instead: long-lived groups - like a catalog must survive a failover. - -`poll_finished` needs the same subscription as `poll_current`. Drop the -subscription once the wait resolves. - -A prototype of the subscription half (about 20 lines in `resume.rs`) turned -all eight #4392 cases into errors with every moq-net test passing. - -Regression tests, on a paused clock (`#[tokio::test(start_paused = true)]`): - -- The #4392 repro as `rs/moq-net/tests/track_abort_over_session.rs`: abort - mid-group over one and two hops on lite-03 through lite-06; the in-flight - read errors. -- The #4365 repro: abort mid-group 2, re-create at group 3 through the origin; - group 2 ends with an error and group 3 arrives. - -## Closes - -- [#4392](https://github.com/moq-dev/moq/issues/4392) - close this issue when the quest finishes -- [#4365](https://github.com/moq-dev/moq/issues/4365) - close this issue when the quest finishes - -## Related - -- [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - tells a resumed latest group when the new copy dropped it -- [Parked reads wake](/quest/m1/parked-read-wakes.md) - edits the same `resume.rs` wakeups; lands after this quest's #4491 -- [#2991](/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md) - extends the same `resume.rs` takeover tests diff --git a/quest/m1/subscribe-drop.md b/quest/m1/subscribe-drop.md index a74d3596df7..6b836be43c5 100644 --- a/quest/m1/subscribe-drop.md +++ b/quest/m1/subscribe-drop.md @@ -32,8 +32,8 @@ Decided: - Datagram groups stay best effort. A publisher counts a datagram as delivered, so a lost one leaves an uncovered hole that waits out the tail grace, as today. -- A resumed group ([Resumed groups](/quest/m1/resume-latest.md)) that is the - new copy's latest ends with the DROP's error when the copy drops it. +- A resumed group that is the new copy's latest ends with the DROP's error + when the copy drops it. - A dropped or aborted group is visible to readers, not silently skipped. #4533 found the Rust model releases an aborted group's sequence and skips it, so a truncated first object is indistinguishable from a group never sent. diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 4026a4f81d4..5c20397676a 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -962,9 +962,9 @@ fn last_group(end: Option) -> Option { /// mid-group therefore reaches an already-handed-out group with no bookkeeping, and a /// reader that never touches the subscription again still picks the continuation up. /// -/// A copy that dies (or never arrives) stalls the group rather than erroring it, the -/// same way a dead segment stalls the track; the loss is only surfaced once no -/// replacement can arrive. +/// A copy that dies waits for a replacement route. A missing continuation subscribes +/// that route from the frame this group needs; a refusal ends the wait, as does a live +/// edge past this group when the route declares no start. pub(crate) struct Group { state: kio::Consumer, /// The logical subscription's live max age budget. @@ -985,6 +985,10 @@ pub(crate) struct Group { /// this group, and this reader's positioned copy. current: Option, + /// Demand held on a replacement copy until this group ends, since the parent + /// subscriber may be idle. + waiting: Option, + /// The segment whose copy died under us, and why. dead: Option<(u64, Error)>, @@ -992,6 +996,27 @@ pub(crate) struct Group { stale_stats: crate::stats::Meter, } +/// A group reader's own subscription on one segment's route, starting at the frame +/// it needs. A floor never widens the parent's request past it, where a live-edge +/// start would erase the parent's continuation floor from the aggregate. +struct Waiting { + segment: u64, + start: Position, + track: track::Consumer, + pending: kio::Pending, +} + +impl Waiting { + fn new(segment: u64, start: Position, track: &track::Consumer) -> Self { + Self { + segment, + start, + track: track.clone(), + pending: track.subscribe(track::Subscription::default().with_start(start)), + } + } +} + struct Current { segment: u64, cap: Option, @@ -1030,6 +1055,12 @@ impl Clone for Group { index: self.index, end: self.end, current, + // Each reader holds its own demand, so dropping the original cannot + // strand a clone still reading the tail. + waiting: self + .waiting + .as_ref() + .map(|w| Waiting::new(w.segment, w.start, &w.track)), dead: self.dead.clone(), stale_stats: self.stale_stats.clone(), } @@ -1052,6 +1083,7 @@ impl Group { index, end: None, current: None, + waiting: None, dead: None, stale_stats: Default::default(), } @@ -1195,6 +1227,51 @@ impl Group { } } + /// Ask the replacement for this group's continuation even when the parent subscriber + /// is idle. + /// + /// The demand outlives the peek: a copy whose header has arrived, even before + /// this reader first looked, still needs it for the tail. It is released only + /// once the group ends or moves routes. + fn poll_copy( + &mut self, + segment: u64, + start: Position, + track: &track::Consumer, + waiter: &kio::Waiter, + ) -> Poll>> { + // A route is reused whatever `start` asks for, since the reader's position only + // moves forward and the segments are disjoint: keeping the earlier floor asks for + // a superset of what a later read needs, and re-subscribing to narrow it would + // drop the demand in between. + if self.waiting.as_ref().is_none_or(|w| w.segment != segment) { + self.waiting = Some(Waiting::new(segment, start, track)); + } + let result = (|| { + if let Poll::Ready(Some(copy)) = track.poll_peek_group(self.sequence, waiter) { + return Poll::Ready(Ok(Some(copy))); + } + // The pending handle owns the demand; accepting it checks refusals without + // moving this group's cursor or asking for historical data via FETCH. + ready!(self.waiting.as_ref().expect("registered above").pending.poll_ok(waiter))?; + // A declared start already tells the peek whether this group is coming, even + // when a newer group's stream overtakes it. Without one, a live edge past + // this group is the only sign it never will. + if ready!(track.poll_start(waiter)).is_none() + && track + .peek_latest() + .is_some_and(|latest| latest.sequence > self.sequence) + { + return Poll::Ready(Err(Error::NotFound)); + } + track.poll_peek_group(self.sequence, waiter).map(Ok) + })(); + if !matches!(result, Poll::Pending | Poll::Ready(Ok(Some(_)))) { + self.waiting = None; + } + result + } + /// Point `current` at the route owning frame `index` of this group. /// /// `Ready(Ok(false))` once the group can produce nothing more, and `Ready(Err(_))` @@ -1206,10 +1283,10 @@ impl Group { frame: self.index, }; let dead = self.dead.as_ref().map(|(segment, _)| *segment); - let sequence = self.sequence; let found = ready!(self.poll_covering(position, dead, waiter)); let Some((segment, track, Some(cap), bound)) = found else { + self.waiting = None; // No segment covers the position, but a latched copy still drains: // a pruned segment's cursor holds exactly the frames it owned (its // cap was its produced edge), so read it dry before giving up. Once @@ -1231,7 +1308,7 @@ impl Group { } // The route may not have delivered this group yet, so wait on its cache. - let Some(group) = ready!(track.poll_peek_group(sequence, waiter)) else { + let Some(group) = ready!(self.poll_copy(segment, position, &track, waiter))? else { // This route will never have it; fall back to whichever segment replaces it. self.dead = Some((segment, Error::NotFound)); continue; @@ -1315,6 +1392,8 @@ impl Group { pub fn poll_read_frame(&mut self, waiter: &kio::Waiter) -> Poll>> { loop { if self.end.is_some_and(|end| self.index >= end) { + // Raising the cap re-registers the demand on the next read. + self.waiting = None; return Poll::Ready(Ok(None)); } if !ready!(self.poll_current(waiter))? { @@ -1334,8 +1413,16 @@ impl Group { return Poll::Ready(Ok(Some(frame))); } Ok(None) if self.roll() => continue, - Ok(None) => return Poll::Ready(Ok(None)), - Err(err) if latency_expired => return Poll::Ready(Err(err)), + Ok(None) => { + self.waiting = None; + return Poll::Ready(Ok(None)); + } + // Terminal, and the reader may be retained past the error, so the + // demand outliving the group would pin upstream demand for good. + Err(err) if latency_expired => { + self.waiting = None; + return Poll::Ready(Err(err)); + } Err(err) => self.bury(err), } } @@ -1344,6 +1431,8 @@ impl Group { pub fn poll_next_frame(&mut self, waiter: &kio::Waiter) -> Poll>> { loop { if self.end.is_some_and(|end| self.index >= end) { + // Raising the cap re-registers the demand on the next read. + self.waiting = None; return Poll::Ready(Ok(None)); } if !ready!(self.poll_current(waiter))? { @@ -1363,8 +1452,16 @@ impl Group { return Poll::Ready(Ok(Some(frame))); } Ok(None) if self.roll() => continue, - Ok(None) => return Poll::Ready(Ok(None)), - Err(err) if latency_expired => return Poll::Ready(Err(err)), + Ok(None) => { + self.waiting = None; + return Poll::Ready(Ok(None)); + } + // Terminal, and the reader may be retained past the error, so the + // demand outliving the group would pin upstream demand for good. + Err(err) if latency_expired => { + self.waiting = None; + return Poll::Ready(Err(err)); + } Err(err) => self.bury(err), } } @@ -1378,7 +1475,10 @@ impl Group { } let current = self.current.as_mut().expect("resolved above"); let Some(cap) = current.cap else { - return current.group.poll_finished(waiter); + let count = ready!(current.group.poll_finished(waiter)); + // A complete copy is cached, so it no longer needs the demand that pulled it. + self.waiting = None; + return Poll::Ready(count); }; // A bounded copy can't declare the end; the continuation does, unless it @@ -1396,9 +1496,10 @@ impl Group { loop { let dead = self.dead.as_ref().map(|(segment, _)| *segment); let Some((segment, track, _, bound)) = ready!(self.poll_covering(seam, dead, waiter)) else { + self.waiting = None; return Poll::Ready(Ok(cap)); }; - match ready!(track.poll_peek_group(self.sequence, waiter)) { + match ready!(self.poll_copy(segment, seam, &track, waiter))? { // The continuation's copy declares the count: its own count // already includes the frames it skipped. Some(continuation) => { @@ -1415,7 +1516,9 @@ impl Group { continue; } let index = self.index; - return continuation.poll_finished(waiter).map_ok(|_| index); + let end = ready!(continuation.poll_finished(waiter)); + self.waiting = None; + return Poll::Ready(end.map(|_| index)); } // This route will never have it; wait for whatever replaces it. None => self.dead = Some((segment, Error::NotFound)), @@ -4318,13 +4421,9 @@ mod test { ); } - /// A route that has run ahead of the seam but has not been given up on still parks. - /// - /// It may yet deliver the missing frames out of order, and its own progress is what - /// moved the resume point past them, so its being ahead is not evidence the frames - /// are gone. Only a route already declared dead strands the reader. + /// Latest-only recovery ends a missing continuation once the replacement is ahead. #[tokio::test] - async fn live_route_ahead_of_the_seam_still_parks() { + async fn live_route_ahead_of_the_seam_ends_the_group() { let (track_a, consumer_a) = track_pair("a"); let (mut track_b, consumer_b) = track_pair("b"); @@ -4346,12 +4445,40 @@ mod test { write_group(&mut track_b, 1, "b1"); assert_eq!(recv(&mut sub), 1); + assert!( + matches!(reading.read_frame().now_or_never(), Some(Err(Error::NotFound))), + "latest recovery cannot request an older continuation" + ); + } + + /// A declared start at or below the seam promises the group, so a newer group + /// overtaking it on the wire does not end the wait. + #[tokio::test] + async fn declared_start_waits_for_an_overtaken_group() { + let (track_a, consumer_a) = track_pair("a"); + let (mut track_b, consumer_b) = track_pair("b"); + + let mut producer = Producer::new(); + producer.takeover(&consumer_a).unwrap(); + let mut sub = producer.consume().subscribe(replay()); + + let mut group = track_a.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(Timestamp::ZERO, b"a0".to_vec()).unwrap(); + group.write_frame(Timestamp::ZERO, b"a1".to_vec()).unwrap(); + + let mut reading = sub.recv_group().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(read(&mut reading), b"a0"); + assert_eq!(read(&mut reading), b"a1"); + + producer.takeover(&consumer_b).unwrap(); + track_b.start_at(0).unwrap(); + write_group(&mut track_b, 1, "b1"); + assert_eq!(recv(&mut sub), 1); assert!( reading.read_frame().now_or_never().is_none(), - "a live route may still fill the seam out of order" + "the declared start still promises group 0" ); - // Once it does, the reader picks up where it left off. let mut group = track_b.create_group(group::Info { sequence: 0 }).unwrap(); group.start_at(2).unwrap(); group.write_frame(Timestamp::ZERO, b"b2".to_vec()).unwrap(); @@ -4996,13 +5123,10 @@ mod test { drop(group); track_a.abort(Error::Dropped).unwrap(); - // While B covers the seam the count is still open: B may serve frame 2. + // Until B advances, it may still serve frame 2. + assert!(reading.finished().now_or_never().is_none()); write_group(&mut track_b, 1, "b1"); assert_eq!(recv(&mut sub), 1); - assert!( - reading.finished().now_or_never().is_none(), - "the seam is still coverable" - ); // Enough failovers prune A and B: nothing can serve the seam anymore, so // the cap is the end. @@ -5064,6 +5188,125 @@ mod test { } } + #[tokio::test(start_paused = true)] + async fn group_only_read_requests_its_continuation() { + group_only_continuation(false).await; + } + + #[tokio::test(start_paused = true)] + async fn group_only_finish_requests_its_continuation() { + group_only_continuation(true).await; + } + + async fn group_only_continuation(finish: bool) { + let (track_a, consumer_a) = track_pair("a"); + let (track_b, consumer_b) = track_pair("b"); + let mut producer = Producer::new(); + producer.takeover(&consumer_a).unwrap(); + let mut sub = producer.consume().subscribe(None); + let mut group = track_a.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(Timestamp::ZERO, b"head".to_vec()).unwrap(); + // `finished` probes the seam past a bounded copy; a read waits past a dead one. + if finish { + producer.takeover(&consumer_b).unwrap(); + } + let mut reading = sub.recv_group().await.unwrap().unwrap(); + // The parent subscriber never asks B for anything. + drop(sub); + let parked = |reading: &mut group::Consumer| match finish { + true => reading.finished().now_or_never().is_none(), + false => reading.read_frame().now_or_never().is_none(), + }; + if !finish { + assert_eq!(read(&mut reading), b"head"); + producer.takeover(&consumer_b).unwrap(); + track_a.abort(Error::Dropped).unwrap(); + } + assert!(parked(&mut reading)); + // Demand starts at the frame the reader needs: its cursor, or the seam past A's copy. + assert_eq!( + track_b.subscription().expect("group-only demand").start, + Some(Position { group: 0, frame: 1 }) + ); + + // The header arriving does not end the wait: the tail still needs the demand. + let mut copy = track_b.create_group(group::Info { sequence: 0 }).unwrap(); + copy.write_frame(Timestamp::ZERO, b"head".to_vec()).unwrap(); + assert!(parked(&mut reading)); + assert!(track_b.subscription().is_some(), "the tail still needs demand"); + + copy.write_frame(Timestamp::ZERO, b"tail".to_vec()).unwrap(); + copy.finish().unwrap(); + if finish { + reading.finished().await.unwrap(); + assert_eq!(read(&mut reading), b"head"); + } + assert_eq!(read(&mut reading), b"tail"); + assert!(reading.read_frame().await.unwrap().is_none()); + assert!(track_b.subscription().is_none(), "the finished group releases demand"); + } + + /// A replacement cached before the reader first looks still needs demand for its + /// tail, and a clone keeps that demand after the original reader is dropped. + #[tokio::test(start_paused = true)] + async fn cached_replacement_holds_demand_across_clones() { + let (track_a, consumer_a) = track_pair("a"); + let (track_b, consumer_b) = track_pair("b"); + let mut producer = Producer::new(); + producer.takeover(&consumer_a).unwrap(); + let mut sub = producer.consume().subscribe(None); + let mut group = track_a.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(Timestamp::ZERO, b"head".to_vec()).unwrap(); + let mut reading = sub.recv_group().await.unwrap().unwrap(); + drop(sub); + assert_eq!(read(&mut reading), b"head"); + + producer.takeover(&consumer_b).unwrap(); + track_a.abort(Error::Dropped).unwrap(); + let mut copy = track_b.create_group(group::Info { sequence: 0 }).unwrap(); + copy.write_frame(Timestamp::ZERO, b"head".to_vec()).unwrap(); + assert!(reading.read_frame().now_or_never().is_none()); + assert!(track_b.subscription().is_some(), "a cached header still needs demand"); + + let mut clone = reading.clone(); + drop(reading); + assert!(track_b.subscription().is_some(), "the clone keeps its own demand"); + + copy.write_frame(Timestamp::ZERO, b"tail".to_vec()).unwrap(); + copy.finish().unwrap(); + assert_eq!(read(&mut clone), b"tail"); + assert!(clone.read_frame().await.unwrap().is_none()); + assert!(track_b.subscription().is_none(), "the finished group releases demand"); + } + + /// A reader that stops at its cap may be retained, so the cap releases the demand + /// on the replacement rather than the drop. + #[tokio::test(start_paused = true)] + async fn capped_read_releases_replacement_demand() { + let (track_a, consumer_a) = track_pair("a"); + let (track_b, consumer_b) = track_pair("b"); + let mut producer = Producer::new(); + producer.takeover(&consumer_a).unwrap(); + let mut sub = producer.consume().subscribe(None); + let mut group = track_a.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(Timestamp::ZERO, b"head".to_vec()).unwrap(); + let mut reading = sub.recv_group().await.unwrap().unwrap(); + drop(sub); + assert_eq!(read(&mut reading), b"head"); + + producer.takeover(&consumer_b).unwrap(); + track_a.abort(Error::Dropped).unwrap(); + let mut copy = track_b.create_group(group::Info { sequence: 0 }).unwrap(); + copy.write_frame(Timestamp::ZERO, b"head".to_vec()).unwrap(); + copy.write_frame(Timestamp::ZERO, b"mid".to_vec()).unwrap(); + reading.set_frames(..2); + assert_eq!(read(&mut reading), b"mid"); + assert!(track_b.subscription().is_some(), "the read still holds demand"); + + assert!(reading.read_frame().now_or_never().unwrap().unwrap().is_none()); + assert!(track_b.subscription().is_none(), "the cap releases demand"); + } + /// `finished()` resolves when the seam's covering route skip-declared the /// group: its segment geometrically covers the continuation, but its /// SUBSCRIBE_START floor proves the group will never arrive, so the cap is diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index ecadbbfb180..a7391e46206 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -17,83 +17,94 @@ fn produce_origin(hop: u64) -> moq_net::origin::Producer { producer } -async fn read_all(group: &mut group::Consumer) -> Vec> { +async fn read_all(group: &mut group::Consumer) -> moq_net::Result>> { let mut frames = Vec::new(); - while let Some(frame) = group.read_frame().await.expect("group aborted") { + while let Some(frame) = group.read_frame().await? { frames.push(frame.payload.to_vec()); } - frames + Ok(frames) } -/// The group in flight when the last reader left comes back whole on rejoin. +/// A rejoin recovers the group reset on leave, whether or not the live edge moved past it. /// /// The relay cancels its idle upstream subscription, resetting that group mid-transfer. /// Resuming the rejoin at the frame where the reset landed would ask upstream for a tail /// whose head is gone, so the group would never reach the returning reader. -#[tokio::test] +#[tokio::test(start_paused = true)] async fn rejoin_recovers_the_group_reset_on_leave() { - for version in ["moq-lite-05", "moq-lite-06"] { - tokio::time::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").await.unwrap(); - - let ts = |ms| Timestamp::from_millis(ms).unwrap(); - let prefs = || track::Subscription::default().with_max_age(Duration::from_secs(10)); - - let mut group = track.append_group().unwrap(); - group.write_frame(ts(0), b"a0".as_ref()).unwrap(); - group.finish().unwrap(); - let mut open = track.append_group().unwrap(); - open.write_frame(ts(100), b"b0".as_ref()).unwrap(); - - let mut sub = remote.track("video").unwrap().subscribe(prefs()).await.unwrap(); - loop { - let mut group = sub.recv_group().await.unwrap().unwrap(); - group.read_frame().await.unwrap().unwrap(); - if group.sequence == 1 { - break; + for advance in [false, true] { + for version in ["moq-lite-05", "moq-lite-06"] { + tokio::time::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").await.unwrap(); + + let ts = |ms| Timestamp::from_millis(ms).unwrap(); + let prefs = || track::Subscription::default().with_max_age(Duration::from_secs(10)); + + let mut group = track.append_group().unwrap(); + group.write_frame(ts(0), b"a0".as_ref()).unwrap(); + group.finish().unwrap(); + let mut open = track.append_group().unwrap(); + open.write_frame(ts(100), b"b0".as_ref()).unwrap(); + + let mut sub = remote.track("video").unwrap().subscribe(prefs()).await.unwrap(); + loop { + let mut group = sub.recv_group().await.unwrap().unwrap(); + group.read_frame().await.unwrap().unwrap(); + if group.sequence == 1 { + break; + } } - } - - // Leave mid-group; the publisher cuts the open group once demand is gone. - drop(sub); - track.unused().await.unwrap(); - open.write_frame(ts(133), b"b1".as_ref()).unwrap(); - open.finish().unwrap(); - - let mut live = track.append_group().unwrap(); - live.write_frame(ts(5000), b"c0".as_ref()).unwrap(); - live.finish().unwrap(); - - let mut sub = remote.track("video").unwrap().subscribe(prefs()).await.unwrap(); - let mut rejoined = Vec::new(); - loop { - let mut group = sub.recv_group().await.unwrap().unwrap(); - let sequence = group.sequence; - rejoined.push((sequence, read_all(&mut group).await)); - if sequence == 2 { - break; + + // Leave mid-group; the publisher cuts the open group once demand is gone. + drop(sub); + track.unused().await.unwrap(); + open.write_frame(ts(133), b"b1".as_ref()).unwrap(); + open.finish().unwrap(); + + if advance { + let mut live = track.append_group().unwrap(); + live.write_frame(ts(5000), b"c0".as_ref()).unwrap(); + live.finish().unwrap(); + } + + let mut sub = remote.track("video").unwrap().subscribe(prefs()).await.unwrap(); + let mut rejoined = Vec::new(); + loop { + let mut group = sub.recv_group().await.unwrap().unwrap(); + let sequence = group.sequence; + rejoined.push((sequence, read_all(&mut group).await)); + if sequence == if advance { 2 } else { 1 } { + break; + } + } + let reset = rejoined.iter().find(|(seq, _)| *seq == 1).expect("reset group"); + // The reader asks for the group it is missing, so the reset group comes back + // whole even once the live edge has moved past it. + assert_eq!( + reset.1.as_ref().unwrap(), + &vec![b"b0".to_vec(), b"b1".to_vec()], + "{version}" + ); + if advance { + assert_eq!(rejoined.last().unwrap().1.as_ref().unwrap(), &vec![b"c0".to_vec()]); } - } - assert!( - rejoined.contains(&(1, vec![b"b0".to_vec(), b"b1".to_vec()])), - "{version}: the reset group never came back whole: {rejoined:?}" - ); - }) - .await - .unwrap_or_else(|_| panic!("{version} timed out")); + }) + .await + .unwrap_or_else(|_| panic!("{version} timed out")); + } } } diff --git a/rs/moq-net/tests/track_abort_over_session.rs b/rs/moq-net/tests/track_abort_over_session.rs new file mode 100644 index 00000000000..4fbc351386f --- /dev/null +++ b/rs/moq-net/tests/track_abort_over_session.rs @@ -0,0 +1,108 @@ +//! A track aborted by its publisher while the broadcast stays announced, read by a +//! subscriber on the far side of one session. + +mod support; + +use std::time::Duration; + +use moq_net::{Hop, Timestamp, Version}; +use support::harness::{MockConnectOptions, connect_mock}; + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + tokio::spawn(support::harness::run(driver)); + producer +} + +/// A group reader alone must drive failover, without polling the next group. +async fn after_abort(version: &str, hops: u32) { + let publisher = produce_origin(1); + let relay = produce_origin(2); + let far = produce_origin(3); + 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 mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.server_publish = Some(relay.consume()); + options.client_subscribe = Some(far.clone()); + let _pair2 = connect_mock(options).await; + let consumer = if hops == 2 { far.consume() } else { relay.consume() }; + consumer.routed("bench").await.unwrap(); + let remote = consumer.request_broadcast("bench").await.unwrap(); + let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap(); + let mut open = track.append_group().unwrap(); + open.write_frame(Timestamp::ZERO, b"head".as_ref()).unwrap(); + let mut group = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(group.read_frame().await.unwrap().unwrap().payload, b"head".as_ref()); + track.abort(moq_net::Error::Cancel).unwrap(); + drop(open); + assert!( + tokio::time::timeout(Duration::from_secs(2), group.read_frame()) + .await + .expect("group-only read hung") + .is_err() + ); +} + +macro_rules! abort_test { + ($name:ident, $version:literal, $hops:literal) => { + #[tokio::test(start_paused = true)] + async fn $name() { + after_abort($version, $hops).await; + } + }; +} +abort_test!(abort_03_one, "moq-lite-03", 1); +abort_test!(abort_03_two, "moq-lite-03", 2); +abort_test!(abort_04_one, "moq-lite-04", 1); +abort_test!(abort_04_two, "moq-lite-04", 2); +abort_test!(abort_05_one, "moq-lite-05", 1); +abort_test!(abort_05_two, "moq-lite-05", 2); +abort_test!(abort_06_one, "moq-lite-06", 1); +abort_test!(abort_06_two, "moq-lite-06", 2); + +async fn recreate(finish: bool) { + let origin = produce_origin(1); + let broadcast = origin.create_broadcast("bench").unwrap(); + broadcast.announce(Default::default()).unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + let remote = origin.consume().request_broadcast("bench").await.unwrap(); + let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap(); + let mut open = track.create_group(moq_net::group::Info { sequence: 2 }).unwrap(); + open.write_frame(Timestamp::ZERO, b"head".as_ref()).unwrap(); + let mut group = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(group.sequence, 2); + group.read_frame().await.unwrap().unwrap(); + track.abort(moq_net::Error::Cancel).unwrap(); + let fresh = broadcast.create_track("video", None).unwrap(); + let mut next = fresh.create_group(moq_net::group::Info { sequence: 3 }).unwrap(); + next.write_frame(Timestamp::from_millis(1).unwrap(), b"next".as_ref()) + .unwrap(); + drop(open); + let result = tokio::time::timeout(Duration::from_secs(2), async { + if finish { + group.finished().await.map(|_| ()) + } else { + group.read_frame().await.map(|_| ()) + } + }) + .await + .expect("group-only failover hung"); + assert!(result.is_err()); + let next = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(next.sequence, 3); +} + +#[tokio::test(start_paused = true)] +async fn recreated_track_skips_old_group() { + recreate(false).await; +} + +#[tokio::test(start_paused = true)] +async fn recreated_track_finishes_old_group() { + recreate(true).await; +}