From 91f19b9c37e7f31a644f098a5a2cec699cb96c50 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 27 Aug 2026 07:52:10 -0700 Subject: [PATCH] test(net): assert a stalled write releases the group it was serving A group being served can expire mid-stream when a single transport write stalls: the publisher stamps the group's cache access once per frame, immediately before writing it, and nothing re-stamps inside the write. A peer whose flow control window stays shut for longer than the whole retention window therefore stops accessing the group and it ages out from under the stream, which resets with Old. That is the intended behavior, not a bug. Holding the group for as long as a wedged peer refuses to read would let any subscriber pin cache indefinitely, which is exactly what the retention window exists to bound. The policy was emergent and unasserted, so lock it in before someone "fixes" it with a retention lease. Adds SinkSession::gated_uni alongside the existing gated_bi so a test can stall a group stream the way flow control does. Co-Authored-By: Claude Opus 5 --- rs/moq-net/src/lite/publisher.rs | 92 +++++++++++++++++++++++++++ rs/moq-net/src/lite/test_transport.rs | 25 +++++++- 2 files changed, 116 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 9955e9068c..0324f58476 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 5c0aaf8728..76c76b8512 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> {