From 8376466dfcb800201b5bf25c5bd19993714aa0ef Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 14:58:19 -0700 Subject: [PATCH 01/11] quest: claim quest/m1/quic/coverage/characterize.md Co-Authored-By: Claude Opus 5.5 From 6b899bdee0e74e87426acf333f312bc2a4182234 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 15:15:04 -0700 Subject: [PATCH 02/11] test(quic): characterize frame error paths and resets Co-Authored-By: Claude Opus 5.5 --- rs/moq-quic/src/connection/mod.rs | 21 + rs/moq-quic/src/tests/characterize/frames.rs | 452 +++++++++++++++++++ rs/moq-quic/src/tests/characterize/mod.rs | 122 +++++ rs/moq-quic/src/tests/characterize/reset.rs | 245 ++++++++++ rs/moq-quic/src/tests/mod.rs | 1 + 5 files changed, 841 insertions(+) create mode 100644 rs/moq-quic/src/tests/characterize/frames.rs create mode 100644 rs/moq-quic/src/tests/characterize/mod.rs create mode 100644 rs/moq-quic/src/tests/characterize/reset.rs diff --git a/rs/moq-quic/src/connection/mod.rs b/rs/moq-quic/src/connection/mod.rs index 3425f00ba2b..6934909c409 100644 --- a/rs/moq-quic/src/connection/mod.rs +++ b/rs/moq-quic/src/connection/mod.rs @@ -3927,6 +3927,27 @@ impl Connection { Some(packet.payload.to_vec()) } + /// Encrypts `payload` as this connection's next packet in `space`, untracked, so tests can + /// send frames a correct peer never would + #[cfg(test)] + pub(crate) fn forge_packet(&mut self, now: Instant, space: SpaceId, payload: &[u8]) -> Vec { + let mut buf = Vec::new(); + let builder = PacketBuilder::new( + now, + space, + self.rem_cids.active(), + &mut buf, + usize::MAX, + 0, + true, + self, + ) + .expect("connection can still send"); + buf.extend_from_slice(payload); + builder.finish(self, now, &mut buf); + buf + } + /// The number of bytes of packets containing retransmittable frames that have not been /// acknowledged or declared lost. #[cfg(test)] diff --git a/rs/moq-quic/src/tests/characterize/frames.rs b/rs/moq-quic/src/tests/characterize/frames.rs new file mode 100644 index 00000000000..fe83ff9a6bc --- /dev/null +++ b/rs/moq-quic/src/tests/characterize/frames.rs @@ -0,0 +1,452 @@ +//! How a connection answers frames a correct peer never sends + +use super::*; + +/// The wire value of a stream ID +fn id(initiator: Side, dir: Dir, index: u64) -> u64 { + StreamId::new(initiator, dir, index).into() +} + +/// Sends `frames` from the client on an established connection, returning how the server +/// closed, or `None` if it is still up +fn server_answer(frames: Frames) -> Option { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = connected(); + client_sends(&mut pair, client_ch, SpaceId::Data, frames); + pair.drive(); + match lost(pair.server_conn_mut(server_ch)) { + None => None, + Some(ConnectionError::TransportError(error)) => Some(error.code), + Some(other) => panic!("server lost with {other:?}"), + } +} + +/// Sends `frames` from the server on an established connection, returning how the client +/// closed, or `None` if it is still up +fn client_answer(frames: Frames) -> Option { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = connected(); + server_sends(&mut pair, server_ch, SpaceId::Data, frames); + pair.drive(); + match lost(pair.client_conn_mut(client_ch)) { + None => None, + Some(ConnectionError::TransportError(error)) => Some(error.code), + Some(other) => panic!("client lost with {other:?}"), + } +} + +#[test] +fn crypto_in_0rtt() { + let _guard = subscribe(); + let mut pair = pair_with(|_| {}); + let config = client_with(|_| {}); + + // A first connection leaves the client a ticket for 0-RTT + let client_ch = pair.begin_connect(config.clone()); + pair.drive(); + pair.server.assert_accept(); + let now = pair.time; + pair.client_conn_mut(client_ch) + .close(now, VarInt(0), Bytes::new()); + pair.drive(); + + let client_ch = pair.begin_connect(config); + assert!(pair.client_conn_mut(client_ch).has_0rtt()); + pair.drive_client(); + client_sends( + &mut pair, + client_ch, + SpaceId::Data, + Frames::default().var(FrameType::CRYPTO, &[0, 1]).raw(&[0]), + ); + pair.drive(); + + let server_ch = pair.server.assert_accept(); + assert_eq!( + lost_code(pair.server_conn_mut(server_ch)), + TransportErrorCode::PROTOCOL_VIOLATION + ); +} + +#[test] +fn stream_data_blocked_on_send_only() { + let frames = Frames::default().var( + FrameType::STREAM_DATA_BLOCKED, + &[id(Side::Server, Dir::Uni, 0), 0], + ); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::STREAM_STATE_ERROR) + ); +} + +/// STREAM_DATA_BLOCKED is only logged, so a stream ID past the limit or never opened is ignored +#[test] +fn stream_data_blocked_on_unopened() { + for id in [ + id(Side::Client, Dir::Bi, 1_000_000), + id(Side::Server, Dir::Bi, 0), + ] { + let frames = Frames::default().var(FrameType::STREAM_DATA_BLOCKED, &[id, 0]); + assert_eq!(server_answer(frames), None); + } +} + +#[test] +fn max_stream_data_on_recv_only() { + let frames = Frames::default().var( + FrameType::MAX_STREAM_DATA, + &[id(Side::Client, Dir::Uni, 0), 1000], + ); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::STREAM_STATE_ERROR) + ); +} + +#[test] +fn max_stream_data_on_unopened() { + let frames = Frames::default().var( + FrameType::MAX_STREAM_DATA, + &[id(Side::Server, Dir::Bi, 0), 1000], + ); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::STREAM_STATE_ERROR) + ); +} + +#[test] +fn stop_sending_on_recv_only() { + let frames = Frames::default().var( + FrameType::STOP_SENDING, + &[id(Side::Client, Dir::Uni, 0), 0], + ); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::STREAM_STATE_ERROR) + ); +} + +#[test] +fn stop_sending_on_unopened() { + let frames = + Frames::default().var(FrameType::STOP_SENDING, &[id(Side::Server, Dir::Bi, 0), 0]); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::STREAM_STATE_ERROR) + ); +} + +#[test] +fn streams_blocked_above_limit() { + for ty in [FrameType::STREAMS_BLOCKED_BIDI, FrameType::STREAMS_BLOCKED_UNI] { + // The largest representable limit is fine + let frames = Frames::default().var(ty, &[1 << 60]); + assert_eq!(server_answer(frames), None); + + let frames = Frames::default().var(ty, &[(1 << 60) + 1]); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::FRAME_ENCODING_ERROR) + ); + } +} + +#[test] +fn max_streams_above_limit() { + for ty in [FrameType::MAX_STREAMS_BIDI, FrameType::MAX_STREAMS_UNI] { + let frames = Frames::default().var(ty, &[1 << 60]); + assert_eq!(server_answer(frames), None); + + let frames = Frames::default().var(ty, &[(1 << 60) + 1]); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::FRAME_ENCODING_ERROR) + ); + } +} + +/// A NEW_CONNECTION_ID frame with sequence `sequence`, retiring those before `retire_prior_to` +fn new_cid(frames: Frames, sequence: u64, retire_prior_to: u64) -> Frames { + let mut cid = [0xab; 9]; + cid[0] = 8; + cid[1..5].copy_from_slice(&(sequence as u32).to_be_bytes()); + frames + .var(FrameType::NEW_CONNECTION_ID, &[sequence, retire_prior_to]) + .raw(&cid) + .raw(&[sequence as u8; RESET_TOKEN_SIZE]) +} + +#[test] +fn new_connection_id_above_limit() { + // The client's CIDs past the server's active_connection_id_limit, none retired + let mut frames = Frames::default(); + for sequence in 1..=8 { + frames = new_cid(frames, sequence, 0); + } + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::CONNECTION_ID_LIMIT_ERROR) + ); +} + +/// The frame decoder rejects it before the connection's own PROTOCOL_VIOLATION check runs +#[test] +fn new_connection_id_retires_unissued() { + let frames = new_cid(Frames::default(), 2, 3); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::FRAME_ENCODING_ERROR) + ); +} + +#[test] +fn new_connection_id_retired_is_discarded() { + // Retiring through sequence 4 then repeating sequence 1 discards the stale frame + let frames = new_cid(new_cid(Frames::default(), 4, 4), 1, 0); + assert_eq!(server_answer(frames), None); +} + +#[test] +fn retire_connection_id_unissued() { + let frames = Frames::default().var(FrameType::RETIRE_CONNECTION_ID, &[1_000]); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::PROTOCOL_VIOLATION) + ); +} + +/// Has the client retire the server CID its forged packet is sent to, and lets the server +/// answer with a replacement CID +fn retire_in_use(pair: &mut Pair, client_ch: ConnectionHandle, server_ch: ConnectionHandle) { + let active = pair.client_conn_mut(client_ch).active_rem_cid_seq(); + client_sends( + pair, + client_ch, + SpaceId::Data, + Frames::default().var(FrameType::RETIRE_CONNECTION_ID, &[active]), + ); + pair.drive_server(); + assert!(lost(pair.server_conn_mut(server_ch)).is_none()); +} + +/// RFC 9000 lets a receiver close when a packet retires its own destination CID; the server +/// instead retires it and issues a replacement, which the client, never having retired the CID +/// it still uses, rejects as over its limit +#[test] +fn retire_connection_id_in_use() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = connected(); + retire_in_use(&mut pair, client_ch, server_ch); + pair.drive_client(); + assert_eq!( + lost_code(pair.client_conn_mut(client_ch)), + TransportErrorCode::CONNECTION_ID_LIMIT_ERROR + ); +} + +/// A connection closed by its own transport error sends CONNECTION_CLOSE to a CID the server +/// retired, and the server answers with a stateless reset. Today the reset replaces the unread +/// close reason with `Reset`, or, once the reason was read, reports a second `ConnectionLost`. +fn stateless_reset_after_local_close(poll_early: bool) { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = connected(); + retire_in_use(&mut pair, client_ch, server_ch); + pair.drive_client(); + if poll_early { + assert_eq!( + lost_code(pair.client_conn_mut(client_ch)), + TransportErrorCode::CONNECTION_ID_LIMIT_ERROR + ); + } + + pair.drive(); + let lost = lost(pair.client_conn_mut(client_ch)); + if poll_early { + assert_eq!(lost, None); + } else { + assert_matches!( + lost, + Some(ConnectionError::TransportError(TransportError { + code: TransportErrorCode::CONNECTION_ID_LIMIT_ERROR, + .. + })) + ); + } +} + +#[test] +#[ignore = "bug: /quest/m1/quic/close-reason-after-reset.md"] +fn stateless_reset_after_local_close_keeps_reason() { + stateless_reset_after_local_close(false); +} + +#[test] +#[ignore = "bug: /quest/m1/quic/close-reason-after-reset.md"] +fn stateless_reset_after_local_close_is_not_a_second_loss() { + stateless_reset_after_local_close(true); +} + +#[test] +fn new_token_from_client() { + let frames = Frames::default() + .var(FrameType::NEW_TOKEN, &[4]) + .raw(b"tokn"); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::PROTOCOL_VIOLATION) + ); +} + +#[test] +fn new_token_empty() { + let frames = Frames::default().var(FrameType::NEW_TOKEN, &[0]); + assert_eq!( + client_answer(frames), + Some(TransportErrorCode::FRAME_ENCODING_ERROR) + ); +} + +#[test] +fn handshake_done_from_client() { + let frames = Frames::default().var(FrameType::HANDSHAKE_DONE, &[]); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::PROTOCOL_VIOLATION) + ); +} + +#[test] +fn stream_offset_at_max() { + let max = VarInt::MAX.into_inner(); + let s = id(Side::Client, Dir::Uni, 0); + + // Ending at 2^62 - 1 is only past flow control, while ending past it is unrepresentable, + // which RFC 9000 lets a receiver report as either error + for offset in [max - 1, max] { + let frames = Frames::default().stream(s, offset, b"x", false); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::FLOW_CONTROL_ERROR) + ); + } +} + +#[test] +fn crypto_offset_at_max() { + let max = VarInt::MAX.into_inner(); + let frames = Frames::default() + .var(FrameType::CRYPTO, &[max, 1]) + .raw(b"x"); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::CRYPTO_BUFFER_EXCEEDED) + ); +} + +#[test] +fn stream_data_past_final_size() { + let s = id(Side::Client, Dir::Uni, 0); + let frames = Frames::default().stream(s, 0, b"abc", true); + let frames = frames.stream(s, 3, b"d", false); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::FINAL_SIZE_ERROR) + ); +} + +#[test] +fn stream_final_size_changed() { + let s = id(Side::Client, Dir::Uni, 0); + let frames = Frames::default().stream(s, 0, b"abc", true); + let frames = frames.stream(s, 0, b"abcde", true); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::FINAL_SIZE_ERROR) + ); +} + +/// Today the reader gets all five bytes and then blocks forever: the shrunken final size never +/// matches what was read, so the stream neither finishes nor errors. +#[test] +#[ignore = "bug: /quest/m1/quic/final-size-below-received.md"] +fn stream_final_size_below_received() { + let s = id(Side::Client, Dir::Uni, 0); + let frames = Frames::default().stream(s, 0, b"abcde", false); + let frames = frames.stream(s, 0, b"abc", true); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::FINAL_SIZE_ERROR) + ); +} + +#[test] +fn reset_final_size_below_received() { + let s = id(Side::Client, Dir::Uni, 0); + let frames = Frames::default().stream(s, 0, b"abcdefghij", false); + let frames = frames.var(FrameType::RESET_STREAM, &[s, 0, 5]); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::FINAL_SIZE_ERROR) + ); +} + +#[test] +fn reset_final_size_differs_from_fin() { + let s = id(Side::Client, Dir::Uni, 0); + let frames = Frames::default().stream(s, 0, b"abc", true); + let frames = frames.var(FrameType::RESET_STREAM, &[s, 0, 5]); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::FINAL_SIZE_ERROR) + ); +} + +#[test] +fn reset_final_size_past_flow_control() { + let s = id(Side::Client, Dir::Uni, 0); + let frames = Frames::default().var(FrameType::RESET_STREAM, &[s, 0, 1 << 40]); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::FLOW_CONTROL_ERROR) + ); +} + +#[test] +fn reset_on_send_only() { + let frames = Frames::default().var( + FrameType::RESET_STREAM, + &[id(Side::Server, Dir::Uni, 0), 0, 0], + ); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::STREAM_STATE_ERROR) + ); +} + +#[test] +fn stream_on_send_only() { + let frames = Frames::default().stream(id(Side::Server, Dir::Uni, 0), + 0, + b"x", + false, + ); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::STREAM_STATE_ERROR) + ); +} + +#[test] +fn stream_above_stream_limit() { + let frames = Frames::default().stream(id(Side::Client, Dir::Uni, 1_000_000), + 0, + b"x", + false, + ); + assert_eq!( + server_answer(frames), + Some(TransportErrorCode::STREAM_LIMIT_ERROR) + ); +} diff --git a/rs/moq-quic/src/tests/characterize/mod.rs b/rs/moq-quic/src/tests/characterize/mod.rs new file mode 100644 index 00000000000..e9618c99b7e --- /dev/null +++ b/rs/moq-quic/src/tests/characterize/mod.rs @@ -0,0 +1,122 @@ +//! Characterization tests: they pin what the connection does today, right or wrong, so a +//! rewrite that changes behavior fails here and has to accept or fix each change in its diff. +//! +//! A pinned bug lands as an `#[ignore]`d test asserting the correct behavior, naming the quest +//! that fixes it. Every pair pins its congestion controller instead of taking the harness +//! default, so a change of default does not churn the goldens. + +use super::*; +use crate::coding::{BufMutExt, Codec}; + +mod frames; +mod reset; + +/// The transport every characterization pair runs: Cubic with deterministic packet numbers +fn transport() -> TransportConfig { + let mut transport = TransportConfig::default(); + transport.congestion_controller_factory(Arc::new(congestion::CubicConfig::default())); + transport.deterministic_packet_numbers(true); + transport +} + +/// A pair whose server runs [`transport`] with `configure` applied +fn pair_with(configure: impl FnOnce(&mut TransportConfig)) -> Pair { + let mut config = server_config(); + let mut transport = transport(); + configure(&mut transport); + config.transport = Arc::new(transport); + Pair::new(Default::default(), config) +} + +/// A client config running [`transport`] with `configure` applied +fn client_with(configure: impl FnOnce(&mut TransportConfig)) -> ClientConfig { + let mut config = client_config(); + let mut transport = transport(); + configure(&mut transport); + config.transport = Arc::new(transport); + config +} + +/// A connected pair whose endpoints both run [`transport`] +fn connected() -> (Pair, ConnectionHandle, ConnectionHandle) { + connected_with(|_| {}, |_| {}) +} + +/// A connected pair whose server and client transports are configured separately +fn connected_with( + server: impl FnOnce(&mut TransportConfig), + client: impl FnOnce(&mut TransportConfig), +) -> (Pair, ConnectionHandle, ConnectionHandle) { + let mut pair = pair_with(server); + let (client_ch, server_ch) = pair.connect_with(client_with(client)); + (pair, client_ch, server_ch) +} + +/// Frames encoded by hand, so tests can send what the frame encoder refuses to +#[derive(Default)] +struct Frames(Vec); + +impl Frames { + /// Appends a frame of type `ty` whose fields are all varints + fn var(mut self, ty: FrameType, fields: &[u64]) -> Self { + ty.encode(&mut self.0); + for &field in fields { + self.0.write_var(field); + } + self + } + + /// Appends a STREAM frame with explicit offset and length + fn stream(mut self, id: u64, offset: u64, data: &[u8], fin: bool) -> Self { + // The STREAM type with its OFF and LEN bits set, plus FIN + self.0.write_var(0x0e | fin as u64); + for field in [id, offset, data.len() as u64] { + self.0.write_var(field); + } + self.raw(data) + } + + /// Appends raw bytes, such as the payload after a frame's length field + fn raw(mut self, bytes: &[u8]) -> Self { + self.0.extend_from_slice(bytes); + self + } +} + +/// Sends `frames` from the client to the server in the client's next packet in `space` +fn client_sends(pair: &mut Pair, ch: ConnectionHandle, space: SpaceId, frames: Frames) { + let now = pair.time; + let packet = pair.client_conn_mut(ch).forge_packet(now, space, &frames.0); + pair.server + .inbound + .push_back((pair.time, None, packet.as_slice().into())); +} + +/// Sends `frames` from the server to the client in the server's next packet in `space` +fn server_sends(pair: &mut Pair, ch: ConnectionHandle, space: SpaceId, frames: Frames) { + let now = pair.time; + let packet = pair.server_conn_mut(ch).forge_packet(now, space, &frames.0); + pair.client + .inbound + .push_back((pair.time, None, packet.as_slice().into())); +} + +/// Drains `conn`'s events, returning the error it was lost with, if any +fn lost(conn: &mut Connection) -> Option { + let mut lost = None; + while let Some(event) = conn.poll() { + if let Event::ConnectionLost { reason } = event { + assert!(lost.is_none(), "lost twice: {lost:?}, then {reason:?}"); + lost = Some(reason); + } + } + lost +} + +/// The transport error code `conn` was lost with, panicking if it is still up or lost otherwise +fn lost_code(conn: &mut Connection) -> TransportErrorCode { + match lost(conn) { + Some(ConnectionError::TransportError(error)) => error.code, + other => panic!("expected a transport error, got {other:?}"), + } +} diff --git a/rs/moq-quic/src/tests/characterize/reset.rs b/rs/moq-quic/src/tests/characterize/reset.rs new file mode 100644 index 00000000000..b040d399f3a --- /dev/null +++ b/rs/moq-quic/src/tests/characterize/reset.rs @@ -0,0 +1,245 @@ +//! How a reset or stopped stream ends for the reader and what credit it gives back + +use super::*; + +const CODE: VarInt = VarInt(7); + +/// Every chunk a server stream yields, and how it ended +fn read_all(pair: &mut Pair, ch: ConnectionHandle, s: StreamId) -> (Vec, ReadError) { + let mut recv = pair.server_recv(ch, s); + let mut chunks = recv.read(true).unwrap(); + let mut data = Vec::new(); + let end = loop { + match chunks.next(usize::MAX) { + Ok(Some(chunk)) => data.extend_from_slice(&chunk.bytes), + Ok(None) => panic!("stream finished"), + Err(e) => break e, + } + }; + let _ = chunks.finalize(); + (data, end) +} + +/// Opens a client uni stream carrying `data`, delivered and accepted by the server +fn open(pair: &mut Pair, client_ch: ConnectionHandle, server_ch: ConnectionHandle, data: &[u8]) -> StreamId { + let s = pair.client_streams(client_ch).open(Dir::Uni).unwrap(); + pair.client_send(client_ch, s).write(data).unwrap(); + pair.drive(); + assert_eq!(pair.server_streams(server_ch).accept(Dir::Uni), Some(s)); + s +} + +#[test] +fn received_reset_reports_the_code_once() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = connected(); + let s = open(&mut pair, client_ch, server_ch, b"hello"); + assert_eq!(pair.server_recv(server_ch, s).received_reset(), Ok(None)); + + pair.client_send(client_ch, s).reset(CODE).unwrap(); + pair.drive(); + + // Observing the reset frees the stream, so every later access finds it closed + assert_eq!(pair.server_recv(server_ch, s).received_reset(), Ok(Some(CODE))); + assert!(pair.server_recv(server_ch, s).received_reset().is_err()); + assert_eq!( + pair.server_recv(server_ch, s).read(true).err(), + Some(ReadableError::ClosedStream) + ); + assert!(pair.server_recv(server_ch, s).stop(CODE).is_err()); +} + +#[test] +fn received_reset_on_closed_streams() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = connected(); + + // Never opened + let unopened = StreamId::new(Side::Client, Dir::Uni, 5); + assert!(pair.server_recv(server_ch, unopened).received_reset().is_err()); + + // Stopped by the reader, even after the peer resets in response + let s = open(&mut pair, client_ch, server_ch, b"hello"); + pair.server_recv(server_ch, s).stop(CODE).unwrap(); + assert!(pair.server_recv(server_ch, s).received_reset().is_err()); + pair.drive(); + assert!(pair.server_recv(server_ch, s).received_reset().is_err()); + + // Finished and fully read + let s = open(&mut pair, client_ch, server_ch, b"hello"); + pair.client_send(client_ch, s).finish().unwrap(); + pair.drive(); + assert_eq!(pair.server_recv(server_ch, s).received_reset(), Ok(None)); + { + let mut recv = pair.server_recv(server_ch, s); + let mut chunks = recv.read(true).unwrap(); + assert!(chunks.next(usize::MAX).unwrap().is_some()); + assert_eq!(chunks.next(usize::MAX), Ok(None)); + let _ = chunks.finalize(); + } + assert!(pair.server_recv(server_ch, s).received_reset().is_err()); +} + +/// A plain RESET_STREAM discards whatever the reader has not read yet, delivered or not +#[test] +fn chunks_end_in_reset() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = connected(); + let s = open(&mut pair, client_ch, server_ch, b"hello"); + + // Part of the stream is read before the reset arrives + { + let mut recv = pair.server_recv(server_ch, s); + let mut chunks = recv.read(true).unwrap(); + assert_eq!(chunks.next(2).unwrap().unwrap().bytes, &b"he"[..]); + let _ = chunks.finalize(); + } + pair.client_send(client_ch, s).reset(CODE).unwrap(); + pair.drive(); + + let mut recv = pair.server_recv(server_ch, s); + let mut chunks = recv.read(true).unwrap(); + assert_eq!(chunks.next(usize::MAX), Err(ReadError::Reset(CODE))); + // The end repeats until the chunks are finalized + assert_eq!(chunks.next(usize::MAX), Err(ReadError::Reset(CODE))); + let _ = chunks.finalize(); + assert_eq!( + pair.server_recv(server_ch, s).read(true).err(), + Some(ReadableError::ClosedStream) + ); +} + +#[test] +fn chunks_end_in_finished() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = connected(); + let s = open(&mut pair, client_ch, server_ch, b"hello"); + pair.client_send(client_ch, s).finish().unwrap(); + pair.drive(); + + let mut recv = pair.server_recv(server_ch, s); + let mut chunks = recv.read(true).unwrap(); + assert_eq!(chunks.next(usize::MAX).unwrap().unwrap().bytes, &b"hello"[..]); + assert_eq!(chunks.next(usize::MAX), Ok(None)); + assert_eq!(chunks.next(usize::MAX), Ok(None)); + let _ = chunks.finalize(); + assert_eq!( + pair.server_recv(server_ch, s).read(true).err(), + Some(ReadableError::ClosedStream) + ); +} + +/// The reader sees exactly the code the writer reset with, across the varint range +#[test] +fn reset_codes() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = connected(); + for code in [VarInt(0), VarInt(63), VarInt(64), VarInt::MAX] { + let s = open(&mut pair, client_ch, server_ch, b"x"); + pair.client_send(client_ch, s).reset(code).unwrap(); + pair.drive(); + assert_eq!(read_all(&mut pair, server_ch, s), (Vec::new(), ReadError::Reset(code))); + + let s = open(&mut pair, client_ch, server_ch, b"x"); + pair.server_recv(server_ch, s).stop(code).unwrap(); + pair.drive(); + assert_eq!( + pair.client_send(client_ch, s).write(b"y"), + Err(WriteError::Stopped(code)) + ); + } +} + +/// Stream ID credit for a reset stream returns only once the reader observes the reset +#[test] +fn reset_returns_stream_credit_once_observed() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = + connected_with(|t| _ = t.max_concurrent_uni_streams(VarInt(1)), |_| {}); + let s = open(&mut pair, client_ch, server_ch, b"hello"); + pair.client_send(client_ch, s).reset(CODE).unwrap(); + pair.drive(); + assert_eq!(pair.client_streams(client_ch).open(Dir::Uni), None); + + assert_eq!(read_all(&mut pair, server_ch, s).1, ReadError::Reset(CODE)); + pair.drive(); + assert!(pair.client_streams(client_ch).open(Dir::Uni).is_some()); +} + +/// The connection window used by `s`'s unread bytes, which fill it +const WINDOW: u32 = 4000; + +/// A connected pair whose server's connection window is filled by an unread client stream +fn blocked() -> (Pair, ConnectionHandle, ConnectionHandle, StreamId) { + let (mut pair, client_ch, server_ch) = + connected_with(|t| _ = t.receive_window(VarInt::from_u32(WINDOW)), |_| {}); + let s = open(&mut pair, client_ch, server_ch, &[0; WINDOW as usize]); + // The window is spent, so nothing more can be written + let other = pair.client_streams(client_ch).open(Dir::Uni).unwrap(); + assert_eq!( + pair.client_send(client_ch, other).write(b"x"), + Err(WriteError::Blocked) + ); + (pair, client_ch, server_ch, s) +} + +/// How many bytes a fresh client stream can write +fn writable(pair: &mut Pair, ch: ConnectionHandle) -> usize { + let s = pair.client_streams(ch).open(Dir::Uni).unwrap(); + pair.client_send(ch, s) + .write(&[0; 2 * WINDOW as usize]) + .unwrap_or(0) +} + +#[test] +fn reset_returns_connection_credit() { + let _guard = subscribe(); + let (mut pair, client_ch, _, s) = blocked(); + pair.client_send(client_ch, s).reset(CODE).unwrap(); + pair.drive(); + assert_eq!(writable(&mut pair, client_ch), WINDOW as usize); +} + +#[test] +fn stop_returns_connection_credit() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch, s) = blocked(); + pair.server_recv(server_ch, s).stop(CODE).unwrap(); + pair.drive(); + assert_eq!(writable(&mut pair, client_ch), WINDOW as usize); +} + +/// 0-RTT data the server rejected never spent its connection window +#[test] +fn zero_rtt_rejection_returns_connection_credit() { + let _guard = subscribe(); + let crypto = Arc::new(server_crypto()); + let server = |crypto: Arc| { + let mut config = ServerConfig::with_crypto(crypto); + let mut transport = transport(); + transport.receive_window(VarInt::from_u32(WINDOW)); + config.transport = Arc::new(transport); + Arc::new(config) + }; + let mut pair = Pair::new(Default::default(), (*server(crypto)).clone()); + let config = client_with(|_| {}); + + let client_ch = pair.begin_connect(config.clone()); + pair.drive(); + pair.server.assert_accept(); + let now = pair.time; + pair.client_conn_mut(client_ch) + .close(now, VarInt(0), Bytes::new()); + pair.drive(); + + // A server with a fresh session cache cannot resume, so it rejects the 0-RTT data + pair.server.set_server_config(Some(server(Arc::new(server_crypto())))); + let client_ch = pair.begin_connect(config); + assert!(pair.client_conn_mut(client_ch).has_0rtt()); + assert_eq!(writable(&mut pair, client_ch), WINDOW as usize); + assert_eq!(writable(&mut pair, client_ch), 0); + pair.drive(); + assert!(!pair.client_conn_mut(client_ch).accepted_0rtt()); + + assert_eq!(writable(&mut pair, client_ch), WINDOW as usize); +} diff --git a/rs/moq-quic/src/tests/mod.rs b/rs/moq-quic/src/tests/mod.rs index 6cc4f307f4d..8d59fb34981 100644 --- a/rs/moq-quic/src/tests/mod.rs +++ b/rs/moq-quic/src/tests/mod.rs @@ -42,6 +42,7 @@ use crate::{ mod util; use util::*; +mod characterize; mod reset_stream_at; mod token; From 2d7fe71bef5ef8ff92023ce90149aa29c54e8d3d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 15:19:54 -0700 Subject: [PATCH 03/11] test(quic): characterize runtime limits and endpoint routing Co-Authored-By: Claude Opus 5.5 --- .../src/tests/characterize/endpoint.rs | 332 ++++++++++++++++++ rs/moq-quic/src/tests/characterize/limits.rs | 207 +++++++++++ rs/moq-quic/src/tests/characterize/mod.rs | 2 + 3 files changed, 541 insertions(+) create mode 100644 rs/moq-quic/src/tests/characterize/endpoint.rs create mode 100644 rs/moq-quic/src/tests/characterize/limits.rs diff --git a/rs/moq-quic/src/tests/characterize/endpoint.rs b/rs/moq-quic/src/tests/characterize/endpoint.rs new file mode 100644 index 00000000000..d549e70d354 --- /dev/null +++ b/rs/moq-quic/src/tests/characterize/endpoint.rs @@ -0,0 +1,332 @@ +//! How an endpoint answers datagrams that match no connection + +use super::*; + +const CID_KEY: u64 = 0x1234_5678; +const REMOTE: &str = "[::2]:7890"; + +/// An endpoint config whose CIDs come from a hashed generator with a known key +fn endpoint_config() -> EndpointConfig { + let mut config = EndpointConfig::default(); + config.cid_generator(Arc::new(|| Box::new(HashedConnectionIdGenerator::from_key(CID_KEY)))); + config +} + +/// A server endpoint running [`endpoint_config`], or a client-only one +fn endpoint(server: bool) -> Endpoint { + let server = server.then(|| { + let mut config = server_config(); + config.transport = Arc::new(transport()); + Arc::new(config) + }); + Endpoint::new(Arc::new(endpoint_config()), server, true) +} + +/// A CID the endpoint would accept as its own, without a connection behind it +fn valid_cid() -> ConnectionId { + HashedConnectionIdGenerator::from_key(CID_KEY).generate_cid() +} + +/// A short-header datagram to `dst_cid`, `len` bytes long +fn short(dst_cid: &[u8], len: usize) -> Vec { + let mut packet = vec![0x40]; + packet.extend_from_slice(dst_cid); + packet.resize(len, 0xaa); + packet +} + +/// A long-header datagram of type `ty` (0 Initial, 2 Handshake), padded to `len` bytes +fn long(ty: u8, version: u32, dst_cid: &[u8], len: usize) -> Vec { + let mut packet = vec![0xc0 | (ty << 4)]; + packet.extend_from_slice(&version.to_be_bytes()); + packet.push(dst_cid.len() as u8); + packet.extend_from_slice(dst_cid); + packet.extend_from_slice(&[4, 1, 2, 3, 4]); + if ty == 0 { + // An empty token + packet.push(0); + } + let remaining = len - packet.len() - 2; + packet.write_var(remaining as u64); + packet.resize(len, 0xaa); + packet +} + +/// What `endpoint` answers `packet` with, at `now` +fn answer(endpoint: &mut Endpoint, now: Instant, packet: &[u8]) -> Answer { + let mut buf = Vec::new(); + match endpoint.handle(now, REMOTE.parse().unwrap(), None, None, packet.into(), &mut buf) { + None => Answer::Nothing, + Some(DatagramEvent::Response(transmit)) => Answer::Response(buf[..transmit.size].to_vec()), + Some(DatagramEvent::NewConnection(incoming)) => { + endpoint.ignore(incoming); + Answer::NewConnection + } + Some(DatagramEvent::ConnectionEvent(..)) => panic!("no connection exists"), + } +} + +#[derive(Debug, PartialEq, Eq)] +enum Answer { + Nothing, + Response(Vec), + NewConnection, +} + +impl Answer { + /// Whether this is `endpoint`'s stateless reset for `cid`: a short header ending in its token + fn is_reset_for(&self, endpoint: &Endpoint, cid: ConnectionId) -> bool { + let Self::Response(packet) = self else { + return false; + }; + let token = ResetToken::new(&*endpoint.config().reset_key, cid); + packet[0] & 0xc0 == 0x40 && packet.ends_with(&token[..]) + } +} + +/// An unknown but valid CID gets one stateless reset per interval, smaller than what incited it +#[test] +fn unknown_cid_gets_stateless_reset() { + let _guard = subscribe(); + let mut server = endpoint(true); + let now = Instant::now(); + let cid = valid_cid(); + let packet = short(&cid, 100); + + let first = answer(&mut server, now, &packet); + assert!(first.is_reset_for(&server, cid), "got {first:?}"); + let Answer::Response(reset) = first else { + unreachable!() + }; + assert!(reset.len() < packet.len()); + + assert_eq!(answer(&mut server, now, &packet), Answer::Nothing); + let later = now + server.config().min_reset_interval; + assert!(matches!(answer(&mut server, later, &packet), Answer::Response(_))); +} + +/// A reset must be smaller than its inciting datagram yet look like a short header, so datagrams +/// of 21 bytes or fewer get none +#[test] +fn small_unknown_packet_gets_no_reset() { + let _guard = subscribe(); + let now = Instant::now(); + let cid = valid_cid(); + assert_eq!(answer(&mut endpoint(true), now, &short(&cid, 21)), Answer::Nothing); + let Answer::Response(reset) = answer(&mut endpoint(true), now, &short(&cid, 22)) else { + panic!("expected a stateless reset"); + }; + assert_eq!(reset.len(), 21); +} + +/// The CID generator rejects a CID it could not have issued, so it gets no reset +#[test] +fn invalid_cid_is_dropped() { + let _guard = subscribe(); + let now = Instant::now(); + assert_eq!( + answer(&mut endpoint(true), now, &short(&[0xab; 8], 100)), + Answer::Nothing + ); +} + +#[test] +fn zero_length_cid_short_packet_is_dropped() { + let _guard = subscribe(); + let mut config = EndpointConfig::default(); + config.cid_generator(Arc::new(|| Box::new(RandomConnectionIdGenerator::new(0)))); + let mut server = Endpoint::new(Arc::new(config), Some(Arc::new(server_config())), true); + assert_eq!(answer(&mut server, Instant::now(), &short(&[], 100)), Answer::Nothing); +} + +/// Only Initials can start a connection, so another long header for an unknown CID is dropped +#[test] +fn long_header_for_unknown_connection_is_dropped() { + let _guard = subscribe(); + let now = Instant::now(); + let packet = long(2, DEFAULT_SUPPORTED_VERSIONS[0], &valid_cid(), 1200); + assert_eq!(answer(&mut endpoint(true), now, &packet), Answer::Nothing); +} + +/// An Initial in a datagram under 1200 bytes is dropped before the endpoint derives its keys, +/// while a client's real Initial is exactly 1200 +#[test] +fn short_initial_is_dropped() { + let _guard = subscribe(); + let packet = long(0, DEFAULT_SUPPORTED_VERSIONS[0], &[0x11; 8], 1199); + assert_eq!(answer(&mut endpoint(true), Instant::now(), &packet), Answer::Nothing); + + let mut pair = pair_with(|_| {}); + pair.begin_connect(client_with(|_| {})); + pair.drive_client(); + let (_, _, datagram) = pair.server.inbound.pop_front().unwrap(); + assert_eq!(datagram.len(), 1200); + let now = pair.time; + assert_eq!(answer(&mut pair.server, now, &datagram), Answer::NewConnection); +} + +/// An Initial whose DCID is shorter than RFC 9000's 8 bytes is refused with a close +#[test] +fn initial_with_short_cid_is_refused() { + let _guard = subscribe(); + let packet = long(0, DEFAULT_SUPPORTED_VERSIONS[0], &[0x11; 7], 1200); + let answer = answer(&mut endpoint(true), Instant::now(), &packet); + let Answer::Response(close) = answer else { + panic!("expected an Initial close, got {answer:?}"); + }; + assert_eq!(close[0] & 0xf0, 0xc0, "a long-header Initial"); +} + +/// Garbage under header protection unmasks to random reserved bits. The endpoint drops an +/// Initial whose bits are invalid, and takes one whose bits pass as a new connection, which +/// then fails to accept. +#[test] +fn undecryptable_initial() { + let _guard = subscribe(); + let mut server = endpoint(true); + let now = Instant::now(); + let mut packet = long(0, DEFAULT_SUPPORTED_VERSIONS[0], &[0x11; 8], 1200); + assert_eq!(answer(&mut server, now, &packet), Answer::Nothing); + + // Vary the header protection sample, 4 bytes past the packet number at byte 22, until the + // reserved bits unmask as zero + let mut buf = Vec::new(); + let incoming = (0..=u8::MAX) + .find_map(|byte| { + packet[26] = byte; + match server.handle(now, REMOTE.parse().unwrap(), None, None, packet[..].into(), &mut buf) { + Some(DatagramEvent::NewConnection(incoming)) => Some(incoming), + None => None, + Some(_) => panic!("unexpected response"), + } + }) + .expect("some sample unmasks valid reserved bits"); + let error = server.accept(incoming, now, &mut buf, None).unwrap_err(); + assert_matches!(error.cause, ConnectionError::TransportError(_)); + assert!(error.response.is_none()); + assert_eq!(server.open_connections(), 0); +} + +#[test] +fn undecodable_datagram_is_dropped() { + let _guard = subscribe(); + let now = Instant::now(); + // A long header cut off inside its version + assert_eq!(answer(&mut endpoint(true), now, &[0xc0, 0, 0]), Answer::Nothing); + // An Initial whose two-byte length field, after its 20 bytes of header, runs past the + // datagram + let mut packet = long(0, DEFAULT_SUPPORTED_VERSIONS[0], &[0x11; 8], 1200); + packet[20..22].copy_from_slice(&[0x7f, 0xff]); + assert_eq!(answer(&mut endpoint(true), now, &packet), Answer::Nothing); +} + +/// Only a server negotiates versions, and only for datagrams big enough to start a connection +#[test] +fn wrong_version_initial() { + let _guard = subscribe(); + let now = Instant::now(); + let packet = long(0, 0x0a1a_2a3a, &[0x11; 8], 1200); + assert!(matches!( + answer(&mut endpoint(true), now, &packet), + Answer::Response(_) + )); + assert_eq!(answer(&mut endpoint(false), now, &packet), Answer::Nothing); + let packet = long(0, 0x0a1a_2a3a, &[0x11; 8], 1199); + assert_eq!(answer(&mut endpoint(true), now, &packet), Answer::Nothing); +} + +/// A client-only endpoint answers both an Initial and a short header for an unknown connection +/// with a stateless reset +#[test] +fn client_endpoint_sends_stateless_reset() { + let _guard = subscribe(); + let now = Instant::now(); + let cid = valid_cid(); + let mut client = endpoint(false); + assert!(answer(&mut client, now, &short(&cid, 100)).is_reset_for(&client, cid)); + + let dst_cid = ConnectionId::new(&[0x11; 8]); + let packet = long(0, DEFAULT_SUPPORTED_VERSIONS[0], &dst_cid, 1200); + let mut client = endpoint(false); + assert!(answer(&mut client, now, &packet).is_reset_for(&client, dst_cid)); +} + +/// A server whose one-byte CID space is three-quarters used +fn exhausted() -> Pair { + let mut config = EndpointConfig::default(); + config.cid_generator(Arc::new(|| Box::new(RandomConnectionIdGenerator::new(1)))); + let config = Arc::new(config); + let mut server_config = server_config(); + server_config.transport = Arc::new(transport()); + let server = Endpoint::new(config, Some(Arc::new(server_config)), true); + let client = Endpoint::new(Arc::new(EndpointConfig::default()), None, true); + Pair::new_from_endpoint(client, server) +} + +/// Uses up the server's CID space with its own outgoing connections, returning how many it made +fn exhaust(pair: &mut Pair) -> usize { + let mut made = 0; + loop { + let remote = SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 1); + match pair + .server + .connect(pair.time, client_with(|_| {}), remote, "localhost") + { + Ok((ch, conn)) => { + pair.server.connections.insert(ch, conn); + made += 1; + } + Err(ConnectError::CidsExhausted) => return made, + Err(e) => panic!("{e}"), + } + } +} + +#[test] +fn cids_exhausted_refuses_connect() { + let _guard = subscribe(); + let mut pair = exhausted(); + // 256 CIDs, a quarter held in reserve + assert_eq!(exhaust(&mut pair), 193); +} + +/// Exhaustion before an Initial arrives drops it without a word +#[test] +fn cids_exhausted_drops_initial() { + let _guard = subscribe(); + let mut pair = exhausted(); + exhaust(&mut pair); + let client_ch = pair.begin_connect(client_with(|_| {})); + pair.drive_client(); + pair.drive_server(); + pair.server.assert_no_accept(); + assert!(pair.client.inbound.is_empty()); + assert!(pair.client_conn_mut(client_ch).is_handshaking()); +} + +/// Exhaustion between an Initial and its accept refuses the connection +#[test] +fn cids_exhausted_refuses_accept() { + let _guard = subscribe(); + let mut pair = exhausted(); + pair.server.handle_incoming = Box::new(|_| IncomingConnectionBehavior::Wait); + let client_ch = pair.begin_connect(client_with(|_| {})); + pair.drive_client(); + pair.drive_server(); + let incoming = pair.server.pop_waiting_incoming(); + + exhaust(&mut pair); + let now = pair.time; + assert_matches!( + pair.server.try_accept(incoming, now), + Err(ConnectionError::CidsExhausted) + ); + pair.drive(); + assert_matches!( + lost(pair.client_conn_mut(client_ch)), + Some(ConnectionError::ConnectionClosed(ConnectionClose { + error_code: TransportErrorCode::CONNECTION_REFUSED, + .. + })) + ); +} diff --git a/rs/moq-quic/src/tests/characterize/limits.rs b/rs/moq-quic/src/tests/characterize/limits.rs new file mode 100644 index 00000000000..f825ee0ec0b --- /dev/null +++ b/rs/moq-quic/src/tests/characterize/limits.rs @@ -0,0 +1,207 @@ +//! Stream and flow control limits changed while a connection is up + +use super::*; + +/// How many more uni streams the client can open right now, opening them all +fn openable(pair: &mut Pair, ch: ConnectionHandle) -> Vec { + iter::from_fn(|| pair.client_streams(ch).open(Dir::Uni)).collect() +} + +/// Finishes each client stream and has the server read it to the end +fn finish_all(pair: &mut Pair, client_ch: ConnectionHandle, server_ch: ConnectionHandle, streams: &[StreamId]) { + for &s in streams { + pair.client_send(client_ch, s).write(b"x").unwrap(); + pair.client_send(client_ch, s).finish().unwrap(); + } + pair.drive(); + while let Some(s) = pair.server_streams(server_ch).accept(Dir::Uni) { + let mut recv = pair.server_recv(server_ch, s); + let mut chunks = recv.read(true).unwrap(); + while chunks.next(usize::MAX).unwrap().is_some() {} + let _ = chunks.finalize(); + } + pair.drive(); +} + +#[test] +fn max_concurrent_streams_raised() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = + connected_with(|t| _ = t.max_concurrent_uni_streams(VarInt(1)), |_| {}); + assert_eq!(openable(&mut pair, client_ch).len(), 1); + + pair.server_conn_mut(server_ch) + .set_max_concurrent_streams(Dir::Uni, VarInt(3)); + assert_eq!(pair.server_conn_mut(server_ch).max_concurrent_streams(Dir::Uni), 3); + pair.drive(); + assert_eq!(openable(&mut pair, client_ch).len(), 2); +} + +/// Lowering the limit keeps the credit already issued, and withholds new credit until enough +/// streams close to fall under it +#[test] +fn max_concurrent_streams_lowered() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = + connected_with(|t| _ = t.max_concurrent_uni_streams(VarInt(3)), |_| {}); + pair.server_conn_mut(server_ch) + .set_max_concurrent_streams(Dir::Uni, VarInt(1)); + assert_eq!(pair.server_conn_mut(server_ch).max_concurrent_streams(Dir::Uni), 3); + pair.drive(); + + let streams = openable(&mut pair, client_ch); + assert_eq!(streams.len(), 3); + finish_all(&mut pair, client_ch, server_ch, &streams); + assert_eq!(pair.server_conn_mut(server_ch).max_concurrent_streams(Dir::Uni), 1); + assert_eq!(openable(&mut pair, client_ch).len(), 1); +} + +/// How many bytes a fresh client stream can write, up to `limit` +fn writable(pair: &mut Pair, ch: ConnectionHandle, limit: usize) -> usize { + let s = pair.client_streams(ch).open(Dir::Uni).unwrap(); + pair.client_send(ch, s).write(&vec![0; limit]).unwrap_or(0) +} + +/// Reads everything the server has buffered on every client uni stream, returning the total +fn read_buffered(pair: &mut Pair, ch: ConnectionHandle, streams: &mut Vec) -> usize { + streams.extend(iter::from_fn(|| pair.server_streams(ch).accept(Dir::Uni))); + let mut total = 0; + for &s in streams.iter() { + let mut recv = pair.server_recv(ch, s); + let Ok(mut chunks) = recv.read(true) else { + continue; + }; + while let Ok(Some(chunk)) = chunks.next(usize::MAX) { + total += chunk.bytes.len(); + } + let _ = chunks.finalize(); + } + total +} + +const WINDOW: u32 = 4000; + +#[test] +fn receive_window_raised() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = + connected_with(|t| _ = t.receive_window(VarInt::from_u32(WINDOW)), |_| {}); + assert_eq!(writable(&mut pair, client_ch, 2 * WINDOW as usize), WINDOW as usize); + pair.drive(); + + pair.server_conn_mut(server_ch) + .set_receive_window(VarInt::from_u32(2 * WINDOW)); + pair.drive(); + assert_eq!(writable(&mut pair, client_ch, 2 * WINDOW as usize), WINDOW as usize); +} + +/// A lowered window is paid down from credit the reader would otherwise return +#[test] +fn receive_window_lowered() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = + connected_with(|t| _ = t.receive_window(VarInt::from_u32(2 * WINDOW)), |_| {}); + assert_eq!(writable(&mut pair, client_ch, 4 * WINDOW as usize), 2 * WINDOW as usize); + pair.drive(); + + pair.server_conn_mut(server_ch) + .set_receive_window(VarInt::from_u32(WINDOW)); + let mut streams = Vec::new(); + assert_eq!(read_buffered(&mut pair, server_ch, &mut streams), 2 * WINDOW as usize); + pair.drive(); + assert_eq!(writable(&mut pair, client_ch, 4 * WINDOW as usize), WINDOW as usize); +} + +/// The send window bounds unacknowledged data, so lowering it blocks writes until acks drain +/// what is in flight +#[test] +fn send_window_lowered_and_raised() { + let _guard = subscribe(); + let (mut pair, client_ch, _) = connected(); + pair.client_conn_mut(client_ch).set_send_window(1000); + assert_eq!(writable(&mut pair, client_ch, 4000), 1000); + assert_eq!(writable(&mut pair, client_ch, 4000), 0); + + pair.client_conn_mut(client_ch).set_send_window(3000); + assert_eq!(writable(&mut pair, client_ch, 4000), 2000); + + // Acknowledged data frees the window again + pair.drive(); + assert_eq!(writable(&mut pair, client_ch, 4000), 3000); +} + +/// A server whose stream windows grew since the ticket the client resumes with +/// +/// The client opens 0-RTT streams with the remembered, smaller windows, and the larger ones +/// arrive with the handshake. Today `set_params` raises only peer-initiated bidi streams, which +/// cannot exist yet, so the client's own streams stay at the old window. The server, believing +/// it granted the new one, never sends MAX_STREAM_DATA, and the stream stalls for good. +fn zero_rtt_window_raise(dir: Dir) { + let _guard = subscribe(); + const OLD: u32 = 1000; + const NEW: u32 = 100_000; + const LEN: usize = 50_000; + + let crypto = Arc::new(server_crypto()); + let server = |window: u32| { + let mut config = ServerConfig::with_crypto(crypto.clone()); + let mut transport = transport(); + transport.stream_receive_window(VarInt::from_u32(window)); + config.transport = Arc::new(transport); + config + }; + let mut pair = Pair::new(Default::default(), server(OLD)); + let config = client_with(|_| {}); + + let client_ch = pair.begin_connect(config.clone()); + pair.drive(); + pair.server.assert_accept(); + let now = pair.time; + pair.client_conn_mut(client_ch) + .close(now, VarInt(0), Bytes::new()); + pair.drive(); + + pair.server.set_server_config(Some(Arc::new(server(NEW)))); + let client_ch = pair.begin_connect(config); + assert!(pair.client_conn_mut(client_ch).has_0rtt()); + let s = pair.client_streams(client_ch).open(dir).unwrap(); + let mut written = pair.client_send(client_ch, s).write(&[0; LEN]).unwrap(); + assert_eq!(written, OLD as usize); + + pair.drive(); + assert!(pair.client_conn_mut(client_ch).accepted_0rtt()); + let server_ch = pair.server.assert_accept(); + assert_eq!(pair.server_streams(server_ch).accept(dir), Some(s)); + + // Keep writing and reading until neither side makes progress + let mut read = 0; + loop { + let before = (written, read); + if let Ok(n) = pair.client_send(client_ch, s).write(&[0; LEN][written..]) { + written += n; + } + let mut recv = pair.server_recv(server_ch, s); + let mut chunks = recv.read(true).unwrap(); + while let Ok(Some(chunk)) = chunks.next(usize::MAX) { + read += chunk.bytes.len(); + } + let _ = chunks.finalize(); + pair.drive(); + if (written, read) == before { + break; + } + } + assert_eq!(read, LEN); +} + +#[test] +#[ignore = "bug: /quest/m1/quic/zero-rtt-window-raise.md"] +fn zero_rtt_bidi_window_raise() { + zero_rtt_window_raise(Dir::Bi); +} + +#[test] +#[ignore = "bug: /quest/m1/quic/zero-rtt-window-raise.md"] +fn zero_rtt_uni_window_raise() { + zero_rtt_window_raise(Dir::Uni); +} diff --git a/rs/moq-quic/src/tests/characterize/mod.rs b/rs/moq-quic/src/tests/characterize/mod.rs index e9618c99b7e..96ee3936275 100644 --- a/rs/moq-quic/src/tests/characterize/mod.rs +++ b/rs/moq-quic/src/tests/characterize/mod.rs @@ -8,7 +8,9 @@ use super::*; use crate::coding::{BufMutExt, Codec}; +mod endpoint; mod frames; +mod limits; mod reset; /// The transport every characterization pair runs: Cubic with deterministic packet numbers From 529170ff768b9dfdac8a220cbd4f4bc2eb352bdd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 15:27:16 -0700 Subject: [PATCH 04/11] test(quic): golden scheduling order and packet/path characterization Co-Authored-By: Claude Opus 5.5 --- rs/moq-quic/src/tests/characterize/mod.rs | 2 + rs/moq-quic/src/tests/characterize/paths.rs | 384 ++++++++++++++++++ .../src/tests/characterize/schedule.rs | 209 ++++++++++ 3 files changed, 595 insertions(+) create mode 100644 rs/moq-quic/src/tests/characterize/paths.rs create mode 100644 rs/moq-quic/src/tests/characterize/schedule.rs diff --git a/rs/moq-quic/src/tests/characterize/mod.rs b/rs/moq-quic/src/tests/characterize/mod.rs index 96ee3936275..beba3ecdf3e 100644 --- a/rs/moq-quic/src/tests/characterize/mod.rs +++ b/rs/moq-quic/src/tests/characterize/mod.rs @@ -11,7 +11,9 @@ use crate::coding::{BufMutExt, Codec}; mod endpoint; mod frames; mod limits; +mod paths; mod reset; +mod schedule; /// The transport every characterization pair runs: Cubic with deterministic packet numbers fn transport() -> TransportConfig { diff --git a/rs/moq-quic/src/tests/characterize/paths.rs b/rs/moq-quic/src/tests/characterize/paths.rs new file mode 100644 index 00000000000..f5b721ed760 --- /dev/null +++ b/rs/moq-quic/src/tests/characterize/paths.rs @@ -0,0 +1,384 @@ +//! Packet-level defenses and path changes + +use std::any::Any; + +use super::*; +use crate::crypto::{self, KeyPair, Keys}; + +const MSG: &[u8] = b"hello"; + +/// Failed decryptions a [`Limited`] connection tolerates +const INTEGRITY_LIMIT: u64 = 3; + +/// Client crypto whose packet keys tolerate only [`INTEGRITY_LIMIT`] failed decryptions +struct Limited(Arc); + +impl crypto::ClientConfig for Limited { + fn start_session( + self: Arc, + version: u32, + server_name: &str, + params: &TransportParameters, + ) -> Result, ConnectError> { + let inner = self.0.clone().start_session(version, server_name, params)?; + Ok(Box::new(LimitedSession(inner))) + } +} + +struct LimitedSession(Box); + +struct LimitedKey(Box); + +impl crypto::PacketKey for LimitedKey { + fn encrypt(&self, packet: u64, buf: &mut [u8], header_len: usize) { + self.0.encrypt(packet, buf, header_len) + } + fn decrypt( + &self, + packet: u64, + header: &[u8], + payload: &mut BytesMut, + ) -> Result<(), crypto::CryptoError> { + self.0.decrypt(packet, header, payload) + } + fn tag_len(&self) -> usize { + self.0.tag_len() + } + fn confidentiality_limit(&self) -> u64 { + self.0.confidentiality_limit() + } + fn integrity_limit(&self) -> u64 { + INTEGRITY_LIMIT + } +} + +fn limit_pair(keys: KeyPair>) -> KeyPair> { + KeyPair { + local: Box::new(LimitedKey(keys.local)), + remote: Box::new(LimitedKey(keys.remote)), + } +} + +fn limit(keys: Keys) -> Keys { + Keys { + header: keys.header, + packet: limit_pair(keys.packet), + } +} + +impl crypto::Session for LimitedSession { + fn initial_keys(&self, dst_cid: ConnectionId, side: Side) -> Keys { + limit(self.0.initial_keys(dst_cid, side)) + } + fn handshake_data(&self) -> Option> { + self.0.handshake_data() + } + fn peer_identity(&self) -> Option> { + self.0.peer_identity() + } + fn early_crypto(&self) -> Option<(Box, Box)> { + self.0.early_crypto() + } + fn early_data_accepted(&self) -> Option { + self.0.early_data_accepted() + } + fn is_handshaking(&self) -> bool { + self.0.is_handshaking() + } + fn read_handshake(&mut self, buf: &[u8]) -> Result { + self.0.read_handshake(buf) + } + fn transport_parameters(&self) -> Result, TransportError> { + self.0.transport_parameters() + } + fn write_handshake(&mut self, buf: &mut Vec) -> Option { + self.0.write_handshake(buf).map(limit) + } + fn next_1rtt_keys(&mut self) -> Option>> { + self.0.next_1rtt_keys().map(limit_pair) + } + fn is_valid_retry(&self, orig_dst_cid: ConnectionId, header: &[u8], payload: &[u8]) -> bool { + self.0.is_valid_retry(orig_dst_cid, header, payload) + } + fn export_keying_material( + &self, + output: &mut [u8], + label: &[u8], + context: &[u8], + ) -> Result<(), crypto::ExportKeyingMaterialError> { + self.0.export_keying_material(output, label, context) + } +} + +/// Packets that fail authentication past the integrity limit kill the connection silently +#[test] +fn aead_integrity_limit() { + let _guard = subscribe(); + let mut pair = pair_with(|_| {}); + let mut config = ClientConfig::new(Arc::new(Limited(Arc::new(client_crypto())))); + config.transport = Arc::new(transport()); + let (client_ch, server_ch) = pair.connect_with(config); + + for sent in 1..=INTEGRITY_LIMIT + 1 { + let now = pair.time; + let mut packet = pair + .server_conn_mut(server_ch) + .forge_packet(now, SpaceId::Data, &[0; 32]); + // Flip a byte of the AEAD tag + *packet.last_mut().unwrap() ^= 0xff; + pair.client + .inbound + .push_back((now, None, packet.as_slice().into())); + pair.drive_client(); + let lost = lost(pair.client_conn_mut(client_ch)); + if sent <= INTEGRITY_LIMIT { + assert_eq!(lost, None); + } else { + assert_matches!( + lost, + Some(ConnectionError::TransportError(TransportError { + code: TransportErrorCode::AEAD_LIMIT_REACHED, + .. + })) + ); + } + } + + // The client sends no CONNECTION_CLOSE + assert!(pair.client_conn_mut(client_ch).is_drained()); + assert!(pair.server.inbound.is_empty()); +} + +/// A replayed packet is discarded before its frames are processed +#[test] +fn duplicate_packet_is_discarded() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = connected(); + pair.drive(); + let s = pair.client_streams(client_ch).open(Dir::Uni).unwrap(); + pair.client_send(client_ch, s).write(MSG).unwrap(); + pair.drive_client(); + assert_eq!(pair.server.inbound.len(), 1); + let duplicate = pair.server.inbound[0].clone(); + pair.server.inbound.push_back(duplicate); + let before = pair.server_conn_mut(server_ch).stats().frame_rx.stream; + pair.drive(); + + assert_eq!(pair.server_conn_mut(server_ch).stats().frame_rx.stream, before + 1); + assert_eq!(pair.server_streams(server_ch).accept(Dir::Uni), Some(s)); + let mut recv = pair.server_recv(server_ch, s); + let mut chunks = recv.read(true).unwrap(); + assert_eq!(chunks.next(usize::MAX).unwrap().unwrap().bytes, MSG); + assert_eq!(chunks.next(usize::MAX), Err(ReadError::Blocked)); + let _ = chunks.finalize(); +} + +/// Rebuilds a Retry datagram with `src_cid`, under a valid integrity tag +fn forge_retry(retry: &[u8], src_cid: &[u8], orig_dst_cid: ConnectionId) -> Vec { + let version = u32::from_be_bytes(retry[1..5].try_into().unwrap()); + let dst_cid_end = 6 + retry[5] as usize; + let src_cid_end = dst_cid_end + 1 + retry[dst_cid_end] as usize; + let mut packet = retry[..dst_cid_end].to_vec(); + packet.push(src_cid.len() as u8); + packet.extend_from_slice(src_cid); + packet.extend_from_slice(&retry[src_cid_end..retry.len() - 16]); + let tag = server_config() + .crypto + .retry_tag(version, orig_dst_cid, &packet); + packet.extend_from_slice(&tag); + packet +} + +/// Rewrites the server's Retry datagram, given the client's original DCID, into the datagrams +/// the client receives instead +type Mangle = fn(&[u8], ConnectionId) -> Vec>; + +/// A Retry whose integrity tag fails +fn bad_tag(retry: &[u8], _: ConnectionId) -> Vec> { + let mut retry = retry.to_vec(); + *retry.last_mut().unwrap() ^= 0xff; + vec![retry] +} + +/// A Retry naming the client's original DCID as the server's new CID +fn same_cid(retry: &[u8], orig_dst_cid: ConnectionId) -> Vec> { + vec![forge_retry(retry, &orig_dst_cid, orig_dst_cid)] +} + +/// The server's Retry, then a second, different one +fn second(retry: &[u8], orig_dst_cid: ConnectionId) -> Vec> { + vec![retry.to_vec(), forge_retry(retry, &[0x22; 8], orig_dst_cid)] +} + +/// Starts a connection the server Retries, queueing for the client what `mangle` makes of the +/// Retry, and returning the client's original DCID +fn retried(mangle: Mangle) -> (Pair, ConnectionHandle, ConnectionId) { + let mut pair = pair_with(|_| {}); + pair.server.handle_incoming = Box::new(validate_incoming); + let client_ch = pair.begin_connect(client_with(|_| {})); + pair.drive_client(); + let initial = &pair.server.inbound[0].2; + let orig_dst_cid = ConnectionId::new(&initial[6..6 + initial[5] as usize]); + pair.drive_server(); + let (time, ecn, retry) = pair.client.inbound.pop_back().unwrap(); + assert!(pair.client.inbound.is_empty()); + assert_eq!(retry[0] & 0xf0, 0xf0, "a Retry"); + for datagram in mangle(&retry, orig_dst_cid) { + pair.client + .inbound + .push_back((time, ecn, datagram.as_slice().into())); + } + (pair, client_ch, orig_dst_cid) +} + +/// The client's next Initial keeps its original DCID unless it accepted a Retry +#[test] +fn invalid_retry_is_discarded() { + let _guard = subscribe(); + for (mangle, accepted) in [(bad_tag as Mangle, false), (same_cid, false), (second, true)] { + let (mut pair, _, orig_dst_cid) = retried(mangle); + pair.drive_client(); + while pair.server.inbound.is_empty() { + pair.time = pair.client.next_wakeup().unwrap(); + pair.drive_client(); + } + let initial = &pair.server.inbound[0].2; + let dst_cid = ConnectionId::new(&initial[6..6 + initial[5] as usize]); + assert_eq!(dst_cid != orig_dst_cid, accepted); + } +} + +/// The client follows the first Retry and ignores the second +#[test] +fn second_retry_is_discarded() { + let _guard = subscribe(); + let (mut pair, client_ch, _) = retried(second); + pair.drive(); + let conn = pair.client_conn_mut(client_ch); + assert!(!conn.is_handshaking()); + assert!(lost(conn).is_none()); +} + +/// A discarded Retry still counts as an authenticated packet, so the client then discards the +/// server's genuine Retry of its retransmitted Initial too, and never connects +#[test] +#[ignore = "bug: /quest/m1/quic/retry-after-invalid-retry.md"] +fn retry_after_invalid_retry() { + let _guard = subscribe(); + for mangle in [bad_tag as Mangle, same_cid] { + let (mut pair, client_ch, _) = retried(mangle); + pair.drive(); + assert!(!pair.client_conn_mut(client_ch).is_handshaking()); + } +} + +/// 0-RTT data sent before a Retry is discarded by the server, so the client sends it again +#[test] +fn zero_rtt_retransmit_after_retry() { + let _guard = subscribe(); + let mut pair = pair_with(|_| {}); + pair.server.handle_incoming = Box::new(validate_incoming); + // Without a NEW_TOKEN to spend, every connection is retried + let mut config = client_with(|_| {}); + config.token_store(Arc::new(NoneTokenStore)); + + let client_ch = pair.begin_connect(config.clone()); + pair.drive(); + pair.server.assert_accept(); + let now = pair.time; + pair.client_conn_mut(client_ch) + .close(now, VarInt(0), Bytes::new()); + pair.drive(); + + let client_ch = pair.begin_connect(config); + assert!(pair.client_conn_mut(client_ch).has_0rtt()); + let s = pair.client_streams(client_ch).open(Dir::Uni).unwrap(); + pair.client_send(client_ch, s).write(MSG).unwrap(); + pair.drive(); + + let server_ch = pair.server.assert_accept(); + assert!(pair.client_conn_mut(client_ch).accepted_0rtt()); + assert_eq!(pair.client_conn_mut(client_ch).stats().frame_tx.stream, 2); + assert_eq!(pair.server_streams(server_ch).accept(Dir::Uni), Some(s)); + let mut recv = pair.server_recv(server_ch, s); + let mut chunks = recv.read(true).unwrap(); + assert_eq!(chunks.next(usize::MAX).unwrap().unwrap().bytes, MSG); + let _ = chunks.finalize(); +} + +/// A packet that arrives from a new address moves the server to it, and when the new address +/// never answers the challenge, the server falls back to the address it validated before +#[test] +fn path_validation_failure_reverts() { + let _guard = subscribe(); + let (mut pair, client_ch, server_ch) = connected(); + pair.drive(); + let old = pair.client.addr; + let new = SocketAddr::new(old.ip(), CLIENT_PORTS.lock().unwrap().next().unwrap()); + + // One client packet arrives from the new address, as if forwarded by an attacker + pair.client_conn_mut(client_ch).ping(); + pair.drive_client(); + let now = pair.time; + pair.server.drive(now, new); + assert_eq!(pair.server_conn_mut(server_ch).remote_address(), new); + + // Nothing the server sends arrives, and the client stays silent + let mut updates = Vec::new(); + while pair.server_conn_mut(server_ch).remote_address() == new { + pair.server.outbound.clear(); + pair.time = pair.server.next_wakeup().expect("validation times out"); + let now = pair.time; + pair.server.drive(now, old); + while let Some(event) = pair.server_conn_mut(server_ch).poll() { + updates.push(event); + } + } + pair.server.outbound.clear(); + assert_eq!(pair.server_conn_mut(server_ch).remote_address(), old); + assert_matches!(updates.as_slice(), [.., Event::PathUpdated]); + + // The connection still works on the old path + let s = pair.client_streams(client_ch).open(Dir::Uni).unwrap(); + pair.client_send(client_ch, s).write(MSG).unwrap(); + pair.drive(); + assert_eq!(pair.server_streams(server_ch).accept(Dir::Uni), Some(s)); +} + +/// The RTT the server measures after the client's port changes +fn rtt_after_rebinding(ip: IpAddr) -> (Duration, Duration) { + let _guard = subscribe(); + let mut pair = pair_with(|_| {}); + pair.latency = Duration::from_millis(50); + pair.client.addr = SocketAddr::new(ip, CLIENT_PORTS.lock().unwrap().next().unwrap()); + pair.server.addr = SocketAddr::new(ip, pair.server.addr.port()); + let (client_ch, server_ch) = pair.connect_with(client_with(|_| {})); + pair.drive(); + let before = pair.server_conn_mut(server_ch).rtt(); + + pair.client.addr = SocketAddr::new(ip, CLIENT_PORTS.lock().unwrap().next().unwrap()); + pair.client_conn_mut(client_ch).ping(); + pair.drive_client(); + let now = pair.time + pair.latency; + pair.time = now; + let new = pair.client.addr; + pair.server.drive(now, new); + assert_eq!(pair.server_conn_mut(server_ch).remote_address(), new); + (before, pair.server_conn_mut(server_ch).rtt()) +} + +/// A port change on the same IPv4 address looks like NAT rebinding and keeps the path's RTT and +/// congestion state +#[test] +fn ipv4_rebinding_keeps_path_state() { + let (before, after) = rtt_after_rebinding(Ipv4Addr::LOCALHOST.into()); + assert_eq!(before, after); +} + +/// The same port change over IPv6 starts the path over from the initial RTT +#[test] +fn ipv6_rebinding_resets_path_state() { + let (before, after) = rtt_after_rebinding(Ipv6Addr::LOCALHOST.into()); + assert_ne!(before, after); + assert_eq!(after, TransportConfig::default().initial_rtt); +} diff --git a/rs/moq-quic/src/tests/characterize/schedule.rs b/rs/moq-quic/src/tests/characterize/schedule.rs new file mode 100644 index 00000000000..51755d25f5d --- /dev/null +++ b/rs/moq-quic/src/tests/characterize/schedule.rs @@ -0,0 +1,209 @@ +//! Golden tests of the order the client sends stream and datagram bytes in +//! +//! Each golden lists the packets the server received carrying stream or datagram data, one per +//! line: `s@+` for a STREAM frame (`!` marks FIN) and `d+` for a +//! DATAGRAM. A scheduler change that reorders bytes shows up as a diff of these lines. + +use super::*; + +const LEN: usize = 3000; + +/// Drives the pair to idle, asserting the server received stream and datagram frames in the +/// packets `golden` lists, one per line +#[track_caller] +fn assert_golden(pair: &mut Pair, golden: &str) { + pair.server.capture_inbound_packets = true; + pair.drive(); + pair.server.capture_inbound_packets = false; + let mut lines = Vec::new(); + for payload in mem::take(&mut pair.server.captured_packets) { + let mut line = Vec::new(); + for frame in frame::Iter::new(payload.into()).unwrap() { + match frame.unwrap() { + Frame::Stream(s) => { + let fin = if s.fin { "!" } else { "" }; + let id = u64::from(s.id); + line.push(format!("s{id}@{}+{}{fin}", s.offset, s.data.len())); + } + Frame::Datagram(d) => line.push(format!("d+{}", d.data.len())), + _ => {} + } + } + if !line.is_empty() { + lines.push(line.join(" ")); + } + } + let golden: Vec<_> = golden + .lines() + .map(str::trim) + .filter(|line| !line.is_empty()) + .collect(); + assert_eq!(lines.join("\n"), golden.join("\n")); +} + +/// Opens a client uni stream at `priority` and queues `LEN` bytes and a FIN on it +fn queue(pair: &mut Pair, ch: ConnectionHandle, priority: i32) -> StreamId { + let s = pair.client_streams(ch).open(Dir::Uni).unwrap(); + pair.client_send(ch, s).set_priority(priority).unwrap(); + assert_eq!(pair.client_send(ch, s).write(&[0; LEN]).unwrap(), LEN); + pair.client_send(ch, s).finish().unwrap(); + s +} + +/// A connected pair with the server's flush of post-handshake traffic out of the way +fn idle(fairness: bool) -> (Pair, ConnectionHandle) { + let (mut pair, client_ch, _) = connected_with(|_| {}, |t| _ = t.send_fairness(fairness)); + pair.drive(); + (pair, client_ch) +} + +/// Higher priorities drain first, whatever order the streams opened in +#[test] +fn priorities() { + let _guard = subscribe(); + let (mut pair, ch) = idle(true); + queue(&mut pair, ch, 0); + queue(&mut pair, ch, 10); + queue(&mut pair, ch, -5); + assert_golden( + &mut pair, + " + s6@0+1424 + s6@1424+1422 + s6@2846+154! s2@0+1264 + s2@1264+1422 + s2@2686+314! s10@0+1104 + s10@1104+1422 + s10@2526+474! + ", + ); +} + +/// Equal-priority streams take turns, a packet at a time +#[test] +fn equal_priority_fair() { + let _guard = subscribe(); + let (mut pair, ch) = idle(true); + for _ in 0..3 { + queue(&mut pair, ch, 0); + } + assert_golden( + &mut pair, + " + s2@0+1424 + s6@0+1424 + s10@0+1424 + s2@1424+1422 + s6@1424+1422 + s10@1424+1422 + s2@2846+154! s6@2846+154! s10@2846+154! + ", + ); +} + +/// Without fairness, equal-priority streams drain in the order they were opened +#[test] +fn equal_priority_unfair() { + let _guard = subscribe(); + let (mut pair, ch) = idle(false); + for _ in 0..3 { + queue(&mut pair, ch, 0); + } + assert_golden( + &mut pair, + " + s2@0+1424 + s2@1424+1422 + s2@2846+154! s6@0+1264 + s6@1264+1422 + s6@2686+314! s10@0+1104 + s10@1104+1422 + s10@2526+474! + ", + ); +} + +/// A lost packet's data waits behind a higher-priority stream queued after the loss +#[test] +fn retransmit_behind_higher_priority() { + let _guard = subscribe(); + let (mut pair, ch) = idle(true); + queue(&mut pair, ch, 0); + + // Lose the first packet of the flight + let now = pair.time; + pair.client.drive(now, pair.server.addr); + let _ = pair.client.outbound.pop_front().unwrap(); + queue(&mut pair, ch, 5); + assert_golden( + &mut pair, + " + s2@1424+1422 + s2@2846+154! + s6@0+1424 + s6@1424+1422 + s6@2846+154! + s2@0+1424 + ", + ); +} + +/// Datagrams fill each packet before stream data, which takes the rest +#[test] +fn datagrams_interleaved() { + let _guard = subscribe(); + let (mut pair, ch) = idle(true); + queue(&mut pair, ch, 0); + for _ in 0..3 { + pair.client_datagrams(ch) + .send(vec![0; 500].into(), false) + .unwrap(); + } + queue(&mut pair, ch, 0); + pair.client_datagrams(ch) + .send(vec![0; 1000].into(), false) + .unwrap(); + assert_golden( + &mut pair, + " + d+500 d+500 s2@0+418 + d+500 s6@0+921 + d+1000 s2@418+419 + s6@921+1422 + s2@837+1422 + s6@2343+657! s2@2259+741! + ", + ); +} + +/// A stream longer than the congestion window loses a packet of its first flight; the lost +/// range goes out again ahead of the stream's unsent data +#[test] +fn retransmit_within_stream() { + let _guard = subscribe(); + let (mut pair, ch) = idle(true); + let s = pair.client_streams(ch).open(Dir::Uni).unwrap(); + assert_eq!(pair.client_send(ch, s).write(&[0; 16_000]).unwrap(), 16_000); + + // Lose the second packet of the first flight + let now = pair.time; + pair.client.drive(now, pair.server.addr); + let _ = pair.client.outbound.remove(1).unwrap(); + assert_golden( + &mut pair, + " + s2@0+1424 + s2@2846+1422 + s2@4268+1422 + s2@5690+1422 + s2@7112+1422 + s2@8534+1422 + s2@9956+1422 + s2@1424+1422 + s2@11378+1422 + s2@12800+1422 + s2@14222+1422 + s2@15644+356 + ", + ); +} From 77f75137ff084cb764f872293367428d5c523ffa Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 15:40:06 -0700 Subject: [PATCH 05/11] style(quic): format characterization tests Co-Authored-By: Claude Opus 5.5 --- .../src/tests/characterize/endpoint.rs | 52 ++++++++++++++--- rs/moq-quic/src/tests/characterize/frames.rs | 26 +++------ rs/moq-quic/src/tests/characterize/limits.rs | 56 +++++++++++++++---- rs/moq-quic/src/tests/characterize/paths.rs | 11 +++- rs/moq-quic/src/tests/characterize/reset.rs | 31 ++++++++-- 5 files changed, 131 insertions(+), 45 deletions(-) diff --git a/rs/moq-quic/src/tests/characterize/endpoint.rs b/rs/moq-quic/src/tests/characterize/endpoint.rs index d549e70d354..d445c20a94d 100644 --- a/rs/moq-quic/src/tests/characterize/endpoint.rs +++ b/rs/moq-quic/src/tests/characterize/endpoint.rs @@ -8,7 +8,9 @@ const REMOTE: &str = "[::2]:7890"; /// An endpoint config whose CIDs come from a hashed generator with a known key fn endpoint_config() -> EndpointConfig { let mut config = EndpointConfig::default(); - config.cid_generator(Arc::new(|| Box::new(HashedConnectionIdGenerator::from_key(CID_KEY)))); + config.cid_generator(Arc::new(|| { + Box::new(HashedConnectionIdGenerator::from_key(CID_KEY)) + })); config } @@ -55,7 +57,14 @@ fn long(ty: u8, version: u32, dst_cid: &[u8], len: usize) -> Vec { /// What `endpoint` answers `packet` with, at `now` fn answer(endpoint: &mut Endpoint, now: Instant, packet: &[u8]) -> Answer { let mut buf = Vec::new(); - match endpoint.handle(now, REMOTE.parse().unwrap(), None, None, packet.into(), &mut buf) { + match endpoint.handle( + now, + REMOTE.parse().unwrap(), + None, + None, + packet.into(), + &mut buf, + ) { None => Answer::Nothing, Some(DatagramEvent::Response(transmit)) => Answer::Response(buf[..transmit.size].to_vec()), Some(DatagramEvent::NewConnection(incoming)) => { @@ -102,7 +111,10 @@ fn unknown_cid_gets_stateless_reset() { assert_eq!(answer(&mut server, now, &packet), Answer::Nothing); let later = now + server.config().min_reset_interval; - assert!(matches!(answer(&mut server, later, &packet), Answer::Response(_))); + assert!(matches!( + answer(&mut server, later, &packet), + Answer::Response(_) + )); } /// A reset must be smaller than its inciting datagram yet look like a short header, so datagrams @@ -112,7 +124,10 @@ fn small_unknown_packet_gets_no_reset() { let _guard = subscribe(); let now = Instant::now(); let cid = valid_cid(); - assert_eq!(answer(&mut endpoint(true), now, &short(&cid, 21)), Answer::Nothing); + assert_eq!( + answer(&mut endpoint(true), now, &short(&cid, 21)), + Answer::Nothing + ); let Answer::Response(reset) = answer(&mut endpoint(true), now, &short(&cid, 22)) else { panic!("expected a stateless reset"); }; @@ -136,7 +151,10 @@ fn zero_length_cid_short_packet_is_dropped() { let mut config = EndpointConfig::default(); config.cid_generator(Arc::new(|| Box::new(RandomConnectionIdGenerator::new(0)))); let mut server = Endpoint::new(Arc::new(config), Some(Arc::new(server_config())), true); - assert_eq!(answer(&mut server, Instant::now(), &short(&[], 100)), Answer::Nothing); + assert_eq!( + answer(&mut server, Instant::now(), &short(&[], 100)), + Answer::Nothing + ); } /// Only Initials can start a connection, so another long header for an unknown CID is dropped @@ -154,7 +172,10 @@ fn long_header_for_unknown_connection_is_dropped() { fn short_initial_is_dropped() { let _guard = subscribe(); let packet = long(0, DEFAULT_SUPPORTED_VERSIONS[0], &[0x11; 8], 1199); - assert_eq!(answer(&mut endpoint(true), Instant::now(), &packet), Answer::Nothing); + assert_eq!( + answer(&mut endpoint(true), Instant::now(), &packet), + Answer::Nothing + ); let mut pair = pair_with(|_| {}); pair.begin_connect(client_with(|_| {})); @@ -162,7 +183,10 @@ fn short_initial_is_dropped() { let (_, _, datagram) = pair.server.inbound.pop_front().unwrap(); assert_eq!(datagram.len(), 1200); let now = pair.time; - assert_eq!(answer(&mut pair.server, now, &datagram), Answer::NewConnection); + assert_eq!( + answer(&mut pair.server, now, &datagram), + Answer::NewConnection + ); } /// An Initial whose DCID is shorter than RFC 9000's 8 bytes is refused with a close @@ -194,7 +218,14 @@ fn undecryptable_initial() { let incoming = (0..=u8::MAX) .find_map(|byte| { packet[26] = byte; - match server.handle(now, REMOTE.parse().unwrap(), None, None, packet[..].into(), &mut buf) { + match server.handle( + now, + REMOTE.parse().unwrap(), + None, + None, + packet[..].into(), + &mut buf, + ) { Some(DatagramEvent::NewConnection(incoming)) => Some(incoming), None => None, Some(_) => panic!("unexpected response"), @@ -212,7 +243,10 @@ fn undecodable_datagram_is_dropped() { let _guard = subscribe(); let now = Instant::now(); // A long header cut off inside its version - assert_eq!(answer(&mut endpoint(true), now, &[0xc0, 0, 0]), Answer::Nothing); + assert_eq!( + answer(&mut endpoint(true), now, &[0xc0, 0, 0]), + Answer::Nothing + ); // An Initial whose two-byte length field, after its 20 bytes of header, runs past the // datagram let mut packet = long(0, DEFAULT_SUPPORTED_VERSIONS[0], &[0x11; 8], 1200); diff --git a/rs/moq-quic/src/tests/characterize/frames.rs b/rs/moq-quic/src/tests/characterize/frames.rs index fe83ff9a6bc..f2255a63261 100644 --- a/rs/moq-quic/src/tests/characterize/frames.rs +++ b/rs/moq-quic/src/tests/characterize/frames.rs @@ -118,10 +118,8 @@ fn max_stream_data_on_unopened() { #[test] fn stop_sending_on_recv_only() { - let frames = Frames::default().var( - FrameType::STOP_SENDING, - &[id(Side::Client, Dir::Uni, 0), 0], - ); + let frames = + Frames::default().var(FrameType::STOP_SENDING, &[id(Side::Client, Dir::Uni, 0), 0]); assert_eq!( server_answer(frames), Some(TransportErrorCode::STREAM_STATE_ERROR) @@ -130,8 +128,7 @@ fn stop_sending_on_recv_only() { #[test] fn stop_sending_on_unopened() { - let frames = - Frames::default().var(FrameType::STOP_SENDING, &[id(Side::Server, Dir::Bi, 0), 0]); + let frames = Frames::default().var(FrameType::STOP_SENDING, &[id(Side::Server, Dir::Bi, 0), 0]); assert_eq!( server_answer(frames), Some(TransportErrorCode::STREAM_STATE_ERROR) @@ -140,7 +137,10 @@ fn stop_sending_on_unopened() { #[test] fn streams_blocked_above_limit() { - for ty in [FrameType::STREAMS_BLOCKED_BIDI, FrameType::STREAMS_BLOCKED_UNI] { + for ty in [ + FrameType::STREAMS_BLOCKED_BIDI, + FrameType::STREAMS_BLOCKED_UNI, + ] { // The largest representable limit is fine let frames = Frames::default().var(ty, &[1 << 60]); assert_eq!(server_answer(frames), None); @@ -427,11 +427,7 @@ fn reset_on_send_only() { #[test] fn stream_on_send_only() { - let frames = Frames::default().stream(id(Side::Server, Dir::Uni, 0), - 0, - b"x", - false, - ); + let frames = Frames::default().stream(id(Side::Server, Dir::Uni, 0), 0, b"x", false); assert_eq!( server_answer(frames), Some(TransportErrorCode::STREAM_STATE_ERROR) @@ -440,11 +436,7 @@ fn stream_on_send_only() { #[test] fn stream_above_stream_limit() { - let frames = Frames::default().stream(id(Side::Client, Dir::Uni, 1_000_000), - 0, - b"x", - false, - ); + let frames = Frames::default().stream(id(Side::Client, Dir::Uni, 1_000_000), 0, b"x", false); assert_eq!( server_answer(frames), Some(TransportErrorCode::STREAM_LIMIT_ERROR) diff --git a/rs/moq-quic/src/tests/characterize/limits.rs b/rs/moq-quic/src/tests/characterize/limits.rs index f825ee0ec0b..61f26367b15 100644 --- a/rs/moq-quic/src/tests/characterize/limits.rs +++ b/rs/moq-quic/src/tests/characterize/limits.rs @@ -8,7 +8,12 @@ fn openable(pair: &mut Pair, ch: ConnectionHandle) -> Vec { } /// Finishes each client stream and has the server read it to the end -fn finish_all(pair: &mut Pair, client_ch: ConnectionHandle, server_ch: ConnectionHandle, streams: &[StreamId]) { +fn finish_all( + pair: &mut Pair, + client_ch: ConnectionHandle, + server_ch: ConnectionHandle, + streams: &[StreamId], +) { for &s in streams { pair.client_send(client_ch, s).write(b"x").unwrap(); pair.client_send(client_ch, s).finish().unwrap(); @@ -32,7 +37,11 @@ fn max_concurrent_streams_raised() { pair.server_conn_mut(server_ch) .set_max_concurrent_streams(Dir::Uni, VarInt(3)); - assert_eq!(pair.server_conn_mut(server_ch).max_concurrent_streams(Dir::Uni), 3); + assert_eq!( + pair.server_conn_mut(server_ch) + .max_concurrent_streams(Dir::Uni), + 3 + ); pair.drive(); assert_eq!(openable(&mut pair, client_ch).len(), 2); } @@ -46,13 +55,21 @@ fn max_concurrent_streams_lowered() { connected_with(|t| _ = t.max_concurrent_uni_streams(VarInt(3)), |_| {}); pair.server_conn_mut(server_ch) .set_max_concurrent_streams(Dir::Uni, VarInt(1)); - assert_eq!(pair.server_conn_mut(server_ch).max_concurrent_streams(Dir::Uni), 3); + assert_eq!( + pair.server_conn_mut(server_ch) + .max_concurrent_streams(Dir::Uni), + 3 + ); pair.drive(); let streams = openable(&mut pair, client_ch); assert_eq!(streams.len(), 3); finish_all(&mut pair, client_ch, server_ch, &streams); - assert_eq!(pair.server_conn_mut(server_ch).max_concurrent_streams(Dir::Uni), 1); + assert_eq!( + pair.server_conn_mut(server_ch) + .max_concurrent_streams(Dir::Uni), + 1 + ); assert_eq!(openable(&mut pair, client_ch).len(), 1); } @@ -86,30 +103,47 @@ fn receive_window_raised() { let _guard = subscribe(); let (mut pair, client_ch, server_ch) = connected_with(|t| _ = t.receive_window(VarInt::from_u32(WINDOW)), |_| {}); - assert_eq!(writable(&mut pair, client_ch, 2 * WINDOW as usize), WINDOW as usize); + assert_eq!( + writable(&mut pair, client_ch, 2 * WINDOW as usize), + WINDOW as usize + ); pair.drive(); pair.server_conn_mut(server_ch) .set_receive_window(VarInt::from_u32(2 * WINDOW)); pair.drive(); - assert_eq!(writable(&mut pair, client_ch, 2 * WINDOW as usize), WINDOW as usize); + assert_eq!( + writable(&mut pair, client_ch, 2 * WINDOW as usize), + WINDOW as usize + ); } /// A lowered window is paid down from credit the reader would otherwise return #[test] fn receive_window_lowered() { let _guard = subscribe(); - let (mut pair, client_ch, server_ch) = - connected_with(|t| _ = t.receive_window(VarInt::from_u32(2 * WINDOW)), |_| {}); - assert_eq!(writable(&mut pair, client_ch, 4 * WINDOW as usize), 2 * WINDOW as usize); + let (mut pair, client_ch, server_ch) = connected_with( + |t| _ = t.receive_window(VarInt::from_u32(2 * WINDOW)), + |_| {}, + ); + assert_eq!( + writable(&mut pair, client_ch, 4 * WINDOW as usize), + 2 * WINDOW as usize + ); pair.drive(); pair.server_conn_mut(server_ch) .set_receive_window(VarInt::from_u32(WINDOW)); let mut streams = Vec::new(); - assert_eq!(read_buffered(&mut pair, server_ch, &mut streams), 2 * WINDOW as usize); + assert_eq!( + read_buffered(&mut pair, server_ch, &mut streams), + 2 * WINDOW as usize + ); pair.drive(); - assert_eq!(writable(&mut pair, client_ch, 4 * WINDOW as usize), WINDOW as usize); + assert_eq!( + writable(&mut pair, client_ch, 4 * WINDOW as usize), + WINDOW as usize + ); } /// The send window bounds unacknowledged data, so lowering it blocks writes until acks drain diff --git a/rs/moq-quic/src/tests/characterize/paths.rs b/rs/moq-quic/src/tests/characterize/paths.rs index f5b721ed760..fe1be554079 100644 --- a/rs/moq-quic/src/tests/characterize/paths.rs +++ b/rs/moq-quic/src/tests/characterize/paths.rs @@ -164,7 +164,10 @@ fn duplicate_packet_is_discarded() { let before = pair.server_conn_mut(server_ch).stats().frame_rx.stream; pair.drive(); - assert_eq!(pair.server_conn_mut(server_ch).stats().frame_rx.stream, before + 1); + assert_eq!( + pair.server_conn_mut(server_ch).stats().frame_rx.stream, + before + 1 + ); assert_eq!(pair.server_streams(server_ch).accept(Dir::Uni), Some(s)); let mut recv = pair.server_recv(server_ch, s); let mut chunks = recv.read(true).unwrap(); @@ -235,7 +238,11 @@ fn retried(mangle: Mangle) -> (Pair, ConnectionHandle, ConnectionId) { #[test] fn invalid_retry_is_discarded() { let _guard = subscribe(); - for (mangle, accepted) in [(bad_tag as Mangle, false), (same_cid, false), (second, true)] { + for (mangle, accepted) in [ + (bad_tag as Mangle, false), + (same_cid, false), + (second, true), + ] { let (mut pair, _, orig_dst_cid) = retried(mangle); pair.drive_client(); while pair.server.inbound.is_empty() { diff --git a/rs/moq-quic/src/tests/characterize/reset.rs b/rs/moq-quic/src/tests/characterize/reset.rs index b040d399f3a..9128f01c495 100644 --- a/rs/moq-quic/src/tests/characterize/reset.rs +++ b/rs/moq-quic/src/tests/characterize/reset.rs @@ -21,7 +21,12 @@ fn read_all(pair: &mut Pair, ch: ConnectionHandle, s: StreamId) -> (Vec, Rea } /// Opens a client uni stream carrying `data`, delivered and accepted by the server -fn open(pair: &mut Pair, client_ch: ConnectionHandle, server_ch: ConnectionHandle, data: &[u8]) -> StreamId { +fn open( + pair: &mut Pair, + client_ch: ConnectionHandle, + server_ch: ConnectionHandle, + data: &[u8], +) -> StreamId { let s = pair.client_streams(client_ch).open(Dir::Uni).unwrap(); pair.client_send(client_ch, s).write(data).unwrap(); pair.drive(); @@ -40,7 +45,10 @@ fn received_reset_reports_the_code_once() { pair.drive(); // Observing the reset frees the stream, so every later access finds it closed - assert_eq!(pair.server_recv(server_ch, s).received_reset(), Ok(Some(CODE))); + assert_eq!( + pair.server_recv(server_ch, s).received_reset(), + Ok(Some(CODE)) + ); assert!(pair.server_recv(server_ch, s).received_reset().is_err()); assert_eq!( pair.server_recv(server_ch, s).read(true).err(), @@ -56,7 +64,11 @@ fn received_reset_on_closed_streams() { // Never opened let unopened = StreamId::new(Side::Client, Dir::Uni, 5); - assert!(pair.server_recv(server_ch, unopened).received_reset().is_err()); + assert!( + pair.server_recv(server_ch, unopened) + .received_reset() + .is_err() + ); // Stopped by the reader, even after the peer resets in response let s = open(&mut pair, client_ch, server_ch, b"hello"); @@ -119,7 +131,10 @@ fn chunks_end_in_finished() { let mut recv = pair.server_recv(server_ch, s); let mut chunks = recv.read(true).unwrap(); - assert_eq!(chunks.next(usize::MAX).unwrap().unwrap().bytes, &b"hello"[..]); + assert_eq!( + chunks.next(usize::MAX).unwrap().unwrap().bytes, + &b"hello"[..] + ); assert_eq!(chunks.next(usize::MAX), Ok(None)); assert_eq!(chunks.next(usize::MAX), Ok(None)); let _ = chunks.finalize(); @@ -138,7 +153,10 @@ fn reset_codes() { let s = open(&mut pair, client_ch, server_ch, b"x"); pair.client_send(client_ch, s).reset(code).unwrap(); pair.drive(); - assert_eq!(read_all(&mut pair, server_ch, s), (Vec::new(), ReadError::Reset(code))); + assert_eq!( + read_all(&mut pair, server_ch, s), + (Vec::new(), ReadError::Reset(code)) + ); let s = open(&mut pair, client_ch, server_ch, b"x"); pair.server_recv(server_ch, s).stop(code).unwrap(); @@ -233,7 +251,8 @@ fn zero_rtt_rejection_returns_connection_credit() { pair.drive(); // A server with a fresh session cache cannot resume, so it rejects the 0-RTT data - pair.server.set_server_config(Some(server(Arc::new(server_crypto())))); + pair.server + .set_server_config(Some(server(Arc::new(server_crypto())))); let client_ch = pair.begin_connect(config); assert!(pair.client_conn_mut(client_ch).has_0rtt()); assert_eq!(writable(&mut pair, client_ch), WINDOW as usize); From 4cd6b1c36d09ba5c9e895f4a617576f6e1fb28bd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 15:40:29 -0700 Subject: [PATCH 06/11] quest: finish quest/m1/quic/coverage/characterize.md Co-Authored-By: Claude Opus 5.5 --- quest/m1/quic/coverage/README.md | 1 - quest/m1/quic/coverage/characterize.md | 44 -------------------------- quest/m1/quic/coverage/sim.md | 4 --- 3 files changed, 49 deletions(-) delete mode 100644 quest/m1/quic/coverage/characterize.md diff --git a/quest/m1/quic/coverage/README.md b/quest/m1/quic/coverage/README.md index a585f2f0565..6ab59e923a4 100644 --- a/quest/m1/quic/coverage/README.md +++ b/quest/m1/quic/coverage/README.md @@ -44,7 +44,6 @@ and the harness defaults to Cubic while production runs BBR3. - [Benchmarks](/quest/m1/quic/coverage/bench.md) - Criterion sweeps of stream scheduling, receive and reset, endpoint routing, and the wire codec, on the baseline - [Coverage report](/quest/m1/quic/coverage/report.md) - `just rs coverage ` and a nightly lcov artifact, with no threshold -- [Characterization tests](/quest/m1/quic/coverage/characterize.md) - golden scheduling order, and every uncovered frame, reset, limit, and endpoint path, on today's harness - [Fuzzing](/quest/m1/quic/coverage/fuzz.md) - a moq-net-style fuzz crate with the upstream parser targets, frame decode, and a connection op-sequence target, run nightly - [Seeded simulation](/quest/m1/quic/coverage/sim.md) - a deterministic, impaired network in `Pair`, BBR3 as the harness default, and property tests over random workloads diff --git a/quest/m1/quic/coverage/characterize.md b/quest/m1/quic/coverage/characterize.md deleted file mode 100644 index 45addc8f65f..00000000000 --- a/quest/m1/quic/coverage/characterize.md +++ /dev/null @@ -1,44 +0,0 @@ -# [M] Characterize moq-quic behavior - -## Goal - -Tests on today's harness pin the behavior the QUIC rewrites touch, so a -rebase onto any of them fails loudly wherever behavior changed. - -- **Scheduling**: golden tests of the exact order today's scheduler sends - stream bytes across priorities, equal-priority streams, retransmits, and - datagrams interleaved with streams. #5711 changes this order on purpose - and must update each golden in its own diff. -- **Frame error paths**: CRYPTO in 0-RTT, STREAM_DATA_BLOCKED or - MAX_STREAM_DATA on the wrong direction or an unopened stream, STOP_SENDING - on a receive-only or unopened stream, STREAMS_BLOCKED and MAX_STREAMS - above the limit (bidi included), NEW_CONNECTION_ID limit and retirement - errors, NEW_TOKEN and HANDSHAKE_DONE from a client, offsets at 2^62, and - final-size errors on data and reset. -- **Resets**: `RecvStream::received_reset`, reset codes, `Chunks` ending in - Reset or Finished, and connection credit returned on reset, stop, and - 0-RTT rejection. -- **Runtime limits**: `set_max_concurrent_streams`, `set_receive_window`, - and `set_send_window` raised and lowered mid-connection, including the - remote-bidi window raise in `set_params`. -- **Endpoint**: unknown and invalid CIDs, long headers for unknown - connections, short, undecodable, and wrong-version initials, the - `cids_exhausted` refusal, and a stateless reset from a client-only - endpoint. -- **Packets and paths**: the AEAD integrity limit, duplicate discard, - invalid Retry handling, 0-RTT retransmit after Retry, path-validation - failure reverting the path, and IPv4 rebinding through - `PathData::from_previous`. - -## Plan - -Decided 2026-10-16: the scheduling tests pin the exact old order rather -than only invariants, so every intentional change in #5711 is visible and -reviewed. If #5711 merges first, write the goldens against the baseline -(b30289127) anyway and update them on rebase, so the PR still shows the -diff. - -The survey's line numbers for each uncovered path come from -`cargo llvm-cov` on the baseline; [the coverage report](/quest/m1/quic/coverage/report.md) -reproduces them. This quest uses today's `Pair` as is; seeded impairments -belong to [the simulation](/quest/m1/quic/coverage/sim.md). diff --git a/quest/m1/quic/coverage/sim.md b/quest/m1/quic/coverage/sim.md index 2b5124000b1..7f37eec4f29 100644 --- a/quest/m1/quic/coverage/sim.md +++ b/quest/m1/quic/coverage/sim.md @@ -28,7 +28,3 @@ Decided 2026-10-16: that asserts Cubic-specific timing pins Cubic explicitly. A test that fails under BBR3 for a real bug stays on BBR3, `#[ignore]`d with a follow-up quest, rather than being hidden on Cubic. - -## Related - -- [Characterization tests](/quest/m1/quic/coverage/characterize.md) - targeted tests on today's harness, which can land before this From 08c75ceb632afc0c4aaaac84cd2de9e00f6685d5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 15:47:50 -0700 Subject: [PATCH 07/11] test(quic): drop the global port counter and latency from path tests Co-Authored-By: Claude Opus 5.5 --- rs/moq-quic/src/tests/characterize/paths.rs | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/rs/moq-quic/src/tests/characterize/paths.rs b/rs/moq-quic/src/tests/characterize/paths.rs index fe1be554079..279efa0280e 100644 --- a/rs/moq-quic/src/tests/characterize/paths.rs +++ b/rs/moq-quic/src/tests/characterize/paths.rs @@ -321,7 +321,7 @@ fn path_validation_failure_reverts() { let (mut pair, client_ch, server_ch) = connected(); pair.drive(); let old = pair.client.addr; - let new = SocketAddr::new(old.ip(), CLIENT_PORTS.lock().unwrap().next().unwrap()); + let new = SocketAddr::new(old.ip(), old.port() + 1); // One client packet arrives from the new address, as if forwarded by an attacker pair.client_conn_mut(client_ch).ping(); @@ -356,19 +356,17 @@ fn path_validation_failure_reverts() { fn rtt_after_rebinding(ip: IpAddr) -> (Duration, Duration) { let _guard = subscribe(); let mut pair = pair_with(|_| {}); - pair.latency = Duration::from_millis(50); - pair.client.addr = SocketAddr::new(ip, CLIENT_PORTS.lock().unwrap().next().unwrap()); + pair.client.addr = SocketAddr::new(ip, pair.client.addr.port()); pair.server.addr = SocketAddr::new(ip, pair.server.addr.port()); let (client_ch, server_ch) = pair.connect_with(client_with(|_| {})); pair.drive(); let before = pair.server_conn_mut(server_ch).rtt(); - pair.client.addr = SocketAddr::new(ip, CLIENT_PORTS.lock().unwrap().next().unwrap()); + let new = SocketAddr::new(ip, pair.client.addr.port() + 1); + pair.client.addr = new; pair.client_conn_mut(client_ch).ping(); pair.drive_client(); - let now = pair.time + pair.latency; - pair.time = now; - let new = pair.client.addr; + let now = pair.time; pair.server.drive(now, new); assert_eq!(pair.server_conn_mut(server_ch).remote_address(), new); (before, pair.server_conn_mut(server_ch).rtt()) From fd1d2f9c1cdf368c52921264aaaf77b1cd8e60b0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 15:57:53 -0700 Subject: [PATCH 08/11] quest: track the four bugs the characterization tests pin The ignored tests now point at quests that exist: the Retry bug in m0, since the shipped release has the same guard, and the other three in the moq-quic epic. Co-Authored-By: Claude Opus 5.5 --- quest/m0/README.md | 1 + quest/m0/retry-after-invalid-retry.md | 34 +++++++++++++++++++++ quest/m1/quic/README.md | 3 ++ quest/m1/quic/close-reason-after-reset.md | 21 +++++++++++++ quest/m1/quic/final-size-below-received.md | 20 ++++++++++++ quest/m1/quic/zero-rtt-window-raise.md | 25 +++++++++++++++ rs/moq-quic/src/tests/characterize/paths.rs | 2 +- 7 files changed, 105 insertions(+), 1 deletion(-) create mode 100644 quest/m0/retry-after-invalid-retry.md create mode 100644 quest/m1/quic/close-reason-after-reset.md create mode 100644 quest/m1/quic/final-size-below-received.md create mode 100644 quest/m1/quic/zero-rtt-window-raise.md diff --git a/quest/m0/README.md b/quest/m0/README.md index cd2eefceabb..85bf351a0e2 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -84,6 +84,7 @@ have fixes on `main`: macOS tests (#5335), drill-sensitivity (#5338), loom - [qmux release backport](/quest/m0/qmux-credit-backport.md) - adopt the 0.5 fix independently of the next breaking cut - [Lite SETUP backport](/quest/m0/lite-setup-backport.md) - `release` closes a moq-lite session on a repeated SETUP parameter or a second Setup Stream, in a later 0.14.x patch (#5521) +- [A discarded Retry no longer blocks the genuine one](/quest/m0/retry-after-invalid-retry.md) - one injected or corrupted Retry no longer kills a handshake, on `main` and then `release` - [Kotlin wrapper POM release](/quest/m0/kt-ffi-pom.md) - `dev.moq:moq` 0.5.2 ships from `release`, its POMs naming moq-ffi at `[0.4.3,0.5)` - [CI cost and latency](/quest/m0/ci/README.md) - PRs stop queueing behind hosted runner caps: macOS and Windows move to the merge queue, caches move to R2, and setup is skipped when no work is selected diff --git a/quest/m0/retry-after-invalid-retry.md b/quest/m0/retry-after-invalid-retry.md new file mode 100644 index 00000000000..722db511f7f --- /dev/null +++ b/quest/m0/retry-after-invalid-retry.md @@ -0,0 +1,34 @@ +# [S] A discarded Retry no longer blocks the genuine one + +## Goal + +A client that discards an invalid Retry (a bad integrity tag, or a source CID +equal to its original destination CID) still follows the server's genuine +Retry and connects. Today the discarded Retry bumps `total_authed_packets`, +so the client then discards the genuine Retry too and never connects: one +injected or corrupted Retry kills a handshake against any server that +validates addresses. + +`retry_after_invalid_retry` in +`rs/moq-quic/src/tests/characterize/paths.rs` must pass with its `#[ignore]` +removed, while `invalid_retry_is_discarded` and `second_retry_is_discarded` +keep passing. + +## Plan + +Decided 2026-10-16 (maintainer): m0, because the shipped release has the same +Retry guard in moq-noq-proto 1.3.5. + +The Retry arm of `process_decrypted_packet` in +`rs/moq-quic/src/connection/mod.rs` uses `total_authed_packets > 1` to mean +"already processed a server Initial or Retry", but every Retry reaches +`on_packet_authenticated` before its validity checks run. Make the guard +mean what RFC 9000 17.2.5.2 says: count only a Retry the client accepts, or +track that state explicitly. The VersionNegotiate arm shares the counter; +check whether a discarded Retry also makes it ignore a genuine Version +Negotiation. + +Land the fix on `main` first. The `release` backport follows CONTRIBUTING.md +as separate PRs: a moq-dev/noq 1.3.x patch to moq-noq-proto, then a moq-noq +bump onto `release`. The PR that lands the main fix adds that backport quest, +modeled on [the Lite SETUP backport](/quest/m0/lite-setup-backport.md). diff --git a/quest/m1/quic/README.md b/quest/m1/quic/README.md index 0f867694617..73786d188a1 100644 --- a/quest/m1/quic/README.md +++ b/quest/m1/quic/README.md @@ -74,6 +74,9 @@ cache limits remain separate; bitrate pacing composes with this memory cap. datagrams join their subscription's send group instead of preempting every stream - [Relay peers get wider limits](/quest/m1/quic/peer-limits.md) - MAX_STREAMS and MAX_DATA are raised after SETUP identifies a cluster peer +- [A FIN below received data is a final size error](/quest/m1/quic/final-size-below-received.md) - a shrunken final size closes with FINAL_SIZE_ERROR instead of blocking the reader forever +- [A stateless reset keeps a closed connection's reason](/quest/m1/quic/close-reason-after-reset.md) - a reset after a local transport error neither replaces the reason nor reports a second loss +- [0-RTT streams adopt a raised stream window](/quest/m1/quic/zero-rtt-window-raise.md) - a server's larger stream window reaches the client's 0-RTT streams, so they no longer stall - [Delete NewReno](/quest/m1/quic/new-reno.md) - the unused public controller goes ## Related diff --git a/quest/m1/quic/close-reason-after-reset.md b/quest/m1/quic/close-reason-after-reset.md new file mode 100644 index 00000000000..364f741aec9 --- /dev/null +++ b/quest/m1/quic/close-reason-after-reset.md @@ -0,0 +1,21 @@ +# [S] A stateless reset keeps a closed connection's reason + +## Goal + +After a connection closes on its own transport error, a later stateless reset +drains it without changing what the application sees: the original close +reason is reported once. Today, while the connection is `Closed`, the reset +replaces the unread reason with `Reset`, or, once the reason was read, +yields a second `ConnectionLost`. + +`stateless_reset_after_local_close_keeps_reason` and +`stateless_reset_after_local_close_is_not_a_second_loss` in +`rs/moq-quic/src/tests/characterize/frames.rs` must pass with their +`#[ignore]` removed. + +## Plan + +`Connection::finish_packet` in `rs/moq-quic/src/connection/mod.rs` preserves +the close reason on a reset only in `Draining` or `Drained`. Extend that to +`Closed`: move to `Drained` without touching `self.error`, so `poll`, which +takes the error once, reports nothing new. diff --git a/quest/m1/quic/final-size-below-received.md b/quest/m1/quic/final-size-below-received.md new file mode 100644 index 00000000000..268e025765d --- /dev/null +++ b/quest/m1/quic/final-size-below-received.md @@ -0,0 +1,20 @@ +# [S] A FIN below received data is a final size error + +## Goal + +A STREAM frame whose FIN sets a final size below data already received closes +the connection with FINAL_SIZE_ERROR, as RFC 9000 4.5 requires. Today the +frame is accepted: the reader gets the bytes past the final size, then +`Blocked` forever, since the shrunken final size never matches what was +read. + +`stream_final_size_below_received` in +`rs/moq-quic/src/tests/characterize/frames.rs` must pass with its `#[ignore]` +removed. + +## Plan + +`Recv::ingest` in `rs/moq-quic/src/connection/streams/recv.rs` checks a FIN +only against an already-known final size. When a FIN arrives first, also +reject an `end` below `self.end`, before recording the size. RESET_STREAM +already makes this check (`reset_final_size_below_received`). diff --git a/quest/m1/quic/zero-rtt-window-raise.md b/quest/m1/quic/zero-rtt-window-raise.md new file mode 100644 index 00000000000..b78c87187cb --- /dev/null +++ b/quest/m1/quic/zero-rtt-window-raise.md @@ -0,0 +1,25 @@ +# [S] 0-RTT streams adopt a raised stream window + +## Goal + +When a server raised its stream receive window between sessions, a client's +0-RTT streams adopt the new window once the server's transport parameters +arrive, and run to completion. Today they keep the remembered smaller window, +and the server, believing the client already has the larger one, never sends +MAX_STREAM_DATA, so bidi and uni streams stall forever. + +`zero_rtt_bidi_window_raise` and `zero_rtt_uni_window_raise` in +`rs/moq-quic/src/tests/characterize/limits.rs` must pass with their +`#[ignore]` removed. + +## Plan + +Decided 2026-10-16 (maintainer): `StreamsState::set_params` in +`rs/moq-quic/src/connection/streams/state.rs` raises the send window of the +client's own 0-RTT streams to `initial_max_stream_data_bidi_remote` (bidi) or +`initial_max_stream_data_uni` (uni), and wakes any writer the new credit +unblocks. RFC 9000 7.4.1 forbids a server that accepts 0-RTT from lowering +these, so the window only grows. + +The same PR deletes the existing loop over peer-initiated bidi streams in +`set_params`, which is unreachable through the public API. diff --git a/rs/moq-quic/src/tests/characterize/paths.rs b/rs/moq-quic/src/tests/characterize/paths.rs index 279efa0280e..caa0f697ff5 100644 --- a/rs/moq-quic/src/tests/characterize/paths.rs +++ b/rs/moq-quic/src/tests/characterize/paths.rs @@ -269,7 +269,7 @@ fn second_retry_is_discarded() { /// A discarded Retry still counts as an authenticated packet, so the client then discards the /// server's genuine Retry of its retransmitted Initial too, and never connects #[test] -#[ignore = "bug: /quest/m1/quic/retry-after-invalid-retry.md"] +#[ignore = "bug: /quest/m0/retry-after-invalid-retry.md"] fn retry_after_invalid_retry() { let _guard = subscribe(); for mangle in [bad_tag as Mangle, same_cid] { From 8a9f040851b9c369ae26c2cddd0ece19f927c61c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 17:31:07 -0700 Subject: [PATCH 09/11] test(quic): a retried handshake must connect, not just stop handshaking Co-Authored-By: Claude Opus 5.5 --- rs/moq-quic/src/tests/characterize/mod.rs | 13 +++++++++++++ rs/moq-quic/src/tests/characterize/paths.rs | 6 ++---- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/rs/moq-quic/src/tests/characterize/mod.rs b/rs/moq-quic/src/tests/characterize/mod.rs index beba3ecdf3e..b887c0deaab 100644 --- a/rs/moq-quic/src/tests/characterize/mod.rs +++ b/rs/moq-quic/src/tests/characterize/mod.rs @@ -117,6 +117,19 @@ fn lost(conn: &mut Connection) -> Option { lost } +/// Drains `conn`'s events, panicking unless it connected and was never lost +fn assert_connected(conn: &mut Connection) { + let mut connected = false; + while let Some(event) = conn.poll() { + match event { + Event::Connected => connected = true, + Event::ConnectionLost { reason } => panic!("lost: {reason:?}"), + _ => {} + } + } + assert!(connected, "never connected"); +} + /// The transport error code `conn` was lost with, panicking if it is still up or lost otherwise fn lost_code(conn: &mut Connection) -> TransportErrorCode { match lost(conn) { diff --git a/rs/moq-quic/src/tests/characterize/paths.rs b/rs/moq-quic/src/tests/characterize/paths.rs index caa0f697ff5..2e0bc284ee2 100644 --- a/rs/moq-quic/src/tests/characterize/paths.rs +++ b/rs/moq-quic/src/tests/characterize/paths.rs @@ -261,9 +261,7 @@ fn second_retry_is_discarded() { let _guard = subscribe(); let (mut pair, client_ch, _) = retried(second); pair.drive(); - let conn = pair.client_conn_mut(client_ch); - assert!(!conn.is_handshaking()); - assert!(lost(conn).is_none()); + assert_connected(pair.client_conn_mut(client_ch)); } /// A discarded Retry still counts as an authenticated packet, so the client then discards the @@ -275,7 +273,7 @@ fn retry_after_invalid_retry() { for mangle in [bad_tag as Mangle, same_cid] { let (mut pair, client_ch, _) = retried(mangle); pair.drive(); - assert!(!pair.client_conn_mut(client_ch).is_handshaking()); + assert_connected(pair.client_conn_mut(client_ch)); } } From 6e83641ca6709240103e07a16ee7467a59cd6300 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 18:54:15 -0700 Subject: [PATCH 10/11] test(quic): stream priority is i64 on main Co-Authored-By: Claude Opus 5.5 --- rs/moq-quic/src/tests/characterize/schedule.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-quic/src/tests/characterize/schedule.rs b/rs/moq-quic/src/tests/characterize/schedule.rs index 51755d25f5d..3d92c0c36c7 100644 --- a/rs/moq-quic/src/tests/characterize/schedule.rs +++ b/rs/moq-quic/src/tests/characterize/schedule.rs @@ -42,7 +42,7 @@ fn assert_golden(pair: &mut Pair, golden: &str) { } /// Opens a client uni stream at `priority` and queues `LEN` bytes and a FIN on it -fn queue(pair: &mut Pair, ch: ConnectionHandle, priority: i32) -> StreamId { +fn queue(pair: &mut Pair, ch: ConnectionHandle, priority: i64) -> StreamId { let s = pair.client_streams(ch).open(Dir::Uni).unwrap(); pair.client_send(ch, s).set_priority(priority).unwrap(); assert_eq!(pair.client_send(ch, s).write(&[0; LEN]).unwrap(), LEN); From 7de8b6f1ff7ac0cf1444c134bec062ba4d911823 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 23:27:40 -0700 Subject: [PATCH 11/11] test(quic): pin datagram order after group scheduling #5770 schedules datagrams in their send group, so ungrouped datagrams now go before equal-priority stream data, and one that does not fit ends the packet instead of letting streams fill it. Co-Authored-By: Claude Opus 5.5 --- .../src/tests/characterize/schedule.rs | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/rs/moq-quic/src/tests/characterize/schedule.rs b/rs/moq-quic/src/tests/characterize/schedule.rs index 3d92c0c36c7..1c13c5029a8 100644 --- a/rs/moq-quic/src/tests/characterize/schedule.rs +++ b/rs/moq-quic/src/tests/characterize/schedule.rs @@ -148,7 +148,7 @@ fn retransmit_behind_higher_priority() { ); } -/// Datagrams fill each packet before stream data, which takes the rest +/// Datagrams go before stream data of equal priority, and one that does not fit ends the packet #[test] fn datagrams_interleaved() { let _guard = subscribe(); @@ -156,22 +156,23 @@ fn datagrams_interleaved() { queue(&mut pair, ch, 0); for _ in 0..3 { pair.client_datagrams(ch) - .send(vec![0; 500].into(), false) + .send(vec![0; 500].into(), None, 0) .unwrap(); } queue(&mut pair, ch, 0); pair.client_datagrams(ch) - .send(vec![0; 1000].into(), false) + .send(vec![0; 1000].into(), None, 0) .unwrap(); assert_golden( &mut pair, " - d+500 d+500 s2@0+418 - d+500 s6@0+921 - d+1000 s2@418+419 - s6@921+1422 - s2@837+1422 - s6@2343+657! s2@2259+741! + d+500 d+500 + d+500 + d+1000 s2@0+421 + s6@0+1424 + s2@421+1422 + s6@1424+1422 + s2@1843+1157! s6@2846+154! ", ); }