From ccab4310ea888f5259aa43044a6512814c8ff7db Mon Sep 17 00:00:00 2001 From: hatoo Date: Mon, 12 Oct 2026 19:42:48 +0900 Subject: [PATCH] fix(quic): suppress authenticated packet replays before recovery and frame dispatch --- src/http3.rs | 4 + src/quic/conn.rs | 14 +- src/quic/conn/tests/receive.rs | 320 +++++++++++++++++++++++++++++++++ src/quic/mod.rs | 1 + src/quic/received.rs | 163 +++++++++++++++++ tests/support/h3_duplicates.rs | 99 ++++++++++ 6 files changed, 600 insertions(+), 1 deletion(-) create mode 100644 src/quic/conn/tests/receive.rs create mode 100644 src/quic/received.rs create mode 100644 tests/support/h3_duplicates.rs diff --git a/src/http3.rs b/src/http3.rs index e6669cd..909a67f 100644 --- a/src/http3.rs +++ b/src/http3.rs @@ -18,6 +18,10 @@ mod recv; #[path = "../tests/support/h3_server.rs"] mod test_server; +#[cfg(test)] +#[path = "../tests/support/h3_duplicates.rs"] +mod duplicate_tests; + use std::collections::VecDeque; use std::net::UdpSocket; use std::os::fd::{FromRawFd, RawFd}; diff --git a/src/quic/conn.rs b/src/quic/conn.rs index 5cbd4f6..9d12dc1 100644 --- a/src/quic/conn.rs +++ b/src/quic/conn.rs @@ -88,6 +88,7 @@ use super::packet::{ ConnectionId, Space, decode_packet_number, encode_packet_number, protect_header, unprotect_header, }; +use super::received::ReceivedPackets; use super::recovery::{Congestion, Rtt, SentFrame, SentPacket, SentPackets, pto_deadline}; use super::stream::{ Dir, RecvStream, SendStream, client_stream_id, is_client_initiated, stream_dir, @@ -130,6 +131,7 @@ struct SpaceState { keys: Option, next_packet_number: u64, largest_received: Option, + received: ReceivedPackets, ack: AckState, sent: SentPackets, /// Handshake bytes the peer has not acknowledged. Held rather than @@ -430,6 +432,7 @@ impl Connection { fn discard_space(&mut self, space: Space) { let s = &mut self.spaces[space as usize]; s.keys = None; + s.received = ReceivedPackets::default(); s.ack = AckState::default(); s.sent = SentPackets::default(); s.crypto_out.clear(); @@ -1124,7 +1127,7 @@ impl Connection { } /// Returns how much of the buffer this packet took, and whether it was - /// decrypted and acted on + /// authenticated (including a duplicate, for idle/reset handling) fn handle_packet(&mut self, now: Instant, buf: &mut [u8]) -> Result<(usize, bool)> { // Nothing in a header has been authenticated, so a packet whose // header cannot be read - a version we do not speak, a connection @@ -1209,6 +1212,13 @@ impl Connection { Ok(p) => p.len(), Err(_) => return Ok((end, false)), }; + // ACK-of-ACK pruning must not let a packet run its frame handlers a + // second time or cancel pending PTO probes. Authenticate first: even a + // known packet number can have a forged header or payload (RFC 9000 + // Section 12.3). Retransmissions in a new packet remain eligible. + if !self.spaces[space as usize].received.insert(pn) { + return Ok((end, true)); + } if generation == Generation::Next { self.rotate_keys(pn, phase); } @@ -2413,6 +2423,8 @@ impl Connection { mod tests { use super::*; + mod receive; + fn client() -> Connection { let tls = crate::tls::client_config(b"h3").unwrap(); Connection::connect( diff --git a/src/quic/conn/tests/receive.rs b/src/quic/conn/tests/receive.rs new file mode 100644 index 0000000..a2cd143 --- /dev/null +++ b/src/quic/conn/tests/receive.rs @@ -0,0 +1,320 @@ +//! Protected packets, including byte-for-byte replays. No packet is injected +//! through handle_frame: header protection and AEAD run on every receipt. + +use super::*; + +fn receiver() -> Connection { + let mut conn = client(); + conn.spaces[Space::Handshake as usize].keys = + Some(initial_keys(&[8; 8], rustls::Side::Client).unwrap()); + conn.one_rtt_remote = Some(initial_keys(&[9; 8], rustls::Side::Client).unwrap().remote); + conn.one_rtt_local = Some(initial_keys(&[9; 8], rustls::Side::Client).unwrap().local); + conn.connected = true; + conn.max_data_peer = 1 << 20; + conn.test_send_limits(128, 1 << 20); + conn +} + +fn protected(conn: &Connection, space: Space, pn: u64, payload: &[u8]) -> Vec { + let dcid = match space { + Space::Initial => conn.original_dcid, + Space::Handshake => ConnectionId::new(&[8; 8]).unwrap(), + Space::Data => ConnectionId::new(&[9; 8]).unwrap(), + }; + let keys = initial_keys(dcid.as_slice(), rustls::Side::Server).unwrap(); + seal( + conn, + space, + pn, + false, + &keys.local, + keys.local.packet.as_ref(), + payload, + ) +} + +fn seal( + conn: &Connection, + space: Space, + pn: u64, + phase: bool, + keys: &DirectionalKeys, + packet_key: &dyn rustls::quic::PacketKey, + payload: &[u8], +) -> Vec { + let mut out = Vec::new(); + if space == Space::Data { + header::put_short_header(&mut out, &conn.local_cid, 4, phase); + } else { + LongHeader { + space, + dcid: conn.local_cid, + scid: ConnectionId::new(&[7; 8]).unwrap(), + token: Vec::new(), + } + .put(&mut out, 4, 4 + payload.len() + TAG_LEN); + } + let pn_offset = out.len(); + out.extend_from_slice(&(pn as u32).to_be_bytes()); + out.extend_from_slice(payload); + let (head, body) = out.split_at_mut(pn_offset + 4); + let tag = packet_key.encrypt_in_place(pn, head, body).unwrap(); + out.extend_from_slice(tag.as_ref()); + protect_header(keys.header.as_ref(), &mut out, pn_offset, 4).unwrap(); + out +} + +fn receive(conn: &mut Connection, now: Instant, wire: &[u8]) { + conn.handle_datagram(now, &mut wire.to_vec()).unwrap(); +} + +#[test] +fn duplicate_packets_preserve_pending_probes_in_every_number_space() { + for space in [Space::Initial, Space::Handshake, Space::Data] { + let mut conn = receiver(); + let now = Instant::now(); + let packet = protected(&conn, space, 0, &[frame::PING as u8]); + receive(&mut conn, now, &packet); + assert_eq!(conn.spaces[space as usize].largest_received, Some(0)); + conn.spaces[space as usize].ack.take_pending(); + conn.pto_count = 3; + conn.pto_probes = 2; + conn.pto_space = space; + receive(&mut conn, now + Duration::from_millis(50), &packet); + assert_eq!((conn.pto_count, conn.pto_probes), (3, 2), "{space:?}"); + // Discarded packets do not generate ACKs; a retransmission in a new + // packet is processed and acknowledged normally. + assert!(!conn.spaces[space as usize].ack.take_pending()); + assert_eq!(conn.pto_space, space); + } +} + +#[test] +fn duplicate_path_challenge_is_not_dispatched_after_ack_of_ack() { + let mut conn = receiver(); + let now = Instant::now(); + let mut payload = vec![frame::PATH_CHALLENGE as u8]; + payload.extend_from_slice(b"original"); + let original = protected(&conn, Space::Data, 0, &payload); + receive(&mut conn, now, &original); + assert_eq!(conn.path_response.take(), Some(*b"original")); + conn.spaces[Space::Data as usize].sent.push(SentPacket { + number: 7, + time_sent: now, + size: 40, + ack_eliciting: false, + ack_largest: Some(0), + frames: Vec::new(), + }); + conn.on_ack(Space::Data, 7, 0, 0, &[], now).unwrap(); + assert!(conn.spaces[Space::Data as usize].ack.ranges().is_empty()); + receive(&mut conn, now + Duration::from_millis(100), &original); + assert_eq!( + conn.path_response, None, + "ACK pruning must not enable replay" + ); + + // The same frame retransmitted in a new packet is still processed. + let retry = protected(&conn, Space::Data, 1, &payload); + receive(&mut conn, now + Duration::from_millis(200), &retry); + assert_eq!(conn.path_response.take(), Some(*b"original")); +} + +#[test] +fn duplicate_stream_does_not_repeat_readable_or_finished_events() { + let mut conn = receiver(); + let now = Instant::now(); + let (id, _) = conn.send_oneshot(b"request").unwrap(); + let mut payload = Vec::new(); + frame::put_stream(&mut payload, id, 0, true, b"response"); + let packet = protected(&conn, Space::Data, 0, &payload); + receive(&mut conn, now, &packet); + assert!(conn.events.contains(&Event::Readable(id))); + assert!(conn.events.contains(&Event::Finished { id, reset: None })); + conn.events.clear(); + let received = conn.data_received; + for _ in 0..64 { + receive(&mut conn, now, &packet); + } + assert!( + conn.events.is_empty(), + "replayed frame events: {:?}", + conn.events + ); + assert_eq!(conn.data_received, received); + assert_eq!(received, 8); +} + +#[test] +fn replayed_ack_only_packet_neither_cancels_probes_nor_creates_ack_loop() { + let mut conn = receiver(); + let now = Instant::now(); + let mut payload = Vec::new(); + frame::put_ack(&mut payload, &[(0, 0)], 0); + let packet = protected(&conn, Space::Data, 0, &payload); + receive(&mut conn, now, &packet); + conn.pto_count = 2; + conn.pto_probes = 1; + conn.needs_send = false; + for _ in 0..8 { + receive(&mut conn, now, &packet); + } + assert!(!conn.spaces[Space::Data as usize].ack.ack_eliciting_pending); + assert!(!conn.needs_send); + assert_eq!((conn.pto_count, conn.pto_probes), (2, 1)); +} + +#[test] +fn unauthenticated_packets_cannot_poison_history_or_refresh_idle() { + let mut conn = receiver(); + let now = Instant::now(); + let original = protected(&conn, Space::Data, 0, &[frame::PING as u8]); + receive(&mut conn, now, &original); + let deadline = conn.idle_deadline; + conn.pto_count = 3; + conn.pto_probes = 2; + conn.spaces[Space::Data as usize].ack.take_pending(); + for pn in [0, 1, 1000] { + let mut forged = protected(&conn, Space::Data, pn, &[frame::PING as u8]); + *forged.last_mut().unwrap() ^= 1; + receive(&mut conn, now + Duration::from_millis(50), &forged); + assert_eq!(conn.idle_deadline, deadline); + assert_eq!(conn.spaces[Space::Data as usize].largest_received, Some(0)); + assert_eq!((conn.pto_count, conn.pto_probes), (3, 2)); + assert!(!conn.spaces[Space::Data as usize].ack.ack_eliciting_pending); + } + let valid = protected(&conn, Space::Data, 1, &[frame::PING as u8]); + receive(&mut conn, now + Duration::from_millis(100), &valid); + assert_eq!(conn.spaces[Space::Data as usize].largest_received, Some(1)); + assert!(conn.spaces[Space::Data as usize].ack.take_pending()); + assert_eq!((conn.pto_count, conn.pto_probes), (0, 0)); +} + +#[test] +fn ack_pruning_preserves_unseen_gaps_and_the_original_ack_delay() { + let mut conn = receiver(); + let now = Instant::now(); + for pn in [0, 2] { + let packet = protected(&conn, Space::Data, pn, &[frame::PING as u8]); + receive(&mut conn, now, &packet); + } + let later = now + Duration::from_millis(50); + let duplicate = protected(&conn, Space::Data, 2, &[frame::PING as u8]); + let ack = &mut conn.spaces[Space::Data as usize].ack; + let delay = ack.delay(later, 3); + assert!(ack.ack_eliciting_pending); + receive(&mut conn, later, &duplicate); + let ack = &mut conn.spaces[Space::Data as usize].ack; + assert_eq!(ack.delay(later, 3), delay); + assert!(ack.take_pending(), "a duplicate cannot cancel an owed ACK"); + ack.trim_below(2); + receive(&mut conn, later, &duplicate); + assert!(conn.spaces[Space::Data as usize].ack.ranges().is_empty()); + let reordered = protected(&conn, Space::Data, 1, &[frame::PING as u8]); + receive(&mut conn, later, &reordered); + assert_eq!(conn.spaces[Space::Data as usize].ack.ranges(), &[(1, 1)]); + assert!(conn.spaces[Space::Data as usize].ack.take_pending()); + assert_eq!(conn.spaces[Space::Data as usize].largest_received, Some(2)); +} + +#[test] +fn history_survives_key_updates_and_accepts_unseen_previous_key_packets() { + let mut conn = receiver(); + let now = Instant::now(); + let old = protected(&conn, Space::Data, 0, &[frame::PING as u8]); + receive(&mut conn, now, &old); + let next = initial_keys(&[10; 8], rustls::Side::Client).unwrap(); + conn.next_1rtt = Some(rustls::quic::PacketKeySet { + local: next.local.packet, + remote: next.remote.packet, + }); + let hp = initial_keys(&[9; 8], rustls::Side::Server).unwrap(); + let next_keys = initial_keys(&[10; 8], rustls::Side::Server).unwrap(); + let new = seal( + &conn, + Space::Data, + 2, + true, + &hp.local, + next_keys.local.packet.as_ref(), + &[frame::PING as u8], + ); + receive(&mut conn, now, &new); + assert!(conn.key_phase); + assert_eq!(conn.rotate_at, 2); + conn.pto_count = 3; + conn.pto_probes = 2; + for packet in [&old, &new] { + receive(&mut conn, now, packet); + assert_eq!((conn.pto_count, conn.pto_probes), (3, 2)); + assert_eq!(conn.rotate_at, 2); + } + // It was delayed, not previously received. Old keys remain usable. + let reordered = protected(&conn, Space::Data, 1, &[frame::PING as u8]); + receive(&mut conn, now, &reordered); + assert_eq!((conn.pto_count, conn.pto_probes), (0, 0)); + assert_eq!(conn.spaces[Space::Data as usize].ack.ranges(), &[(0, 2)]); + assert!(conn.key_phase); +} + +#[test] +fn evicted_replays_are_dropped_but_new_packet_retransmissions_still_progress() { + let mut conn = receiver(); + let now = Instant::now(); + let (id, _) = conn.send_oneshot(b"request").unwrap(); + let mut response = Vec::new(); + frame::put_stream(&mut response, id, 0, true, b"response"); + let delayed = protected(&conn, Space::Data, 1, &response); + let original = protected(&conn, Space::Data, 0, &[frame::PING as u8]); + receive(&mut conn, now, &original); + for pn in (2..=1024).step_by(2) { + let packet = protected(&conn, Space::Data, pn, &[frame::PING as u8]); + receive(&mut conn, now, &packet); + } + conn.pto_count = 3; + conn.pto_probes = 2; + for packet in [&original, &delayed] { + receive(&mut conn, now, packet); + assert_eq!((conn.pto_count, conn.pto_probes), (3, 2)); + assert_eq!(conn.data_received, 0); + } + let retry = protected(&conn, Space::Data, 1025, &response); + receive(&mut conn, now, &retry); + assert_eq!(conn.data_received, 8); + assert_eq!(conn.unanswered, 0); + assert_eq!( + conn.events + .iter() + .filter(|e| matches!(e, Event::Finished { .. })) + .count(), + 1 + ); +} + +#[test] +fn a_duplicate_coalesced_packet_does_not_hide_the_next_number_space() { + let mut conn = receiver(); + let now = Instant::now(); + let initial = protected(&conn, Space::Initial, 0, &[frame::PING as u8]); + let handshake = protected(&conn, Space::Handshake, 0, &[frame::PING as u8]); + receive(&mut conn, now, &initial); + let mut coalesced = initial.clone(); + coalesced.extend_from_slice(&handshake); + receive(&mut conn, now, &coalesced); + assert_eq!( + conn.spaces[Space::Handshake as usize].largest_received, + Some(0) + ); + assert!(conn.spaces[Space::Handshake as usize].ack.take_pending()); +} + +#[test] +fn a_new_authenticated_malformed_packet_still_reports_its_error() { + let mut conn = receiver(); + // PATH_CHALLENGE with its eight-byte body missing. + let mut malformed = protected(&conn, Space::Data, 0, &[frame::PATH_CHALLENGE as u8]); + assert!( + conn.handle_datagram(Instant::now(), &mut malformed) + .is_err() + ); +} diff --git a/src/quic/mod.rs b/src/quic/mod.rs index 5e05a67..64d70c0 100644 --- a/src/quic/mod.rs +++ b/src/quic/mod.rs @@ -12,6 +12,7 @@ pub mod crypto; pub mod frame; pub mod header; pub mod packet; +mod received; pub mod recovery; pub mod stream; pub mod transport; diff --git a/src/quic/received.rs b/src/quic/received.rs new file mode 100644 index 0000000..18cbaa3 --- /dev/null +++ b/src/quic/received.rs @@ -0,0 +1,163 @@ +//! Bounded duplicate history, independent of ACK serialization and ACK-of-ACK. + +#[derive(Default)] +pub(super) struct ReceivedPackets { + /// Inclusive ranges, largest first. In-order traffic uses just one range, + /// regardless of the connection's age or packet rate. + ranges: Vec<(u64, u64)>, + /// Forgotten history cannot become eligible for processing again. This + /// floor also discards unseen gaps below it (RFC 9000 Section 12.3). + minimum: u64, +} + +impl ReceivedPackets { + // Bound sparse history while allowing substantial reordering. This limit + // counts gaps rather than packets, so a long in-order burst cannot evict a + // delayed packet by itself. At most 4096 bytes of range storage per space. + const MAX_RANGES: usize = 256; + + /// Record an authenticated packet, returning true only for a new number. + pub(super) fn insert(&mut self, pn: u64) -> bool { + if pn < self.minimum { + return false; + } + let at = self.ranges.partition_point(|r| r.0 > pn); + if self.ranges.get(at).is_some_and(|r| pn <= r.1) { + return false; + } + let above = at > 0 && self.ranges[at - 1].0.checked_sub(1) == Some(pn); + let below = self + .ranges + .get(at) + .is_some_and(|r| r.1.checked_add(1) == Some(pn)); + match (above, below) { + (true, true) => { + self.ranges[at - 1].0 = self.ranges[at].0; + self.ranges.remove(at); + } + (true, false) => self.ranges[at - 1].0 = pn, + (false, true) => self.ranges[at].1 = pn, + (false, false) => { + // Evict before inserting to keep capacity, not just length, + // bounded. A new oldest range can be discarded immediately. + if self.ranges.len() == Self::MAX_RANGES { + if at == Self::MAX_RANGES { + self.minimum = pn + 1; + return false; + } + self.minimum = self.ranges.pop().unwrap().1 + 1; + } + self.ranges.insert(at, (pn, pn)); + } + } + true + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeSet; + + #[test] + fn contiguous_traffic_preserves_old_gaps_and_uses_one_range() { + let mut history = ReceivedPackets::default(); + assert!(history.insert(0)); + for pn in 2..10_000 { + assert!(history.insert(pn)); + } + assert_eq!(history.ranges, [(2, 9999), (0, 0)]); + assert!(history.insert(1)); + assert_eq!(history.ranges, [(0, 9999)]); + for pn in [0, 1, 300, 9999] { + assert!(!history.insert(pn)); + } + assert_eq!(history.minimum, 0); + } + + #[test] + fn evicted_ranges_and_their_gaps_never_become_new_again() { + let mut history = ReceivedPackets::default(); + for pn in (0..=1024).step_by(2) { + assert!(history.insert(pn)); + } + assert_eq!(history.ranges.len(), ReceivedPackets::MAX_RANGES); + assert_eq!(history.minimum, 513); + assert!(history.ranges.capacity() <= ReceivedPackets::MAX_RANGES); + for pn in 0..513 { + assert!(!history.insert(pn)); + } + // The gap immediately below the oldest retained range can still join it. + assert!(history.insert(513)); + assert!(!history.insert(512)); + assert_eq!(history.minimum, 513); + assert!(history.insert(515)); + assert_eq!(history.ranges.len(), ReceivedPackets::MAX_RANGES - 1); + } + + #[test] + fn older_new_range_at_capacity_advances_the_floor_without_growing() { + let mut history = ReceivedPackets::default(); + for pn in (1000..1512).step_by(2) { + assert!(history.insert(pn)); + } + assert!(!history.insert(500)); + assert_eq!(history.minimum, 501); + assert!(!history.insert(499)); + assert!(history.insert(999)); + assert_eq!(history.ranges.len(), ReceivedPackets::MAX_RANGES); + assert!(history.ranges.capacity() <= ReceivedPackets::MAX_RANGES); + } + + #[test] + fn packet_number_boundaries_do_not_wrap() { + for top in [(1 << 62) - 1, u64::MAX] { + let mut history = ReceivedPackets::default(); + for pn in [top, top - 2, top - 1, 0, 1] { + assert!(history.insert(pn)); + assert!(!history.insert(pn)); + } + assert_eq!(history.ranges, [(top - 2, top), (0, 1)]); + } + } + + #[test] + fn bounded_history_matches_an_independent_set_model() { + for seed in [1_u64, 0x132, 0x5eed] { + let mut history = ReceivedPackets::default(); + let mut seen = BTreeSet::new(); + let mut floor = 0; + let mut rng = seed; + for step in 0..8000 { + rng ^= rng << 13; + rng ^= rng >> 7; + rng ^= rng << 17; + let pn = if step < 1000 { step * 2 } else { rng % 16_000 }; + let fresh = pn >= floor && seen.insert(pn); + let mut ranges = Vec::<(u64, u64)>::new(); + for &n in &seen { + if let Some(last) = ranges.last_mut() + && last.1 + 1 == n + { + last.1 = n; + } else { + ranges.push((n, n)); + } + } + if ranges.len() > ReceivedPackets::MAX_RANGES { + floor = ranges.remove(0).1 + 1; + seen = seen.split_off(&floor); + } + assert_eq!( + history.insert(pn), + fresh && pn >= floor, + "seed {seed}, step {step}" + ); + ranges.reverse(); + assert_eq!(history.ranges, ranges); + assert_eq!(history.minimum, floor); + assert!(history.ranges.capacity() <= ReceivedPackets::MAX_RANGES); + } + } + } +} diff --git a/tests/support/h3_duplicates.rs b/tests/support/h3_duplicates.rs new file mode 100644 index 0000000..694ea1c --- /dev/null +++ b/tests/support/h3_duplicates.rs @@ -0,0 +1,99 @@ +//! Replay real Quinn ciphertext immediately and after later response packets. +//! The worker must retain exactly one status and latency sample per response. + +use std::collections::VecDeque; +use std::net::UdpSocket; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +#[test] +fn immediate_and_delayed_response_replays_preserve_exact_worker_accounting() { + let upstream = super::test_server::h3_server_addr(); + let relay = UdpSocket::bind("127.0.0.1:0").unwrap(); + relay + .set_read_timeout(Some(Duration::from_millis(10))) + .unwrap(); + let addr = relay.local_addr().unwrap(); + let stop = Arc::new(AtomicBool::new(false)); + let stopped = stop.clone(); + let thread = std::thread::spawn(move || { + let mut client = None; + let mut held = VecDeque::new(); + let mut immediate = 0; + let mut delayed = 0; + let mut sent_bytes = 0; + let mut buf = [0_u8; 65536]; + while !stopped.load(Ordering::SeqCst) { + let (n, from) = match relay.recv_from(&mut buf) { + Ok(packet) => packet, + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + ) => + { + continue; + } + Err(e) => panic!("relay receive: {e}"), + }; + if from != upstream { + if let Some(previous) = client { + assert_eq!(previous, from, "one benchmark connection"); + } + client = Some(from); + relay.send_to(&buf[..n], upstream).unwrap(); + continue; + } + let to = client.expect("a response follows a request"); + relay.send_to(&buf[..n], to).unwrap(); + sent_bytes += n as u64; + // Long-header handshake packets pass unchanged. Short-header + // packets are independently encrypted by Quinn, not this fixture. + if n > 0 && buf[0] & 0x80 == 0 { + for _ in 0..3 { + relay.send_to(&buf[..n], to).unwrap(); + sent_bytes += n as u64; + immediate += 1; + } + held.push_back(buf[..n].to_vec()); + if held.len() > 4 { + let packet = held.pop_front().unwrap(); + relay.send_to(&packet, to).unwrap(); + sent_bytes += packet.len() as u64; + delayed += 1; + } + } + } + (immediate, delayed, sent_bytes) + }); + let target = + crate::target::parse_target(&format!("https://{addr}/"), "GET", &[], None, false).unwrap(); + let stats = crate::http3::run_worker( + &target, + 1, + crate::budget::Budget::Requests(512), + Duration::from_secs(5), + Some(Duration::from_secs(5)), + 8, + ); + stop.store(true, Ordering::SeqCst); + let (immediate, delayed, sent_bytes) = thread.join().unwrap(); + let stats = stats.unwrap(); + assert!( + immediate >= 3 && delayed > 0, + "replays: {immediate} immediate, {delayed} delayed" + ); + assert_eq!( + (stats.completed, stats.errors, stats.connect_errors), + (512, 0, 0) + ); + assert_eq!(stats.latencies_ns.len(), 512); + assert!(stats.latencies_ns.iter().all(|&n| n > 0)); + assert_eq!(stats.status_counts[200], 512); + assert_eq!(stats.status_counts.iter().sum::(), 512); + // The existing byte counter measures UDP payload, including replayed + // ciphertext. Some final copies can remain unread when the budget ends. + assert!(stats.bytes_received > 512 * 11); + assert!(stats.bytes_received <= sent_bytes); +}