diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 9955e9068..0324f5847 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -2392,6 +2392,98 @@ mod serve_group_test { "rank 0 must reach the transport as send order 255: {priorities:?}", ); } + + /// A subscriber that stops reading must not pin the group it was being served. + /// + /// The publisher stamps the group's cache access once per frame, immediately + /// before writing it, so a delivery in progress gets a full `latency_max` of + /// grace per frame handed out. Nothing re-stamps inside the write itself: a peer + /// whose flow control window stays shut for longer than the whole retention + /// window lets the group expire mid-stream and the stream resets with `Old`. + /// That is the point. Holding the group for as long as a wedged peer refuses to + /// read would let any subscriber pin cache indefinitely. + #[tokio::test(start_paused = true)] + async fn stalled_write_releases_the_group() { + let gate = kio::Producer::new(true); + let session = SinkSession::gated_uni(gate.consume()); + let log = session.log.clone(); + + let track_priority = kio::Producer::new(0u8); + let subscription = Subscription { + session, + id: 0, + track_name: "test".into(), + priority: PriorityQueue::default(), + track_priority: track_priority.consume(), + track_priority_seen: 0, + version: Version::Lite06Wip, + timescale: Some(crate::Timescale::default()), + }; + + let mut track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); + let mut group = track.create_group(group::Info { sequence: 0 }).unwrap(); + group + .write_frame(Timestamp::from_millis(0).unwrap(), b"first".as_slice()) + .unwrap(); + + // The live edge moves on, so the served group is demoted and expirable. + track + .create_group(group::Info { sequence: 1 }) + .unwrap() + .finish() + .unwrap(); + + let handle = subscription.priority.insert(Priority::new(0, 0)); + let mut serve = std::pin::pin!(subscription.serve_group(0, handle, group.consume())); + + // Write the header and the first frame, leaving the task awaiting the next. + assert!(futures::poll!(serve.as_mut()).is_pending()); + + // From here every write blocks, the way a shut flow control window does. + *gate.write().ok().expect("gate open") = false; + + group + .write_frame(Timestamp::from_millis(10).unwrap(), b"second".as_slice()) + .unwrap(); + group + .write_frame(Timestamp::from_millis(20).unwrap(), b"third".as_slice()) + .unwrap(); + group.finish().unwrap(); + + // The publisher takes "second" (stamping the group) and blocks writing it. + assert!(futures::poll!(serve.as_mut()).is_pending()); + + // The write stays blocked well past the retention window while the source + // keeps publishing, which is what runs the expiry scan. + for sequence in 2..8u64 { + tokio::time::advance(crate::track::DEFAULT_LATENCY_MAX / 2).await; + track.create_group(group::Info { sequence }).unwrap().finish().unwrap(); + assert!(futures::poll!(serve.as_mut()).is_pending()); + } + + *gate.write().ok().expect("gate open") = true; + let res = serve.await; + assert!( + matches!(res, Err(Error::Old)), + "a wedged peer must not hold an expired group open: {res:?}" + ); + + // The reason reaches the peer, so it reads as a truncated group rather than + // a routine cancel and it can re-request the sequence. + assert_eq!(log.resets(), vec![Error::Old.to_code()]); + + // Only the untaken tail is lost: the frame already handed to the publisher + // owns its payload, so the release can't reclaim it mid-write. + let writes = log.writes.lock().unwrap(); + assert!( + writes.windows(b"second".len()).any(|w| w == b"second"), + "the in-flight frame still reached the wire" + ); + assert!( + !writes.windows(b"third".len()).any(|w| w == b"third"), + "the untaken tail was released with the group" + ); + } } #[cfg(test)] diff --git a/rs/moq-net/src/lite/test_transport.rs b/rs/moq-net/src/lite/test_transport.rs index 5c0aaf872..76c76b851 100644 --- a/rs/moq-net/src/lite/test_transport.rs +++ b/rs/moq-net/src/lite/test_transport.rs @@ -302,6 +302,8 @@ pub struct SinkSession { /// Set by [`Self::gated_bi`]. `None` parks `open_bi` itself forever, which is all /// a test driving only uni streams needs. bi_gate: Option>, + /// Set by [`Self::gated_uni`]. `None` writes immediately. + uni_gate: Option>, /// The ALPN to report, for a test that needs a specific negotiated version rather /// than the SETUP-negotiated fallback an absent one selects. protocol: Option<&'static str>, @@ -316,6 +318,7 @@ impl SinkSession { Self { log, bi_gate: None, + uni_gate: None, protocol: None, stats: Arc::new(Mutex::new(SinkStats::default())), } @@ -347,6 +350,22 @@ impl SinkSession { Self { log: Log::default(), bi_gate: Some(gate), + uni_gate: None, + protocol: None, + stats: Arc::new(Mutex::new(SinkStats::default())), + } + } + + /// Serve uni streams, holding every write while `gate` reads false. + /// + /// Groups travel on uni streams, so this is how a test stalls delivery the way + /// QUIC flow control does: the publisher keeps its consumer and its place in the + /// group, but no byte reaches the wire until the gate reopens. + pub fn gated_uni(gate: kio::Consumer) -> Self { + Self { + log: Log::default(), + bi_gate: None, + uni_gate: Some(gate), protocol: None, stats: Arc::new(Mutex::new(SinkStats::default())), } @@ -381,7 +400,11 @@ impl web_transport_trait::Session for SinkSession { } async fn open_uni(&self) -> Result { - Ok(SinkSend::new(self.log.clone())) + Ok(SinkSend { + log: self.log.clone(), + gate: self.uni_gate.clone(), + finished: false, + }) } fn send_datagram(&self, _payload: bytes::Bytes) -> Result<(), Self::Error> {