From 44d1be6fe7bd3b9c93084bd9cd49476e1a188896 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 17:38:25 -0700 Subject: [PATCH 1/4] fix(net): a mid-group takeover retires its copy in place A takeover that retired a copy mid-group parked its warm cache, which drops every segment and splices the cache as a fresh one. A reader already mid-group got a second cursor over the cache and was handed the open group again. Swap the segment's track for its cache instead: readers on it keep their cursor, newcomers read the cache, and a cursor that died with its source picks the cache up past what its reader already has. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/origin.rs | 47 +++++++++++++++++++++++++++----- rs/moq-net/src/model/resume.rs | 49 +++++++++++++++++++++++++++++++++- 2 files changed, 89 insertions(+), 7 deletions(-) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 94aef76377..b56cca0803 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2459,10 +2459,10 @@ impl TrackIo { self.dead = Some(copy); } - /// Retire `copy` as a takeover replaces it. Mid-group, park its delivery warm, open - /// group included, so a reader arriving later still gets the group's head and the - /// takeover continues it. Between groups its own cache keeps everything finished, so - /// there is nothing to do. + /// Retire `copy` as a takeover replaces it. Mid-group, its segment is served from a + /// warm cache of its delivery instead, open group included, so a reader arriving + /// later still gets the group's head and the takeover continues it. Between groups + /// its own cache keeps everything finished, so there is nothing to do. fn retire(&mut self, copy: &track::Consumer) -> Result<(), Error> { let Some(held) = self.held.take() else { return Ok(()); @@ -2474,7 +2474,7 @@ impl TrackIo { return Ok(()); }; self.head = None; - self.resume.park(&warm.track)?; + self.resume.retire(&warm.track)?; self.warm = Some(warm); Ok(()) } @@ -6080,7 +6080,8 @@ mod tests { /// next route only for the frames past the break. The front keeps the head it /// spliced, so a reader arriving afterwards still gets the whole group: whether the /// old route dies before the new one takes over (a reconnect) or after (a cheaper - /// route preempting it). + /// route preempting it). A reader joining mid-outage gets it too, and the reader + /// already mid-group is not handed it a second time. #[tokio::test] async fn a_takeover_keeps_the_open_group_head_for_later_readers() { tokio::time::pause(); @@ -6124,8 +6125,18 @@ mod tests { // Either the first route's session ends and reconnects, announcing the same route // afresh while the old one is still held, or a cheaper route preempts the live one. let mut first = Some((group, source, upstream, dynamic)); + let mut joining = None; if dies_first { drop(first.take()); + // The reader sees its route die and stalls. Polled in place: yielding would let + // the front give up on a path with no route left before the replacement lands. + assert!(futures::FutureExt::now_or_never(subscription.recv_group()).is_none()); + let track = resolved.track("log").unwrap(); + joining = Some(tokio::spawn(async move { + let mut joined = track.subscribe(None).await.unwrap(); + let group = joined.recv_group().await.unwrap().expect("track ended"); + (joined, group) + })); } let cost = if dies_first { 5 } else { 0 }; let replacement_server = producer @@ -6171,13 +6182,37 @@ mod tests { for expect in [b"a", b"b", b"c"] { assert_eq!(read(&mut late).await, expect, "dies_first={dies_first}"); } + let mut joined = None; + if let Some(joining) = joining { + let (subscription, mut group) = tokio::time::timeout(Duration::from_secs(1), joining) + .await + .expect("the reader joining mid-outage never got the group") + .unwrap(); + assert_eq!(group.sequence, 0); + for expect in [b"a", b"b", b"c"] { + assert_eq!(read(&mut group).await, expect); + } + joined = Some((subscription, group)); + } + + // The reader that was mid-group already has it. + assert!( + tokio::time::timeout(Duration::from_secs(1), subscription.recv_group()) + .await + .is_err(), + "dies_first={dies_first}: the group was handed out twice" + ); // The continuation still flows to both. next_group.write_frame(crate::Timestamp::ZERO, b"d".as_ref()).unwrap(); assert_eq!(read(&mut reading).await, b"d"); assert_eq!(read(&mut late).await, b"d"); + if let Some((_, group)) = &mut joined { + assert_eq!(read(group).await, b"d"); + } next_group.finish().unwrap(); drop(( + joined, reading, late, subscription, diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 5c20397676..00f5de30e0 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -50,6 +50,8 @@ struct Segment { /// A parked track's warm cache (see [`Producer::park`]), which live readers hold /// until the next segment's copy resolves its start. warm: bool, + /// `track` is a retired source's cache (see [`Producer::retire`]). + cache: bool, } impl Segment { @@ -328,6 +330,7 @@ impl ResumeState { track, ask: start, warm: false, + cache: false, }); self.epoch += 1; self.prune(); @@ -500,6 +503,27 @@ impl Producer { Ok(()) } + /// Serve the newest segment's range from `cache`, a copy of what its source delivered, + /// as that source retires. + /// + /// The segment keeps its place, so a reader already on it keeps its cursor rather than + /// being handed the cached groups a second time. A reader yet to arrive, or whose + /// cursor died with the source, reads the cache instead. + pub(crate) fn retire(&mut self, cache: impl super::origin_impl::Consume) -> Result<()> { + let track = cache.consume(); + let mut state = self.state.write().map_err(|_| Error::Dropped)?; + if state.finished || state.abort.is_some() { + return Err(Error::Closed); + } + let Some(last) = state.segments.last_mut() else { + return Err(Error::Dropped); + }; + last.track = track; + last.cache = true; + state.epoch += 1; + Ok(()) + } + /// Whether any segment is spliced in, and so whether there is anything for /// [`Self::release`] to drop. /// @@ -1532,6 +1556,12 @@ struct SegmentSub { id: u64, start: Option, end: Option, + /// Whether a reconcile already saw the segment's source retire for its cache; see + /// [`Producer::retire`]. + cache: bool, + /// Groups below this were handed out before the cursor subscribed: a cursor + /// re-subscribed onto a retired source's cache must not surface them again. + floor: u64, /// Where the source is asked to start; see [`Segment::ask`]. ask: Option, sub: SubState, @@ -1583,7 +1613,7 @@ impl SegmentSub { /// The first group this segment can serve, for the underlying read cursor. fn first_group(&self) -> u64 { - self.start.map_or(0, |start| start.group) + self.start.map_or(0, |start| start.group).max(self.floor) } /// The exclusive group cap this segment can serve, for the underlying read @@ -1804,6 +1834,21 @@ impl Subscriber { if let Some(warm) = &mut existing.warm { warm.next = next; } + if segment.cache && !existing.cache { + existing.cache = true; + // The source retired for its cache. A live cursor keeps reading the + // source, which still holds whatever it owes; one that died with it + // picks the cache up past what this reader already has. + if matches!(existing.sub, SubState::Done(Err(_))) { + existing.sub = SubState::Pending(segment.track.subscribe(slice( + &self.last_prefs, + segment.ask, + segment.end, + ))); + existing.terminal = None; + existing.floor = self.next_sequence; + } + } if existing.end != segment.end { // The boundary bounds the drift anchor as well as the demand; the // anchor follows in `refresh_anchor`. @@ -1829,6 +1874,8 @@ impl Subscriber { id: segment.id, start: segment.start, end: segment.end, + cache: segment.cache, + floor: 0, ask: segment.ask, sub: SubState::Pending(sub), terminal: None, From 3132677dad7c1041af9d61809fa61c83558b42dd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 20:21:09 -0700 Subject: [PATCH 2/4] test(relay): a GOAWAY mid-group never hands the split group back Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/tests/goaway_cluster.rs | 96 ++++++++++++++++++++++++++++ 1 file changed, 96 insertions(+) diff --git a/rs/moq-relay/tests/goaway_cluster.rs b/rs/moq-relay/tests/goaway_cluster.rs index d9cb9511c6..6eec66e154 100644 --- a/rs/moq-relay/tests/goaway_cluster.rs +++ b/rs/moq-relay/tests/goaway_cluster.rs @@ -118,6 +118,102 @@ async fn drain_session_with_zero_timeout_closes_at_once_inner() { let _ = within("client observes the close", client_connection.closed()).await; } +#[test] +fn cluster_continues_a_group_split_by_goaway() { + run_cluster_test(cluster_continues_a_group_split_by_goaway_inner()); +} + +/// A GOAWAY that lands mid-group hands the rest of the group to the reader already +/// holding it, and the next group after it, never the split group a second time. +async fn cluster_continues_a_group_split_by_goaway_inner() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + + let upstream_origin = moq_tokio::origin::spawn(); + let broadcast = upstream_origin.create_broadcast("cam").expect("create broadcast"); + broadcast.announce(Default::default()).expect("announce"); + let track = broadcast.create_track("video", None).expect("create track"); + + let (port_a, mut accepted_a, _handle_a) = spawn_upstream(upstream_origin.clone()).await; + let (port_b, mut accepted_b, _handle_b) = spawn_upstream(upstream_origin.clone()).await; + + let mut client_config = moq_tokio::connect::Config::default(); + client_config.tls.insecure = Some(true); + client_config.goaway.handover = Duration::from_secs(2); + let client = client_config.init(Default::default()).expect("client init"); + + let mut cluster_config = cluster::Config::default(); + cluster_config.connect = vec![Peer::new(format!("tcp://127.0.0.1:{port_a}/"))]; + let cluster = cluster::Cluster::new(cluster::Options::new(cluster_config)) + .expect("cluster init") + .with_client(client); + let started = cluster.clone().start().await.expect("cluster start"); + let cluster_run = tokio::spawn(started.run()); + + let session_a = within("A accepts", accepted_a.recv()).await.expect("A accepts"); + let consumer = cluster.origin.consume(); + within("routed via A", consumer.routed("cam")).await.expect("routed"); + let bc = consumer.request_broadcast("cam").await.expect("broadcast resolves"); + let mut sub = within("subscribe", bc.track("video").expect("track").subscribe(None)) + .await + .expect("subscribe"); + + let write = |group: &mut moq_net::group::Producer, payload: &'static str| { + group + .write_frame(moq_net::Timestamp::ZERO, payload.as_bytes()) + .expect("write frame"); + }; + // A group's sequence and its next frame, or `None` at its end. + async fn read(group: &mut moq_net::group::Consumer) -> (u64, Option) { + let frame = within("read frame", group.read_frame()).await.expect("read"); + (group.sequence, frame.map(|frame| frame.payload)) + } + + // The first group is open, its first frame read, when A redirects to B. + let mut split = track.append_group().expect("append group"); + write(&mut split, "a"); + let mut reading = within("recv the split group", sub.recv_group()) + .await + .expect("recv") + .expect("track ended"); + assert_eq!(read(&mut reading).await, (0, Some("a".into()))); + + let mut announced = consumer.announced(); + session_a + .drain() + .send(moq_net::goaway::Goaway::redirect(format!("tcp://127.0.0.1:{port_b}/"))) + .expect("send goaway"); + let _session_b = within("B accepts", accepted_b.recv()).await.expect("B accepts"); + within("routed via B", async { + loop { + let update = announced.next().await.expect("announce update"); + if update.kind.is_active() && update.route.cost != moq_net::origin::Cost::DRAIN { + break; + } + } + }) + .await; + + write(&mut split, "b"); + split.finish().expect("finish"); + assert_eq!(read(&mut reading).await, (0, Some("b".into()))); + assert_eq!(read(&mut reading).await, (0, None)); + + let mut after = track.append_group().expect("append group"); + write(&mut after, "c"); + after.finish().expect("finish"); + let mut next = within("recv the next group", sub.recv_group()) + .await + .expect("recv") + .expect("track ended"); + assert_eq!( + read(&mut next).await, + (1, Some("c".into())), + "the split group came back" + ); + + cluster_run.abort(); +} + #[test] fn cluster_reconnects_on_empty_uri_goaway() { run_cluster_test(cluster_reconnects_on_empty_uri_goaway_inner()); From f52a63f831118a99193ca082f91eff2136bc867a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 20:45:28 -0700 Subject: [PATCH 3/4] fix(net): a subscriber hands its cursors off to a park's cache Retiring in place kept the continuation's mid-group start, so after enough takeovers of one open group no segment could hand its head to a newcomer. Keep the park, which leaves one unbounded cache segment, and make the subscriber treat it as a handoff instead: drop the cursors it replaced and start the cache's cursor past what the reader was already handed. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/origin.rs | 10 +-- rs/moq-net/src/model/resume.rs | 140 ++++++++++++++++++++++----------- 2 files changed, 99 insertions(+), 51 deletions(-) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index b56cca0803..436e273ddf 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2459,10 +2459,10 @@ impl TrackIo { self.dead = Some(copy); } - /// Retire `copy` as a takeover replaces it. Mid-group, its segment is served from a - /// warm cache of its delivery instead, open group included, so a reader arriving - /// later still gets the group's head and the takeover continues it. Between groups - /// its own cache keeps everything finished, so there is nothing to do. + /// Retire `copy` as a takeover replaces it. Mid-group, park its delivery warm, open + /// group included, so a reader arriving later still gets the group's head and the + /// takeover continues it. Between groups its own cache keeps everything finished, so + /// there is nothing to do. fn retire(&mut self, copy: &track::Consumer) -> Result<(), Error> { let Some(held) = self.held.take() else { return Ok(()); @@ -2474,7 +2474,7 @@ impl TrackIo { return Ok(()); }; self.head = None; - self.resume.retire(&warm.track)?; + self.resume.park(&warm.track)?; self.warm = Some(warm); Ok(()) } diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 00f5de30e0..322fab5bf4 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -50,8 +50,6 @@ struct Segment { /// A parked track's warm cache (see [`Producer::park`]), which live readers hold /// until the next segment's copy resolves its start. warm: bool, - /// `track` is a retired source's cache (see [`Producer::retire`]). - cache: bool, } impl Segment { @@ -330,7 +328,6 @@ impl ResumeState { track, ask: start, warm: false, - cache: false, }); self.epoch += 1; self.prune(); @@ -503,27 +500,6 @@ impl Producer { Ok(()) } - /// Serve the newest segment's range from `cache`, a copy of what its source delivered, - /// as that source retires. - /// - /// The segment keeps its place, so a reader already on it keeps its cursor rather than - /// being handed the cached groups a second time. A reader yet to arrive, or whose - /// cursor died with the source, reads the cache instead. - pub(crate) fn retire(&mut self, cache: impl super::origin_impl::Consume) -> Result<()> { - let track = cache.consume(); - let mut state = self.state.write().map_err(|_| Error::Dropped)?; - if state.finished || state.abort.is_some() { - return Err(Error::Closed); - } - let Some(last) = state.segments.last_mut() else { - return Err(Error::Dropped); - }; - last.track = track; - last.cache = true; - state.epoch += 1; - Ok(()) - } - /// Whether any segment is spliced in, and so whether there is anything for /// [`Self::release`] to drop. /// @@ -1556,11 +1532,8 @@ struct SegmentSub { id: u64, start: Option, end: Option, - /// Whether a reconcile already saw the segment's source retire for its cache; see - /// [`Producer::retire`]. - cache: bool, - /// Groups below this were handed out before the cursor subscribed: a cursor - /// re-subscribed onto a retired source's cache must not surface them again. + /// Groups below this were handed out before the cursor subscribed: a warm cache + /// taking over a reader's cursors must not surface them again. floor: u64, /// Where the source is asked to start; see [`Segment::ask`]. ask: Option, @@ -1824,6 +1797,17 @@ impl Subscriber { for s in &mut self.segments { s.pruned = !segments.iter().any(|n| n.id == s.id); } + // A park replaces every segment with a cache of what they delivered (see + // [`Producer::park`]). Cursors still draining the replaced segments would hand + // that content out a second time, so the cache takes over from them, past what + // this reader was already handed. Groups they parked at the cap are in the + // cache too, and park again from there. + let parked = segments + .iter() + .any(|n| n.warm && !self.segments.iter().any(|s| s.id == n.id)); + if parked { + self.segments.retain(|s| !s.pruned); + } self.segments.retain(|s| !s.retired()); let nexts: Vec<_> = segments.iter().skip(1).map(|next| Some(next.track.clone())).collect(); @@ -1834,21 +1818,6 @@ impl Subscriber { if let Some(warm) = &mut existing.warm { warm.next = next; } - if segment.cache && !existing.cache { - existing.cache = true; - // The source retired for its cache. A live cursor keeps reading the - // source, which still holds whatever it owes; one that died with it - // picks the cache up past what this reader already has. - if matches!(existing.sub, SubState::Done(Err(_))) { - existing.sub = SubState::Pending(segment.track.subscribe(slice( - &self.last_prefs, - segment.ask, - segment.end, - ))); - existing.terminal = None; - existing.floor = self.next_sequence; - } - } if existing.end != segment.end { // The boundary bounds the drift anchor as well as the demand; the // anchor follows in `refresh_anchor`. @@ -1874,8 +1843,8 @@ impl Subscriber { id: segment.id, start: segment.start, end: segment.end, - cache: segment.cache, - floor: 0, + // A cache taking over (see above) starts past what this reader has. + floor: if segment.warm { self.next_sequence } else { 0 }, ask: segment.ask, sub: SubState::Pending(sub), terminal: None, @@ -3297,6 +3266,85 @@ mod test { assert_eq!(recv(&mut sub), 1); } + /// A park that replaces a reader's segments with a cache of what they delivered + /// hands that reader nothing twice, however many times one open group (a catalog + /// taking deltas) changes route. A newcomer still gets the whole group, head first, + /// after the takeovers have pruned every segment that started at its head. + #[tokio::test] + async fn repeated_parks_mid_group_hand_the_head_out_once() { + let mut producer = Producer::new(); + let (track, consumer) = track_pair("a"); + producer.takeover(&consumer).unwrap(); + let mut sub = producer.consume().subscribe(None); + let mut group = track.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(Timestamp::ZERO, b"0".to_vec()).unwrap(); + let mut reading = sub.recv_group().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(read(&mut reading), b"0"); + + let mut routes = vec![(track, consumer, group)]; + let mut heads = Vec::new(); + let rounds = MAX_SEGMENTS as u64 + 1; + for frame in 1..=rounds { + // The cache holds the whole group so far, the way the origin rebuilds it. + let (cache, cache_consumer) = track_pair("cache"); + let mut whole = cache.create_group(group::Info { sequence: 0 }).unwrap(); + for f in 0..frame { + whole.write_frame(Timestamp::ZERO, f.to_string().into_bytes()).unwrap(); + } + producer.park(&cache_consumer).unwrap(); + + let (next, next_consumer) = track_pair("next"); + producer.takeover(&next_consumer).unwrap(); + let mut continuation = next.create_group(group::Info { sequence: 0 }).unwrap(); + continuation.start_at(frame).unwrap(); + continuation + .write_frame(Timestamp::ZERO, frame.to_string().into_bytes()) + .unwrap(); + + assert_eq!(read(&mut reading), frame.to_string().as_bytes()); + recv_pending(&mut sub); + routes.push((next, next_consumer, continuation)); + // The origin finishes a cache once the takeover lands, but holds its open + // group: dropping that would clear its frames. + cache.finish().unwrap(); + heads.push(whole); + } + + let mut late = producer.consume().subscribe(None); + let mut group = late.recv_group().now_or_never().unwrap().unwrap().unwrap(); + for frame in 0..=rounds { + assert_eq!(read(&mut group), frame.to_string().as_bytes()); + } + } + + /// A group held back at the reader's cap when a park replaces its segments comes + /// from the cache once the cap rises, exactly once. + #[tokio::test] + async fn a_park_reoffers_a_group_held_at_the_cap_once() { + let (mut track_a, consumer_a) = track_pair("a"); + let mut producer = Producer::new(); + producer.takeover(&consumer_a).unwrap(); + let mut sub = producer.consume().subscribe(replay()); + sub.end_at(..1); + write_group(&mut track_a, 0, "0"); + write_group(&mut track_a, 1, "1"); + assert_eq!(recv(&mut sub), 0); + recv_pending(&mut sub); + + let (mut cache, cache_consumer) = track_pair("cache"); + write_group(&mut cache, 0, "0"); + write_group(&mut cache, 1, "1"); + producer.park(&cache_consumer).unwrap(); + let (mut track_b, consumer_b) = track_pair("b"); + producer.takeover(&consumer_b).unwrap(); + write_group(&mut track_b, 2, "2"); + + sub.end_at(..3); + assert_eq!(recv(&mut sub), 1); + assert_eq!(recv(&mut sub), 2); + recv_pending(&mut sub); + } + /// A parked beyond-cap group must not block in-range groups that arrive /// behind it: a relay can ingest a burst micro-reordered (newest first). #[tokio::test] From 1b4641e11d814d8ef1a9334743fbbcbf6de441a9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 20:50:58 -0700 Subject: [PATCH 4/4] test(relay): gate on B's route, not A's replay Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/tests/goaway_cluster.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/rs/moq-relay/tests/goaway_cluster.rs b/rs/moq-relay/tests/goaway_cluster.rs index 6eec66e154..874aa53d9a 100644 --- a/rs/moq-relay/tests/goaway_cluster.rs +++ b/rs/moq-relay/tests/goaway_cluster.rs @@ -177,7 +177,10 @@ async fn cluster_continues_a_group_split_by_goaway_inner() { .expect("track ended"); assert_eq!(read(&mut reading).await, (0, Some("a".into()))); + // The cursor replays A's route first. After the GOAWAY, A's re-prices to DRAIN, so + // the next route at any other cost is B's. let mut announced = consumer.announced(); + within("A's route", announced.next()).await.expect("announce update"); session_a .drain() .send(moq_net::goaway::Goaway::redirect(format!("tcp://127.0.0.1:{port_b}/")))