fix(moq-net): close accepted group request demand cleanly

go/wrapper/publish.go

@@ -409,9 +409,8 @@ func (r *GroupRequest) Abort(errorCode uint16) error {
 //
 // It is weak: holding it does not keep the request alive. The last caller to
 // leave withdraws the request, so once unused, demand never returns: drop the
-// request. Waits fail once the request is answered: ErrClosed if it was
-// dropped, otherwise the error the accept or reject left for the waiting
-// fetches.
+// request. Waits fail with ErrClosed after the request is accepted or dropped,
+// and keep the rejection reason after it is rejected.
 type GroupDemand struct {
 	inner *ffi.MoqGroupDemand
 }

py/moq-rs/moq/publish.py

@@ -204,9 +204,9 @@ class GroupDemand:
 
     Returned by ``GroupRequest.demand()``. Weak: holding it does not keep the
     request alive. The last caller to leave withdraws the request, so once
-    unused, demand never returns: drop the request. Waits raise once the
-    request is answered: ``moq.Error.Closed`` if it was dropped, otherwise the
-    error the accept or reject left for the waiting fetches.
+    unused, demand never returns: drop the request. Waits raise
+    ``moq.Error.Closed`` after the request is accepted or dropped, and keep
+    the rejection reason after it is rejected.
     """
 
     def __init__(self, inner: MoqGroupDemand) -> None:

quest/m1/README.md

@@ -168,7 +168,6 @@ above them.
 - [Refuse rewinds](/quest/m1/refuse-rewinds.md) - a timed track refuses a group whose first frame is earlier than the previous group's, in Rust and js/net, on every hop
 - [Feed fetch index](/quest/m1/feed-fetch-index.md) - held feed fetches wake per group and are capped per peer session, on main then release
 - [JS pending tail](/quest/m1/js-pending-tail.md) - a JS reader holds for a track's pending tail like Rust
-- [Group demand after accept](/quest/m1/group-demand-accept.md) - an accepted group request's demand ends cleanly instead of failing with NotFound
 - [WebGPU on Safari](/quest/m1/webgpu-safari.md) - the WebGPU renderer is verified on Safari 26 for macOS and iOS
 - [A/V sync across a break](/quest/m1/watch-break-av-sync.md) - `@moq/watch` never plays pre-break audio out of sync with pre-break video
 - [moqsrc reconnect](/quest/m1/moqsrc-reconnect.md) - `moqsrc` redials after losing its relay and resumes on the same pads

rs/moq-ffi/src/demand.rs

@@ -49,8 +49,8 @@ impl MoqTrackDemand {
 ///
 /// Weak: holding it does not keep the request alive. The last caller to leave withdraws the
 /// request, and a later fetch of the group queues a fresh one, so once unused, demand never
-/// returns: drop the request. Waits fail once the request is answered: with `Closed` if it was
-/// dropped, otherwise with the error the accept or reject left for the waiting fetches.
+/// returns: drop the request. Waits fail with `Closed` after the request is accepted or dropped,
+/// and keep the rejection reason after it is rejected.
 #[derive(uniffi::Object)]
 pub struct MoqGroupDemand {
 	inner: moq_net::group::Demand,

rs/moq-ffi/src/test.rs

@@ -1659,33 +1659,26 @@ async fn group_request_demand_ends_when_the_fetcher_leaves() {
 	assert!(matches!(request.demand(), Err(MoqError::Closed)));
 }
 
-/// An accepted request's demand ends with the `NotFound` the accept leaves for joined fetches
-/// it can't cover, not `Closed`.
+/// An accepted request's demand closes cleanly while its fetch gets the group.
 #[tokio::test]
-async fn group_request_demand_fails_after_accept() {
+async fn group_request_demand_closes_after_accept() {
+	tokio::time::pause();
 	let broadcast = MoqBroadcastProducer::new().unwrap();
 	let track = broadcast.publish_track("events".into(), None).unwrap();
 	let dynamic = track.dynamic().unwrap();
 	let consumer = broadcast.consume().unwrap();
 
 	let fetch = tokio::spawn(async move { consumer.fetch_group("events".into(), 5, None).await });
-	let request = tokio::time::timeout(TIMEOUT, dynamic.requested_group())
-		.await
-		.expect("timed out waiting for group request")
-		.unwrap();
+	let request = dynamic.requested_group().await.unwrap();
 	let demand = request.demand().unwrap();
 	let group = request.accept().unwrap();
 	group.finish().unwrap();
 
-	let used = tokio::time::timeout(TIMEOUT, demand.used())
-		.await
-		.expect("timed out waiting for an accepted demand to end");
-	assert!(matches!(used, Err(MoqError::NotFound)), "got {used:?}");
-	tokio::time::timeout(TIMEOUT, fetch)
-		.await
-		.expect("timed out waiting for fetch")
-		.expect("fetch task panicked")
-		.unwrap();
+	for result in [demand.used().await, demand.unused().await] {
+		assert!(matches!(result, Err(MoqError::Closed)), "got {result:?}");
+	}
+	assert!(!demand.is_used());
+	fetch.await.expect("fetch task panicked").unwrap();
 }
 
 #[tokio::test]

rs/moq-net/src/model/group.rs

@@ -1104,6 +1104,9 @@ impl Producer {
 }
 
 /// A cloneable, watch-only handle to a group's readers or pending fetch callers.
+///
+/// An accepted or dropped request ends waits with [`Error::Dropped`]; a rejected request keeps
+/// its rejection reason.
 #[derive(Clone)]
 pub struct Demand {
 	sequence: u64,

rs/moq-net/src/model/track.rs

@@ -486,18 +486,19 @@ pub(crate) struct PendingFetch {
 	// has taken it, its range is already on the wire.
 	frame_start: u64,
 
-	// Result channel back to the joined fetches. Written only on rejection; a
-	// successful accept resolves them through the track cache instead. Dropping
-	// every producer without writing (a vanished handler) closes the channel,
-	// which a [`Fetching`] reads as [`Error::NotFound`].
+	// Result channel for joined fetches. An accept resolves covered callers through
+	// the track cache and refuses uncovered ones. Dropping every producer closes
+	// an unserved attempt, which a [`Fetching`] reads as [`Error::NotFound`].
 	result: kio::Producer<FetchOutcome>,
 }
 
-/// The result of a fetch attempt. Stays empty on success (the group lands in the
-/// track cache); a handler writes `rejected` to fail every joined fetch.
+/// Whether a fetch attempt was accepted, or why it was rejected.
 #[derive(Default)]
 pub(crate) struct FetchOutcome {
 	pub(crate) rejected: Option<Error>,
+	// An accept still refuses joined fetches the cached group cannot cover, without
+	// giving the request's demand watcher a rejection reason.
+	accepted: bool,
 }
 
 impl TrackState {
@@ -3443,7 +3444,7 @@ impl group::Request {
 		// 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);
+			outcome.accepted = true;
 		}
 		res
 	}
@@ -3626,10 +3627,11 @@ impl kio::Task for Fetching {
 			};
 		};
 
-		// A written rejection fails every joined fetch. The channel closing without
-		// one means the attempt was dropped unserved (its handlers went away).
+		// An answered attempt fails joined fetches the cache cannot cover. The channel
+		// closing without an answer means its handlers went away.
 		let err = match result.outcome.poll(waiter, |outcome| match &outcome.rejected {
 			Some(err) => Poll::Ready(err.clone()),
+			None if outcome.accepted => Poll::Ready(Error::NotFound),
 			None => Poll::Pending,
 		}) {
 			Poll::Ready(Ok(err)) => err,
@@ -10617,6 +10619,75 @@ mod test {
 		assert!(matches!(demand.poll_unused(&kio::Waiter::noop()), Poll::Ready(Ok(()))));
 		request.reject(Error::Cancel);
 		assert!(matches!(demand.used().now_or_never(), Some(Err(Error::Cancel))));
+
+		// The async close helper reports clean acceptance separately from rejection.
+		for accepted in [true, false] {
+			let producer = track_producer("test", Info::timed());
+			let dynamic = producer.dynamic();
+			let consumer = producer.consume();
+			let fetch = consumer.fetch_group(3, None);
+			let request = dynamic.requested_group().now_or_never().unwrap().unwrap();
+			let demand = request.demand();
+			let group = if accepted {
+				Some(request.accept(Info::timed()).unwrap())
+			} else {
+				request.reject(Error::NotFound);
+				None
+			};
+			assert!(matches!(
+				(accepted, demand.closed().now_or_never()),
+				(true, Some(Error::Dropped)) | (false, Some(Error::NotFound))
+			));
+			drop(fetch);
+			if let Some(group) = group {
+				group.finish().unwrap();
+			}
+		}
+	}
+
+	#[test]
+	fn accepted_group_request_demand_ends_cleanly() {
+		let producer = track_producer("test", Info::timed());
+		let dynamic = producer.dynamic();
+		let consumer = producer.consume();
+		let mut fetch = consumer.fetch_group(3, None);
+		let waiter = kio::Waiter::noop();
+		let Poll::Ready(Ok(request)) = dynamic.poll_requested_group(&waiter) else {
+			panic!("the fetch request is ready");
+		};
+		let demand = request.demand();
+		let group = request.accept(Info::timed()).unwrap();
+
+		assert!(!demand.is_used());
+		assert!(matches!(demand.poll_used(&waiter), Poll::Ready(Err(Error::Dropped))));
+		assert!(matches!(demand.poll_unused(&waiter), Poll::Ready(Err(Error::Dropped))));
+		assert!(
+			matches!(kio::Task::poll(&mut *fetch, &waiter), Poll::Ready(Ok(_))),
+			"the fetch still gets the accepted group"
+		);
+		group.finish().unwrap();
+	}
+
+	#[test]
+	fn rejected_group_request_demand_keeps_its_reason() {
+		let producer = track_producer("test", Info::timed());
+		let dynamic = producer.dynamic();
+		let consumer = producer.consume();
+		let mut fetch = consumer.fetch_group(3, None);
+		let waiter = kio::Waiter::noop();
+		let Poll::Ready(Ok(request)) = dynamic.poll_requested_group(&waiter) else {
+			panic!("the fetch request is ready");
+		};
+		let demand = request.demand();
+		request.reject(Error::NotFound);
+
+		assert!(!demand.is_used());
+		assert!(matches!(demand.poll_used(&waiter), Poll::Ready(Err(Error::NotFound))));
+		assert!(matches!(demand.poll_unused(&waiter), Poll::Ready(Err(Error::NotFound))));
+		assert!(matches!(
+			kio::Task::poll(&mut *fetch, &waiter),
+			Poll::Ready(Err(Error::NotFound))
+		));
 	}
 
 	#[moq_net_sim::test]

swift/Sources/Moq/Dynamic.swift

@@ -41,8 +41,8 @@ public final class TrackRequest: Sendable {
 ///
 /// Returned by `GroupRequest.demand()`. Weak: holding it does not keep the request alive. The last
 /// caller to leave withdraws the request, so once unused, demand never returns: drop the request.
-/// Waits throw once the request is answered: `MoqError.Closed` if it was dropped, otherwise the
-/// error the accept or reject left for the waiting fetches.
+/// Waits throw `MoqError.Closed` after the request is accepted or dropped, and keep the rejection
+/// reason after it is rejected.
 public final class GroupDemand: Sendable {
     let ffi: MoqGroupDemand