From dc58b95fc0ea738b4dc2dd66742d63259d4d7bb2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 22:30:27 -0700 Subject: [PATCH 1/4] chore: claim IETF fetch abandonment quest Co-Authored-By: GPT-6 From bb7e4854b8f8a5e30a96beb90b26311e1a998f75 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 22:46:14 -0700 Subject: [PATCH 2/4] fix(net): cancel abandoned IETF group fetches Co-Authored-By: GPT-6 --- doc/concept/standard.md | 5 +- quest/m1/README.md | 1 - quest/m1/demand-everywhere.md | 5 +- quest/m1/ietf-fetch-abandonment.md | 32 --- rs/moq-net/src/ietf/subscriber.rs | 306 ++++++++++++++++++++++++++--- 5 files changed, 290 insertions(+), 59 deletions(-) delete mode 100644 quest/m1/ietf-fetch-abandonment.md diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 7da4a25b11..bb065dc998 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -46,7 +46,10 @@ on moq-lite. On drafts 14–19, the Rust publisher answers a standalone `FETCH` within one group from the cache. A relay fetches a missing group upstream with a `FETCH` of that one whole group, and an upstream refusal is the refusal the fetcher -sees. A range touching several groups is refused with `NOT_SUPPORTED`, as is +sees. Once its last reader leaves, the relay cancels the upstream fetch, even +before `FETCH_OK`, and aborts an incomplete group instead of caching it as whole. +Drafts 14–16 use `FETCH_CANCEL`; drafts 17–19 stop and reset the request stream. +A range touching several groups is refused with `NOT_SUPPORTED`, as is any `FETCH` on draft-20 and later, which moved the range into `LOCATION_FILTER`. A standalone `FETCH` carries no timestamps, since no `SUBSCRIBE_OK` declared a timescale for it. diff --git a/quest/m1/README.md b/quest/m1/README.md index a29b80e341..bfcac0ac5c 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -51,7 +51,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 -- [IETF FETCH abandonment](/quest/m1/ietf-fetch-abandonment.md) - the Rust moq-transport subscriber cancels a group FETCH once its last reader leaves - [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/demand-everywhere.md b/quest/m1/demand-everywhere.md index 9db4a1b3aa..2378c60c55 100644 --- a/quest/m1/demand-everywhere.md +++ b/quest/m1/demand-everywhere.md @@ -41,6 +41,5 @@ its race still needs an owner. Public API: breaking on `dev` in moq-net and `@moq/net`. Wire: none. -## Related - -- [IETF FETCH abandonment](/quest/m1/ietf-fetch-abandonment.md) - first new consumer of group request demand; can ship on `poll_unused` without waiting for this +When main's IETF FETCH cancellation integrates into dev, migrate its request +watch from `poll_unused` to `demand().poll_unused` along with the other handlers. diff --git a/quest/m1/ietf-fetch-abandonment.md b/quest/m1/ietf-fetch-abandonment.md deleted file mode 100644 index 6f3977b781..0000000000 --- a/quest/m1/ietf-fetch-abandonment.md +++ /dev/null @@ -1,32 +0,0 @@ -# [S] An abandoned IETF group FETCH is cancelled - -## Goal - -When the last reader leaves a group that the Rust moq-transport subscriber is -fetching upstream, it cancels that FETCH (STOP_SENDING plus a reset of the -request stream, or FETCH_CANCEL on drafts that define it), before and after -FETCH_OK, the same as an idle SUBSCRIBE and the lite FETCH after #4691. The -partial group aborts with `Error::Cancel` and never ends clean. - -## Plan - -Facts (`origin/main`, 2026-10-01): `run_group_fetch` in -`rs/moq-net/src/ietf/subscriber.rs` sends the FETCH and waits only for the -request stream to close or the group slot to finish. It never watches demand. -It is cancelled only when its whole subscription ends. The subscribe path -already polls `poll_unused` and calls `cancel_subscribe`. Rust already encodes -and decodes `ietf::FetchCancel` (0x17, `rs/moq-net/src/ietf/fetch.rs`) and the -adapter routes it, but `run_group_fetch` never sends it. JS never issues an IETF FETCH (`fetchGroup` refuses on -moq-transport), so it is out of scope. - -Decided (2026-10-01): cancel after the answer too, matching SUBSCRIBE; the -fetched group is not filled into the cache for absent readers. Reuse -`group::Request::poll_unused` from #4691, or `demand()` if -[Demand everywhere](/quest/m1/demand-everywhere.md) lands first. Send the -existing `FetchCancel` on the drafts that define it (14 to 16) and reset the -request stream on later drafts, which removed FETCH_CANCEL. - -Tests: a mock-time test abandons a group FETCH before and after FETCH_OK and -asserts the upstream stream is cancelled. - -Public API: none. Wire: none. diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index a8a7dad47e..dc2c64d6ee 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -3024,7 +3024,7 @@ where // Registered before the FETCH goes out, since its fetch stream can overtake FETCH_OK. let slot = kio::Producer::new(GroupFetch::Pending); - let _registered = GroupFetchEntry::new(&self.state, fetch_id, slot.clone()); + let registered = GroupFetchEntry::new(&self.state, fetch_id, slot.clone()); let mut stream = match Stream::open(&mut self.session.clone(), self.version).await { Ok(stream) => stream, @@ -3056,13 +3056,33 @@ where fill_timeout: false, }) .await?; - self.read_group_fetch_response(&mut stream).await + Ok::<_, Error>(()) } .await; + let res = match res { + Err(err) => Some(Err(err)), + Ok(()) => { + let mut response = std::pin::pin!(self.read_group_fetch_response(&mut stream)); + kio::wait(|waiter| { + // A refusal retires the peer's request, so it wins over abandonment. + if let Poll::Ready(res) = waiter.poll_future(response.as_mut()) { + return Poll::Ready(Some(res)); + } + request.poll_unused(waiter).map(|()| None) + }) + .await + } + }; let ok = match res { - Ok(ok) => ok, - Err(err) => { + Some(Ok(ok)) => ok, + None => { + request.reject(Error::Cancel); + drop(registered); + self.cancel_group_fetch(stream, fetch_id).await; + return; + } + Some(Err(err)) => { tracing::debug!(%err, group = sequence, "group fetch refused"); request.reject(err); let _ = stream.writer.close().await; @@ -3097,6 +3117,8 @@ where let info = track::Info::default() .with_timescale(Timescale::MICRO) .with_max_age(self.origin.default_max_age()); + // Joined fetches still count until they pick the accepted group up from the cache. + let joined = request.result.clone(); let mut producer = match request.accept(info) { Ok(producer) => producer, // Already served by a concurrent fetch, or the track closed. @@ -3112,6 +3134,7 @@ where let _ = stream.writer.close().await; return; } + let demand = producer.clone(); if let Ok(mut state) = slot.write() { *state = GroupFetch::Ready { producer, @@ -3121,9 +3144,9 @@ where }; } - // Hold the request open until its fetch stream is done: closing our side first is - // what a draft-14-16 adapter reads as cancelling the FETCH. A publisher that fails - // after FETCH_OK resets the request instead and owes no fetch stream, so the group + // Keep the request open until its data stream finishes or every reader leaves. + // A publisher that fails after FETCH_OK resets the request instead and owes no fetch + // stream, so the group // it left waiting is aborted. A FIN is not that: the fetch stream can trail it. let mut open = true; let reset = kio::wait(|waiter| { @@ -3136,22 +3159,54 @@ where } } slot.poll(waiter, |state| match &**state { - GroupFetch::Done => Poll::Ready(()), - _ => Poll::Pending, + GroupFetch::Done => Poll::Ready(None), + _ => { + if joined.poll_unused(waiter).is_ready() + && demand.poll_unused(waiter).is_ready() + && demand.abort_unused(Error::Cancel) + { + Poll::Ready(Some(Error::Cancel)) + } else { + Poll::Pending + } + } }) - .map(|_| None) + .map(|res| res.ok().flatten()) }) .await; - if let Some(err) = reset - && let Ok(mut state) = slot.write() - && matches!(*state, GroupFetch::Ready { .. }) - && let GroupFetch::Ready { producer, .. } = std::mem::replace(&mut *state, GroupFetch::Done) - { - let _ = producer.abort(err); + if let Some(err) = reset { + let cancel = matches!(err, Error::Cancel); + let _ = demand.abort(err); + drop(registered); + if cancel { + self.cancel_group_fetch(stream, fetch_id).await; + return; + } } let _ = stream.writer.close().await; } + /// Cancel a FETCH using the negotiated draft's existing cancellation signal. + async fn cancel_group_fetch(&self, mut stream: Stream, request_id: RequestId) { + stream.reader.abort(&Error::Cancel); + match self.version { + Version::Draft14 | Version::Draft15 | Version::Draft16 => { + // The adapter has no transport reset, so deliver FETCH_CANCEL without a + // subsequent writer drop discarding unacknowledged control bytes. + let res = async { + stream.writer.encode(&ietf::FetchCancel::ID).await?; + stream.writer.encode(&ietf::FetchCancel { request_id }).await?; + stream.writer.close().await + } + .await; + if let Err(err) = res { + tracing::debug!(%err, "failed to cancel group fetch"); + } + } + _ => stream.writer.abort(&Error::Cancel), + } + } + /// Read the answer to a group FETCH: FETCH_OK, or the publisher's refusal as an error. async fn read_group_fetch_response(&self, stream: &mut Stream) -> Result { let type_id: u64 = stream.reader.decode().await?; @@ -3206,20 +3261,32 @@ where .await; let (producer, timescale, start, end) = taken?; + let closed = producer.clone(); let mut producer = crate::recv::Group::new(producer); - let res = match self - .recv_group_fetch_objects(stream, &mut producer, start, end, timescale) + let res = { + let mut receive = + std::pin::pin!(self.recv_group_fetch_objects(stream, &mut producer, start, end, timescale)); + kio::wait(|waiter| { + if let Poll::Ready(res) = waiter.poll_future(receive.as_mut()) { + return Poll::Ready(res); + } + closed.poll_closed(waiter).map(Err) + }) .await - { + }; + + // Completion and abandonment share the slot lock, so a complete group is never + // aborted between finishing it and publishing Done to the request handler. + let mut state = slot.write().ok(); + let res = match res { Ok(()) => producer.finish(), Err(err) => { let _ = producer.abort(err.clone()); Err(err) } }; - - if let Ok(mut state) = slot.write() { - *state = GroupFetch::Done; + if let Some(state) = state.as_mut() { + **state = GroupFetch::Done; } res } @@ -7692,6 +7759,201 @@ mod joining_fetch_tests { assert!(writes.is_empty(), "a refused join must not write SUBSCRIBE"); } + #[derive(Clone, Copy, PartialEq, Eq)] + enum FetchStage { + Unanswered, + Accepted, + Receiving, + Complete, + } + + /// Last-reader cancellation covers a pending answer, an accepted group waiting for + /// its stream, and a stream stalled after its first complete frame. + async fn abandon_group_fetch(version: Version, stage: FetchStage) { + const GROUP: u64 = 4; + let session = ScriptedSession::new(Vec::new()); + let (tasks, _task_set) = TaskSet::new(); + let subscriber = Subscriber::new( + crate::time::Clock::tokio(), + session.clone(), + crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), + Control::new(None, false), + None, + peer::PeerSetup::default(), + crate::Hop::new(1).unwrap(), + None, + version, + tasks, + Default::default(), + ); + let track = track::Producer::new( + std::sync::Arc::new(crate::broadcast::Info::default()), + "video", + track::Info::default(), + ); + let dynamic = track.dynamic(); + let consumer = track.consume(); + let mut fetch = Box::pin(consumer.fetch_group(GROUP, None)); + assert!(futures::poll!(fetch.as_mut()).is_pending()); + let request = dynamic.requested_group().await.expect("group request"); + let outcome = request.result.clone(); + let mut run = Box::pin(subscriber.clone().run_group_fetch( + Path::new("broadcast").to_owned(), + "video".into(), + request, + None, + )); + assert!(futures::poll!(run.as_mut()).is_pending()); + let slot = subscriber.state.lock().group_fetches[&RequestId(1)].clone(); + + if stage == FetchStage::Unanswered { + drop(fetch); + assert!( + futures::poll!(run.as_mut()).is_ready(), + "{version:?}: unanswered FETCH stays alive" + ); + assert!(matches!(outcome.read().rejected, Some(Error::Cancel))); + } else { + session.push(&message_bytes( + ietf::FetchOk::ID, + &ietf::FetchOk { + request_id: matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) + .then_some(RequestId(1)), + group_order: GroupOrder::Ascending, + end_of_track: false, + end_location: ietf::Location { + group: GROUP + 1, + object: 0, + }, + }, + version, + )); + assert!(futures::poll!(run.as_mut()).is_pending()); + let group = fetch.await.expect("accepted group"); + let observed = { + let state = slot.read(); + let GroupFetch::Ready { producer, .. } = &*state else { + panic!("accepted slot") + }; + producer.clone() + }; + if stage == FetchStage::Accepted { + drop(group); + assert!( + futures::poll!(run.as_mut()).is_ready(), + "{version:?}: accepted FETCH stays alive" + ); + } else { + let mut objects = Vec::new(); + if version == Version::Draft14 { + GROUP.encode(&mut objects, version).unwrap(); + 0u64.encode(&mut objects, version).unwrap(); // subgroup + 0u64.encode(&mut objects, version).unwrap(); // object + 0u8.encode(&mut objects, version).unwrap(); // priority + Vec::::new().encode(&mut objects, version).unwrap(); + } else { + ietf::FetchObject::Object { + subgroup: ietf::FetchSubgroup::Zero, + group: Some(GROUP), + object: Some(0), + priority: Some(0), + properties: None, + } + .encode(&mut objects, version) + .unwrap(); + } + 1u64.encode(&mut objects, version).unwrap(); + objects.push(b'x'); + let data = ScriptedSession::new(objects); + let (_, recv) = data.clone().open_bi().await.unwrap(); + let mut reader = Reader::new(recv, version); + let mut receiving = subscriber.clone(); + let mut fill = Box::pin(receiving.recv_group_fetch(&mut reader, slot.clone())); + assert!(futures::poll!(fill.as_mut()).is_pending()); + let mut group = group; + assert_eq!(group.read_frame().await.unwrap().unwrap().payload.as_ref(), b"x"); + assert!( + futures::poll!(run.as_mut()).is_pending(), + "the reader still wants the group" + ); + drop(group); + if stage == FetchStage::Complete { + data.close(crate::lite::test_transport::Close::Fin); + assert!(matches!(futures::poll!(fill.as_mut()), Poll::Ready(Ok(())))); + assert!(futures::poll!(run.as_mut()).is_ready()); + assert!(observed.is_finished()); + assert!(!observed.is_aborted()); + let mut cached = consumer.fetch_group(GROUP, None).await.expect("complete group cached"); + assert_eq!(cached.read_frame().await.unwrap().unwrap().payload.as_ref(), b"x"); + assert!(cached.read_frame().await.unwrap().is_none()); + } else { + assert!( + futures::poll!(run.as_mut()).is_ready(), + "{version:?}: partial FETCH stays alive" + ); + assert!(matches!(futures::poll!(fill.as_mut()), Poll::Ready(Err(Error::Cancel)))); + } + } + if stage != FetchStage::Complete { + assert!( + matches!(observed.poll_closed(&kio::Waiter::noop()), Poll::Ready(Error::Cancel)), + "partial group must abort" + ); + } + } + + if stage == FetchStage::Complete { + assert!(session.log.stops().is_empty()); + assert!(session.log.resets().is_empty()); + assert!(subscriber.state.lock().group_fetches.is_empty()); + return; + } + assert_eq!(session.log.stops(), [crate::ietf::error::CANCELLED]); + let messages = decode_messages(&session.log, version); + let cancels: Vec<_> = messages.iter().filter(|(id, _)| *id == ietf::FetchCancel::ID).collect(); + if matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) { + assert_eq!(cancels.len(), 1, "legacy FETCH_CANCEL"); + let mut body = cancels[0].1.clone(); + assert_eq!( + ietf::FetchCancel::decode_msg(&mut body, version).unwrap().request_id, + RequestId(1) + ); + assert!(session.log.resets().is_empty(), "deliver FETCH_CANCEL before closing"); + } else { + assert!(cancels.is_empty(), "FETCH_CANCEL was removed"); + assert_eq!(session.log.resets(), [crate::ietf::error::CANCELLED]); + } + assert!(subscriber.state.lock().group_fetches.is_empty(), "request retired"); + } + + #[tokio::test(start_paused = true)] + async fn an_abandoned_group_fetch_before_fetch_ok_is_cancelled() { + for version in JOINING_DRAFTS { + abandon_group_fetch(version, FetchStage::Unanswered).await; + } + } + + #[tokio::test(start_paused = true)] + async fn an_abandoned_group_fetch_after_fetch_ok_is_cancelled() { + for version in JOINING_DRAFTS { + abandon_group_fetch(version, FetchStage::Accepted).await; + } + } + + #[tokio::test(start_paused = true)] + async fn an_abandoned_partial_group_fetch_is_cancelled() { + for version in JOINING_DRAFTS { + abandon_group_fetch(version, FetchStage::Receiving).await; + } + } + + #[tokio::test(start_paused = true)] + async fn a_complete_group_fetch_is_cached_when_the_reader_leaves() { + for version in JOINING_DRAFTS { + abandon_group_fetch(version, FetchStage::Complete).await; + } + } + /// A publisher that resets the request after FETCH_OK owes no fetch stream, so the /// group it accepted is aborted instead of left open for every reader to wait on. /// Without that, `run_group_fetch` never returns. From cdc8dacab61188a18f7794a28f80c0c7d1edb586 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 22:49:48 -0700 Subject: [PATCH 3/4] fix(net): use the group fetch slot poll contract Co-Authored-By: GPT-6 --- rs/moq-net/src/ietf/subscriber.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index dc2c64d6ee..e35b97e559 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -3158,20 +3158,22 @@ where Poll::Pending => {} } } + let mut abandoned = false; slot.poll(waiter, |state| match &**state { - GroupFetch::Done => Poll::Ready(None), + GroupFetch::Done => Poll::Ready(()), _ => { if joined.poll_unused(waiter).is_ready() && demand.poll_unused(waiter).is_ready() && demand.abort_unused(Error::Cancel) { - Poll::Ready(Some(Error::Cancel)) + abandoned = true; + Poll::Ready(()) } else { Poll::Pending } } }) - .map(|res| res.ok().flatten()) + .map(|_| abandoned.then_some(Error::Cancel)) }) .await; if let Some(err) = reset { From b6e875c57f9edd6cfb5a84b5acb84ca81ba4cc52 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 09:28:59 -0700 Subject: [PATCH 4/4] fix(net): keep a written group cached when the FETCH request resets The abandonment change aborted the group on any request reset, including one that arrives after the fetch stream already wrote the group whole. Restore the guard so only a group still waiting for its stream is lost, and track abandonment explicitly instead of matching Error::Cancel. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/subscriber.rs | 63 +++++++++++++++++++------------ 1 file changed, 39 insertions(+), 24 deletions(-) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 6fa74a78cd..be6cfc35d3 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -3223,9 +3223,10 @@ where // Keep the request open until its data stream finishes or every reader leaves. // A publisher that fails after FETCH_OK resets the request instead and owes no fetch - // stream, so the group - // it left waiting is aborted. A FIN is not that: the fetch stream can trail it. + // stream, so the group it left waiting is aborted. A FIN is not that: the fetch + // stream can trail it. let mut open = true; + let mut abandoned = false; let reset = kio::wait(|waiter| { if open { let mut cx = std::task::Context::from_waker(waiter.waker()); @@ -3235,32 +3236,32 @@ where Poll::Pending => {} } } - let mut abandoned = false; slot.poll(waiter, |state| match &**state { GroupFetch::Done => Poll::Ready(()), - _ => { - if joined.poll_unused(waiter).is_ready() - && demand.poll_unused(waiter).is_ready() - && demand.abort_unused(Error::Cancel) - { - abandoned = true; - Poll::Ready(()) - } else { - Poll::Pending - } + _ if joined.poll_unused(waiter).is_ready() + && demand.poll_unused(waiter).is_ready() + && demand.abort_unused(Error::Cancel) => + { + abandoned = true; + Poll::Ready(()) } + _ => Poll::Pending, }) - .map(|_| abandoned.then_some(Error::Cancel)) + .map(|_| None) }) .await; - if let Some(err) = reset { - let cancel = matches!(err, Error::Cancel); - let _ = demand.abort(err); + if abandoned { drop(registered); - if cancel { - self.cancel_group_fetch(stream, fetch_id).await; - return; - } + self.cancel_group_fetch(stream, fetch_id).await; + return; + } + // A group already written stays cached; only one still waiting for its stream is lost. + if let Some(err) = reset + && let Ok(mut state) = slot.write() + && matches!(*state, GroupFetch::Ready { .. }) + && let GroupFetch::Ready { producer, .. } = std::mem::replace(&mut *state, GroupFetch::Done) + { + let _ = producer.abort(err); } let _ = stream.writer.close().await; } @@ -7915,12 +7916,15 @@ mod joining_fetch_tests { Accepted, Receiving, Complete, + /// Complete, then the publisher resets the request stream. + CompleteReset, } /// Last-reader cancellation covers a pending answer, an accepted group waiting for /// its stream, and a stream stalled after its first complete frame. async fn abandon_group_fetch(version: Version, stage: FetchStage) { const GROUP: u64 = 4; + let complete = matches!(stage, FetchStage::Complete | FetchStage::CompleteReset); let session = ScriptedSession::new(Vec::new()); let (tasks, _task_set) = TaskSet::new(); let subscriber = Subscriber::new( @@ -8028,9 +8032,12 @@ mod joining_fetch_tests { "the reader still wants the group" ); drop(group); - if stage == FetchStage::Complete { + if complete { data.close(crate::lite::test_transport::Close::Fin); assert!(matches!(futures::poll!(fill.as_mut()), Poll::Ready(Ok(())))); + if stage == FetchStage::CompleteReset { + session.close(crate::lite::test_transport::Close::Reset); + } assert!(futures::poll!(run.as_mut()).is_ready()); assert!(observed.is_finished()); assert!(!observed.is_aborted()); @@ -8045,7 +8052,7 @@ mod joining_fetch_tests { assert!(matches!(futures::poll!(fill.as_mut()), Poll::Ready(Err(Error::Cancel)))); } } - if stage != FetchStage::Complete { + if !complete { assert!( matches!(observed.poll_closed(&kio::Waiter::noop()), Poll::Ready(Error::Cancel)), "partial group must abort" @@ -8053,7 +8060,7 @@ mod joining_fetch_tests { } } - if stage == FetchStage::Complete { + if complete { assert!(session.log.stops().is_empty()); assert!(session.log.resets().is_empty()); assert!(subscriber.state.lock().group_fetches.is_empty()); @@ -8105,6 +8112,14 @@ mod joining_fetch_tests { } } + /// A request reset after the group is written must not abort the cached group. + #[tokio::test(start_paused = true)] + async fn a_complete_group_fetch_survives_a_request_reset() { + for version in JOINING_DRAFTS { + abandon_group_fetch(version, FetchStage::CompleteReset).await; + } + } + /// A publisher that resets the request after FETCH_OK owes no fetch stream, so the /// group it accepted is aborted instead of left open for every reader to wait on. /// Without that, `run_group_fetch` never returns.