diff --git a/quest/m1/README.md b/quest/m1/README.md index 3ab218d653d..a9ed35f1c1f 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -161,7 +161,6 @@ per-frame CPU growth since #5926. - [A moq-transport request token is one auth per distinct token](/quest/m1/ietf-token-auth.md) - requests naming the same token share one verification and grant, on any request type, ending with the last such request - [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 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 - [Native rung disable](/quest/m1/simulcast-rung-disable.md) - OBS stops sending the top renditions its grant cannot fund, as the Rust/FFI publishers do, and a shaped uplink through moq-ffi proves the resume - [Audio jitter target](/quest/m1/audio-jitter-target/README.md) - the playout target is default in both languages; a manual browser proof and a native trace replay remain - [A/V clock](/quest/m1/av-clock.md) - the audio playhead drives Sync.reference while audio plays, through per-track sync handles 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 79cca9bdd1f..86353dc37f0 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -276,7 +276,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. @@ -298,7 +299,7 @@ enum Fill { /// the join to the next group boundary. Failed { sequence: u64 }, - /// 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. @@ -373,7 +374,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) { @@ -382,8 +383,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. /// /// [`Fill::Failed`] stays: a joining FETCH is not in PUBLISH_DONE's Stream Count, so its /// head can land after the end, and the tail it owed may already have been dropped. @@ -3865,10 +3869,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 @@ -15694,6 +15703,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. + #[kio::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,