diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 56fa1292aea..7ea12f74c96 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1887,12 +1887,14 @@ where match end { End::Fetch(request) => { - let fetch = self.clone().run_group_fetch( - broadcast_path.to_owned(), - track_name.clone(), - request, - timescale, - ); + let subscriber = self.clone(); + let broadcast = broadcast_path.to_owned(); + let name = track_name.clone(); + let fetch = async move { + if let Some(request) = subscriber.wait_group_head(request_id, request).await { + subscriber.run_group_fetch(broadcast, name, request, timescale).await; + } + }; group_fetches.push(fetch); } End::Unused => match track.abort_unused(Error::Cancel) { @@ -1927,6 +1929,7 @@ where { fetch_done = true; } + let _ = group_fetches.poll(waiter); poll_settled(&mut settle, waiter, &fill, count) }) .await; @@ -2613,7 +2616,7 @@ where /// /// Waiting is what keeps the two streams from interleaving into one producer. It ends /// with the subscription, so a publisher that promises a fill and never delivers one - /// costs this stream and nothing else. Meanwhile this stream stops holding the + /// costs this group and nothing else. Meanwhile this stream stops holding the /// subscription's end open: the fill's own stream holds it if its header arrived, and /// the grace gives up on it if not. async fn claim_fill( @@ -3000,6 +3003,40 @@ where Ok(true) } + /// Wait for the subscription's own joined head before FETCHing that group again. + /// Racing the head can cache the group first and refuse the live tail that ends it. + /// Older history still fetches immediately; a settled fill leaves any remaining miss + /// to a FETCH, while the cache answers a head that landed. + async fn wait_group_head(&self, subscribe_id: RequestId, mut request: group::Request) -> Option { + let needed = kio::wait(|waiter| { + if request.poll_resolved(waiter).is_ready() || request.poll_unused(waiter).is_ready() { + return Poll::Ready(false); + } + + let state = self.state.lock(); + let Some(held) = state.subscribes.get(&subscribe_id) else { + return Poll::Ready(true); + }; + if held.largest.is_none_or(|largest| largest.group != request.sequence()) { + return Poll::Ready(true); + } + // Register under the same lock as the read, so the head settling cannot miss + // waking this request. + match held.fill.poll_ref(waiter, |fill| match **fill { + Fill::Requested | Fill::Serving(_) | Fill::Active => Poll::Pending, + Fill::Ready { .. } | Fill::Done => Poll::Ready(()), + }) { + Poll::Pending => Poll::Pending, + Poll::Ready(_) => Poll::Ready(true), + } + }) + .await; + if !needed || request.poll_resolved(&kio::Waiter::noop()).is_ready() { + return None; + } + Some(request) + } + /// Fetch a group from the publisher to fill a cache miss, with a standalone (or /// draft-20 filtered) FETCH for this subscription's track. It asks from the frame the reader wants through the end /// of the group, so a publisher that evicted the prefix can still answer. @@ -7713,7 +7750,10 @@ mod joining_fetch_tests { impl JoinRun { async fn start(version: Version, start: Option, ok: Vec, fetch: Vec) -> Self { - let session = ScriptedSession::per_stream(vec![ok, fetch]); + Self::subscribe(version, start, ScriptedSession::per_stream(vec![ok, fetch])).await + } + + async fn subscribe(version: Version, start: Option, session: ScriptedSession) -> Self { let (tasks, _task_set) = crate::util::TaskSet::new(); let subscriber = Subscriber::new( crate::time::Clock::tokio(), @@ -7853,6 +7893,203 @@ mod joining_fetch_tests { } } + /// A reader missing the head of the group the subscription joined waits for the + /// subscription's own fill, which is bringing that head. A cache-miss FETCH racing it + /// could cache the group first, refusing the tail that ends it for live readers. Only a + /// fill that settled without the head leaves the miss to a FETCH. + #[tokio::test(start_paused = true)] + async fn a_miss_waits_for_the_subscriptions_own_fill() { + let version = Version::Draft22; + let largest = Some(ietf::Location { group: 4, object: 1 }); + let run = JoinRun::start(version, None, subscribe_ok(version, largest), fetch_ok(version)).await; + let fetches = || { + decode_messages(&run.session.log, version) + .iter() + .filter(|(id, _)| *id == ietf::Fetch::ID) + .count() + }; + assert!(run.fill_outstanding(), "the subscription asked for a fill"); + + let mut fetch = Box::pin(run._hold.1.fetch_group(4, None)); + assert!(futures::poll!(fetch.as_mut()).is_pending()); + settle().await; + assert_eq!(fetches(), 0, "the miss waits for the subscription's fill"); + + // The fill fails, so the head is the miss's to fetch after all. + let fill = run + .subscriber + .state + .lock() + .subscribes + .values() + .next() + .unwrap() + .fill + .clone(); + *fill.write().ok().unwrap() = Fill::Done; + settle().await; + assert_eq!(fetches(), 1, "a settled fill leaves the miss to a FETCH"); + } + + /// The head landing ends the wait, and the cache answers the miss: no FETCH duplicates + /// the group the fill wrote. + #[tokio::test(start_paused = true)] + async fn a_miss_is_answered_by_the_head_its_fill_lands() { + let version = Version::Draft22; + let largest = Some(ietf::Location { group: 4, object: 1 }); + let run = JoinRun::start(version, None, subscribe_ok(version, largest), fetch_ok(version)).await; + let fetches = || { + decode_messages(&run.session.log, version) + .iter() + .filter(|(id, _)| *id == ietf::Fetch::ID) + .count() + }; + + let mut fetch = Box::pin(run._hold.1.fetch_group(4, None)); + assert!(futures::poll!(fetch.as_mut()).is_pending()); + settle().await; + assert_eq!(fetches(), 0, "the miss waits for the subscription's fill"); + + // The fill writes the head, as `run_fill` does. + let (mut track, fill) = { + let state = run.subscriber.state.lock(); + let held = state.subscribes.values().next().unwrap(); + (held.producer.clone().unwrap(), held.fill.clone()) + }; + let mut head = None; + open_fill_group(&mut track, &mut head, 4).unwrap(); + let (sequence, next, producer) = head.unwrap(); + *fill.write().ok().unwrap() = Fill::Ready { + sequence, + next, + producer: producer.into_inner(), + }; + settle().await; + assert_eq!(fetches(), 0, "the cache answers the miss"); + let Poll::Ready(Ok(group)) = futures::poll!(fetch.as_mut()) else { + panic!("the miss must resolve from the cache"); + }; + assert_eq!(group.sequence, 4); + } + + /// Only the joined group waits for the subscription's fill. A head that never arrives + /// leaves its fill outstanding for as long as the subscription lasts, which must not hold + /// back older groups it brings nothing for. + #[tokio::test(start_paused = true)] + async fn a_miss_for_older_history_skips_the_fill_wait() { + let version = Version::Draft22; + let largest = Some(ietf::Location { group: 4, object: 1 }); + let run = JoinRun::start(version, None, subscribe_ok(version, largest), fetch_ok(version)).await; + let fetches = || { + decode_messages(&run.session.log, version) + .iter() + .filter(|(id, _)| *id == ietf::Fetch::ID) + .count() + }; + assert!(run.fill_outstanding(), "the subscription asked for a fill"); + + let mut fetch = Box::pin(run._hold.1.fetch_group(3, None)); + assert!(futures::poll!(fetch.as_mut()).is_pending()); + settle().await; + assert_eq!(fetches(), 1, "group 3 is fetched without waiting for group 4's head"); + } + + /// A failed head still permits a cache-miss FETCH while PUBLISH_DONE waits for its + /// reordered data streams, rather than leaving that request unpolled until teardown. + #[tokio::test(start_paused = true)] + async fn a_settled_head_wait_fetches_during_publish_done_grace() { + let version = Version::Draft22; + let largest = Some(ietf::Location { group: 4, object: 1 }); + let run = JoinRun::subscribe(version, None, ScriptedSession::new(subscribe_ok(version, largest))).await; + let fetches = || { + decode_messages(&run.session.log, version) + .iter() + .filter(|(id, _)| *id == ietf::Fetch::ID) + .count() + }; + let _fetch = run._hold.1.fetch_group(4, None); + settle().await; + assert_eq!(fetches(), 0, "the miss waits for the subscription's fill"); + let fill = run + .subscriber + .state + .lock() + .subscribes + .values() + .next() + .unwrap() + .fill + .clone(); + + run.session.push(&message_bytes( + ietf::PublishDone::ID, + &ietf::PublishDone { + request_id: None, + status_code: ietf::PublishDoneStatus::TrackEnded.code(version), + stream_count: 0, + reason_phrase: "done".into(), + }, + version, + )); + // The scripted reader needs an explicit wake after push. Starting the head + // wakes its parked miss, so the driver reads PUBLISH_DONE and enters the grace. + *fill.write().ok().unwrap() = Fill::Active; + settle().await; + assert!(!run.serving.is_finished()); + + *fill.write().ok().unwrap() = Fill::Done; + settle().await; + assert_eq!(fetches(), 1, "a settled head permits a FETCH during the grace"); + } + + /// Withdrawing the last reader releases a pending head wait without a FETCH. + #[tokio::test(start_paused = true)] + async fn a_head_wait_ends_when_its_reader_leaves() { + let version = Version::Draft22; + let largest = Some(ietf::Location { group: 4, object: 1 }); + let run = JoinRun::start(version, None, subscribe_ok(version, largest), fetch_ok(version)).await; + let (id, track) = { + let state = run.subscriber.state.lock(); + let (id, held) = state.subscribes.iter().next().unwrap(); + (*id, held.producer.clone().unwrap()) + }; + let dynamic = track.dynamic(); + let fetch = run._hold.1.fetch_group(4, None); + let request = dynamic.requested_group().await.unwrap(); + let mut waiting = Box::pin(run.subscriber.wait_group_head(id, request)); + assert!(futures::poll!(waiting.as_mut()).is_pending()); + + drop(fetch); + assert!(matches!(futures::poll!(waiting.as_mut()), Poll::Ready(None))); + assert!(dynamic.poll_requested_group(&kio::Waiter::noop()).is_pending()); + assert!(run.fill_outstanding(), "cancelling a miss does not cancel the join"); + } + + /// Aborting the track releases a pending head wait with the track's error. + #[tokio::test(start_paused = true)] + async fn a_head_wait_ends_when_its_track_aborts() { + let version = Version::Draft22; + let largest = Some(ietf::Location { group: 4, object: 1 }); + let run = JoinRun::start(version, None, subscribe_ok(version, largest), fetch_ok(version)).await; + let (id, track) = { + let state = run.subscriber.state.lock(); + let (id, held) = state.subscribes.iter().next().unwrap(); + (*id, held.producer.clone().unwrap()) + }; + let dynamic = track.dynamic(); + let mut fetch = Box::pin(run._hold.1.fetch_group(4, None)); + let request = dynamic.requested_group().await.unwrap(); + let mut waiting = Box::pin(run.subscriber.wait_group_head(id, request)); + assert!(futures::poll!(waiting.as_mut()).is_pending()); + + track.abort(Error::Cancel).unwrap(); + assert!(matches!(futures::poll!(waiting.as_mut()), Poll::Ready(None))); + assert!(matches!( + futures::poll!(fetch.as_mut()), + Poll::Ready(Err(Error::Cancel)) + )); + } + /// A frame-level start is refused before SUBSCRIBE is written, rather than rounded down. #[tokio::test(start_paused = true)] async fn a_frame_level_start_never_reaches_the_wire() { diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index e1c20656ef8..3dbacb22fe4 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -3210,6 +3210,18 @@ impl group::Request { } } + /// Ready once the cache or a terminal track state answers this request. + pub(crate) fn poll_resolved(&mut self, waiter: &kio::Waiter) -> Poll<()> { + let _ = ready!(self.state.poll_ref(waiter, |state| { + state.poll_fetch_cached(self.sequence, self.frame_start) + })); + // Readers check the track before the result channel, including on its close. + // They already have an answer, so retire the attempt without rejecting them. + self.done = true; + self.remove(); + Poll::Ready(()) + } + /// Poll for the request becoming unused (every waiting [`Consumer::fetch_group`] /// dropped), so a handler can stop serving and drop the request. ///