From f422a12e5fb9b643c5726dbd5bf99d9ed57817ec Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 30 Oct 2026 06:29:57 -0700 Subject: [PATCH 1/2] chore(quest): claim quest/m1/ietf-fill-end-cancel.md Quest-Claim: 534eabcf27b132b4 From 95083ca83e935e93dd4f58e2216319f25bd34556 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 30 Oct 2026 06:35:44 -0700 Subject: [PATCH 2/2] fix(ietf): cancel an unclaimed fill head when a subscription ends Only TRACK_ENDED makes an unclaimed head its whole group. Any other end, such as draft-18 SUBSCRIPTION_ENDED, leaves the group open upstream, so the Rust subscriber now aborts the head instead of finishing it, as the JS subscriber does since #7089. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/ietf-fill-end-cancel.md | 21 ------ rs/moq-net/src/ietf/subscriber.rs | 107 +++++++++++++++++++++++++++--- 3 files changed, 99 insertions(+), 30 deletions(-) delete mode 100644 quest/m1/ietf-fill-end-cancel.md diff --git a/quest/m1/README.md b/quest/m1/README.md index fe863117b99..5ce62a8c0ef 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -153,7 +153,6 @@ removed). - [moq-quic's API improves on quinn's](/quest/m1/quic-api/README.md) - moq-quic and both QUIC drivers ship one mirrored API: `send::`/`recv::` namespaces, one error per direction, consuming terminal operations, plain configs, no quinn baggage - [On-demand JavaScript FETCH](/quest/m1/js-fetch-on-demand.md) - browser publishers answer IETF FETCH for evicted or uncached groups through the JS ranges request surface - [Rust FETCH parity](/quest/m1/ietf-fetch-parity.md) - Rust finishes an answered joining FETCH's head, refuses a lost head with DOES_NOT_EXIST, and sends no timestamps without TIMESCALE, as the JS publisher does -- [Rust cancels an unclaimed fill head when a subscription ends](/quest/m1/ietf-fill-end-cancel.md) - SUBSCRIPTION_ENDED cancels, not finishes, a fill no group was sent for, as in JS - [SUBSCRIBE honors the properties opt-out](/quest/m1/subscribe-timestamps.md) - the Rust publisher sends untimed objects to a SUBSCRIBE that declines Track Properties, and the Rust subscriber asks for them when it needs timestamps - [A resumed untimed copy keeps DYNAMIC_GROUPS](/quest/m1/ietf-dynamic-groups-resume.md) - opting out of properties on a draft-22 resume no longer forgets DYNAMIC_GROUPS - [Rust marks a lost listed group unknown](/quest/m1/ietf-publisher-unknown-range.md) - the Rust IETF publisher sends End of Unknown Range for a listed group it can't serve, after the publisher's open quests land diff --git a/quest/m1/ietf-fill-end-cancel.md b/quest/m1/ietf-fill-end-cancel.md deleted file mode 100644 index 9f4be21be03..00000000000 --- a/quest/m1/ietf-fill-end-cancel.md +++ /dev/null @@ -1,21 +0,0 @@ -# [S] Rust cancels an unclaimed fill head when a subscription ends - -## Goal - -When SUBSCRIPTION_ENDED arrives while a fill head is still unclaimed, Rust's -moq-transport subscriber cancels that fill instead of finishing it, as the JS -subscriber does since [moq-dev/moq#7089](https://github.com/moq-dev/moq/pull/7089), -so a reader never sees a clean end for a group it was never sent. - -## Plan - -Planned 2026-10-30 (maintainer, `/quest-plan`). The finish is `Fill::end` in -`rs/moq-net/src/ietf/subscriber.rs`. Add a test that ends a subscription with -an unclaimed fill head and expects the cancel, in both languages' terms. - -[IETF fill refill](/quest/m2/ietf-fill-refill.md) would later remove this -fill handoff entirely; this fix is for the release cut, not a reason to wait. - -## Related - -- [IETF fill refill](/quest/m2/ietf-fill-refill.md) - replaces the fill claim and handoff this touches diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index f70b2ec62b2..fb6dc55d4c4 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -274,7 +274,8 @@ enum Fill { /// The tail is what ends the group, and a publisher serving the subscription's range /// opens a stream for it even when the group ended at the join point, since that empty /// stream is how the group ends. One that opens none instead leaves this head unfinished - /// until the publisher ends the subscription, which is what publishes it. + /// until the publisher ends the subscription: TRACK_ENDED publishes it, and any other + /// end cancels it. /// /// Nothing shorter is safe to infer. A later group arriving looks like proof that no /// tail is coming, but streams are independent: the tail's own can still be behind it. @@ -290,7 +291,7 @@ enum Fill { /// and gets dropped, which degrades the join to the next group boundary. Done, - /// The publisher ended the subscription cleanly, so no tail is coming. Unlike + /// The publisher ended the track, so no tail is coming. Unlike /// [`Fill::Done`], a head still being written is the whole group once it lands: a /// draft-18 joining FETCH is not in PUBLISH_DONE's Stream Count, so the settle can /// end before it does. @@ -349,7 +350,7 @@ impl Fill { /// Release a head nothing claimed, publishing the objects it did carry. /// /// The tail is what normally ends the group, so this is the fallback for when none is - /// coming: the publisher ended the subscription, or the tail that arrived could not be + /// coming: the publisher ended the track, or the tail that arrived could not be /// stitched. Finishing rather than aborting, because the head is a valid prefix of the /// group: it starts at the group's first object and has no holes. fn release(&mut self) { @@ -358,8 +359,11 @@ impl Fill { } } - /// The publisher ended the subscription cleanly: release the head, and any still being - /// written when it lands. + /// The publisher ended the track: release the head, and any still being written when it + /// lands. + /// + /// Only TRACK_ENDED, since any other end, such as SUBSCRIPTION_ENDED, leaves the group + /// open upstream and so [`Self::abort`]s the head instead. fn end(&mut self) { self.release(); *self = Fill::Ended; @@ -3818,10 +3822,15 @@ impl Subscriber { poll_settled(&mut settle, waiter, &fill, count) }) .await; - // The publisher ended without a tail for the head, so the head - // is the whole group, even one that lands after this. + // No tail is coming for the head. Only TRACK_ENDED makes it the + // whole group, even one that lands after this. Any other end + // leaves the group open upstream, so the head is cut short + // rather than passed off as the whole group. if let Ok(mut fill) = fill.write() { - fill.end(); + match &end { + Ok(()) => fill.end(), + Err(err) => fill.abort(err.clone()), + } } } // Only TRACK_ENDED ends the track. Any other status ends just this @@ -15458,6 +15467,88 @@ mod joining_fetch_tests { } } + /// A head nothing claimed when the publisher ends the subscription is the whole group only + /// on TRACK_ENDED. SUBSCRIPTION_ENDED leaves the group open upstream, so finishing the head + /// would pass a prefix off as the whole group: it is cancelled instead. + #[moq_clock::test] + async fn an_unclaimed_head_finishes_only_when_the_track_ends() { + let largest = Some(ietf::Location { group: 4, object: 0 }); + for version in JOINING_DRAFTS { + for (status_code, whole) in [(ietf::PublishDoneStatus::TrackEnded.code(), true), (0x3, false)] { + let mut ok = subscribe_ok(version, largest); + ok.extend(message_bytes( + ietf::PublishDone::ID, + &ietf::PublishDone { + status_code, + stream_count: 0, + reason_phrase: "done".into(), + }, + version, + )); + let mut run = JoinRun::start(version, None, ok, fetch_ok(version)).await; + let fetch_id = *run + .subscriber + .state + .lock() + .fetches + .keys() + .next() + .expect("joining FETCH"); + + let mut objects = Vec::new(); + let w = &mut Encoder::new(&mut objects, version.into()); + w.varint(ietf::FetchHeader::TYPE).unwrap(); + ietf::FetchHeader { request_id: fetch_id }.encode(w, version).unwrap(); + ietf::FetchObject::Object { + subgroup: ietf::FetchSubgroup::Zero, + group: Some(4), + object: Some(0), + priority: Some(0), + properties: None, + } + .encode(w, version) + .unwrap(); + w.bytes(b"x").unwrap(); + let data = ScriptedSession::new(objects); + data.close(crate::lite::test_transport::Close::Fin); + let (_, recv) = data.clone().open_bi(Default::default()).await.unwrap(); + let mut reader = Reader::new(crate::transport::boxed::RecvStream::new(recv), version); + + // The whole head lands before the grace runs out, and no tail ever claims it. + run.subscriber.clone().recv_fill(&mut reader).await.expect("fill"); + assert!( + matches!( + *run.subscriber + .state + .lock() + .subscribes + .values() + .next() + .unwrap() + .fill + .read(), + Fill::Ready { .. } + ), + "{version}/{status_code}: the head waits for its tail" + ); + + let mut subscriber = (&mut run._hold.2).await.expect("subscribed"); + let mut group = subscriber.recv_group().await.unwrap().unwrap(); + assert_eq!(group.sequence, 4, "{version}/{status_code}"); + assert_eq!(group.read_frame().await.unwrap().unwrap().payload.as_ref(), b"x"); + + while !run.subscriber.state.lock().subscribes.is_empty() { + settle().await; + } + let end = group.read_frame().await; + match whole { + true => assert!(matches!(end, Ok(None)), "{version}: the head was cut short: {end:?}"), + false => assert!(end.is_err(), "{version}: the head read as the whole group: {end:?}"), + } + } + } + } + /// SUBSCRIPTION_ENDED ends only our subscription, not the track, so the copy is released /// for a later reader to subscribe afresh rather than lingering as if the track ended. Its /// reader sees an error, not a clean end: the fresh copy's groups continue past this one's,