diff --git a/quest/m1/README.md b/quest/m1/README.md index 99c369dcf0..64ea0a9d76 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -91,7 +91,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 b765a7cec8..daaa269223 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}, @@ -2290,10 +2290,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 { @@ -2338,6 +2343,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)] @@ -2478,6 +2498,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. @@ -2757,6 +2783,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)); diff --git a/rs/moq-net/tests/wide_sequence.rs b/rs/moq-net/tests/wide_sequence.rs index 536b8c39a7..48c258aa9c 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", None).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();