diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index acf098a..04976f2 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -375,6 +375,58 @@ mod tests { } } + #[test] + fn pending_removal_and_sequence_reuse_preserve_expiry_and_capacity() { + let start = Instant::now(); + // Remove the head, middle or tail, then reuse that sequence at a later + // deadline. Observe expiry and capacity, not private list links. + for removed in 0..3 { + let mut pending_map = PendingMap::new(3); + for seq in 0..3 { + pending_map.preflight_insert(seq).unwrap(); + pending_map + .commit_insert(pending(seq, start + Duration::from_secs(u64::from(seq)))); + } + assert!(matches!( + pending_map.preflight_insert(3), + Err(ClientError::PendingLimitExceeded { .. }) + )); + assert_eq!(pending_map.remove(removed).unwrap().wire_seq, removed); + pending_map.preflight_insert(removed).unwrap(); + let reused_deadline = start + Duration::from_secs(4); + pending_map.commit_insert(pending(removed, reused_deadline)); + + let remaining = (0..3).filter(|seq| *seq != removed).collect::>(); + for (index, seq) in remaining.iter().enumerate() { + assert_eq!( + pending_map.next_timeout_deadline(), + Some(start + Duration::from_secs(u64::from(*seq))) + ); + let expired = pending_map.drain_expired_bounded(start + Duration::from_secs(3), 1); + assert_eq!( + expired + .probes + .iter() + .map(|probe| probe.wire_seq) + .collect::>(), + vec![*seq] + ); + assert_eq!(expired.more_due, index == 0); + } + assert_eq!(pending_map.next_timeout_deadline(), Some(reused_deadline)); + let expired = pending_map.drain_expired_bounded(reused_deadline, 3); + assert_eq!(expired.probes.len(), 1); + assert_eq!(expired.probes[0].wire_seq, removed); + assert!(!expired.more_due); + assert!(pending_map.is_empty()); + assert_eq!(pending_map.next_timeout_deadline(), None); + pending_map.preflight_insert(removed).unwrap(); + pending_map.commit_insert(pending(removed, reused_deadline)); + assert_eq!(pending_map.remove(removed).unwrap().wire_seq, removed); + assert!(pending_map.is_empty()); + } + } + // Public loss/reply events cannot reveal leaked retention bookkeeping. #[test] fn timed_out_retention_stays_bounded_after_repeated_removals() { diff --git a/crates/irtt-client/tests/linux_transport.rs b/crates/irtt-client/tests/linux_transport.rs index 68bed05..2aa9699 100644 --- a/crates/irtt-client/tests/linux_transport.rs +++ b/crates/irtt-client/tests/linux_transport.rs @@ -14,6 +14,8 @@ use irtt_proto::{decode_request, Clock, DecodedRequestKind, ReceivedStats, Stamp use irtt_server::{ServerConfig, ServerCore}; use nix::sys::socket::{recvmsg, ControlMessageOwned, MsgFlags}; +const TEST_TTL: u32 = 37; + // A compliant packet peer uses the real server core. Its receive-side IP_TOS // observation is independent of the client/server parameter agreement. fn peer(dscp_allowed: bool, probes: usize) -> (SocketAddr, thread::JoinHandle>) { @@ -24,6 +26,7 @@ fn peer(dscp_allowed: bool, probes: usize) -> (SocketAddr, thread::JoinHandle (SocketAddr, thread::JoinHandle( std::os::fd::AsRawFd::as_raw_fd(&socket), @@ -54,6 +57,18 @@ fn peer(dscp_allowed: bool, probes: usize) -> (SocketAddr, thread::JoinHandle None, }) .expect("Linux must report IP_TOS"); + let ttl = msg + .cmsgs() + .unwrap() + .find_map(|message| match message { + ControlMessageOwned::Ipv4Ttl(ttl) => Some(ttl), + _ => None, + }) + .expect("Linux must report IP_TTL"); + assert_eq!( + ttl, TEST_TTL as i32, + "configured TTL must survive OPEN, ECHO and CLOSE" + ); markings.push(marking); let source = SocketAddr::from((source.ip(), source.port())); if let Some(reply) = core.handle_datagram(source, &bytes[..size]).unwrap() { @@ -79,6 +94,7 @@ fn config(addr: SocketAddr) -> ClientConfig { length: 128, open_timeouts: vec![Duration::from_secs(1)], socket_config: SocketConfig { + ttl: Some(TEST_TTL), recv_timeout: Some(Duration::from_secs(1)), ..SocketConfig::default() }, diff --git a/crates/irtt-integration-tests/tests/managed_behavior.rs b/crates/irtt-integration-tests/tests/managed_behavior.rs index 8ef85b2..1144aa6 100644 --- a/crates/irtt-integration-tests/tests/managed_behavior.rs +++ b/crates/irtt-integration-tests/tests/managed_behavior.rs @@ -1,10 +1,6 @@ mod support; -use std::{ - collections::HashMap, - future::{poll_fn, Future}, - time::Duration, -}; +use std::{collections::HashMap, time::Duration}; use irtt_client::{managed::*, ClientConfig, ClientEvent}; use irtt_server::ServerConfig; @@ -15,31 +11,30 @@ use tokio::time::timeout; // on a scheduler, a send cursor, or a private event-publication hook. #[tokio::test(flavor = "current_thread")] async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() { - timeout(Duration::from_secs(5), async { + timeout(Duration::from_secs(10), async { let server = InTreeServer::start(ServerConfig::default()); for pacing in [ManagedPacing::Burst, ManagedPacing::Staggered] { - let config = ManagedClientConfig { - client: ClientConfig { - duration: Some(Duration::from_millis(600)), - interval: Duration::from_millis(100), - probe_timeout: Duration::from_millis(100), - open_timeouts: vec![Duration::from_millis(200)], - ..ClientConfig::default() - }, - pacing, - ..ManagedClientConfig::default() - }; - let targets = - ["first", "second"].map(|id| ManagedTargetConfig::new(id, server.addr.to_string())); - let (task, handle) = ManagedClient::task(config, targets.into()).unwrap(); - let mut events = handle.subscribe().unwrap(); - let mut task = Box::pin(task); - let mut sends: HashMap> = HashMap::new(); - let mut release_batches = Vec::new(); - let outcome = poll_fn(|cx| { - let result = task.as_mut().poll(cx); - let mut released = 0; - while let Ok(event) = events.try_recv() { + let mut observed_pacing = false; + for _ in 0..3 { + let config = ManagedClientConfig { + client: ClientConfig { + duration: Some(Duration::from_millis(600)), + interval: Duration::from_millis(100), + probe_timeout: Duration::from_millis(100), + open_timeouts: vec![Duration::from_millis(200)], + ..ClientConfig::default() + }, + pacing, + ..ManagedClientConfig::default() + }; + let targets = ["first", "second"] + .map(|id| ManagedTargetConfig::new(id, server.addr.to_string())); + let (task, handle) = ManagedClient::task(config, targets.into()).unwrap(); + let mut events = handle.subscribe().unwrap(); + let run = tokio::spawn(task); + let mut sends: HashMap> = HashMap::new(); + let mut actual_sends = Vec::new(); + while let Ok(event) = events.recv().await { if let ManagedEvent::Client { target, event: @@ -52,7 +47,7 @@ async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() { { let scheduled = scheduled_at.expect("managed probes expose their cadence"); assert!(sent_at.mono >= scheduled); - released += 1; + actual_sends.push((target.clone(), sent_at.mono)); let previous = sends.entry(target).or_default(); if let Some(last) = previous.last() { let elapsed: Duration = scheduled.duration_since(*last); @@ -65,33 +60,43 @@ async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() { previous.push(scheduled); } } - if released > 0 { - release_batches.push(released); + let outcome = run.await.unwrap(); + assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete); + assert_eq!(outcome.successful_target_outcomes, 2); + assert_eq!(sends.len(), 2); + for target in outcome.recent_target_outcomes.iter() { + assert!(sends[&target.target].len() >= 2); + assert!(target.replies_received >= 2); + } + // Use only the initial pair: later stagger gates can legitimately + // compress after a delayed wake. Startup may instead serve one + // target twice before the other opens, so retry a fresh run then. + if actual_sends.len() < 2 || actual_sends[0].0 == actual_sends[1].0 { + continue; + } + let gap = actual_sends[1].1.duration_since(actual_sends[0].1); + match pacing { + ManagedPacing::Staggered => { + assert!( + gap >= Duration::from_millis(50), + "first stagger gate was bypassed: {gap:?}" + ); + observed_pacing = true; + } + ManagedPacing::Burst => { + // A pause between burst sends makes this observation + // inconclusive; it must not fail an otherwise valid run. + observed_pacing = gap < Duration::from_millis(25); + } + } + if observed_pacing { + break; } - result - }) - .await; - assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete); - assert_eq!(outcome.successful_target_outcomes, 2); - assert_eq!(sends.len(), 2); - // Drive the public Future and observe the events each poll releases. - // Burst serves ready targets together; staggered releases one before - // yielding to its next gate. Descheduling cannot stretch/compress an - // observed wall-time gap into a different pacing classification. - match pacing { - ManagedPacing::Burst => assert!( - release_batches.iter().filter(|count| **count == 2).count() >= 2, - "burst must repeatedly release both ready targets: {release_batches:?}" - ), - ManagedPacing::Staggered => assert!( - release_batches.iter().all(|count| *count == 1), - "staggered must release targets separately: {release_batches:?}" - ), - } - for target in outcome.recent_target_outcomes.iter() { - assert!(sends[&target.target].len() >= 2); - assert!(target.replies_received >= 2); } + assert!( + observed_pacing, + "could not observe {pacing:?} startup pacing in three fresh runs" + ); } }) .await