diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index d4f3216a6d..9ad9737a22 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -411,6 +411,20 @@ impl TrackState { next_sequence: u64, end_sequence: Option, ) -> Poll>> { + // An abort before the end settled cut the track off. One after it is a clean + // end (see `is_complete`): the cached groups were everything the end promised, + // so the cursor ends as `poll_recv_group` does, not with the abort. + if let Some(err) = &self.abort + && !self.settled + { + return Poll::Ready(Err(err.clone())); + } + // Nothing more can arrive once the last producer is gone: the track was sealed + // (dropped with its end declared) or aborted with that end settled. Either closed + // the channel, so parking is not an option: the consumer would be handed the + // abort, or `Dropped`, instead of the end it reached. + let closed = self.sealed || self.abort.is_some(); + // If the exclusive end is already at or below where we'd resume, no // group can ever satisfy this call until the cap rises. Pending (not // None) so the consumer is parked rather than told the stream is over. @@ -418,8 +432,8 @@ impl TrackState { if let Some(end) = end_sequence && end <= next_sequence { - if let Some(err) = &self.abort { - return Poll::Ready(Err(err.clone())); + if closed { + return Poll::Ready(Ok(None)); } return Poll::Pending; } @@ -440,16 +454,12 @@ impl TrackState { } // No in-range group is cached. Decide whether more could ever arrive. - if let Some(err) = &self.abort { - return Poll::Ready(Err(err.clone())); - } // `final_sequence` is one past the last possible sequence. If our // floor is already at/past it, nothing else can land in range. - // A sealed track produces nothing more either: the last producer - // dropped with the boundary already declared, so a gap below it is - // the end, not a wait. Cached in-range groups were returned above. - // This is the cursor the ordered and spliced readers use. - if self.sealed || self.final_sequence.is_some_and(|fin| next_sequence >= fin) { + // A closed track produces nothing more either, so a gap below its + // boundary is the end, not a wait. Cached in-range groups were + // returned above. This is the cursor the ordered and spliced readers use. + if closed || self.final_sequence.is_some_and(|fin| next_sequence >= fin) { return Poll::Ready(Ok(None)); } Poll::Pending @@ -1195,8 +1205,14 @@ fn commit_abort(mut state: kio::Mut<'_, TrackState>, err: Error) { // Decided before the cache goes: the groups below the end are the evidence. state.settled = state.is_settled(); state.abort = Some(err); - state.clear_cache(); - state.datagrams.clear(); + // A settled end is a clean end: the track holds every group it promised, so keep + // them for consumers still draining, as `finish()` does. Clearing here would abort + // the finished groups a slower reader has not pulled yet, and it would then see the + // track end short of them with no error. + if !state.settled { + state.clear_cache(); + state.datagrams.clear(); + } state.close(); } @@ -1506,6 +1522,11 @@ impl Producer { /// of the leftover cache. Child groups are independent: a consumer that already pulled /// a [`group::Consumer`] keeps its own handle and can finish reading it. /// + /// Unless the declared end had settled: the final sequence from + /// [`finish_at`](Self::finish_at) was reached and every group below it finished. The + /// track then already holds everything it promised, so it ends cleanly and keeps the + /// cached groups for consumers still draining, as [`finish`](Self::finish) does. + /// /// [`finish`](Self::finish) is deliberately not terminal: it declares the final /// sequence, and lower-numbered groups may still be written afterwards. pub fn abort(self, err: Error) -> Result<()> { @@ -4116,8 +4137,9 @@ impl Ordered { /// the cap rises or is removed. /// /// Returns `Poll::Ready(Ok(Some(group)))` when a group is available, - /// `Poll::Ready(Ok(None))` when the track is finished, - /// `Poll::Ready(Err(e))` when the track has been aborted, or + /// `Poll::Ready(Ok(None))` when the track is finished (including an abort after its + /// declared end settled, see [`Producer::abort`]), + /// `Poll::Ready(Err(e))` when the track has been aborted short of its end, or /// `Poll::Pending` when no group is available yet. pub fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll>> { self.inner.poll_next_group(waiter) @@ -6953,6 +6975,83 @@ mod test { assert!(matches!(res, Ok(None))); } + /// The settled end stands on the ordered cursor too: a reader that starts after the + /// abort gets every group below the end, then the clean end, not the abort. + #[tokio::test] + async fn abort_after_the_end_settles_ends_clean_for_an_ordered_reader() { + let mut producer = track_producer("test", None); + let mut consumer = producer.subscribe(None).ordered(); + + for sequence in 0..2 { + let group = producer.create_group(group::Info { sequence }).unwrap(); + group.finish().unwrap(); + } + producer.finish_at(2).unwrap(); + producer.abort(Error::Timeout).unwrap(); + + assert_eq!(drain_ordered(&mut consumer), [0, 1]); + let end = consumer + .next_group() + .now_or_never() + .expect("should not block") + .map(|group| group.map(|group| group.sequence)); + assert!(matches!(end, Ok(None)), "expected the clean end, got {end:?}"); + } + + /// The settled end stands on the ordered cursor even where it has nothing to read: + /// capped below the end, and with a gap below the cap. Parking is not an option on a + /// closed track (the channel would turn it into the abort), so these end clean. + #[tokio::test] + async fn abort_after_the_end_settles_ends_clean_below_the_cap() { + let mut producer = track_producer("test", None); + let mut consumer = producer.subscribe(None).ordered(); + for sequence in 0..2 { + producer + .create_group(group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + } + producer.finish_at(2).unwrap(); + producer.abort(Error::Timeout).unwrap(); + + consumer.set_groups(..1); + assert_eq!(drain_ordered(&mut consumer), [0]); + let end = consumer + .next_group() + .now_or_never() + .expect("should not block") + .map(|group| group.map(|group| group.sequence)); + assert!( + matches!(end, Ok(None)), + "capped at the end: expected the clean end, got {end:?}" + ); + + let mut producer = track_producer("test", None); + let mut consumer = producer.subscribe(None).ordered(); + for sequence in [0, 2] { + producer + .create_group(group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + } + producer.finish_at(3).unwrap(); + producer.abort(Error::Timeout).unwrap(); + + consumer.set_groups(..2); + assert_eq!(drain_ordered(&mut consumer), [0]); + let end = consumer + .next_group() + .now_or_never() + .expect("should not block") + .map(|group| group.map(|group| group.sequence)); + assert!( + matches!(end, Ok(None)), + "gap below the cap: expected the clean end, got {end:?}" + ); + } + #[tokio::test] async fn recv_group_finishes_without_waiting_for_gaps() { let producer = track_producer("test", None); diff --git a/rs/moq-net/tests/settled_end_survives_abort.rs b/rs/moq-net/tests/settled_end_survives_abort.rs new file mode 100644 index 0000000000..2ef3826438 --- /dev/null +++ b/rs/moq-net/tests/settled_end_survives_abort.rs @@ -0,0 +1,89 @@ +//! A track whose declared end has settled holds every group it promised: the end was +//! reached and each group below it finished. An abort that lands afterwards, such as +//! the session dying, ends the track cleanly but must not throw those groups away: a +//! consumer that has not read them yet still gets them, then the clean end. + +mod support; + +use moq_net::{Error, Hop, Timestamp}; + +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 +} + +const GROUPS: u64 = 2; + +/// Publish `GROUPS` finished groups, declare the end at `GROUPS`, then abort the track. +/// Returns the sequences a consumer that starts reading only now receives, in arrival +/// or in sequence order, and how the read ended. +async fn round(abort: bool, ordered: bool) -> (Vec, Option) { + let origin = produce_origin(1); + let broadcast = origin.create_broadcast("bcast").unwrap(); + let mut track = broadcast.create_track("video", None).unwrap(); + // From the first group with a replay window: a late reader is owed the whole track, + // not the live edge. + let subscription = moq_net::track::Subscription::default() + .with_start(moq_net::track::Position::group(0)) + .with_max_age(std::time::Duration::from_secs(30)); + let mut consumer = track.subscribe(subscription); + + for _ in 0..GROUPS { + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, b"frame".as_slice()).unwrap(); + group.finish().unwrap(); + } + track.finish_at(GROUPS).unwrap(); + // Every group below the end is finished: the end has settled and the track is + // complete. Nothing has been read yet. + if abort { + track.abort(Error::Session(moq_net::SessionError::Cancel)).unwrap(); + } + + let mut got = Vec::new(); + if ordered { + let mut consumer = consumer.ordered(); + loop { + match consumer.next_group().await { + Ok(Some(group)) => got.push(group.sequence), + Ok(None) => return (got, None), + Err(err) => return (got, Some(err)), + } + } + } + loop { + match consumer.recv_group().await { + Ok(Some(group)) => got.push(group.sequence), + Ok(None) => return (got, None), + Err(err) => return (got, Some(err)), + } + } +} + +fn assert_whole((got, err): (Vec, Option), what: &str) { + assert!( + err.is_none() && got == (0..GROUPS).collect::>(), + "{what}: got {got:?} of {GROUPS} groups, err={err:?} (every group had finished before \ + the end, so a late reader gets all of them and then the clean end)" + ); +} + +/// The groups a settled track holds outlive an abort: a late reader gets all of them. +#[tokio::test] +async fn an_abort_after_the_end_settled_keeps_the_groups_for_a_late_reader() { + assert_whole(round(true, false).await, "arrival order, aborted after the end settled"); +} + +/// The same in sequence order: the ordered cursor ends clean too, not with the abort. +#[tokio::test] +async fn an_abort_after_the_end_settled_keeps_the_groups_for_a_late_ordered_reader() { + assert_whole(round(true, true).await, "sequence order, aborted after the end settled"); +} + +/// Control: with no abort the same late reader gets every group, then the clean end. +#[tokio::test] +async fn a_settled_track_delivers_every_group_to_a_late_reader() { + assert_whole(round(false, false).await, "arrival order, no abort"); + assert_whole(round(false, true).await, "sequence order, no abort"); +}