From 40cbf23262be0001bf72c0412e322373acac42f7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 22:39:44 -0700 Subject: [PATCH 1/2] chore(quest): claim quest/m1/c/request-refuse.md Quest-Claim: df4e10e3bd2f40dc From 6eeb14e5fa06f238069437faed2c15ac43165296 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 22:47:39 -0700 Subject: [PATCH 2/2] fix(moq-c): refuse only the readers waiting on a group request moq_group_request_free and moq_group_request_abort called Fill::drop_groups, recording the group as dropped for every later reader. They now call Fill::refuse_groups, so only the readers waiting now fail as not found and a later reader asks again, as Rust's fill paths do since #6171. Co-Authored-By: Claude Opus 5.5 --- quest/m1/c/README.md | 1 - quest/m1/c/request-refuse.md | 18 ---------- rs/moq-c/src/api.rs | 8 ++--- rs/moq-c/src/publish.rs | 10 +++--- rs/moq-c/src/test.rs | 70 ++++++++++++++++++++++++++++++++++++ 5 files changed, 80 insertions(+), 27 deletions(-) delete mode 100644 quest/m1/c/request-refuse.md diff --git a/quest/m1/c/README.md b/quest/m1/c/README.md index 00986136419..c1dd070c070 100644 --- a/quest/m1/c/README.md +++ b/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 diff --git a/quest/m1/c/request-refuse.md b/quest/m1/c/request-refuse.md deleted file mode 100644 index f625abb9561..00000000000 --- a/quest/m1/c/request-refuse.md +++ /dev/null @@ -1,18 +0,0 @@ -# [S] moq-c refuses only the waiting readers - -## Goal - -`moq_group_request_free` and `moq_group_request_abort` refuse the group to -the readers waiting on that request, as Rust's `Fill::refuse_groups` does, -instead of marking it dropped for every later reader. - -## Plan - -[moq-dev/moq#6171](https://github.com/moq-dev/moq/pull/6171) added -`Fill::refuse_groups` and moved Rust's fill paths onto it; moq-c's -`group_request_free` (`rs/moq-c/src/publish.rs:893`) still calls -`drop_groups`. The C signatures stay; only the behavior changes, so note it -in the PR's Impact. If the generated C API replaces the hand-written group -request calls first, delete this quest. - -Public API: moq-c behavior change, no signature change. Wire: none. diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index cd9fdd6f263..2ba5404dcb9 100644 --- a/rs/moq-c/src/api.rs +++ b/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)] diff --git a/rs/moq-c/src/publish.rs b/rs/moq-c/src/publish.rs index aac4d2e2c0e..dceb3a406b0 100644 --- a/rs/moq-c/src/publish.rs +++ b/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(()) } diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 0eec252a148..f36680f2b60 100644 --- a/rs/moq-c/src/test.rs +++ b/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());