From e038ee1a5ce40113b4589cb690fbe2125209eecf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 16:51:57 -0700 Subject: [PATCH 1/7] quest: claim fetch-abandonment Co-Authored-By: Claude Opus 5.5 From 423a6296c4410b0542cfc1598d77c4e885644c97 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 17:16:22 -0700 Subject: [PATCH 2/7] fix(net): cancel an abandoned lite FETCH upstream The Rust lite subscriber held its upstream FETCH until the publisher answered, even after every downstream fetch_group gave up. Watch the group request's demand for the whole wait and reset the FETCH stream once it leaves, so the release crosses every relay hop. Adds group::Request::poll_unused, which also withdraws the abandoned attempt so a later fetch queues a fresh one. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/fetch-abandonment.md | 29 ------------- rs/moq-net/src/lite/subscriber.rs | 24 +++++++++++ rs/moq-net/src/model/track.rs | 52 ++++++++++++++++++++++ rs/moq-net/tests/fetch_cancel.rs | 72 +++++++++++++++++++++++++++++++ 5 files changed, 148 insertions(+), 30 deletions(-) delete mode 100644 quest/m1/fetch-abandonment.md create mode 100644 rs/moq-net/tests/fetch_cancel.rs diff --git a/quest/m1/README.md b/quest/m1/README.md index 928f1394a2..21130343c5 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -48,7 +48,6 @@ QUIC studies there on that rule. - [Reader end parity](/quest/m1/reader-end-parity.md) - JS readers see a track's end once the newest group reaches the declared end, as Rust readers do - [JS cache window](/quest/m1/js-prune-cache-window.md) - js/net ages idle groups on a cache window and keeps max_age as media-time staleness, as Rust does - [Cross-relay bursts](/quest/m1/cross-relay-bursts.md) - bursty small-group tracks cross two relays without lost groups, unanswered FETCHes, or stalls -- [FETCH abandonment](/quest/m1/fetch-abandonment.md) - Rust and JS lite subscribers close an upstream FETCH once demand leaves, so an abandoned FETCH is released across relays - [Watch and publish under CSP](/quest/m1/csp-assets.md) - blob workers stay the default; strict-CSP apps host the files and set a base URL - [More tests under load](/quest/m1/test-flakes-2/README.md) - the second round of load-only failures, one quest per flake, fixed at the cause - [A week of nightly interop after #4529](/quest/m1/interop-week.md) - the nightly traces the stall diagnosis reads diff --git a/quest/m1/fetch-abandonment.md b/quest/m1/fetch-abandonment.md deleted file mode 100644 index 107ac99d70..0000000000 --- a/quest/m1/fetch-abandonment.md +++ /dev/null @@ -1,29 +0,0 @@ -# [S] An abandoned FETCH is released along the mesh - -## Goal - -When nobody wants a FETCH's group any more, the lite subscriber in Rust and in -`@moq/net` closes its upstream FETCH stream, so a relay stops pinning the -request and the publisher behind it stops serving. With -[#4531](https://github.com/moq-dev/moq/pull/4531)'s publisher side, an -abandoned FETCH is then released across every hop. - -## Plan - -#4531 made the lite publisher stop serving a FETCH once the requester FINs or -resets, but the Rust subscriber's `FetchServeRun` never watches demand, so a -relay holds its upstream FETCH until it is answered. Watch demand for the -whole wait and close the stream when it leaves, the same way the subscriber -already drops an unwanted SUBSCRIBE. Check `@moq/net`'s subscriber for the same -gap; #4531 found it FINs an abandoned FETCH, so confirm it does so on demand -loss and not only on close. - -Decided with the maintainer: both subscribers, proven end to end. A relay-chain -test (publisher, relay, subscriber) abandons a FETCH mid-wait and asserts the -upstream publisher's serve ends, which #4531 could only test in-crate. - -Public API: none. Wire: none. - -## Related - -- [Cross-relay bursts](/quest/m1/cross-relay-bursts.md) - unanswered FETCHes across relays diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 1af5cf08a7..f327d13ae4 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -4139,6 +4139,16 @@ enum FetchRunState { Done, } +impl FetchRunState { + /// The downstream request, while it waits on the publisher's answer. + fn request(&self) -> Option<&group::Request> { + match self { + Self::Open { request } | Self::Send { request, .. } | Self::Answer { request, .. } => request.as_ref(), + Self::Ingest { .. } | Self::Done => None, + } + } +} + impl FetchServeRun { fn new(serve: TrackServe, request: group::Request, timescale: Option) -> Self { let session = serve.subscriber.session.clone(); @@ -4157,6 +4167,20 @@ impl kio::Task for FetchServeRun { fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { let mut cx = std::task::Context::from_waker(waiter.waker()); loop { + // Until answered, a fetch nobody waits on any more is cancelled upstream, so + // the publisher stops serving it (and a relay there releases its own FETCH). + if let Some(request) = self.state.request() + && request.poll_unused(waiter).is_ready() + { + tracing::debug!(track = %self.serve.name, group = self.group, "fetch abandoned"); + if let FetchRunState::Send { stream, .. } | FetchRunState::Answer { stream, .. } = + std::mem::replace(&mut self.state, FetchRunState::Done) + { + stream.writer.abort(&Error::Cancel); + } + return Poll::Ready(()); + } + match &mut self.state { FetchRunState::Open { request } => { tracing::info!(broadcast = %self.serve.subscriber.log_path(&self.serve.path), track = %self.serve.name, group = self.group, "fetch started"); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index ee4d90299b..8c6182808d 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -3125,6 +3125,25 @@ impl group::Request { } } + /// Poll for the request becoming unused (every waiting [`Consumer::fetch_group`] + /// dropped), so a handler can stop serving and drop the request. + /// + /// Once ready, a later fetch of the group starts a fresh request rather than + /// joining this abandoned one. + pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> { + loop { + ready!(self.result.poll_unused(waiter)); + // Fetches join under the fetch lock, so re-checking under it makes the + // withdrawal atomic with a join racing the last one leaving. + let mut fetch = self.fetch.lock(); + if self.result.is_used() { + continue; + } + fetch.remove_if(&self.sequence, |pending| pending.result.same_channel(&self.result)); + return Poll::Ready(()); + } + } + /// Remove this attempt from the fetch state, unless a newer attempt for the same /// sequence has already replaced it. fn remove(&self) { @@ -8009,6 +8028,39 @@ mod test { assert!(matches!(pending.await, Err(Error::Dropped))); } + /// A request stays wanted while any joined fetch waits, and once the last leaves it + /// is withdrawn: a later fetch starts a fresh request instead of joining it. + #[tokio::test] + async fn fetch_request_unused_once_every_fetch_leaves() { + let producer = track_producer("test", None); + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + + let first = consumer.fetch_group(5, None); + let req = dynamic + .requested_group() + .now_or_never() + .expect("should not block") + .unwrap(); + let second = consumer.fetch_group(5, None); + + drop(first); + assert!(req.poll_unused(&kio::Waiter::noop()).is_pending()); + drop(second); + assert!(req.poll_unused(&kio::Waiter::noop()).is_ready()); + + let retry = consumer.fetch_group(5, None); + let fresh = dynamic + .requested_group() + .now_or_never() + .expect("the retry queues a fresh request") + .unwrap(); + drop(req); + assert!(fresh.poll_unused(&kio::Waiter::noop()).is_pending()); + fresh.accept(None).unwrap().finish().unwrap(); + assert_eq!(retry.await.unwrap().sequence, 5); + } + #[tokio::test] async fn fetch_reject_does_not_poison_retry() { let producer = track_producer("test", None); diff --git a/rs/moq-net/tests/fetch_cancel.rs b/rs/moq-net/tests/fetch_cancel.rs new file mode 100644 index 0000000000..328ea50adc --- /dev/null +++ b/rs/moq-net/tests/fetch_cancel.rs @@ -0,0 +1,72 @@ +//! A FETCH nobody wants anymore is cancelled all the way to the publisher. +//! +//! Lite has no FETCH_OK, so until a publisher answers, each relay on the path holds +//! a FETCH stream and the group request behind it. When the reader gives up, every +//! hop must let go; otherwise a publisher that never answers pins them for good. + +mod support; + +use std::time::Duration; + +use moq_net::{Hop, Version, origin, track}; +use support::harness::peer; + +fn produce_origin(hop: u64) -> origin::Producer { + let (producer, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); + tokio::spawn(support::harness::run(driver)); + producer +} + +/// `relays` relays between the publisher and the reader. +async fn abandoned_fetch_reaches_the_publisher(version: &str, relays: u64) { + let version: Version = version.parse().unwrap(); + let nodes: Vec<_> = (1..=relays + 1).map(produce_origin).collect(); + let mut _pairs = Vec::new(); + for pair in nodes.windows(2) { + _pairs.push(peer(version, &pair[0], &pair[1]).await); + } + + let broadcast = nodes[0].create_broadcast("room").unwrap(); + let mut dynamic = broadcast.dynamic(); + broadcast.announce(Default::default()).unwrap(); + tokio::time::sleep(Duration::from_secs(1)).await; + + let consumer = nodes[relays as usize] + .consume() + .request_broadcast("room") + .await + .unwrap(); + let mut waiting = Box::pin(consumer.track("video").unwrap().fetch_group(0, None)); + + // Answer the TRACK_INFO, but hold the FETCH unanswered. + let request = tokio::select! { + request = dynamic.requested_track() => request.expect("the track request reaches the publisher"), + _ = &mut waiting => panic!("nothing answered the track"), + }; + let groups = request.dynamic(); + let _track = request.accept(track::Info::default()); + let request = tokio::select! { + request = groups.requested_group() => request.expect("the fetch reaches the publisher"), + _ = &mut waiting => panic!("nothing answered the fetch"), + }; + + // The reader gives up. + drop(waiting); + tokio::time::timeout(Duration::from_secs(5), kio::wait(|waiter| request.poll_unused(waiter))) + .await + .unwrap_or_else(|_| panic!("{version:?} over {relays} relays: the publisher's fetch is still wanted")); +} + +#[tokio::test(start_paused = true)] +async fn abandoned_fetch_reaches_the_publisher_lite05() { + for relays in [1, 2] { + abandoned_fetch_reaches_the_publisher("moq-lite-05", relays).await; + } +} + +#[tokio::test(start_paused = true)] +async fn abandoned_fetch_reaches_the_publisher_lite06() { + for relays in [1, 2] { + abandoned_fetch_reaches_the_publisher("moq-lite-06", relays).await; + } +} From bc4d2f93d6dcb20d1c5fd26aae3f24b64666bbef Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 17:43:56 -0700 Subject: [PATCH 3/7] fix(net): cancel a FETCH abandoned mid-response Watch demand through ingest too: once the last reader of a fetched group leaves, reset the FETCH stream and abort the partial group with Cancel instead of caching it as complete. @moq/net did the cancel but closed the truncated group cleanly; it now aborts it. Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/subscriber.test.ts | 24 +++++++++ js/net/src/lite/subscriber.ts | 6 ++- rs/moq-net/src/lite/subscriber.rs | 44 +++++++++++----- rs/moq-net/src/model/group.rs | 5 ++ rs/moq-net/tests/fetch_cancel.rs | 84 +++++++++++++++++++++++------- 5 files changed, 130 insertions(+), 33 deletions(-) diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index de0190a75f..0fbdb8140e 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -925,3 +925,27 @@ test.each([ subscriber.close(); }); + +// A reader leaving partway through a fetched group cancels the FETCH: the truncated group +// must not end clean, as a FIN would make it read whole. +test("the last reader leaving mid-response cancels the fetch", async () => { + const { quic, streams } = fakeSession(); + const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n)); + + const fetch = subscriber.fetchGroup(Path.from("room"), "video", 0); + await drainUntil(() => streams.length === 1); + await answerTrackInfo(streams[0]); + await drainUntil(() => streams.length === 2); + await streams[1].reading; + + // One frame of the group: a zero timestamp delta, then the sized payload. + streams[1].inbound.enqueue(new Uint8Array([0, 4, ...new TextEncoder().encode("head")])); + const group = await fetch; + expect(new TextDecoder().decode((await group.readFrame())?.payload)).toBe("head"); + group.close(); + + const err = fromTransport(await streams[1].aborted) as StreamError; + expect(err.code).toBe(StreamCode.Cancel); + + subscriber.close(); +}); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 3ac633580c..ba81f25468 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -853,11 +853,13 @@ export class Subscriber { pending ??= stream.reader.decodeMaybe(decode); const next = await race([pending, group.closed, unused]); if (next === idle) { - if (!group.isClosed && group.used.peek()) { + if (group.isClosed) break; + if (group.used.peek()) { unused = group.unused().then((): typeof idle => idle); continue; } - break; + // Abandoned mid-group: the truncated group must never end clean. + throw new StreamError(StreamCode.Cancel, { message: "cancel" }); } pending = undefined; if (!next || next instanceof Error) break; diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index f327d13ae4..a339bd39c5 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -4133,6 +4133,9 @@ enum FetchRunState { Ingest { stream: Stream, producer: group::Producer, + // The accepted request's demand: fetches that joined it but have not yet + // picked the group up from the cache. + joined: kio::Producer, // Boxed so the other states stay small; one allocation per fetch. ingest: Box, }, @@ -4140,11 +4143,17 @@ enum FetchRunState { } impl FetchRunState { - /// The downstream request, while it waits on the publisher's answer. - fn request(&self) -> Option<&group::Request> { + /// Whether nobody wants the group any more: no fetch waits on the request, and + /// once accepted, nobody reads the group either. + fn poll_abandoned(&self, waiter: &kio::Waiter) -> bool { match self { - Self::Open { request } | Self::Send { request, .. } | Self::Answer { request, .. } => request.as_ref(), - Self::Ingest { .. } | Self::Done => None, + Self::Open { request } | Self::Send { request, .. } | Self::Answer { request, .. } => request + .as_ref() + .is_some_and(|request| request.poll_unused(waiter).is_ready()), + Self::Ingest { producer, joined, .. } => { + joined.poll_unused(waiter).is_ready() && producer.poll_unused(waiter).is_ready() + } + Self::Done => false, } } } @@ -4167,16 +4176,20 @@ impl kio::Task for FetchServeRun { fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { let mut cx = std::task::Context::from_waker(waiter.waker()); loop { - // Until answered, a fetch nobody waits on any more is cancelled upstream, so - // the publisher stops serving it (and a relay there releases its own FETCH). - if let Some(request) = self.state.request() - && request.poll_unused(waiter).is_ready() - { + // A fetch nobody wants any more is cancelled upstream, so the publisher stops + // serving it (and a relay there releases its own FETCH). A group cut short is + // aborted, never cached as complete. + if self.state.poll_abandoned(waiter) { tracing::debug!(track = %self.serve.name, group = self.group, "fetch abandoned"); - if let FetchRunState::Send { stream, .. } | FetchRunState::Answer { stream, .. } = - std::mem::replace(&mut self.state, FetchRunState::Done) - { - stream.writer.abort(&Error::Cancel); + match std::mem::replace(&mut self.state, FetchRunState::Done) { + FetchRunState::Send { stream, .. } | FetchRunState::Answer { stream, .. } => { + stream.writer.abort(&Error::Cancel); + } + FetchRunState::Ingest { stream, producer, .. } => { + stream.writer.abort(&Error::Cancel); + let _ = producer.abort(Error::Cancel); + } + FetchRunState::Open { .. } | FetchRunState::Done => {} } return Poll::Ready(()); } @@ -4302,6 +4315,9 @@ impl kio::Task for FetchServeRun { let group_info = track::Info::default() .with_timescale(self.timescale.unwrap_or_default()) .with_max_age(self.serve.subscriber.origin.default_max_age()); + // The joined fetches pick the group up from the cache only when next + // polled, so their demand outlives the request until then. + let joined = request.result.clone(); let mut producer = match request.accept(group_info) { Ok(producer) => producer, Err(err) => { @@ -4323,6 +4339,7 @@ impl kio::Task for FetchServeRun { self.state = FetchRunState::Ingest { stream, producer, + joined, ingest: Box::new(FrameIngest::new(&self.serve.subscriber, self.timescale)), }; } @@ -4330,6 +4347,7 @@ impl kio::Task for FetchServeRun { stream, producer, ingest, + .. } => { let res = ready!(ingest.poll(&mut stream.reader, producer, waiter)); let FetchRunState::Ingest { producer, .. } = diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index bb725dcba1..0d9e2cf95c 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -970,6 +970,11 @@ impl Producer { self.state.unused().await.map_err(|_| self.abort_reason()) } + /// Poll for the group becoming unused (every consumer dropped). + pub(crate) fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> { + self.state.poll_unused(waiter).map(|_| ()) + } + /// The recorded abort reason, or [`Error::Dropped`] if the group closed without one. fn abort_reason(&self) -> Error { self.state.read().abort.clone().unwrap_or(Error::Dropped) diff --git a/rs/moq-net/tests/fetch_cancel.rs b/rs/moq-net/tests/fetch_cancel.rs index 328ea50adc..2dab3e8beb 100644 --- a/rs/moq-net/tests/fetch_cancel.rs +++ b/rs/moq-net/tests/fetch_cancel.rs @@ -1,14 +1,16 @@ //! A FETCH nobody wants anymore is cancelled all the way to the publisher. //! -//! Lite has no FETCH_OK, so until a publisher answers, each relay on the path holds -//! a FETCH stream and the group request behind it. When the reader gives up, every -//! hop must let go; otherwise a publisher that never answers pins them for good. +//! Each relay on the path holds a FETCH stream and the group behind it. When the +//! reader gives up, every hop must let go, whether the publisher has yet to answer +//! (lite has no FETCH_OK, so one that never answers would pin them for good) or is +//! partway through the group. A group cut short is aborted, never cached as whole. mod support; use std::time::Duration; -use moq_net::{Hop, Version, origin, track}; +use bytes::Bytes; +use moq_net::{Hop, Timestamp, Version, origin, track}; use support::harness::peer; fn produce_origin(hop: u64) -> origin::Producer { @@ -17,9 +19,19 @@ fn produce_origin(hop: u64) -> origin::Producer { producer } +/// Where the reader gives up. +#[derive(Clone, Copy, Debug)] +enum Stage { + /// Before the publisher answers the FETCH. + Unanswered, + /// After the first frame, with the rest of the group still to come. + MidResponse, +} + /// `relays` relays between the publisher and the reader. -async fn abandoned_fetch_reaches_the_publisher(version: &str, relays: u64) { +async fn abandoned_fetch_reaches_the_publisher(version: &str, relays: u64, stage: Stage) { let version: Version = version.parse().unwrap(); + let ctx = format!("{version:?} over {relays} relays, {stage:?}"); let nodes: Vec<_> = (1..=relays + 1).map(produce_origin).collect(); let mut _pairs = Vec::new(); for pair in nodes.windows(2) { @@ -36,9 +48,10 @@ async fn abandoned_fetch_reaches_the_publisher(version: &str, relays: u64) { .request_broadcast("room") .await .unwrap(); - let mut waiting = Box::pin(consumer.track("video").unwrap().fetch_group(0, None)); + let track = consumer.track("video").unwrap(); + let mut waiting = Box::pin(track.fetch_group(0, None)); - // Answer the TRACK_INFO, but hold the FETCH unanswered. + // Answer the TRACK_INFO, then take the FETCH. let request = tokio::select! { request = dynamic.requested_track() => request.expect("the track request reaches the publisher"), _ = &mut waiting => panic!("nothing answered the track"), @@ -50,23 +63,58 @@ async fn abandoned_fetch_reaches_the_publisher(version: &str, relays: u64) { _ = &mut waiting => panic!("nothing answered the fetch"), }; - // The reader gives up. - drop(waiting); - tokio::time::timeout(Duration::from_secs(5), kio::wait(|waiter| request.poll_unused(waiter))) - .await - .unwrap_or_else(|_| panic!("{version:?} over {relays} relays: the publisher's fetch is still wanted")); + match stage { + Stage::Unanswered => { + drop(waiting); + tokio::time::timeout(Duration::from_secs(5), kio::wait(|waiter| request.poll_unused(waiter))) + .await + .unwrap_or_else(|_| panic!("{ctx}: the publisher's fetch is still wanted")); + } + Stage::MidResponse => { + let mut group = request.accept(None).unwrap(); + group.write_frame(Timestamp::ZERO, Bytes::from_static(b"head")).unwrap(); + + let mut fetched = waiting.await.unwrap(); + let frame = fetched.read_frame().await.unwrap().unwrap(); + assert_eq!(&frame.payload[..], b"head", "{ctx}"); + drop(fetched); + + // The publisher stops serving the group: nobody reads it any more. + tokio::time::timeout(Duration::from_secs(5), group.unused()) + .await + .unwrap_or_else(|_| panic!("{ctx}: the publisher still serves the fetch")) + .unwrap(); + + // No hop cached the head as the whole group: a fresh fetch reads it all. + group.write_frame(Timestamp::ZERO, Bytes::from_static(b"tail")).unwrap(); + group.finish().unwrap(); + let mut fetched = tokio::time::timeout(Duration::from_secs(5), track.fetch_group(0, None)) + .await + .unwrap_or_else(|_| panic!("{ctx}: the refetch stalled")) + .unwrap(); + let mut payloads = Vec::new(); + while let Some(frame) = fetched.read_frame().await.unwrap() { + payloads.push(frame.payload); + } + assert_eq!(payloads, [&b"head"[..], &b"tail"[..]], "{ctx}"); + } + } +} + +async fn abandoned_fetches_reach_the_publisher(version: &str) { + for stage in [Stage::Unanswered, Stage::MidResponse] { + for relays in [1, 2] { + abandoned_fetch_reaches_the_publisher(version, relays, stage).await; + } + } } #[tokio::test(start_paused = true)] async fn abandoned_fetch_reaches_the_publisher_lite05() { - for relays in [1, 2] { - abandoned_fetch_reaches_the_publisher("moq-lite-05", relays).await; - } + abandoned_fetches_reach_the_publisher("moq-lite-05").await; } #[tokio::test(start_paused = true)] async fn abandoned_fetch_reaches_the_publisher_lite06() { - for relays in [1, 2] { - abandoned_fetch_reaches_the_publisher("moq-lite-06", relays).await; - } + abandoned_fetches_reach_the_publisher("moq-lite-06").await; } From bd33b36566c733d9f178bc34c572ee29c125141e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 18:11:56 -0700 Subject: [PATCH 4/7] fix(net): let a FETCH's completion win over abandonment Check ingest abandonment only while the group is still short of its FIN, and abort atomically with a new reader arriving. The accepted request writes NotFound for joined fetches it can't cover, since the subscriber now keeps the result channel open to track their demand. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/subscriber.rs | 59 ++++++++++++++++++------------- rs/moq-net/src/model/group.rs | 21 +++++++++-- rs/moq-net/src/model/track.rs | 14 ++++++-- rs/moq-net/tests/fetch_cancel.rs | 36 ++++++++++++++++++- 4 files changed, 100 insertions(+), 30 deletions(-) diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index a339bd39c5..41bf3ce820 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -4143,17 +4143,11 @@ enum FetchRunState { } impl FetchRunState { - /// Whether nobody wants the group any more: no fetch waits on the request, and - /// once accepted, nobody reads the group either. - fn poll_abandoned(&self, waiter: &kio::Waiter) -> bool { + /// The downstream request, while it waits on the publisher's answer. + fn request(&self) -> Option<&group::Request> { match self { - Self::Open { request } | Self::Send { request, .. } | Self::Answer { request, .. } => request - .as_ref() - .is_some_and(|request| request.poll_unused(waiter).is_ready()), - Self::Ingest { producer, joined, .. } => { - joined.poll_unused(waiter).is_ready() && producer.poll_unused(waiter).is_ready() - } - Self::Done => false, + Self::Open { request } | Self::Send { request, .. } | Self::Answer { request, .. } => request.as_ref(), + Self::Ingest { .. } | Self::Done => None, } } } @@ -4176,20 +4170,16 @@ impl kio::Task for FetchServeRun { fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { let mut cx = std::task::Context::from_waker(waiter.waker()); loop { - // A fetch nobody wants any more is cancelled upstream, so the publisher stops - // serving it (and a relay there releases its own FETCH). A group cut short is - // aborted, never cached as complete. - if self.state.poll_abandoned(waiter) { + // A fetch nobody waits on any more is cancelled upstream, so the publisher stops + // serving it (and a relay there releases its own FETCH). Ingest has its own check. + if let Some(request) = self.state.request() + && request.poll_unused(waiter).is_ready() + { tracing::debug!(track = %self.serve.name, group = self.group, "fetch abandoned"); - match std::mem::replace(&mut self.state, FetchRunState::Done) { - FetchRunState::Send { stream, .. } | FetchRunState::Answer { stream, .. } => { - stream.writer.abort(&Error::Cancel); - } - FetchRunState::Ingest { stream, producer, .. } => { - stream.writer.abort(&Error::Cancel); - let _ = producer.abort(Error::Cancel); - } - FetchRunState::Open { .. } | FetchRunState::Done => {} + if let FetchRunState::Send { stream, .. } | FetchRunState::Answer { stream, .. } = + std::mem::replace(&mut self.state, FetchRunState::Done) + { + stream.writer.abort(&Error::Cancel); } return Poll::Ready(()); } @@ -4346,10 +4336,29 @@ impl kio::Task for FetchServeRun { FetchRunState::Ingest { stream, producer, + joined, ingest, - .. } => { - let res = ready!(ingest.poll(&mut stream.reader, producer, waiter)); + let Poll::Ready(res) = ingest.poll(&mut stream.reader, producer, waiter) else { + // Still short of its end, so completion always wins. Once nobody + // wants the rest (no joined fetch left to pick it up, no reader), + // cancel upstream and abort the truncated group, never caching it + // as complete. The abort is atomic with a new reader arriving. + if joined.poll_unused(waiter).is_pending() || producer.poll_unused(waiter).is_pending() { + return Poll::Pending; + } + if !producer.abort_unused(Error::Cancel) { + continue; + } + tracing::debug!(track = %self.serve.name, group = self.group, "fetch abandoned"); + let FetchRunState::Ingest { stream, .. } = + std::mem::replace(&mut self.state, FetchRunState::Done) + else { + unreachable!() + }; + stream.writer.abort(&Error::Cancel); + return Poll::Ready(()); + }; let FetchRunState::Ingest { producer, .. } = std::mem::replace(&mut self.state, FetchRunState::Done) else { diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 0d9e2cf95c..009d61f08d 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -783,13 +783,30 @@ impl Producer { self.close_aborted(err) } + /// Abort with `err` only while nothing consumes the group, returning whether it is + /// closed. Consumer creation and the check share a lock, so a reader arriving after + /// [`poll_unused`](Self::poll_unused) keeps the group alive instead of reading the abort. + pub(crate) fn abort_unused(&self, err: Error) -> bool { + match self.state.write_unused() { + kio::Unused::Idle(guard) => { + self.commit_abort(guard, err); + true + } + kio::Unused::Closed => true, + kio::Unused::Used => false, + } + } + fn close_aborted(&self, err: Error) -> Result<()> { - let mut guard = modify(&self.state)?; + self.commit_abort(modify(&self.state)?, err); + Ok(()) + } + + fn commit_abort(&self, mut guard: kio::Mut<'_, GroupState>, err: Error) { guard.abort = Some(err); self.alive.aborted.store(true, Ordering::Release); guard.release(); guard.close(); - Ok(()) } /// Abort a write that would grow the group past its budget, holding the lock already diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 8c6182808d..47701a3d3c 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -3101,6 +3101,12 @@ impl group::Request { let res = TrackState::modify(&self.state) .and_then(|mut state| state.insert_group_request(self.sequence, self.frame_start, info.into())); self.remove(); + // A joined fetch the cached group can't cover (a wider start that joined after + // the range went on the wire) fails rather than waits. Closing the channel says + // the same, but a handler tracking the joined demand keeps it open. + if let Ok(mut outcome) = self.result.write() { + outcome.rejected = Some(Error::NotFound); + } res } @@ -8167,6 +8173,9 @@ mod test { assert!(futures::poll!(widest.as_mut()).is_pending()); assert_eq!(request.frame_start(), 2, "the in-flight range is already on the wire"); + // A handler may keep tracking the joined demand past the accept, as the lite + // subscriber does, which holds the result channel open. + let _joined = request.result.clone(); let mut group = request.accept(None).unwrap(); // A handler numbers the frames from where it was asked to start, as `serve_fetch` // does; that offset is what makes the cached group too narrow for `widest`. @@ -8174,13 +8183,14 @@ mod test { group .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"from2")) .unwrap(); - group.finish().unwrap(); // The two callers it covers resolve; the one it doesn't fails cleanly rather - // than being handed a group that starts above what it asked for. + // than being handed a group that starts above what it asked for, even while + // the group is still being written. assert_eq!(narrow.await.unwrap().index(), 5); assert_eq!(wider.await.unwrap().index(), 2); assert!(matches!(widest.await, Err(Error::NotFound))); + group.finish().unwrap(); } #[tokio::test] diff --git a/rs/moq-net/tests/fetch_cancel.rs b/rs/moq-net/tests/fetch_cancel.rs index 2dab3e8beb..5e2d45ff7c 100644 --- a/rs/moq-net/tests/fetch_cancel.rs +++ b/rs/moq-net/tests/fetch_cancel.rs @@ -26,6 +26,8 @@ enum Stage { Unanswered, /// After the first frame, with the rest of the group still to come. MidResponse, + /// Once every frame is read, as the FIN arrives. + Drained, } /// `relays` relays between the publisher and the reader. @@ -98,11 +100,43 @@ async fn abandoned_fetch_reaches_the_publisher(version: &str, relays: u64, stage } assert_eq!(payloads, [&b"head"[..], &b"tail"[..]], "{ctx}"); } + Stage::Drained => { + let mut group = request.accept(None).unwrap(); + group.write_frame(Timestamp::ZERO, Bytes::from_static(b"head")).unwrap(); + group.write_frame(Timestamp::ZERO, Bytes::from_static(b"tail")).unwrap(); + + let mut fetched = waiting.await.unwrap(); + for expected in [&b"head"[..], b"tail"] { + let frame = fetched.read_frame().await.unwrap().unwrap(); + assert_eq!(&frame.payload[..], expected, "{ctx}"); + } + + // The FIN reaches the reader's node at the instant the reader leaves: the + // group is whole, so it must be cached as such rather than cancelled. + const LATENCY: Duration = Duration::from_millis(100); + _pairs.last().unwrap().client_transport.set_latency(LATENCY); + group.finish().unwrap(); + tokio::time::sleep(LATENCY).await; + drop(fetched); + tokio::time::sleep(Duration::from_secs(1)).await; + + // The publisher can no longer serve it, so only a cached copy answers. + group.abort(moq_net::Error::Cancel).unwrap(); + let mut fetched = tokio::time::timeout(Duration::from_secs(5), track.fetch_group(0, None)) + .await + .unwrap_or_else(|_| panic!("{ctx}: the refetch stalled")) + .unwrap_or_else(|err| panic!("{ctx}: the whole group was not cached: {err}")); + let mut payloads = Vec::new(); + while let Some(frame) = fetched.read_frame().await.unwrap() { + payloads.push(frame.payload); + } + assert_eq!(payloads, [&b"head"[..], &b"tail"[..]], "{ctx}"); + } } } async fn abandoned_fetches_reach_the_publisher(version: &str) { - for stage in [Stage::Unanswered, Stage::MidResponse] { + for stage in [Stage::Unanswered, Stage::MidResponse, Stage::Drained] { for relays in [1, 2] { abandoned_fetch_reaches_the_publisher(version, relays, stage).await; } From 3820db2d2ff9dba549bd14a5e4cfbeeeeece2e43 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 18:37:06 -0700 Subject: [PATCH 5/7] fix(net): keep a fetch's cache hit wanted until it is polled A fetch_group that hits the cache now holds the group consumer until polled, so an abandoned upstream fetch can't abort the group in between. A cache lookup never hands out a consumer for a group aborted mid-lookup, and a rejection re-checks the cache an accept may have just filled. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/group.rs | 8 +++ rs/moq-net/src/model/track.rs | 131 +++++++++++++++++++++++++++------- 2 files changed, 114 insertions(+), 25 deletions(-) diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 009d61f08d..f4850f816d 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -953,6 +953,14 @@ impl Producer { } } + /// Create a consumer, or `None` once the group is aborted. Paired with + /// [`abort_unused`](Self::abort_unused): a consumer either exists in time to decline + /// the abort, or sees it here and is never handed out. + pub(crate) fn try_consume(&self) -> Option { + let consumer = self.consume(); + (!self.state.read().is_closed()).then_some(consumer) + } + /// Register for the first-frame timestamp while the group is still unstamped. pub(crate) fn poll_timestamp(&self, waiter: &kio::Waiter) -> Poll<()> { match self.state.poll(waiter, |state| { diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 47701a3d3c..dc0a4bd201 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -635,12 +635,16 @@ impl TrackState { /// end-of-stream. The handler side (a rejection, or no [`Dynamic`] at all) lives /// in [`FetchState`]; [`Fetching`] polls both. fn poll_fetch_cached(&self, sequence: u64, frame_start: u64) -> Poll> { - if let Some(group) = self.covering_group(sequence, frame_start) { + // A group aborted between the lookup and the consume (an abandoned fetch cut + // short) is a miss, not a hit that fails on its first read. + if let Some(group) = self.covering_group(sequence, frame_start) + && let Some(consumer) = group.try_consume() + { // A cache hit refreshes the group: it resets both its age (expiry keys // off the last access) and its standing against the pool-wide average, // so the eviction walk keeps it over never-read groups. group.cache_refresh(); - return Poll::Ready(Ok(group.consume())); + return Poll::Ready(Ok(consumer)); } if let Some(err) = &self.abort { @@ -2824,13 +2828,20 @@ impl Consumer { // Queue a request only when the group isn't already resolvable from the track // (cached, aborted, or past-final all resolve through `Fetching::poll` without // a queue entry). - let (fetch, unresolved) = { + let (fetch, cached) = { let state = state.read(); ( state.fetch.clone(), - state.poll_fetch_cached(sequence, options.frame_start).is_pending(), + state.poll_fetch_cached(sequence, options.frame_start), ) }; + let unresolved = cached.is_pending(); + // Hold a hit until the caller polls: the held consumer keeps the group wanted, + // so an abandoned fetch filling it can't abort it in between. + let hit = match cached { + Poll::Ready(Ok(group)) => Some(Box::new(group)), + _ => None, + }; if unresolved { let mut fetch = fetch.lock(); @@ -2871,6 +2882,7 @@ impl Consumer { sequence, frame_start: options.frame_start, result, + hit, }, stats: self.stats.clone(), }) @@ -3191,8 +3203,12 @@ enum FetchingKind { // This caller's own start, so a cached group that begins above it is a miss // rather than a short answer. frame_start: u64, - // The joined attempt's result channel; `None` when no handler existed to queue on. + // The joined attempt's result channel; `None` on a cache hit, or when no handler + // existed to queue on. result: Option>, + // The group already cached when the fetch was made, held so it stays wanted. + // Boxed: a group consumer dwarfs the rest of the handle. + hit: Option>, }, /// A spliced track's fetch: waits for a segment, then fetches from it. Spliced(kio::Pending), @@ -3202,14 +3218,15 @@ impl kio::Pollable for Fetching { type Output = Result; fn poll(&self, waiter: &kio::Waiter) -> Poll { - let (state, fetch, sequence, frame_start, result) = match &self.inner { + let (state, fetch, sequence, frame_start, result, hit) = match &self.inner { FetchingKind::Plain { state, fetch, sequence, frame_start, result, - } => (state, fetch, *sequence, *frame_start, result.as_ref()), + hit, + } => (state, fetch, *sequence, *frame_start, result.as_ref(), hit), FetchingKind::Spliced(spliced) => { // A fetched group is metered here (once), at the tagged handle: the // spliced source track it comes from is the origin's own, untagged. @@ -3218,22 +3235,26 @@ impl kio::Pollable for Fetching { } }; + // Hand back a consumer sitting where the caller asked rather than at the group's + // own start. Coverage was checked first, so this can only skip frames they excluded. + let resolve = |mut group: group::Consumer| { + group.start_at(frame_start); + group.with_meter(self.stats.meter()) + }; + if let Some(group) = hit { + return Poll::Ready(Ok(resolve((**group).clone()))); + } + // Track side: the cached group, the abort error, or past-final. The outer // error is the channel closing without any of those. - match state.poll(waiter, |state| state.poll_fetch_cached(sequence, frame_start)) { - Poll::Ready(Ok(res)) => { - return Poll::Ready(res.map(|mut group| { - // Hand back a consumer sitting where the caller asked rather than at - // the group's own start. Coverage was checked above, so this can only - // skip frames they excluded. - group.start_at(frame_start); - group.with_meter(self.stats.meter()) - })); - } - Poll::Ready(Err(closed)) => { - return Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))); - } - Poll::Pending => {} + let cached = + |waiter: &kio::Waiter| match state.poll(waiter, |state| state.poll_fetch_cached(sequence, frame_start)) { + Poll::Ready(Ok(res)) => Poll::Ready(res.map(resolve)), + Poll::Ready(Err(closed)) => Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))), + Poll::Pending => Poll::Pending, + }; + if let Poll::Ready(res) = cached(waiter) { + return Poll::Ready(res); } // Handler side. @@ -3251,13 +3272,19 @@ impl kio::Pollable for Fetching { // A written rejection fails every joined fetch. The channel closing without // one means the attempt was dropped unserved (its handlers went away). - match result.poll(waiter, |outcome| match &outcome.rejected { + let err = match result.poll(waiter, |outcome| match &outcome.rejected { Some(err) => Poll::Ready(err.clone()), None => Poll::Pending, }) { - Poll::Ready(Ok(err)) => Poll::Ready(Err(err)), - Poll::Ready(Err(_closed)) => Poll::Ready(Err(Error::NotFound)), - Poll::Pending => Poll::Pending, + Poll::Ready(Ok(err)) => err, + Poll::Ready(Err(_closed)) => Error::NotFound, + Poll::Pending => return Poll::Pending, + }; + // An accept caches the group before it answers the channel, which may land + // between the track check above and this one: look again before failing. + match cached(waiter) { + Poll::Ready(res) => Poll::Ready(res), + Poll::Pending => Poll::Ready(Err(err)), } } } @@ -8067,6 +8094,60 @@ mod test { assert_eq!(retry.await.unwrap().sequence, 5); } + /// A fetch that hits the cache holds the group until polled, so a handler that aborts + /// the group once nobody wants it (an abandoned upstream fetch) leaves it alone. + #[tokio::test] + async fn fetch_hit_keeps_the_group_wanted() { + let producer = track_producer("test", None); + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + + let first = consumer.fetch_group(5, None); + let req = dynamic + .requested_group() + .now_or_never() + .expect("should not block") + .unwrap(); + let mut group = req.accept(None).unwrap(); + group + .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head")) + .unwrap(); + drop(first.await.unwrap()); + + let hit = consumer.fetch_group(5, None); + assert!(!group.abort_unused(Error::Cancel), "the pending hit wants the group"); + let mut fetched = hit.await.unwrap(); + assert_eq!(&fetched.read_frame().await.unwrap().unwrap().payload[..], b"head"); + } + + /// Once a group nobody wanted is aborted, a later fetch misses rather than reading + /// the abort, and queues a fresh request. + #[tokio::test] + async fn fetch_misses_an_abandoned_group() { + let producer = track_producer("test", None); + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + + let first = consumer.fetch_group(5, None); + let req = dynamic + .requested_group() + .now_or_never() + .expect("should not block") + .unwrap(); + let group = req.accept(None).unwrap(); + drop(first.await.unwrap()); + assert!(group.abort_unused(Error::Cancel)); + + let retry = consumer.fetch_group(5, None); + let req = dynamic + .requested_group() + .now_or_never() + .expect("the retry queues a fresh request") + .unwrap(); + req.accept(None).unwrap().finish().unwrap(); + assert!(retry.await.unwrap().read_frame().await.unwrap().is_none()); + } + #[tokio::test] async fn fetch_reject_does_not_poison_retry() { let producer = track_producer("test", None); From 086a810c8c9e424d16647319ef93108b1aa2c5a2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 18:57:28 -0700 Subject: [PATCH 6/7] fix(net): mint a group consumer only while the group is open try_consume dropped the rejected consumer while still holding the group lock, deadlocking when an abort won the race. Use kio's consume_open, whose closed check and count share the write_unused lock. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/group.rs | 36 ++++++++++++++++++++++++++--------- 1 file changed, 27 insertions(+), 9 deletions(-) diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index f4850f816d..e199da5f88 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -930,11 +930,22 @@ impl Producer { /// Create a new consumer for the group. pub fn consume(&self) -> Consumer { + self.consumer(self.state.consume()) + } + + /// Create a consumer, or `None` once the group is aborted. Paired with + /// [`abort_unused`](Self::abort_unused): the closed check and the count share its + /// lock, so a consumer either exists in time to decline the abort or is never made. + pub(crate) fn try_consume(&self) -> Option { + self.state.weak().try_consume().map(|state| self.consumer(state)) + } + + fn consumer(&self, state: kio::Consumer) -> Consumer { Consumer { info: self.info, track: self.track.clone(), inner: ConsumerKind::Plain(Plain { - state: self.state.consume(), + state, index: 0, end: None, prefetch: Prefetch::default(), @@ -953,14 +964,6 @@ impl Producer { } } - /// Create a consumer, or `None` once the group is aborted. Paired with - /// [`abort_unused`](Self::abort_unused): a consumer either exists in time to decline - /// the abort, or sees it here and is never handed out. - pub(crate) fn try_consume(&self) -> Option { - let consumer = self.consume(); - (!self.state.read().is_closed()).then_some(consumer) - } - /// Register for the first-frame timestamp while the group is still unstamped. pub(crate) fn poll_timestamp(&self, waiter: &kio::Waiter) -> Poll<()> { match self.state.poll(waiter, |state| { @@ -1999,6 +2002,21 @@ mod test { assert!(matches!(result, Err(crate::Error::Cancel))); } + #[test] + fn abort_unused_pairs_with_try_consume() { + let producer = Info { sequence: 0 }.produce(); + + let consumer = producer.try_consume().expect("open"); + assert!( + !producer.abort_unused(crate::Error::Cancel), + "a reader declines the abort" + ); + drop(consumer); + + assert!(producer.abort_unused(crate::Error::Cancel)); + assert!(producer.try_consume().is_none(), "an aborted group mints no reader"); + } + #[test] fn abort_clears_cached_frames() { let mut producer = Info { sequence: 0 }.produce(); From 84e965dc50568a15e0fce02c8496bba58fc1c04d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 19:25:47 -0700 Subject: [PATCH 7/7] fix(net): start an accepted fetch group where its request did The group was cached at offset 0 until the handler called start_at, so a wider fetch looking it up in between got a reader that later skipped the head it asked for. Start it before it is visible. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 39 ++++++++++++++++++++++++++++++++--- 1 file changed, 36 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index dc0a4bd201..d526934dff 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1230,7 +1230,11 @@ impl TrackState { // An evicted sequence can be re-fetched; a live one is a duplicate. self.claim_sequence(sequence, frame_start)?; - let group = group::Producer::new(group::Info { sequence }, info, self.cache.clone()); + let mut group = group::Producer::new(group::Info { sequence }, info, self.cache.clone()); + // Start where the request did before the group is visible: a fetch looking it up + // in between would otherwise see it begin at 0 and get a reader that later skips + // the head it asked for. + group.start_at(frame_start)?; // A backfill exists because someone is fetching it right now: stamp that // access so the eviction walk can't kill it before the fetch resolves. // It is also invisible to arrival-order subscribers: fetched on demand, @@ -3089,8 +3093,10 @@ impl group::Request { /// The first frame of the group the consumer wants; 0 is the whole group. /// - /// A handler serving this must [`start_at`](group::Producer::start_at) it, so the frames - /// it writes carry the indices they have in the group rather than restarting at 0. + /// The group [`accept`](Self::accept) returns already starts here, so the frames a + /// handler writes carry the indices they have in the group rather than restarting at + /// 0. A handler serving from elsewhere moves it with + /// [`start_at`](group::Producer::start_at) before the first frame. /// /// There is no end: the handler fetches through the end of the group so the result /// is cacheable for anyone (see [`group::Fetch::frame_start`]). @@ -8120,6 +8126,33 @@ mod test { assert_eq!(&fetched.read_frame().await.unwrap().unwrap().payload[..], b"head"); } + /// An accepted group starts where its request did before anyone can see it, so a + /// wider fetch arriving before the handler writes misses instead of being handed a + /// reader that would skip the head it asked for. + #[tokio::test] + async fn accepted_group_starts_at_the_request() { + let producer = track_producer("test", None); + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + + let _tail = consumer.fetch_group(5, group::Fetch::default().with_frame_start(3)); + let req = dynamic + .requested_group() + .now_or_never() + .expect("should not block") + .unwrap(); + let _group = req.accept(None).unwrap(); + + let whole = consumer.fetch_group(5, None); + let req = dynamic + .requested_group() + .now_or_never() + .expect("the wider fetch misses and queues its own request") + .unwrap(); + assert_eq!(req.frame_start(), 0); + assert!(kio::Pollable::poll(&*whole, &kio::Waiter::noop()).is_pending()); + } + /// Once a group nobody wanted is aborted, a later fetch misses rather than reading /// the abort, and queues a fresh request. #[tokio::test]