fix(moq-c): refuse only the readers waiting on a group request

quest/m1/c/README.md

@@ -69,4 +69,3 @@ requests internally and keeps its ABI.
 - [moq-c package](/quest/m1/c/package.md) - the generated header ships as `moq-c` 0.8.0 with `moq::c`, pkg-config, and a release workflow
 - [C consumers](/quest/m1/c/consumers.md) - the C interop client and `doc/lib/c` samples move onto the generated API
 - [Async failure contract](/quest/m1/c/async-failure-contract.md) - a generated async call that fails early fires no callback, and one that started always delivers its terminal callback
-- [Refuse only waiting readers](/quest/m1/c/request-refuse.md) - freeing or aborting a group request refuses its readers instead of dropping the group for everyone

rs/moq-c/src/api.rs

@@ -3146,10 +3146,10 @@ pub extern "C" fn moq_group_request_accept(request: u32) -> i32 {
 	})
 }
 
-/// Reject a group request, failing the waiting fetches as not found.
+/// Reject a group request, failing the readers waiting on it now as not found.
 ///
-/// The error code is not delivered. Consumes the request handle. Returns a zero on success,
-/// or a negative code on failure.
+/// A later reader of the group asks again. The error code is not delivered. Consumes the
+/// request handle. Returns a zero on success, or a negative code on failure.
 #[unsafe(no_mangle)]
 pub extern "C" fn moq_group_request_abort(request: u32, error_code: u16) -> i32 {
 	ffi::enter(move || {
@@ -3158,7 +3158,7 @@ pub extern "C" fn moq_group_request_abort(request: u32, error_code: u16) -> i32
 	})
 }
 
-/// Free a group request without accepting it, which rejects it.
+/// Free a group request without accepting it, which rejects it like [moq_group_request_abort].
 ///
 /// Returns a zero on success, or a negative code if the handle is unknown.
 #[unsafe(no_mangle)]

rs/moq-c/src/publish.rs

@@ -871,16 +871,18 @@ impl Publish {
 		self.groups.insert(group)
 	}
 
-	/// Reject a group request, failing every fetch waiting on it. A fill has no error code
-	/// to carry, so `error_code` is not delivered: the fetches fail as not found.
+	/// Reject a group request like [`Self::group_request_free`]; `error_code` is not delivered.
 	pub fn group_request_abort(&mut self, request: Id, _error_code: u16) -> Result<(), Error> {
 		self.group_request_free(request)
 	}
 
-	/// Drop a group request, which rejects it.
+	/// Drop a group request, failing the readers waiting on it as not found. The group is
+	/// refused rather than dropped, so a later reader asks again.
 	pub fn group_request_free(&mut self, request: Id) -> Result<(), Error> {
 		let request = self.group_request.remove(request).ok_or(Error::NotFound)?;
-		request.fill.drop_groups(request.sequence..request.sequence + 1);
+		request
+			.fill
+			.refuse_groups(request.sequence..request.sequence + 1, moq_net::Error::NotFound);
 		Ok(())
 	}
 

rs/moq-c/src/test.rs

@@ -3380,6 +3380,76 @@ fn a_group_request_nobody_reads_is_withdrawn() {
 	assert_eq!(moq_origin_close(origin), 0);
 }
 
+/// Rejecting a group request fails only the readers waiting on it: the group is not named
+/// dropped, so a later subscriber asks again.
+#[test]
+fn a_rejected_group_request_binds_only_its_readers() {
+	let origin = id(moq_origin_create());
+	let broadcast = publish_broadcast(origin, b"track-dynamic-refused");
+	let name = b"data";
+	let track =
+		id(unsafe { moq_publish_track(broadcast, name.as_ptr() as *const c_char, name.len(), std::ptr::null()) });
+	let group_cb = Callback::new();
+	let dynamic = id(unsafe { moq_publish_track_dynamic(track, Some(channel_callback), group_cb.ptr) });
+
+	// The live feed ran from 10 and ended, so only a fill can serve group 6.
+	{
+		let mut state = State::lock();
+		let producer = state.publish.track_producer(Id::try_from(track).unwrap()).unwrap();
+		producer.start_at(10).unwrap();
+		let mut group = producer.create_group(moq_net::group::Info { sequence: 10 }).unwrap();
+		group
+			.write_frame(moq_net::Timestamp::from_millis(10).unwrap(), &b"live"[..])
+			.unwrap();
+		group.finish().unwrap();
+		producer.finish().unwrap();
+	}
+	let subscribe = || {
+		let subscription = moq_net::track::Subscription::default()
+			.with_max_delay(None)
+			.with_range(6..7);
+		let mut state = State::lock();
+		let producer = state.publish.track_producer(Id::try_from(track).unwrap()).unwrap();
+		producer.subscribe(subscription)
+	};
+	let waiter = kio::Waiter::noop();
+
+	let mut sequence = 0u64;
+	for reject in [
+		|request| moq_group_request_abort(request, 9),
+		|request| moq_group_request_free(request),
+	] {
+		let mut reader = subscribe();
+		assert!(reader.poll_recv_group(&waiter).is_pending());
+		let request = id(group_cb.recv());
+		assert_eq!(unsafe { moq_group_request_sequence(request, &mut sequence) }, 0);
+		assert_eq!(sequence, 6);
+		assert_eq!(reject(request), 0);
+	}
+
+	let mut reader = subscribe();
+	assert!(reader.poll_recv_group(&waiter).is_pending());
+	let request = id(group_cb.recv());
+	let group = id(moq_group_request_accept(request));
+	let payload = b"fetched";
+	assert_eq!(
+		unsafe { moq_publish_group_frame(group, payload.as_ptr(), payload.len(), 6_000) },
+		0
+	);
+	assert_eq!(moq_publish_group_finish(group), 0);
+	let std::task::Poll::Ready(Ok(Some(group))) = reader.poll_recv_group(&waiter) else {
+		panic!("the accepted group reaches the reader");
+	};
+	assert_eq!(group.sequence, 6);
+
+	drop(reader);
+	assert_eq!(moq_publish_dynamic_cancel(dynamic), 0);
+	assert_eq!(group_cb.recv_terminal(), 0);
+	assert_eq!(moq_publish_track_finish(track), 0);
+	assert_eq!(moq_publish_close(broadcast), 0);
+	assert_eq!(moq_origin_close(origin), 0);
+}
+
 #[test]
 fn track_request_dynamic_survives_accept() {
 	let origin = id(moq_origin_create());