test(client): cover TTL and pending expiry without poll coupling

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::<Vec<_>>();
+            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<_>>(),
+                    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() {

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<Vec<u8>>) {
@@ -24,6 +26,7 @@ fn peer(dscp_allowed: bool, probes: usize) -> (SocketAddr, thread::JoinHandle<Ve
     socket2::SockRef::from(&socket)
         .set_recv_tos_v4(true)
         .unwrap();
+    nix::sys::socket::setsockopt(&socket, nix::sys::socket::sockopt::Ipv4RecvTtl, &true).unwrap();
     let addr = socket.local_addr().unwrap();
     let task = thread::spawn(move || {
         let mut core = ServerCore::new(
@@ -35,7 +38,7 @@ fn peer(dscp_allowed: bool, probes: usize) -> (SocketAddr, thread::JoinHandle<Ve
         // OPEN, every ECHO, then CLOSE.
         for _ in 0..probes + 2 {
             let mut bytes = [0; 2048];
-            let mut control = nix::cmsg_space!(u8);
+            let mut control = nix::cmsg_space!(u8, i32);
             let mut iov = [IoSliceMut::new(&mut bytes)];
             let msg = recvmsg::<nix::sys::socket::SockaddrIn>(
                 std::os::fd::AsRawFd::as_raw_fd(&socket),
@@ -54,6 +57,18 @@ fn peer(dscp_allowed: bool, probes: usize) -> (SocketAddr, thread::JoinHandle<Ve
                     _ => 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()
         },

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<TargetInstance, Vec<_>> = 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<TargetInstance, Vec<_>> = 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