diff --git a/quest/m1/README.md b/quest/m1/README.md index c68db533774..5c87864e946 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -74,7 +74,6 @@ above them. - [Draft-20 FETCH edge-case backport](/quest/m1/fetch-edge-cases.md) - backport the final-object and response-group fixes to a later 0.17 patch after main lands - [Subscribe-close backport](/quest/m1/subscribe-close-backport.md) - the same fix and the `race` loser fix on `release`, so closing a subscription during setup leaves no unhandled rejection in a later 0.17 patch - [Claim answer epochs](/quest/m1/claim-answer-epochs.md) - a claim answers each output by publishing it at its exact path under its own epoch, so a restarted output never reaches a returning reader as the old one's cache -- [Malformed group FETCH](/quest/m1/group-fetch-malformed.md) - an object missing its Timestamp in a relay's group FETCH ends the whole TIMESCALE track, as subscribe and joining FETCH already do - [FFI shape](/quest/m1/ffi-shape/README.md) - the bindings mirror Rust's layers: net at the root, then media, json, flate, audio, and video namespaces built from the handle below - [Bump web-transport-iroh for the capsule close](/quest/m1/iroh-capsule-bump.md) - condition: moq-dev/web-transport#419 ships in a release, then moq's iroh HTTP/3 client reports a peer's close capsule - [Early drops](/quest/m1/subscribe-drop-early.md) - Rust and JS publishers drop a sequence the moment they give it up, and forward an upstream drop diff --git a/quest/m1/group-fetch-malformed.md b/quest/m1/group-fetch-malformed.md deleted file mode 100644 index be769088ddf..00000000000 --- a/quest/m1/group-fetch-malformed.md +++ /dev/null @@ -1,17 +0,0 @@ -# [XS] A malformed object in a group FETCH ends its track - -## Goal - -On a TIMESCALE track, a subgroup object without a Timestamp in a standalone -group FETCH, which a relay sends to fill a cache miss (`run_group_fetch` in -`rs/moq-net/src/ietf/subscriber.rs`), ends the whole track with -`MalformedTrack`, as it already does in `recv_group` and a joining FETCH's -fill (`recv_fill`). A regression test pins it. - -## Plan - -Decided 2026-10-14 by the maintainer, from -[#4822](https://github.com/moq-dev/moq/pull/4822)'s untimed follow-ups: the -publisher broke the track however the object arrived, so failing only the -group is not enough. js/net already ends the track -([#4968](https://github.com/moq-dev/moq/pull/4968)). diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 915229459e6..530e03ce931 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -346,6 +346,7 @@ enum GroupFetch { /// Accepted into the track cache, waiting for the fetch stream to write it. Ready { producer: group::Producer, + track: kio::Producer, timescale: Option, /// The first Object ID requested, which the producer starts at. start: u64, @@ -4439,6 +4440,7 @@ where // Joined fetches still count until they pick the accepted group up from the cache. let joined = request.result.clone(); // The track was accepted before any fetch was asked, so it keeps that info. + let track = request.state.clone(); let mut producer = match request.accept(None) { Ok(producer) => producer, // Already served by a concurrent fetch, or the track closed. @@ -4458,6 +4460,7 @@ where if let Ok(mut state) = slot.write() { *state = GroupFetch::Ready { producer, + track, timescale, start, end, @@ -4573,10 +4576,11 @@ where Poll::Ready(match std::mem::replace(&mut *state, GroupFetch::Receiving) { GroupFetch::Ready { producer, + track, timescale, start, end, - } => Ok((producer, timescale, start, end)), + } => Ok((producer, track, timescale, start, end)), other => { *state = other; Err(Error::Unsupported) @@ -4588,7 +4592,7 @@ where } }) .await; - let (producer, timescale, start, end) = taken?; + let (producer, track, timescale, start, end) = taken?; producer.set_wire(group::Wire { sparse: false, first_object: start == 0, @@ -4620,6 +4624,9 @@ where Ok(()) => producer.finish(), Err(err) => { let _ = producer.abort(err.clone()); + if matches!(err, Error::MalformedTrack) { + let _ = track::TrackState::abort(&track, Error::MalformedTrack); + } Err(err) } }; @@ -12547,15 +12554,14 @@ mod stitch_tests { "timed", track::Info::default().with_timescale(Timescale::MICRO), ); - let group = track.create_group(group::Info { sequence: SEQUENCE }).unwrap(); - let slot = kio::Producer::new(GroupFetch::Ready { - producer: group, - timescale: Some(Timescale::MICRO), - start: 0, - end: None, - }); + run.track = track.clone(); + let (slot, _) = run.slot(Some(Timescale::MICRO), None).await; let res = run.subscriber.recv_group_fetch(&mut run.stream, slot).await; assert!(matches!(res, Err(Error::MalformedTrack)), "{res:?}"); + assert!(matches!( + futures::FutureExt::now_or_never(track.closed()), + Some(Error::MalformedTrack) + )); } /// FETCH_OK's End Location inside the group bounds the stream: one that runs past it @@ -12571,16 +12577,10 @@ mod stitch_tests { let payloads: Vec<&[u8]> = [b"a", b"b", b"c", b"d"][..count].iter().map(|p| &p[..]).collect(); let mut run = GroupFetchRun::new(version, group_fetch_objects(SEQUENCE, 0, &payloads)).await; - let group = run.track.create_group(group::Info { sequence: SEQUENCE }).unwrap(); - let mut consumer = group.consume(); - let slot = kio::Producer::new(GroupFetch::Ready { - producer: group, - timescale: None, - start: 0, - end: Some(END), - }); + let (slot, mut consumer) = run.slot(None, Some(END)).await; let res = run.subscriber.recv_group_fetch(&mut run.stream, slot).await; assert_eq!(res.is_ok(), complete, "{version} {count} objects: {res:?}"); + assert!(futures::FutureExt::now_or_never(run.track.closed()).is_none()); let mut read = 0; let end = loop { @@ -12719,14 +12719,7 @@ mod stitch_tests { #[moq_net_sim::test] async fn a_group_fetch_refuses_a_datagram_object() { let mut run = GroupFetchRun::new(VERSION, datagram_object(SEQUENCE)).await; - let group = run.track.create_group(group::Info { sequence: SEQUENCE }).unwrap(); - let mut consumer = group.consume(); - let slot = kio::Producer::new(GroupFetch::Ready { - producer: group, - timescale: None, - start: 0, - end: None, - }); + let (slot, mut consumer) = run.slot(None, None).await; let res = run.subscriber.recv_group_fetch(&mut run.stream, slot).await; assert!(matches!(res, Err(Error::NotFetchable)), "{res:?}"); @@ -12836,6 +12829,29 @@ mod stitch_tests { } } + /// Accept the cache miss through the same request state as a wire FETCH. + async fn slot( + &self, + timescale: Option, + end: Option, + ) -> (kio::Producer, group::Consumer) { + let dynamic = self.track.dynamic(); + let mut fetch = Box::pin(self.track.consume().fetch_group(SEQUENCE, None)); + assert!(futures::poll!(fetch.as_mut()).is_pending()); + let request = dynamic.requested_group().await.expect("group request"); + let track = request.state.clone(); + let producer = request.accept(None).unwrap(); + let consumer = producer.consume(); + let slot = kio::Producer::new(GroupFetch::Ready { + producer, + track, + timescale, + start: 0, + end, + }); + (slot, consumer) + } + /// Decode the stream into a fresh group numbered from `start`. async fn recv_objects(&mut self, start: u64, end: Option) -> Result<(), Error> { let mut group = self.track.create_group(group::Info { sequence: SEQUENCE }).unwrap(); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 522c42b2375..873ca6b7ccc 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1702,6 +1702,12 @@ impl TrackState { producer.write().map_err(|r| r.abort.clone().unwrap_or(Error::Dropped)) } + /// Abort shared track state without extending its publisher lifetime. + pub(crate) fn abort(producer: &kio::Producer, err: Error) -> Result<()> { + commit_abort(Self::modify(producer)?, err); + Ok(()) + } + /// Insert a group fetched for a [`group::Request`], setting the track's [`Info`] /// if it isn't accepted yet. The group's timescale comes from that info, so a /// fetch can serve an as-yet-unaccepted track (e.g. a relay with no live @@ -2257,8 +2263,7 @@ impl Producer { /// [`finish`](Self::finish) is deliberately not terminal: it declares the final /// sequence, and lower-numbered groups may still be written afterwards. pub fn abort(self, err: Error) -> Result<()> { - commit_abort(self.modify()?, err); - Ok(()) + TrackState::abort(&self.state, err) } /// Abort an unused track, returning the producer unchanged if consumers remain.