From bca5620565c45eabb2bde4c8090e38c7c2e481c7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:23:33 -0700 Subject: [PATCH 1/3] fix(net): reset a subscription that cannot encode a later group A group past the QUIC varint limit resets that subscribe stream with INTERNAL_ERROR. A later group used to write nothing and leave the subscriber waiting. Announce compression keeps a tie literal where hop 100 sizes differently. Co-Authored-By: Grok 4.7 --- quest/m1/README.md | 1 - quest/m1/varint-test-gaps.md | 20 ---------- rs/moq-net/src/lite/compress.rs | 31 +++++++++++++++ rs/moq-net/src/lite/publisher.rs | 38 +++++++++++++++++-- rs/moq-net/tests/wide_sequence.rs | 63 ++++++++++++++++++++++++++++++- 5 files changed, 127 insertions(+), 26 deletions(-) delete mode 100644 quest/m1/varint-test-gaps.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 21d2307fea..1b87f9d2f0 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -93,7 +93,6 @@ blocks. The quests that gated m0 lines moved under them. - [JS request window](/quest/m1/js-request-window.md) - @moq/net grants MAX_REQUEST_ID as requests close, so a peer never stalls after ~21k requests - [Relay session limits](/quest/m1/relay-session-limits.md) - moq-relay sets per-session request limits, tighter for clients than peers, and the bindings name a refused request - [Churn with held subscriptions](/quest/m1/session-churn-held.md) - opening and closing a request costs the same with 1 or 1,024 held subscriptions -- [Wide varint tests](/quest/m1/varint-test-gaps.md) - wide sequences and both varint forms are tested where they differ - [Export linger](/quest/m1/export-ts-linger.md) - `moq export ts --linger` fails on an export error while the broadcast is live, and lingers once it closes or is replaced - [Archive](/quest/m1/archive/README.md) - record selected tracks to any object_store and replay them over FETCH or derived HLS; the catalog entry and format may break in place, since no archives exist - [In-band auth](/quest/m1/auth/README.md) - a session tells its peer what it may publish and subscribe to, unions tokens presented in band, and fails loud on an out-of-scope publish diff --git a/quest/m1/varint-test-gaps.md b/quest/m1/varint-test-gaps.md deleted file mode 100644 index 30f4b11087..0000000000 --- a/quest/m1/varint-test-gaps.md +++ /dev/null @@ -1,20 +0,0 @@ -# [S] Wide varints are tested where the two forms differ - -## Goal - -The plain-`u64` varint codec (#4437) has tests for the cases its first round -left open: - -- The wide-sequence relay test asserts the error a lite-06 reader sees for a - group of `1 << 62` from a lite-07 publisher, not any error. Today the reader - only sees `Stream(Internal)` over the wire, so decide what it should see. -- A subscription whose groups turn wide partway through fails only that - subscription, and the session and other tracks keep going. -- Announce compression is tested where the QUIC and leading-ones forms size - the same value differently. - -## Plan - -Extend `rs/moq-net/tests/wide_sequence.rs` and the lite announce encode tests. - -Public API: none. Wire: none. diff --git a/rs/moq-net/src/lite/compress.rs b/rs/moq-net/src/lite/compress.rs index 257323bd52..be281cc8d7 100644 --- a/rs/moq-net/src/lite/compress.rs +++ b/rs/moq-net/src/lite/compress.rs @@ -566,6 +566,37 @@ mod tests { assert_eq!(wire_hops, chain(2, &[5], 1)); } + /// Hop 100 is 2 bytes as a QUIC varint and 1 byte as leading-ones. Sharing that + /// single hop from 128 announcements back costs 4 bytes in both forms (the base + /// is 2 bytes either way). The literal chain is 5 bytes in QUIC and 4 in + /// leading-ones, so QUIC would compress and leading-ones ties. A tie stays literal. + #[test] + fn a_tie_stays_literal_where_hop_100_sizes_differ() { + let quic = Form::Quic; + let leading = Form::LeadingOnes { seven: true }; + assert_eq!(varint_size(100, quic), 2); + assert_eq!(varint_size(100, leading), 1); + + let base = 128; + let compressed = |form| varint_size(base, form) + varint_size(0, form) + varint_size(1, form); + let literal = |form| varint_size(0, form) * 2 + varint_size(1, form) + varint_size(100, form); + assert!(compressed(quic) < literal(quic)); + assert_eq!(compressed(leading), literal(leading)); + + let mut encoder = AnnounceEncoder::new(VERSION); + let mut decoder = AnnounceDecoder::default(); + start(&mut encoder, &mut decoder, "origin", &[100]); + for n in 1..128 { + let suffix = format!("f{n}"); + start(&mut encoder, &mut decoder, &suffix, &[1]); + } + let (_, wire_path, wire_hops) = encoder.start(Path::new("again").to_owned(), hops(&[100])); + assert_eq!(wire_hops, HopsRef::literal(hops(&[100]))); + let (got_path, got_hops) = decoder.start(wire_path, wire_hops).unwrap(); + assert_eq!(got_path.as_str(), "again"); + assert_eq!(got_hops, hops(&[100])); + } + #[test] fn encoder_sends_literal_when_nothing_is_shared() { let mut encoder = AnnounceEncoder::new(VERSION); diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 54e5e44fee..b095b225ea 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3,7 +3,7 @@ use std::{ collections::{BTreeSet, HashMap}, ops::{Bound, ControlFlow}, sync::{ - Arc, + Arc, Mutex, atomic::{AtomicU64, AtomicUsize, Ordering}, }, task::{Context, Poll, ready}, @@ -2256,10 +2256,15 @@ struct Subscription { /// /// A group is pending from the moment it is queued until its stream opens, or it gives /// up first (expired, or the open failed) and is never counted. +/// +/// `unencodable` rides along because every group machine and the run loop already share +/// this handle. A header that cannot be encoded writes nothing, so the group-stream +/// reset is not a stream the subscriber can pin to this subscription. #[derive(Default)] struct Opens { pending: AtomicU64, opened: AtomicU64, + unencodable: Mutex>, } impl Subscription { @@ -2300,6 +2305,21 @@ impl Subscription { self.track_priority_seen } + /// Remember an encode failure the group stream could not carry. + /// + /// The first one wins. The run loop resets the subscribe stream with it. + fn note_unencodable(&self, err: &Error) { + let mut slot = self.opens.unencodable.lock().unwrap(); + if slot.is_none() { + *slot = Some(err.clone()); + } + } + + /// Take the encode failure, if a group recorded one. + fn take_unencodable(&self) -> Option { + self.opens.unencodable.lock().unwrap().take() + } + /// Test shim: drive one group stream like the old `serve_group`, surfacing the /// error the machine otherwise swallows after aborting the stream. #[cfg(test)] @@ -2440,6 +2460,12 @@ impl TrackRun { // Drive the in-flight group machines; completions just retire. let _ = self.children.poll(waiter); + // A group whose header does not fit this peer's varint wrote nothing. The + // reset is on a stream the subscriber never tied to this subscription, so + // forward it here or the reader waits forever. + if let Some(err) = self.ctx.take_unencodable() { + return Poll::Ready(Err(err)); + } // The first group waits for the source to resolve where its feed starts; datagrams // keep flowing meanwhile. @@ -2892,9 +2918,13 @@ impl kio::Task for GroupServe { type Output = (); fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { - // The machine owns its outcome: the stream was aborted with the reason (or - // reset by the writer's Drop), which is all the subscriber sees. - ready!(self.poll_serve(waiter)).map(|()| ()).unwrap_or(()); + // Old, Lagged, and Evicted reset the group stream, which is all the subscriber + // needs once the header is on the wire. An encode failure can write nothing, so + // that reset has no header to pin it to this subscription. The run loop reads + // it off the shared slot and resets the subscribe stream. + if let Err(err @ Error::Encode(_)) = ready!(self.poll_serve(waiter)) { + self.ctx.note_unencodable(&err); + } Poll::Ready(()) } } diff --git a/rs/moq-net/tests/wide_sequence.rs b/rs/moq-net/tests/wide_sequence.rs index 6fa8898a36..0254d39b36 100644 --- a/rs/moq-net/tests/wide_sequence.rs +++ b/rs/moq-net/tests/wide_sequence.rs @@ -57,7 +57,68 @@ async fn wide_sequence_fails_only_its_subscription() { }) .await .expect("the wide subscription hung instead of failing"); - assert!(result.is_err(), "lite-06 cannot carry the sequence: {result:?}"); + // The group does not fit a QUIC varint. Encode failures have no dedicated stream + // code, so the subscribe stream resets with INTERNAL_ERROR and the session stays up. + assert!( + matches!(result, Err(moq_net::Error::Stream(moq_net::StreamError::Internal))), + "lite-06 should see an internal stream error, not {result:?}" + ); + + let mut sub = remote.track("narrow").unwrap().subscribe(None).await.unwrap(); + let mut group = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(group.read_frame().await.unwrap().unwrap().payload, b"narrow".as_ref()); +} + +/// A subscription that starts inside the QUIC range and later receives a group of +/// `1 << 62` fails only that subscription. The session and the other track continue. +#[moq_net_sim::test] +async fn a_later_wide_group_fails_only_that_subscription() { + const WIDE: u64 = 1 << 62; + + let publisher = produce_origin(1); + let relay = produce_origin(2); + let client = produce_origin(3); + + let broadcast = publisher.create_broadcast("bench").unwrap(); + let wide = broadcast.create_track("wide", None).unwrap(); + let narrow = broadcast.create_track("narrow", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + let mut group = wide.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, b"first".as_ref()).unwrap(); + let mut group = narrow.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, b"narrow".as_ref()).unwrap(); + + let mut options = MockConnectOptions::new("moq-lite-07-wip".parse::().unwrap()); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(relay.clone()); + let _upstream = connect_mock(options).await; + let mut options = MockConnectOptions::new("moq-lite-06".parse::().unwrap()); + options.server_publish = Some(relay.consume()); + options.client_subscribe = Some(client.clone()); + let _downstream = connect_mock(options).await; + + let consumer = client.consume(); + consumer.routed("bench").await.unwrap(); + let remote = consumer.request_broadcast("bench").await.unwrap(); + + let mut sub = remote.track("wide").unwrap().subscribe(None).await.unwrap(); + let mut group = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(group.sequence, 0); + assert_eq!(group.read_frame().await.unwrap().unwrap().payload, b"first".as_ref()); + + let mut group = wide.create_group(moq_net::group::Info { sequence: WIDE }).unwrap(); + group.write_frame(Timestamp::ZERO, b"wide".as_ref()).unwrap(); + + let result = moq_net_sim::timeout(Duration::from_secs(5), sub.recv_group()) + .await + .expect("the wide group hung the subscription instead of failing it"); + match result { + Err(moq_net::Error::Stream(moq_net::StreamError::Internal)) => {} + Err(err) => { + panic!("the later wide group should fail the subscription with an internal stream error, not {err:?}") + } + Ok(_) => panic!("the later wide group should fail the subscription with an internal stream error, got a group"), + } let mut sub = remote.track("narrow").unwrap().subscribe(None).await.unwrap(); let mut group = sub.recv_group().await.unwrap().unwrap(); From 70cf082cc7cd71378cd513cab81378183a654e5a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 16:25:32 -0700 Subject: [PATCH 2/3] fix(net): only an unencodable group header fails the subscription A mid-group encode failure already resets a stream the subscriber can pin to this subscription, so it should end that group only. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/publisher.rs | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index e911ced15a..99b4d3afd4 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -2745,6 +2745,10 @@ impl GroupServe { frame_start: self.frame_start, }; if let Err(err) = writer.buffer(&lite::DataType::Group).and_then(|()| writer.buffer(&msg)) { + // The header wrote nothing, so the group-stream reset has nothing to pin + // it to this subscription. The run loop reads the slot and resets the + // subscribe stream instead. + self.ctx.note_unencodable(&err); self.state = GroupState::Done; writer.abort(&err); return Poll::Ready(Err(err)); @@ -2918,13 +2922,9 @@ impl kio::Task for GroupServe { type Output = (); fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { - // Old, Lagged, and Evicted reset the group stream, which is all the subscriber - // needs once the header is on the wire. An encode failure can write nothing, so - // that reset has no header to pin it to this subscription. The run loop reads - // it off the shared slot and resets the subscribe stream. - if let Err(err @ Error::Encode(_)) = ready!(self.poll_serve(waiter)) { - self.ctx.note_unencodable(&err); - } + // The machine owns its outcome: the stream was aborted with the reason (or + // reset by the writer's Drop), which is all the subscriber sees. + ready!(self.poll_serve(waiter)).map(|()| ()).unwrap_or(()); Poll::Ready(()) } } From af060414d0fe0247bc5f08d6729b96ad152cd208 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 21:16:04 -0700 Subject: [PATCH 3/3] test(net): pass the epoch to request_broadcast in the later-wide-group test Main made the epoch argument required. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/tests/wide_sequence.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-net/tests/wide_sequence.rs b/rs/moq-net/tests/wide_sequence.rs index 1ce2aa386b..48c258aa9c 100644 --- a/rs/moq-net/tests/wide_sequence.rs +++ b/rs/moq-net/tests/wide_sequence.rs @@ -99,7 +99,7 @@ async fn a_later_wide_group_fails_only_that_subscription() { let consumer = client.consume(); consumer.routed("bench").await.unwrap(); - let remote = consumer.request_broadcast("bench").await.unwrap(); + let remote = consumer.request_broadcast("bench", None).await.unwrap(); let mut sub = remote.track("wide").unwrap().subscribe(None).await.unwrap(); let mut group = sub.recv_group().await.unwrap().unwrap();