diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 94aef76377..436e273ddf 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -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..322fab5bf4 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -1532,6 +1532,9 @@ struct SegmentSub { id: u64, start: Option, end: Option, + /// 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, sub: SubState, @@ -1583,7 +1586,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 @@ -1794,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(); @@ -1829,6 +1843,8 @@ impl Subscriber { id: segment.id, start: segment.start, end: segment.end, + // 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, @@ -3250,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] diff --git a/rs/moq-relay/tests/goaway_cluster.rs b/rs/moq-relay/tests/goaway_cluster.rs index d9cb9511c6..874aa53d9a 100644 --- a/rs/moq-relay/tests/goaway_cluster.rs +++ b/rs/moq-relay/tests/goaway_cluster.rs @@ -118,6 +118,105 @@ 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()))); + + // 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}/"))) + .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());