From 47d1f59d559dfb517b694cc9ae36fad0b9623ab7 Mon Sep 17 00:00:00 2001 From: hatoo Date: Wed, 21 Oct 2026 20:29:41 +0900 Subject: [PATCH] test(quic): cover receive-credit recovery and queue lifecycle --- src/quic/conn.rs | 2 + src/quic/conn/tests/credit.rs | 173 ++++++++++++++++++++ src/quic/conn/tests/credit_wire.rs | 246 +++++++++++++++++++++++++++++ 3 files changed, 421 insertions(+) create mode 100644 src/quic/conn/tests/credit.rs create mode 100644 src/quic/conn/tests/credit_wire.rs diff --git a/src/quic/conn.rs b/src/quic/conn.rs index 0159fc5..4a25508 100644 --- a/src/quic/conn.rs +++ b/src/quic/conn.rs @@ -2423,6 +2423,8 @@ impl Connection { mod tests { use super::*; + mod credit; + mod credit_wire; mod receive; fn client() -> Connection { diff --git a/src/quic/conn/tests/credit.rs b/src/quic/conn/tests/credit.rs new file mode 100644 index 0000000..00c0195 --- /dev/null +++ b/src/quic/conn/tests/credit.rs @@ -0,0 +1,173 @@ +//! Queue lifecycle tests use real receive streams, frame budgets and wire bytes. +use super::*; + +fn fixture(count: usize) -> (Connection, Vec) { + let mut conn = client(); + conn.stream_window = 64; + conn.max_streams_bidi = count as u64 + 16; + conn.max_data_local = 1 << 40; + conn.max_data_window = 1 << 40; + let mut ids: Vec<_> = (0..count).map(|_| conn.open_bi().unwrap()).collect(); + for id in [3, 7, 11] { + conn.on_stream(id, 0, b"", false).unwrap(); + ids.push(id); + } + (conn, ids) +} + +fn flush(conn: &mut Connection, budget: usize) -> Vec<(u64, u64)> { + // A nonzero starting position catches errors in the packet-room calculation. + let mut out = vec![0xaa; 13]; + let mut frames = Vec::new(); + conn.fill_data_payload(&mut out, budget, 13, &mut frames, true) + .unwrap(); + assert!(out.len() - 13 <= budget); + let wire: Vec<_> = frame::Iter::new(&out[13..]) + .map(|frame| match frame.unwrap() { + Frame::MaxStreamData { id, limit } => (id, limit), + other => panic!("unexpected credit packet: {other:?}"), + }) + .collect(); + let ledger: Vec<_> = frames + .into_iter() + .map(|frame| match frame { + SentFrame::MaxStreamData { id, limit } => (id, limit), + other => panic!("unexpected credit ledger: {other:?}"), + }) + .collect(); + assert_eq!(wire, ledger); + wire +} + +#[test] +fn credit_queue_matches_lifo_model_across_duplicate_losses_and_packet_budgets() { + let (mut conn, ids) = fixture(64); + let mut model = Vec::new(); + let mut seed = 0x123456789abcdefu64; + let mut emitted = 0; + for step in 0..10_000 { + seed ^= seed << 13; + seed ^= seed >> 7; + seed ^= seed << 17; + let id = ids[seed as usize % ids.len()]; + conn.on_frame_lost(Space::Data, SentFrame::MaxStreamData { id, limit: 1 }); + if !model.contains(&id) { + model.push(id); + } + if step % 7 == 0 { + let budget = [0, 23, 24, 30, 47, 1200][(seed >> 24) as usize % 6]; + let mut room = budget; + let mut expected = Vec::new(); + while room >= 24 { + let Some(id) = model.pop() else { break }; + expected.push((id, 64)); + let mut frame = Vec::new(); + frame::put_max_stream_data(&mut frame, id, 64); + room -= frame.len(); + } + let actual = flush(&mut conn, budget); + emitted += actual.len(); + assert_eq!(actual, expected, "step {step}"); + } + assert_eq!(conn.credit_pending, model, "step {step}"); + } + assert_eq!( + flush(&mut conn, 65_536), + model + .into_iter() + .rev() + .map(|id| (id, 64)) + .collect::>() + ); + assert!(emitted > 1000); +} + +#[test] +fn credit_consumption_failure_and_repeated_reads_keep_one_current_grant() { + let (mut conn, ids) = fixture(1); + for id in ids { + conn.on_stream(id, 0, &[1; 32], false).unwrap(); + assert!( + conn.consume(id, |_| anyhow::bail!("application rejected data")) + .is_err() + ); + assert!(conn.credit_pending.is_empty()); + assert_eq!( + conn.consume(id, |data| { + assert_eq!(data, &[1; 32]); + Ok(()) + }) + .unwrap(), + 32 + ); + for _ in 0..10 { + assert_eq!(conn.consume(id, |_| panic!("no new bytes")).unwrap(), 0); + conn.on_frame_lost(Space::Data, SentFrame::MaxStreamData { id, limit: 64 }); + } + assert_eq!(conn.credit_pending, vec![id]); + assert!(flush(&mut conn, 23).is_empty()); + assert_eq!(conn.credit_pending, vec![id]); + assert_eq!(flush(&mut conn, 24), vec![(id, 96)]); + // An old lost grant must be regenerated from the newly consumed offset. + conn.on_stream(id, 32, &[2; 32], false).unwrap(); + conn.consume(id, |_| Ok(())).unwrap(); + conn.on_frame_lost(Space::Data, SentFrame::MaxStreamData { id, limit: 64 }); + assert_eq!(flush(&mut conn, 24), vec![(id, 128)]); + conn.on_frame_lost(Space::Data, SentFrame::MaxStreamData { id, limit: 96 }); + assert_eq!(flush(&mut conn, 24), vec![(id, 128)]); + } +} + +#[test] +fn credit_fin_and_reset_before_emission_discard_grants_after_budget_deferral() { + let (mut conn, ids) = fixture(2); + for (n, id) in ids.into_iter().enumerate() { + conn.on_stream(id, 0, &[1; 32], false).unwrap(); + conn.consume(id, |_| Ok(())).unwrap(); + assert!(flush(&mut conn, 0).is_empty()); + if n % 2 == 0 { + conn.on_stream(id, 32, b"", true).unwrap(); + } else { + conn.on_reset(id, 0, 32).unwrap(); + } + assert!(flush(&mut conn, 24).is_empty()); + assert!(conn.credit_pending.is_empty()); + conn.on_frame_lost(Space::Data, SentFrame::MaxStreamData { id, limit: 64 }); + assert!(flush(&mut conn, 24).is_empty()); + assert!(conn.credit_pending.is_empty()); + } +} + +#[test] +fn credit_retirement_and_sparse_send_acks_do_not_poison_reused_buffers() { + let (mut conn, _) = fixture(0); + conn.params.initial_max_stream_data_bidi_remote = 64; + for n in 0..16 { + let (id, len) = conn.send_oneshot(b"abcdefgh").unwrap(); + assert_eq!(len, 8); + conn.send_mut(id).unwrap().on_sent(0, 8, true); + conn.on_stream(id, 0, &[3; 32], false).unwrap(); + conn.consume(id, |_| Ok(())).unwrap(); + assert_eq!(conn.credit_pending, vec![id]); + conn.retire(id, 0x10c); + conn.on_frame_acked(SentFrame::Stream { + id, + offset: 4, + len: 4, + fin: true, + }); + assert!(conn.stream_mut(id).is_some()); + conn.on_frame_acked(SentFrame::Stream { + id, + offset: 0, + len: 4, + fin: false, + }); + assert!(conn.stream_mut(id).is_none()); + // A credit lost after retirement must not resurrect the old stream. + conn.on_frame_lost(Space::Data, SentFrame::MaxStreamData { id, limit: 64 }); + assert!(flush(&mut conn, 1200).is_empty()); + assert!(conn.credit_pending.is_empty()); + assert_eq!(conn.spare_bufs.len(), 1, "cycle {n}"); + } +} diff --git a/src/quic/conn/tests/credit_wire.rs b/src/quic/conn/tests/credit_wire.rs new file mode 100644 index 0000000..5d8042e --- /dev/null +++ b/src/quic/conn/tests/credit_wire.rs @@ -0,0 +1,246 @@ +//! An independent Quinn server consumes the actual encrypted credit grants. +//! Small transport windows deliberately exercise the transport API; the CLI's +//! roughly 1 GiB receive window is not changed by this fixture. +use super::*; +use std::net::UdpSocket; +use std::sync::atomic::{AtomicUsize, Ordering}; + +const BODY: usize = 32 * 1024; +const STREAMS: usize = 4; + +fn byte(stream: u64, offset: usize) -> u8 { + (stream as usize ^ offset ^ (offset >> 8)) as u8 +} + +fn run(disrupt: bool) { + let (address_tx, address_rx) = std::sync::mpsc::channel(); + let completed = Arc::new(AtomicUsize::new(0)); + let done = completed.clone(); + let server = std::thread::spawn(move || { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async move { + tokio::time::timeout(Duration::from_secs(20), async move { + let certified = + rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); + let key = rustls::pki_types::PrivatePkcs8KeyDer::from( + certified.signing_key.serialize_der(), + ); + let mut tls = rustls::ServerConfig::builder_with_provider(Arc::new( + rustls::crypto::ring::default_provider(), + )) + .with_safe_default_protocol_versions() + .unwrap() + .with_no_client_auth() + .with_single_cert(vec![certified.cert.der().clone()], key.into()) + .unwrap(); + tls.alpn_protocols = vec![b"credit-test".to_vec()]; + let crypto = + quinn::crypto::rustls::QuicServerConfig::try_from(Arc::new(tls)).unwrap(); + let endpoint = quinn::Endpoint::server( + quinn::ServerConfig::with_crypto(Arc::new(crypto)), + "127.0.0.1:0".parse().unwrap(), + ) + .unwrap(); + address_tx.send(endpoint.local_addr().unwrap()).unwrap(); + let connection = endpoint.accept().await.unwrap().await.unwrap(); + let mut tasks = tokio::task::JoinSet::new(); + let mut send = connection.open_uni().await.unwrap(); + let uni_done = done.clone(); + tasks.spawn(async move { + let id = u64::from(send.id()); + let body: Vec<_> = (0..BODY).map(|n| byte(id, n)).collect(); + send.write_all(&body).await.unwrap(); + send.finish().unwrap(); + assert_eq!(send.stopped().await.unwrap(), None); + uni_done.fetch_add(1, Ordering::SeqCst); + }); + for _ in 0..STREAMS { + let (mut send, mut recv) = connection.accept_bi().await.unwrap(); + let done = done.clone(); + tasks.spawn(async move { + let id = u64::from(send.id()); + assert_eq!(recv.read_to_end(8).await.unwrap(), vec![id as u8]); + let body: Vec<_> = (0..BODY).map(|n| byte(id, n)).collect(); + send.write_all(&body).await.unwrap(); + send.finish().unwrap(); + assert_eq!(send.stopped().await.unwrap(), None); + done.fetch_add(1, Ordering::SeqCst); + }); + } + while let Some(result) = tasks.join_next().await { + result.unwrap(); + } + // All responses including FIN have been acknowledged by shb. + endpoint.close(0u32.into(), b"complete"); + }) + .await + .expect("independent credit peer timed out"); + }); + }); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket.connect(address_rx.recv().unwrap()).unwrap(); + socket.set_nonblocking(true).unwrap(); + let mut conn = Connection::connect( + crate::tls::client_config(b"credit-test").unwrap(), + "localhost", + LocalParamsInput { + initial_max_data: 1 << 20, + initial_max_stream_data: 1024, + initial_max_streams_uni: 1, + max_idle_timeout_ms: 10_000, + handshake_timeout_ms: 5_000, + }, + ) + .unwrap(); + let mut opened = false; + let mut received = std::collections::BTreeMap::::new(); + let mut finished = std::collections::BTreeSet::new(); + let mut limits = std::collections::BTreeMap::::new(); + let mut held = None; + let mut held_after = None; + let mut dropped_limit = None; + let mut recovered = false; + let mut credit_packets = 0; + let mut credits = 0; + let mut duplicate_sent = false; + let mut queue_peak = 0; + let deadline = std::time::Instant::now() + Duration::from_secs(15); + let mut buf = [0u8; 65536]; + let mut out = Vec::new(); + while completed.load(Ordering::SeqCst) < STREAMS + 1 { + assert!( + std::time::Instant::now() < deadline, + "credit client stalled: {} {received:?}", + conn.debug_state() + ); + while let Ok(n) = socket.recv(&mut buf) { + conn.handle_datagram(Instant::now(), &mut buf[..n]).unwrap(); + } + if !opened && conn.connected { + for n in 0..STREAMS { + let id = n as u64 * 4; + assert_eq!(conn.send_oneshot(&[id as u8]), Some((id, 1))); + } + opened = true; + } + while let Some(event) = conn.poll_event() { + match event { + Event::Readable(id) => { + let offset = received.entry(id).or_default(); + conn.consume(id, |data| { + for (n, value) in data.iter().enumerate() { + assert_eq!(*value, byte(id, *offset + n)); + } + *offset += data.len(); + Ok(()) + }) + .unwrap(); + } + Event::Finished { id, reset } => { + assert_eq!(reset, None); + finished.insert(id); + } + Event::Lost(reason) => { + assert_eq!(completed.load(Ordering::SeqCst), STREAMS + 1, "{reason}") + } + _ => {} + } + } + queue_peak = queue_peak.max(conn.credit_pending.len()); + if conn.poll_timeout().is_some_and(|at| Instant::now() >= at) { + conn.handle_timeout(Instant::now()); + } + loop { + out.clear(); + if conn.poll_transmit(Instant::now(), &mut out, None).unwrap() == 0 { + break; + } + let mut grants = 0; + let mut target_limit = None; + if out[0] & 0x80 == 0 { + // Decode what was actually sealed, independently of SentFrame. + let mut packet = out.clone(); + let pn_offset = 1 + conn.peer_cid.len(); + let keys = conn.one_rtt_local.as_ref().unwrap(); + let (_, pn_len) = crate::quic::packet::unprotect_header( + keys.header.as_ref(), + &mut packet, + pn_offset, + ) + .unwrap(); + let pn = conn.spaces[Space::Data as usize].next_packet_number - 1; + let (head, payload) = packet.split_at_mut(pn_offset + pn_len); + let plaintext = keys.packet.decrypt_in_place(pn, head, payload).unwrap(); + for frame in frame::Iter::new(plaintext) { + if let Frame::MaxStreamData { id, limit } = frame.unwrap() { + assert!(limit >= *limits.get(&id).unwrap_or(&1024)); + assert_eq!(limit, received[&id] as u64 + 1024); + limits.insert(id, limit); + if id == 0 { + target_limit = Some(limit); + } + grants += 1; + } + } + } + if grants > 0 { + credits += grants; + credit_packets += 1; + if disrupt && !recovered { + if let Some(limit) = target_limit { + // A single early loss can be masked by the fresh grant + // from consuming the rest of the initial window. Drop + // every fresh limit for stream 0 until it is blocked. + // Only recovery can repeat its current, unadvanced limit. + if dropped_limit != Some(limit) { + dropped_limit = Some(limit); + continue; + } + recovered = true; + } + } else if disrupt && held_after.is_none() { + // Hold another grant behind newer ones, then replay twice. + held = Some(out.clone()); + held_after = Some(credit_packets); + continue; + } + } + socket.send(&out).unwrap(); + } + if held_after.is_some_and(|after| credit_packets > after) + && let Some(packet) = held.take() + { + socket.send(&packet).unwrap(); + socket.send(&packet).unwrap(); + duplicate_sent = true; + } + std::thread::sleep(Duration::from_micros(100)); + } + server.join().unwrap(); + assert_eq!(received.len(), STREAMS + 1); + assert!(received.values().all(|n| *n == BODY)); + assert_eq!(finished.len(), STREAMS); + assert_eq!(limits.len(), STREAMS + 1); + assert!(credits > 50); + assert!(queue_peak > 0); + assert_eq!(duplicate_sent, disrupt); + assert_eq!(recovered, disrupt); + eprintln!( + "credit-wire: disrupted={disrupt} bytes={} streams={} grants={credits} peak={queue_peak}", + BODY * (STREAMS + 1), + STREAMS + 1 + ); +} + +#[test] +fn credit_wire_quinn_consumes_every_byte_beyond_many_receive_windows() { + run(false); +} + +#[test] +fn credit_wire_quinn_recovers_lost_reordered_and_duplicate_grants() { + run(true); +}