test: protect durable behavior identified in review

crates/irtt-client/src/managed/task.rs

@@ -612,6 +612,8 @@ enum OpenSessionFailureCleanup {
 /// The sole authoritative zero-or-many target driver.
 #[must_use = "ManagedClientTask must be awaited or deliberately dropped"]
 pub struct ManagedClientTask {
+    #[cfg(test)]
+    timeout_inspections: std::cell::RefCell<Vec<usize>>,
     state: DriverState,
     lifecycle: ManagedLifecycle,
     config: ManagedClientConfig,
@@ -1833,6 +1835,8 @@ impl ManagedClientTask {
     }
 
     fn target_has_due_timeout(&self, index: usize, now: Instant) -> bool {
+        #[cfg(test)]
+        self.timeout_inspections.borrow_mut().push(index);
         match &self.targets[index].state {
             TargetState::Active { client } => {
                 self.state != DriverState::Stopping
@@ -2318,6 +2322,8 @@ fn build_task(
         stop: Arc::clone(&stop),
     };
     let task = ManagedClientTask {
+        #[cfg(test)]
+        timeout_inspections: std::cell::RefCell::new(Vec::new()),
         state: DriverState::NotStarted,
         lifecycle: ManagedLifecycle::NotStarted,
         config,
@@ -2387,3 +2393,41 @@ fn close_timeout_failure() -> ManagedTargetFailure {
         message: Arc::from("best-effort close did not become writable before its deadline"),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // Poll latency through public events cannot distinguish an O(n) discovery
+    // scan from a bounded scan reliably. This tiny inspection trace measures
+    // work directly, including empty targets, and checks eventual coverage
+    // without asserting a cursor value or a particular visitation order.
+    #[test]
+    fn timeout_discovery_has_bounded_work_and_eventually_visits_every_target() {
+        let count = TIMEOUT_WORK_BUDGET * 2 + 1;
+        let targets = (0..count)
+            .map(|i| ManagedTargetConfig::new(i.to_string(), "127.0.0.1:2112"))
+            .collect();
+        let (mut task, _handle) = ManagedClient::task(
+            ManagedClientConfig {
+                max_live_target_generations: count,
+                ..ManagedClientConfig::default()
+            },
+            targets,
+        )
+        .unwrap();
+        let mut visited = HashSet::new();
+        for _ in 0..count.div_ceil(TIMEOUT_WORK_BUDGET) {
+            task.timeout_inspections.borrow_mut().clear();
+            task.poll_timeout_pass(Instant::now());
+            let inspected = task.timeout_inspections.borrow();
+            assert!(inspected.len() <= TIMEOUT_WORK_BUDGET);
+            visited.extend(inspected.iter().copied());
+        }
+        assert_eq!(
+            visited.len(),
+            count,
+            "timeout discovery must reach all targets"
+        );
+    }
+}

crates/irtt-client/src/session/machine.rs

@@ -1133,3 +1133,173 @@ fn config_duration_to_ns(field: &str, duration: Duration) -> Result<i64, ClientE
         reason: format!("{field} is too large to encode as nanoseconds"),
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use irtt_proto::{FLAG_OPEN, FLAG_REPLY};
+
+    fn opened() -> SessionMachine {
+        let mut machine = SessionMachine::new(
+            ClientConfig {
+                max_pending_probes: 2,
+                clock: Clock::Wall,
+                stamp_at: irtt_proto::StampAt::Receive,
+                ..ClientConfig::default()
+            },
+            "127.0.0.1:2112".parse().unwrap(),
+        )
+        .unwrap();
+        let acceptance = machine
+            .prepare_open_acceptance(
+                OpenReply {
+                    flags: FLAG_OPEN | FLAG_REPLY,
+                    token: 1,
+                    params: machine.requested.clone(),
+                },
+                ClientTimestamp::now(),
+            )
+            .unwrap();
+        machine.commit_open(acceptance);
+        machine
+    }
+
+    fn send(machine: &mut SessionMachine) -> ProbeSent {
+        let probe = machine.prepare_probe().unwrap();
+        let preflight = machine.preflight_probe_commit(&probe).unwrap();
+        let commit = machine
+            .finalize_probe_commit(preflight, ClientTimestamp::now())
+            .unwrap();
+        machine.commit_probe_sent(commit, ClientTimestamp::now(), probe.bytes.len())
+    }
+
+    fn reply(machine: &mut SessionMachine, seq: u32) -> ClientEvent {
+        let events = machine
+            .process_echo_reply(
+                EchoReply {
+                    flags: FLAG_REPLY,
+                    token: 1,
+                    sequence: seq,
+                    recv_count: None,
+                    recv_window: None,
+                    timestamps: TimestampFields {
+                        recv_wall: Some(1),
+                        ..TimestampFields::default()
+                    },
+                    payload: vec![],
+                },
+                32,
+                ClientTimestamp::now(),
+                ReceiveMeta::default(),
+            )
+            .unwrap();
+        events.into_iter().next().unwrap()
+    }
+
+    // Advancing a real client through 2^32 submissions is unreasonable. Only
+    // move the send/receive sequence epoch; construct completion/timeout history through
+    // ordinary machine transitions and assert emitted behavior on reuse.
+    #[test]
+    fn sequence_wrap_reuses_completed_and_timed_out_ids_as_new_probes() {
+        for timed_out in [false, true] {
+            let mut machine = opened();
+            assert_eq!(send(&mut machine).seq, 0);
+            if timed_out {
+                let deadline = machine.next_probe_timeout_deadline().unwrap();
+                assert!(matches!(
+                    machine.poll_timeouts_at(deadline).unwrap()[0],
+                    ClientEvent::EchoLoss { seq: 0, .. }
+                ));
+            } else {
+                assert!(matches!(
+                    reply(&mut machine, 0),
+                    ClientEvent::EchoReply { seq: 0, .. }
+                ));
+            }
+            let session = machine.open_session_mut().unwrap();
+            session.next_wire_seq = u32::MAX;
+            session.highest_received_seq = Some(u32::MAX - 1);
+            assert_eq!(send(&mut machine).seq, u32::MAX);
+            assert!(matches!(
+                reply(&mut machine, u32::MAX),
+                ClientEvent::EchoReply { seq: u32::MAX, .. }
+            ));
+            assert_eq!(send(&mut machine).seq, 0);
+            assert!(matches!(
+                reply(&mut machine, 0),
+                ClientEvent::EchoReply { seq: 0, .. }
+            ));
+            assert!(matches!(
+                reply(&mut machine, 0),
+                ClientEvent::DuplicateReply { seq: 0, .. }
+            ));
+            assert_eq!(send(&mut machine).seq, 1);
+            assert!(matches!(
+                reply(&mut machine, 1),
+                ClientEvent::EchoReply { seq: 1, .. }
+            ));
+            assert!(
+                matches!(
+                    reply(&mut machine, 0),
+                    ClientEvent::DuplicateReply { seq: 0, .. }
+                ),
+                "reused sequence must receive a fresh completion retention lifetime"
+            );
+            assert_eq!(send(&mut machine).seq, 2);
+            assert!(matches!(
+                reply(&mut machine, 2),
+                ClientEvent::EchoReply { seq: 2, .. }
+            ));
+            assert!(
+                matches!(
+                    reply(&mut machine, 0),
+                    ClientEvent::LateReply {
+                        sent_at: None,
+                        rtt: None,
+                        ..
+                    }
+                ),
+                "evicted reused sequence must not resurrect a previous epoch's timeout"
+            );
+        }
+    }
+
+    // A recoverable send failure that consumes an OPT_ID, followed by a
+    // plausible wrong-ID completion, cannot be induced reliably with public
+    // sockets. Exercise that failure boundary directly, without a socket hook
+    // or inspecting the stored correlation flag/metadata.
+    #[test]
+    fn failed_submission_disables_later_tx_endpoints_for_the_session() {
+        let mut machine = opened();
+        for attempt in 0..3 {
+            if attempt == 1 {
+                machine.invalidate_kernel_tx_correlation();
+            }
+            let prepared = machine.prepare_probe().unwrap();
+            let preflight = machine.preflight_probe_commit(&prepared).unwrap();
+            let pre_send = ClientTimestamp::now();
+            let commit = machine.finalize_probe_commit(preflight, pre_send).unwrap();
+            std::thread::sleep(Duration::from_millis(1));
+            let sent =
+                machine.commit_probe_sent(commit, ClientTimestamp::now(), prepared.bytes.len());
+            machine.record_kernel_tx_timestamp(sent.seq, pre_send.wall);
+            let event = reply(&mut machine, sent.seq);
+            let ClientEvent::EchoReply {
+                one_way: Some(one_way),
+                ..
+            } = event
+            else {
+                panic!("expected measured reply")
+            };
+            let expected_wall = if attempt == 0 {
+                pre_send.wall
+            } else {
+                sent.sent_at.wall
+            };
+            let wall_ns =
+                i128::try_from(expected_wall.duration_since(UNIX_EPOCH).unwrap().as_nanos())
+                    .unwrap();
+            assert_eq!(one_way.client_to_server.unwrap().as_nanos(), 1 - wall_ns);
+        }
+    }
+}

crates/irtt-integration-tests/tests/managed_behavior.rs

@@ -16,8 +16,8 @@ async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() {
         for pacing in [ManagedPacing::Burst, ManagedPacing::Staggered] {
             let config = ManagedClientConfig {
                 client: ClientConfig {
-                    duration: Some(Duration::from_millis(200)),
+                    duration: Some(Duration::from_millis(600)),
-                    interval: Duration::from_millis(20),
+                    interval: Duration::from_millis(100),
                     probe_timeout: Duration::from_millis(100),
                     open_timeouts: vec![Duration::from_millis(200)],
                     ..ClientConfig::default()
@@ -31,6 +31,7 @@ async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() {
             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,
@@ -44,11 +45,12 @@ 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);
+                    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);
                         assert!(!elapsed.is_zero());
-                        assert_eq!(elapsed.as_nanos() % Duration::from_millis(20).as_nanos(), 0);
+                        assert_eq!(elapsed.as_nanos() % Duration::from_millis(100).as_nanos(), 0);
                     }
                     previous.push(scheduled);
                 }
@@ -57,6 +59,19 @@ async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() {
             assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete);
             assert_eq!(outcome.successful_target_outcomes, 2);
             assert_eq!(sends.len(), 2);
+            actual_sends.sort_by_key(|(_, at)| *at);
+            let cross_target_gaps = actual_sends.windows(2)
+                .filter(|pair| pair[0].0 != pair[1].0)
+                .map(|pair| pair[1].1.duration_since(pair[0].1)).collect::<Vec<_>>();
+            assert!(cross_target_gaps.len() >= 2);
+            // Allow timer wake jitter while separating the nominal 50 ms
+            // stagger spacing from back-to-back burst sends.
+            match pacing {
+                ManagedPacing::Burst => assert!(cross_target_gaps.iter().filter(|gap| **gap < Duration::from_millis(25)).count() >= 2,
+                    "burst must repeatedly send different targets back-to-back: {cross_target_gaps:?}"),
+                ManagedPacing::Staggered => assert!(cross_target_gaps.iter().all(|gap| *gap >= Duration::from_millis(35)),
+                    "staggered must space different targets across the interval: {cross_target_gaps:?}"),
+            }
             for target in outcome.recent_target_outcomes.iter() {
                 assert!(sends[&target.target].len() >= 2);
                 assert!(target.replies_received >= 2);

crates/irtt-server/tests/protocol_contracts.rs

@@ -316,3 +316,42 @@ fn negotiated_timestamp_modes_emit_only_the_selected_fields_in_order() {
         }
     }
 }
+
+#[test]
+fn duration_close_waits_for_rate_allowance_after_the_deadline() {
+    let refill = Duration::from_secs(4);
+    let maximum = Duration::from_millis(1);
+    let mut server = ServerCore::new(
+        ServerConfig::default()
+            .with_min_send_interval(refill)
+            .with_burst_allowance(1)
+            .with_idle_timeout(Duration::MAX)
+            .with_max_test_duration(maximum),
+    );
+    let opened = open(
+        &mut server,
+        &Params {
+            received_stats: ReceivedStats::Both,
+            ..Params::with_protocol_defaults()
+        },
+    );
+    let first = echo(&mut server, &opened, 0).unwrap();
+    assert_eq!(first.flags & FLAG_CLOSE, 0);
+    thread::sleep(maximum + Duration::from_secs(2));
+    assert!(
+        echo(&mut server, &opened, 63).is_none(),
+        "exhausted allowance must suppress the deadline-crossing reply"
+    );
+    assert_eq!(
+        server.session_count(),
+        1,
+        "rate refusal must not release the session"
+    );
+    thread::sleep(refill);
+    let closed = echo(&mut server, &opened, 1).unwrap();
+    assert_ne!(closed.flags & FLAG_CLOSE, 0);
+    assert_eq!(closed.sequence, 1);
+    assert_eq!(closed.recv_count, Some(2));
+    assert_eq!(closed.recv_window, Some(3));
+    assert_eq!(server.session_count(), 0);
+}

crates/irtt-server/tests/runtime.rs

@@ -155,3 +155,109 @@ async fn wildcard_construction_is_refused_without_reply_source_selection() {
         Err(ServerRuntimeError::WildcardSourceSelectionUnsupported { .. })
     ));
 }
+
+// A current-thread runtime cannot receive while this test holds its thread.
+// Queue real packets before delaying the server: no synthetic ancillary data or
+// transport hooks. A short queue delay selects kernel wall time; a delay beyond
+// the documented one-second sanity bound falls back to userspace. Monotonic and
+// midpoint timestamps must always remain at the later userspace observation.
+#[cfg(target_os = "linux")]
+#[tokio::test(flavor = "current_thread")]
+async fn queued_echoes_select_kernel_receive_wall_only_when_plausible() {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    let ns = || {
+        i64::try_from(
+            SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_nanos(),
+        )
+        .unwrap()
+    };
+    for (placement, delay, kernel_expected) in [
+        (StampAt::Both, Duration::from_millis(100), true),
+        (StampAt::Both, Duration::from_millis(1100), false),
+        (StampAt::Midpoint, Duration::from_millis(100), false),
+    ] {
+        let mut server = Server::bind("127.0.0.1:0".parse().unwrap(), unthrottled())
+            .await
+            .unwrap();
+        let addr = server.local_addr().unwrap();
+        let client = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        client.connect(addr).await.unwrap();
+        let (stop_tx, stop_rx) = oneshot::channel();
+        let run = tokio::spawn(async move {
+            server
+                .run(async {
+                    let _ = stop_rx.await;
+                })
+                .await
+        });
+        let params = Params {
+            stamp_at: placement,
+            clock: Clock::Both,
+            received_stats: ReceivedStats::Both,
+            ..Params::with_protocol_defaults()
+        };
+        let open = encode_request(
+            RequestToEncode::Open {
+                params: &params,
+                no_test: false,
+            },
+            None,
+        )
+        .unwrap();
+        client.send(&open).await.unwrap();
+        let mut buffer = [0; 2048];
+        let len = timeout(Duration::from_secs(2), client.recv(&mut buffer))
+            .await
+            .unwrap()
+            .unwrap();
+        let opened = decode_open_reply(&buffer[..len], None).unwrap();
+        let echo = encode_request(
+            RequestToEncode::Echo {
+                token: opened.token,
+                sequence: 0,
+                params: &opened.params,
+                payload: &[],
+            },
+            None,
+        )
+        .unwrap();
+        client.writable().await.unwrap();
+        let before_send = ns();
+        client.try_send(&echo).unwrap(); // No await from submission through resume marker.
+        let queued_by = ns();
+        std::thread::sleep(delay);
+        let resumed_at = ns();
+        let len = timeout(Duration::from_secs(2), client.recv(&mut buffer))
+            .await
+            .unwrap()
+            .unwrap();
+        let reply = decode_echo_reply(&buffer[..len], &opened.params, None).unwrap();
+        assert_eq!(reply.recv_count, Some(1));
+        assert_eq!(reply.recv_window, Some(1));
+        let timestamps = reply.timestamps;
+        if kernel_expected {
+            let receive = timestamps.recv_wall.unwrap();
+            assert!(
+                receive >= before_send && receive <= queued_by,
+                "receive wall must identify the kernel arrival before the queue delay"
+            );
+            let wall_elapsed = timestamps.send_wall.unwrap() - receive;
+            let mono_elapsed = timestamps.send_mono.unwrap() - timestamps.recv_mono.unwrap();
+            assert!(
+                wall_elapsed - mono_elapsed >= 50_000_000,
+                "queue delay must not move the monotonic endpoint"
+            );
+        } else {
+            let receive = timestamps.recv_wall.or(timestamps.midpoint_wall).unwrap();
+            assert!(
+                receive >= resumed_at,
+                "fallback and midpoint must use userspace time"
+            );
+        }
+        stop_tx.send(()).unwrap();
+        run.await.unwrap().unwrap();
+    }
+}

crates/irtt-server/tests/session_behavior.rs

@@ -1,4 +1,7 @@
-use std::{net::SocketAddr, time::Duration};
+use std::{
+    net::{SocketAddr, SocketAddrV6},
+    time::Duration,
+};
 
 use irtt_proto::{decode_echo_reply, decode_open_reply, encode_request, Params, RequestToEncode};
 use irtt_server::{ServerConfig, ServerCore};
@@ -116,3 +119,143 @@ fn zero_burst_and_zero_idle_timeout_silently_refuse_echoes() {
         assert!(server.handle_datagram(peer, &echo).unwrap().is_none());
     }
 }
+
+#[test]
+fn ipv6_flow_labels_share_a_session_but_zones_do_not() {
+    let endpoint = |flow, scope| {
+        SocketAddr::V6(SocketAddrV6::new(
+            "fe80::1234".parse().unwrap(),
+            40000,
+            flow,
+            scope,
+        ))
+    };
+    let mut server =
+        ServerCore::new(ServerConfig::default().with_min_send_interval(Duration::ZERO));
+    let params = Params {
+        received_stats: irtt_proto::ReceivedStats::Both,
+        ..Params::with_protocol_defaults()
+    };
+    let packet = encode_request(
+        RequestToEncode::Open {
+            params: &params,
+            no_test: false,
+        },
+        None,
+    )
+    .unwrap();
+    let reply = server
+        .handle_datagram(endpoint(1, 2), &packet)
+        .unwrap()
+        .unwrap();
+    let opened = decode_open_reply(reply.bytes(), None).unwrap();
+    let echo = encode_request(
+        RequestToEncode::Echo {
+            token: opened.token,
+            sequence: 0,
+            params: &opened.params,
+            payload: &[],
+        },
+        None,
+    )
+    .unwrap();
+    let close = encode_request(
+        RequestToEncode::Close {
+            token: opened.token,
+        },
+        None,
+    )
+    .unwrap();
+    assert!(server
+        .handle_datagram(endpoint(1, 3), &echo)
+        .unwrap()
+        .is_none());
+    assert!(server
+        .handle_datagram(endpoint(1, 3), &close)
+        .unwrap()
+        .is_none());
+    assert_eq!(server.session_count(), 1);
+    let reply = server
+        .handle_datagram(endpoint(99, 2), &echo)
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        decode_echo_reply(reply.bytes(), &opened.params, None)
+            .unwrap()
+            .recv_count,
+        Some(1)
+    );
+    assert!(server
+        .handle_datagram(endpoint(99, 2), &close)
+        .unwrap()
+        .is_none());
+    assert_eq!(server.session_count(), 0);
+}
+
+#[test]
+fn oversized_authenticated_echo_does_not_advance_session_state() {
+    let key = b"oversized-echo-test";
+    let peer = "127.0.0.1:40000".parse().unwrap();
+    let mut server = ServerCore::new(
+        ServerConfig::default()
+            .with_hmac_key(key)
+            .with_max_packet_length(96)
+            .with_burst_allowance(1)
+            .with_min_send_interval(Duration::from_secs(60))
+            .with_max_test_duration(Duration::from_millis(1))
+            .with_idle_timeout(Duration::MAX),
+    );
+    let params = Params {
+        length: 96,
+        received_stats: irtt_proto::ReceivedStats::Both,
+        ..Params::with_protocol_defaults()
+    };
+    let open = encode_request(
+        RequestToEncode::Open {
+            params: &params,
+            no_test: false,
+        },
+        Some(key),
+    )
+    .unwrap();
+    let reply = server.handle_datagram(peer, &open).unwrap().unwrap();
+    let opened = decode_open_reply(reply.bytes(), Some(key)).unwrap();
+    let oversized_params = Params {
+        length: 128,
+        ..opened.params.clone()
+    };
+    let oversized = encode_request(
+        RequestToEncode::Echo {
+            token: opened.token,
+            sequence: 63,
+            params: &oversized_params,
+            payload: &[],
+        },
+        Some(key),
+    )
+    .unwrap();
+    assert!(oversized.len() > 96);
+    assert!(server.handle_datagram(peer, &oversized).unwrap().is_none());
+    // If the oversized request started maximum duration, the first valid
+    // echo would now close. Oversleep preserves this distinction.
+    std::thread::sleep(Duration::from_millis(2001));
+    let valid = encode_request(
+        RequestToEncode::Echo {
+            token: opened.token,
+            sequence: 0,
+            params: &opened.params,
+            payload: &[],
+        },
+        Some(key),
+    )
+    .unwrap();
+    let reply = server
+        .handle_datagram(peer, &valid)
+        .unwrap()
+        .expect("oversized request must not consume allowance");
+    let reply = decode_echo_reply(reply.bytes(), &opened.params, Some(key)).unwrap();
+    assert_eq!(reply.flags & irtt_proto::FLAG_CLOSE, 0);
+    assert_eq!(reply.recv_count, Some(1));
+    assert_eq!(reply.recv_window, Some(1));
+    assert_eq!(server.session_count(), 1);
+}

crates/irtt-stats/tests/rolling_properties.rs

@@ -1,6 +1,6 @@
 //! Independent count and time windows must match a filtered replay through
 //! StatsCollector, while cumulative accounting retains the complete history.
-//! Generated events use non-decreasing timestamps, as required by the collector.
+//! Generated events have non-decreasing normalized window timestamps.
 
 use std::time::{Duration, Instant, UNIX_EPOCH};
 
@@ -170,7 +170,7 @@ fn build_event(op: &Op, seq: u32, clock_ms: u64, base: Instant) -> ClientEvent {
         Op::Loss => ClientEvent::EchoLoss {
             seq,
             sent_at: ts_at(base, clock_ms),
-            timeout_at: ts_at(base, clock_ms).mono,
+            timeout_at: ts_at(base, clock_ms).mono - Duration::from_millis(10),
         },
         Op::Warning => ClientEvent::Warning {
             kind: irtt_client::WarningKind::UntrackedReply,
@@ -233,6 +233,11 @@ proptest! {
         // scratch after every step.
         let mut history: Vec<(ClientEvent, u64)> = Vec::new();
 
+        // Force a loss whose clamped time evicts the initial send, while its
+        // raw pre-send deadline would retain it. Every generated case must
+        // therefore distinguish the two normalization choices.
+        let ops = [Op::Send, Op::Advance(u16::try_from(time_limit_ms + 5).unwrap()), Op::Loss, Op::Advance(10), Op::Warning]
+            .into_iter().chain(ops).collect::<Vec<_>>();
         for (idx, op) in ops.iter().enumerate() {
             if let Op::Advance(delta) = op {
                 clock_ms += u64::from(*delta);