From 6c19b0aeaf0e4e482b5a41124756b830130bd8b0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 28 Sep 2026 22:20:18 -0700 Subject: [PATCH 1/6] chore(quest): claim resume-latest Co-Authored-By: GPT-6 From d9e513ac2f5d89a46fbafc6dc16844c456024aeb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 29 Sep 2026 00:51:43 -0700 Subject: [PATCH 2/6] Subscribe resumed group waits at the live edge Co-Authored-By: GPT-6 Astra --- doc/concept/moq-lite.md | 11 ++ quest/m1/README.md | 1 - quest/m1/resume-latest.md | 74 ---------- quest/m1/subscribe-drop.md | 3 +- rs/moq-net/src/model/resume.rs | 140 +++++++++++++----- rs/moq-net/tests/rejoin.rs | 143 ++++++++++--------- rs/moq-net/tests/track_abort_over_session.rs | 108 ++++++++++++++ 7 files changed, 305 insertions(+), 175 deletions(-) delete mode 100644 quest/m1/resume-latest.md create mode 100644 rs/moq-net/tests/track_abort_over_session.rs diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index c2947245ed4..6f0e449d266 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -47,6 +47,17 @@ nonzero effective maximum age, or one second when no maximum age is set. moq-lite 05 and 06 instead account for group sequences using received headers and `SUBSCRIBE_DROP`. +## 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 opens a latest subscription even +when the caller is not polling for the next group. If that copy refuses the +subscription or has advanced past the group, the read ends with an error. If +the group is still latest, its remaining frames can continue from that copy. +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/README.md b/quest/m1/README.md index 19dad3cbf70..724e00deed6 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -35,7 +35,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Watch refusal](/quest/m1/watch-refusal.md) - `` shows an origin refusal as an error instead of sitting offline - [moqsink recoverable errors](/quest/m1/moqsink-keyframe-latch.md) - a leading delta or a timestamp rewind drops frames until a keyframe instead of invalidating a moqsink pad - [Splice edge cases](/quest/m1/splice-edges.md) - an unstamped successor, a pruned segment's boundary group, and a warm head during a takeover are each handled correctly -- [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 - [Track tail hardening](/quest/m1/track-tail-hardening.md) - Rust and JS wait out a track's tail by the same rules, with the known hang, count, truncation, grace, and memory holes closed - [JS close end](/quest/m1/js-track-close-end.md) - a JS track's clean close ends at its own last group, not a sibling producer's - [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 diff --git a/quest/m1/resume-latest.md b/quest/m1/resume-latest.md deleted file mode 100644 index f594bfc566c..00000000000 --- a/quest/m1/resume-latest.md +++ /dev/null @@ -1,74 +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 -- [Splice edge cases](/quest/m1/splice-edges.md) - the same splice code and test harness diff --git a/quest/m1/subscribe-drop.md b/quest/m1/subscribe-drop.md index 3f93210d63d..b4d6d8b93cf 100644 --- a/quest/m1/subscribe-drop.md +++ b/quest/m1/subscribe-drop.md @@ -32,8 +32,7 @@ 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. Update `drafts/draft-lcurley-moq-lite.md` (SUBSCRIBE_DROP, SUBSCRIBE_END, the lite-07 changelog), `doc/concept/moq-lite.md`, and the Rust and JS lite diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 4669dd10234..23308d00372 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -911,9 +911,8 @@ 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 requests +/// that route's latest group; a refusal or a live edge past this group ends the wait. pub(crate) struct Group { state: kio::Consumer, /// The logical subscription's live max age budget. @@ -934,6 +933,9 @@ pub(crate) struct Group { /// this group, and this reader's positioned copy. current: Option, + /// Demand held while a group-only reader waits on a replacement copy. + waiting: Option<(u64, kio::Pending)>, + /// The segment whose copy died under us, and why. dead: Option<(u64, Error)>, @@ -979,6 +981,7 @@ impl Clone for Group { index: self.index, end: self.end, current, + waiting: None, dead: self.dead.clone(), stale_stats: self.stale_stats.clone(), } @@ -1001,6 +1004,7 @@ impl Group { index, end: None, current: None, + waiting: None, dead: None, stale_stats: Default::default(), } @@ -1144,6 +1148,38 @@ impl Group { } } + /// Ask the replacement for its live edge even when the parent subscriber is idle. + fn poll_copy( + &mut self, + segment: u64, + track: &track::Consumer, + waiter: &kio::Waiter, + ) -> Poll>> { + if let Poll::Ready(Some(copy)) = track.poll_peek_group(self.sequence, waiter) { + self.waiting = None; + return Poll::Ready(Ok(Some(copy))); + } + if self.waiting.as_ref().is_none_or(|(id, _)| *id != segment) { + self.waiting = Some((segment, track.subscribe(None))); + } + let result = (|| { + // 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").1.poll_ok(waiter))?; + if 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 result.is_ready() { + 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(_))` @@ -1155,10 +1191,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 @@ -1180,7 +1216,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, &track, waiter))? else { // This route will never have it; fall back to whichever segment replaces it. self.dead = Some((segment, Error::NotFound)); continue; @@ -1345,9 +1381,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, &track, waiter))? { // The continuation's copy declares the count: its own count // already includes the frames it skipped. Some(continuation) => { @@ -4068,13 +4105,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"); @@ -4097,15 +4130,9 @@ mod test { 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" + matches!(reading.read_frame().now_or_never(), Some(Err(Error::NotFound))), + "latest recovery cannot request an older continuation" ); - - // 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(); - assert_eq!(read(&mut reading), b"b2"); } #[tokio::test] @@ -4746,13 +4773,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. @@ -4814,10 +4838,62 @@ mod test { } } + #[tokio::test(start_paused = true)] + async fn group_only_read_requests_latest_continuation() { + group_only_continuation(false).await; + } + + #[tokio::test(start_paused = true)] + async fn group_only_finish_requests_latest_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(); + if finish { + producer.takeover(&consumer_b).unwrap(); + } + let mut reading = sub.recv_group().await.unwrap().unwrap(); + drop(sub); + if !finish { + assert_eq!(read(&mut reading), b"head"); + } + if !finish { + producer.takeover(&consumer_b).unwrap(); + } + if !finish { + track_a.abort(Error::Dropped).unwrap(); + } + if finish { + assert!(reading.finished().now_or_never().is_none()); + } else { + assert!(reading.read_frame().now_or_never().is_none()); + } + assert_eq!(track_b.subscription().expect("group-only demand").start, None); + 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"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"); + } else { + assert_eq!(read(&mut reading), b"tail"); + } + assert!(track_b.subscription().is_none(), "resolved wait 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 - /// the end. Polled without draining, and woken by the successor's track (the + /// SUBSCRIBE_START floor proves the group will never arrive. Advancing the + /// latest group makes that loss an error. Polled without draining (the /// seam probe parks on the peek), not just the segment list. #[tokio::test] async fn finished_resolves_when_the_successor_skips_the_seam() { @@ -4840,18 +4916,18 @@ mod test { assert!(reading.finished().now_or_never().is_none(), "the seam is coverable"); // B declares it starts at group 1 and produces it: group 0's continuation - // is skipped for good, so the cap is the end. + // is skipped for good, so its missing continuation is an error. track_b.start_at(1).unwrap(); write_group(&mut track_b, 1, "b1"); assert_eq!(recv(&mut sub), 1); - assert_eq!( + assert!(matches!( reading .finished() .now_or_never() .expect("a skip-declared seam must resolve the count") - .unwrap(), - 2 - ); + .unwrap_err(), + Error::NotFound + )); } /// A reader that already latched a pruned segment's copy keeps draining it: the diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index ecadbbfb180..b0fd77ecb76 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 latest group, but errors if its missing tail is now historical. /// /// 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"); + if advance { + // A copy that already arrived can drain; a missing old tail errors + // once the replacement's latest advances beyond it. + if let Ok(frames) = &reset.1 { + assert_eq!(frames, &vec![b"b0".to_vec(), b"b1".to_vec()]); + } + assert_eq!(rejoined.last().unwrap().1.as_ref().unwrap(), &vec![b"c0".to_vec()]); + } else { + assert_eq!(reset.1.as_ref().unwrap(), &vec![b"b0".to_vec(), b"b1".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; +} From 2dfdf0eccbbd3e9cb3e7991e42966ae34ceb5cf3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 29 Sep 2026 10:40:46 -0700 Subject: [PATCH 3/6] Keep group-only demand through the tail, trust a declared start The temporary subscription was released as soon as the replacement's group header arrived, letting an idle relay cancel upstream before the tail. It now lives until the group ends or moves routes. A live edge past the missing group now ends it only when the copy declares no start. On lite-06 the SUBSCRIBE_START already answers whether the group is coming, so a newer group overtaking it on the wire no longer errors a group still in flight. Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 9 ++- rs/moq-net/src/model/resume.rs | 114 ++++++++++++++++++++++++--------- rs/moq-net/tests/rejoin.rs | 17 +++-- 3 files changed, 102 insertions(+), 38 deletions(-) diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 6f0e449d266..ce3488af35d 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -51,9 +51,12 @@ and `SUBSCRIBE_DROP`. Rust origin readers can keep reading an in-flight group after its source fails. A group reader waiting on a replacement copy opens a latest subscription even -when the caller is not polling for the next group. If that copy refuses the -subscription or has advanced past the group, the read ends with an error. If -the group is still latest, its remaining frames can continue from that copy. +when the caller is not polling for the next group, and holds it until the group +ends. If that copy refuses the subscription, the read ends with an error. On +moq-lite 06, `SUBSCRIBE_START` says whether the group is still coming, so a +newer group overtaking it on the wire does not end the wait. Older versions and +local copies declare no start, so a copy that has advanced past the group ends +the read with an error. Otherwise its remaining frames continue from that copy. 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. diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 23308d00372..0b34de61f99 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -912,7 +912,8 @@ fn last_group(end: Option) -> Option { /// reader that never touches the subscription again still picks the continuation up. /// /// A copy that dies waits for a replacement route. A missing continuation requests -/// that route's latest group; a refusal or a live edge past this group ends the wait. +/// that route's latest group; 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. @@ -1149,6 +1150,9 @@ impl Group { } /// Ask the replacement for its live edge even when the parent subscriber is idle. + /// + /// The demand outlives the peek: a copy whose header has arrived still needs it + /// for the tail, so it is released only once the group ends or moves routes. fn poll_copy( &mut self, segment: u64, @@ -1156,7 +1160,6 @@ impl Group { waiter: &kio::Waiter, ) -> Poll>> { if let Poll::Ready(Some(copy)) = track.poll_peek_group(self.sequence, waiter) { - self.waiting = None; return Poll::Ready(Ok(Some(copy))); } if self.waiting.as_ref().is_none_or(|(id, _)| *id != segment) { @@ -1166,15 +1169,19 @@ impl Group { // 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").1.poll_ok(waiter))?; - if track - .peek_latest() - .is_some_and(|latest| latest.sequence > self.sequence) + // 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 result.is_ready() { + if !matches!(result, Poll::Pending | Poll::Ready(Ok(Some(_)))) { self.waiting = None; } result @@ -1319,7 +1326,10 @@ impl Group { return Poll::Ready(Ok(Some(frame))); } Ok(None) if self.roll() => continue, - Ok(None) => return Poll::Ready(Ok(None)), + Ok(None) => { + self.waiting = None; + return Poll::Ready(Ok(None)); + } Err(err) if latency_expired => return Poll::Ready(Err(err)), Err(err) => self.bury(err), } @@ -1348,7 +1358,10 @@ impl Group { return Poll::Ready(Ok(Some(frame))); } Ok(None) if self.roll() => continue, - Ok(None) => return Poll::Ready(Ok(None)), + Ok(None) => { + self.waiting = None; + return Poll::Ready(Ok(None)); + } Err(err) if latency_expired => return Poll::Ready(Err(err)), Err(err) => self.bury(err), } @@ -1363,7 +1376,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 @@ -1391,7 +1407,9 @@ impl Group { let mut continuation = track.guard_group(continuation, self.subscription.clone(), self.anchor.clone(), bound); continuation.set_stale_meter(self.stale_stats.clone()); - return continuation.poll_finished(waiter); + let count = ready!(continuation.poll_finished(waiter)); + self.waiting = None; + return Poll::Ready(count); } // This route will never have it; wait for whatever replaces it. None => self.dead = Some((segment, Error::NotFound)), @@ -4135,6 +4153,40 @@ mod test { ); } + /// 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(), + "the declared start still promises group 0" + ); + + 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(); + assert_eq!(read(&mut reading), b"b2"); + } + #[tokio::test] async fn takeover_after_empty_segment_keeps_live_edge() { let (track_a, consumer_a) = track_pair("a"); @@ -4856,44 +4908,46 @@ mod test { 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"); - } - if !finish { producer.takeover(&consumer_b).unwrap(); - } - if !finish { track_a.abort(Error::Dropped).unwrap(); } - if finish { - assert!(reading.finished().now_or_never().is_none()); - } else { - assert!(reading.read_frame().now_or_never().is_none()); - } + assert!(parked(&mut reading)); assert_eq!(track_b.subscription().expect("group-only demand").start, None); + + // 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"); - } else { - assert_eq!(read(&mut reading), b"tail"); } - assert!(track_b.subscription().is_none(), "resolved wait releases demand"); + 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"); } /// `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. Advancing the - /// latest group makes that loss an error. Polled without draining (the + /// SUBSCRIBE_START floor proves the group will never arrive, so the cap is + /// the end. Polled without draining, and woken by the successor's track (the /// seam probe parks on the peek), not just the segment list. #[tokio::test] async fn finished_resolves_when_the_successor_skips_the_seam() { @@ -4916,18 +4970,18 @@ mod test { assert!(reading.finished().now_or_never().is_none(), "the seam is coverable"); // B declares it starts at group 1 and produces it: group 0's continuation - // is skipped for good, so its missing continuation is an error. + // is skipped for good, so the cap is the end. track_b.start_at(1).unwrap(); write_group(&mut track_b, 1, "b1"); assert_eq!(recv(&mut sub), 1); - assert!(matches!( + assert_eq!( reading .finished() .now_or_never() .expect("a skip-declared seam must resolve the count") - .unwrap_err(), - Error::NotFound - )); + .unwrap(), + 2 + ); } /// A reader that already latched a pruned segment's copy keeps draining it: the diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index b0fd77ecb76..325c4c00fe5 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -25,7 +25,8 @@ async fn read_all(group: &mut group::Consumer) -> moq_net::Result>> Ok(frames) } -/// A rejoin recovers the latest group, but errors if its missing tail is now historical. +/// A rejoin recovers the reset group, unless its missing tail is historical on a version +/// that cannot declare where the replacement starts. /// /// 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 @@ -93,10 +94,16 @@ async fn rejoin_recovers_the_group_reset_on_leave() { } let reset = rejoined.iter().find(|(seq, _)| *seq == 1).expect("reset group"); if advance { - // A copy that already arrived can drain; a missing old tail errors - // once the replacement's latest advances beyond it. - if let Ok(frames) = &reset.1 { - assert_eq!(frames, &vec![b"b0".to_vec(), b"b1".to_vec()]); + // lite-06 declares where the replacement starts, so the reset group is + // still promised. lite-05 declares nothing, and a live edge past the + // missing tail is the only sign it will never come. + match version { + "moq-lite-06" => assert_eq!(reset.1.as_ref().unwrap(), &vec![b"b0".to_vec(), b"b1".to_vec()]), + _ => assert!( + matches!(reset.1, Err(moq_net::Error::NotFound)), + "{version}: {:?}", + reset.1 + ), } assert_eq!(rejoined.last().unwrap().1.as_ref().unwrap(), &vec![b"c0".to_vec()]); } else { From 97ba9d978df555646f8384fbed368cf02c4fb0b4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 29 Sep 2026 15:03:50 -0700 Subject: [PATCH 4/6] Subscribe a group reader from its own continuation A live-edge start absorbs every floor in the aggregated subscription, so a group reader's demand erased a re-splice's continuation floor. The reader now subscribes from the frame it needs, which never widens a parent floor at or below it, and the reset group in rejoin.rs comes back whole on lite-05 as well as lite-06. Demand is also registered before the peek, so a replacement cached before the first poll still has demand for its tail, and each clone holds its own subscription. Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 19 +++--- rs/moq-net/src/model/resume.rs | 107 +++++++++++++++++++++++++++------ rs/moq-net/tests/rejoin.rs | 23 +++---- 3 files changed, 104 insertions(+), 45 deletions(-) diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index ce3488af35d..f28bd6f730c 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -50,16 +50,15 @@ and `SUBSCRIBE_DROP`. ## 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 opens a latest subscription even -when the caller is not polling for the next group, and holds it until the group -ends. If that copy refuses the subscription, the read ends with an error. On -moq-lite 06, `SUBSCRIBE_START` says whether the group is still coming, so a -newer group overtaking it on the wire does not end the wait. Older versions and -local copies declare no start, so a copy that has advanced past the group ends -the read with an error. Otherwise its remaining frames continue from that copy. -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. +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 diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 0b34de61f99..52d1e27ab97 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -911,9 +911,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 waits for a replacement route. A missing continuation requests -/// that route's latest group; a refusal ends the wait, as does a live edge past this -/// group when the route declares no start. +/// 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. @@ -934,8 +934,9 @@ pub(crate) struct Group { /// this group, and this reader's positioned copy. current: Option, - /// Demand held while a group-only reader waits on a replacement copy. - waiting: Option<(u64, kio::Pending)>, + /// 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)>, @@ -944,6 +945,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, @@ -982,7 +1004,12 @@ impl Clone for Group { index: self.index, end: self.end, current, - waiting: None, + // 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(), } @@ -1149,26 +1176,29 @@ impl Group { } } - /// Ask the replacement for its live edge even when the parent subscriber is idle. + /// 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 still needs it - /// for the tail, so it is released only once the group ends or moves routes. + /// 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>> { - if let Poll::Ready(Some(copy)) = track.poll_peek_group(self.sequence, waiter) { - return Poll::Ready(Ok(Some(copy))); - } - if self.waiting.as_ref().is_none_or(|(id, _)| *id != segment) { - self.waiting = Some((segment, track.subscribe(None))); + 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").1.poll_ok(waiter))?; + 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. @@ -1223,7 +1253,7 @@ impl Group { } // The route may not have delivered this group yet, so wait on its cache. - let Some(group) = ready!(self.poll_copy(segment, &track, 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; @@ -1400,7 +1430,7 @@ impl Group { self.waiting = None; return Poll::Ready(Ok(cap)); }; - match ready!(self.poll_copy(segment, &track, 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) => { @@ -4891,12 +4921,12 @@ mod test { } #[tokio::test(start_paused = true)] - async fn group_only_read_requests_latest_continuation() { + async fn group_only_read_requests_its_continuation() { group_only_continuation(false).await; } #[tokio::test(start_paused = true)] - async fn group_only_finish_requests_latest_continuation() { + async fn group_only_finish_requests_its_continuation() { group_only_continuation(true).await; } @@ -4925,7 +4955,11 @@ mod test { track_a.abort(Error::Dropped).unwrap(); } assert!(parked(&mut reading)); - assert_eq!(track_b.subscription().expect("group-only demand").start, None); + // 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(); @@ -4944,6 +4978,39 @@ mod test { 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"); + } + /// `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 325c4c00fe5..a7391e46206 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -25,8 +25,7 @@ async fn read_all(group: &mut group::Consumer) -> moq_net::Result>> Ok(frames) } -/// A rejoin recovers the reset group, unless its missing tail is historical on a version -/// that cannot declare where the replacement starts. +/// 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 @@ -93,21 +92,15 @@ async fn rejoin_recovers_the_group_reset_on_leave() { } } 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 { - // lite-06 declares where the replacement starts, so the reset group is - // still promised. lite-05 declares nothing, and a live edge past the - // missing tail is the only sign it will never come. - match version { - "moq-lite-06" => assert_eq!(reset.1.as_ref().unwrap(), &vec![b"b0".to_vec(), b"b1".to_vec()]), - _ => assert!( - matches!(reset.1, Err(moq_net::Error::NotFound)), - "{version}: {:?}", - reset.1 - ), - } assert_eq!(rejoined.last().unwrap().1.as_ref().unwrap(), &vec![b"c0".to_vec()]); - } else { - assert_eq!(reset.1.as_ref().unwrap(), &vec![b"b0".to_vec(), b"b1".to_vec()]); } }) .await From 6c08334957b35c68e8dac7f0568561d008914a43 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 29 Sep 2026 22:14:25 -0700 Subject: [PATCH 5/6] Release group-only demand on a terminal read The recovery demand is unbounded, so a reader retained past a latency-expired error pinned upstream demand for as long as it lived. The clean-end path already released it; the expiry path now does too. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/resume.rs | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 76d1e621659..a27ff59c5fb 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -1231,6 +1231,10 @@ impl Group { 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)); } @@ -1402,7 +1406,12 @@ impl Group { self.waiting = None; return Poll::Ready(Ok(None)); } - Err(err) if latency_expired => return Poll::Ready(Err(err)), + // 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), } } @@ -1434,7 +1443,12 @@ impl Group { self.waiting = None; return Poll::Ready(Ok(None)); } - Err(err) if latency_expired => return Poll::Ready(Err(err)), + // 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), } } From 17cee348d37f66798ab05584937a1ac3a2f4d482 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 16:57:10 -0700 Subject: [PATCH 6/6] Release group-only demand at the reader's cap A reader capped by set_frames returned Ok(None) at the cap without dropping its replacement subscription, so a retained reader pinned upstream demand until it was dropped. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/resume.rs | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index c3d5420f86d..5c20397676a 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -1392,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))? { @@ -1429,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))? { @@ -5275,6 +5279,34 @@ mod test { 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