fix(net): cancel an abandoned lite FETCH upstream, before or after the answer

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();
+});

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;

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

rs/moq-net/src/lite/subscriber.rs

@@ -4133,12 +4133,25 @@ enum FetchRunState<S: crate::transport::poll::Session> {
 	Ingest {
 		stream: Stream<S, Version>,
 		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<track::FetchOutcome>,
 		// Boxed so the other states stay small; one allocation per fetch.
 		ingest: Box<FrameIngest>,
 	},
 	Done,
 }
 
+impl<S: crate::transport::poll::Session> FetchRunState<S> {
+	/// 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<S: crate::transport::poll::Session> FetchServeRun<S> {
 	fn new(serve: TrackServe<S>, request: group::Request, timescale: Option<Timescale>) -> Self {
 		let session = serve.subscriber.session.clone();
@@ -4157,6 +4170,20 @@ impl<S: crate::transport::poll::Session> kio::Task for FetchServeRun<S> {
 	fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> {
 		let mut cx = std::task::Context::from_waker(waiter.waker());
 		loop {
+			// 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");
+				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");
@@ -4278,6 +4305,9 @@ impl<S: crate::transport::poll::Session> kio::Task for FetchServeRun<S> {
 					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) => {
@@ -4299,15 +4329,36 @@ impl<S: crate::transport::poll::Session> kio::Task for FetchServeRun<S> {
 					self.state = FetchRunState::Ingest {
 						stream,
 						producer,
+						joined,
 						ingest: Box::new(FrameIngest::new(&self.serve.subscriber, self.timescale)),
 					};
 				}
 				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 {

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
@@ -913,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<Consumer> {
+		self.state.weak().try_consume().map(|state| self.consumer(state))
+	}
+
+	fn consumer(&self, state: kio::Consumer<GroupState>) -> 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(),
@@ -970,6 +998,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)
@@ -1969,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();