refactor(irtt-client): move cadence into managed runner

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

@@ -18,3 +18,6 @@ pub use task::*;
 
 #[cfg(all(feature = "tokio", test))]
 mod blocking_tests;
+
+#[cfg(feature = "tokio")]
+mod schedule;

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

@@ -0,0 +1,228 @@
+use std::time::{Duration, Instant};
+
+use crate::{error::ClientError, session::NegotiatedParams};
+
+#[derive(Debug)]
+pub(crate) struct ProbeSchedule {
+    end_at: Option<Instant>,
+    interval: Duration,
+    next_send_at: Option<Instant>,
+}
+
+#[derive(Debug)]
+pub(crate) struct ScheduleCommit {
+    pub(crate) scheduled_at: Instant,
+    pub(crate) next_send_at: Option<Instant>,
+}
+
+impl ProbeSchedule {
+    pub(crate) fn new(
+        start_at: Instant,
+        negotiated: &NegotiatedParams,
+    ) -> Result<Self, ClientError> {
+        let interval_ns = u64::try_from(negotiated.params.interval_ns)
+            .expect("validated positive negotiated interval");
+        let interval = Duration::from_nanos(interval_ns);
+        let end_at = if negotiated.params.duration_ns > 0 {
+            let duration_ns = u64::try_from(negotiated.params.duration_ns)
+                .expect("validated positive negotiated duration");
+            Some(
+                start_at
+                    .checked_add(Duration::from_nanos(duration_ns))
+                    .ok_or_else(|| ClientError::NegotiationRejected {
+                        reason: "duration is too large to schedule".to_owned(),
+                    })?,
+            )
+        } else {
+            None
+        };
+
+        Ok(Self {
+            end_at,
+            interval,
+            next_send_at: Some(start_at),
+        })
+    }
+
+    pub(crate) fn next_send_deadline(&self) -> Option<Instant> {
+        self.next_send_at
+    }
+
+    pub(crate) fn interval(&self) -> Duration {
+        self.interval
+    }
+
+    pub(crate) fn permit_probe_at(&mut self, now: Instant) -> bool {
+        if self.next_send_at.is_none() {
+            return false;
+        }
+        if self.end_at.is_some_and(|end| now >= end) {
+            self.next_send_at = None;
+            return false;
+        }
+        true
+    }
+
+    pub(crate) fn preflight_managed_commit(
+        &self,
+        scheduled_at: Instant,
+        permission_at: Instant,
+    ) -> Result<ScheduleCommit, ClientError> {
+        let (scheduled_at, next_send_at) =
+            advance_cadence(scheduled_at, self.interval, permission_at)?;
+        Ok(self.finish_commit(scheduled_at, next_send_at))
+    }
+
+    pub(crate) fn commit(&mut self, commit: ScheduleCommit) {
+        self.next_send_at = commit.next_send_at;
+    }
+
+    /// Reclaim slots crossed while a successful socket submission was in flight.
+    /// This follows the send commit; it never advances a failed submission.
+    pub(crate) fn skip_missed_slots_at(&mut self, now: Instant) -> Result<(), ClientError> {
+        if let Some(deadline) = self.next_send_at.filter(|deadline| *deadline <= now) {
+            let commit = self.preflight_managed_commit(deadline, now)?;
+            self.commit(commit);
+        }
+        Ok(())
+    }
+
+    pub(crate) fn is_finished(&self) -> bool {
+        self.next_send_at.is_none()
+    }
+
+    fn finish_commit(&self, scheduled_at: Instant, next_send_at: Instant) -> ScheduleCommit {
+        ScheduleCommit {
+            scheduled_at,
+            next_send_at: if self.end_at.is_some_and(|end| next_send_at >= end) {
+                None
+            } else {
+                Some(next_send_at)
+            },
+        }
+    }
+}
+
+pub(crate) fn advance_cadence(
+    deadline: Instant,
+    interval: Duration,
+    now: Instant,
+) -> Result<(Instant, Instant), ClientError> {
+    if interval.is_zero() {
+        return Err(ClientError::InvalidConfig {
+            reason: "probe interval must be greater than zero".to_owned(),
+        });
+    }
+
+    let elapsed_slots = now
+        .checked_duration_since(deadline)
+        .map_or(0, |elapsed| elapsed.as_nanos() / interval.as_nanos());
+    let scheduled_offset = duration_from_nanos(
+        interval
+            .as_nanos()
+            .checked_mul(elapsed_slots)
+            .ok_or(ClientError::DurationOverflow)?,
+    )?;
+    let next_offset = duration_from_nanos(
+        interval
+            .as_nanos()
+            .checked_mul(
+                elapsed_slots
+                    .checked_add(1)
+                    .ok_or(ClientError::DurationOverflow)?,
+            )
+            .ok_or(ClientError::DurationOverflow)?,
+    )?;
+    let scheduled_at = deadline
+        .checked_add(scheduled_offset)
+        .ok_or(ClientError::DurationOverflow)?;
+    let next_at = deadline
+        .checked_add(next_offset)
+        .ok_or(ClientError::DurationOverflow)?;
+    Ok((scheduled_at, next_at))
+}
+
+fn duration_from_nanos(nanos: u128) -> Result<Duration, ClientError> {
+    const NANOS_PER_SECOND: u128 = 1_000_000_000;
+    let seconds =
+        u64::try_from(nanos / NANOS_PER_SECOND).map_err(|_| ClientError::DurationOverflow)?;
+    let subsec_nanos = u32::try_from(nanos % NANOS_PER_SECOND)
+        .expect("nanosecond remainder is always less than one second");
+    Ok(Duration::new(seconds, subsec_nanos))
+}
+
+pub(crate) fn instant_abs_diff(left: Instant, right: Instant) -> Duration {
+    left.checked_duration_since(right)
+        .or_else(|| right.checked_duration_since(left))
+        .unwrap_or(Duration::ZERO)
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use irtt_proto::Params;
+
+    fn negotiated(duration_ns: i64) -> NegotiatedParams {
+        NegotiatedParams {
+            params: Params {
+                interval_ns: 10_000_000,
+                duration_ns,
+                ..Params::default()
+            },
+            restrictions: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn missed_slots_skip_on_the_absolute_grid_only_after_commit() {
+        let start = Instant::now();
+        let mut schedule = ProbeSchedule::new(start, &negotiated(0)).unwrap();
+        for (now_ms, intended_ms, next_ms) in [(0, 0, 10), (45, 40, 50), (50, 50, 60)] {
+            let deadline = schedule.next_send_deadline().unwrap();
+            let commit = schedule
+                .preflight_managed_commit(deadline, start + Duration::from_millis(now_ms))
+                .unwrap();
+            assert_eq!(
+                commit.scheduled_at,
+                start + Duration::from_millis(intended_ms)
+            );
+            // A failed or pending submission discards this preflight.
+            assert_eq!(schedule.next_send_deadline(), Some(deadline));
+            schedule.commit(commit);
+            assert_eq!(
+                schedule.next_send_deadline(),
+                Some(start + Duration::from_millis(next_ms))
+            );
+        }
+        schedule
+            .skip_missed_slots_at(start + Duration::from_millis(85))
+            .unwrap();
+        assert_eq!(
+            schedule.next_send_deadline(),
+            Some(start + Duration::from_millis(90))
+        );
+    }
+
+    #[test]
+    fn finite_duration_excludes_its_end_even_when_a_slot_is_missed() {
+        let start = Instant::now();
+        let mut schedule = ProbeSchedule::new(start, &negotiated(20_000_000)).unwrap();
+        assert!(schedule.permit_probe_at(start + Duration::from_millis(19)));
+        let commit = schedule
+            .preflight_managed_commit(start, start + Duration::from_millis(19))
+            .unwrap();
+        schedule.commit(commit);
+        assert!(schedule.is_finished());
+
+        let mut missed = ProbeSchedule::new(start, &negotiated(20_000_000)).unwrap();
+        assert!(!missed.permit_probe_at(start + Duration::from_millis(20)));
+        assert!(missed.is_finished());
+    }
+
+    #[test]
+    fn cadence_rejects_overflow_before_commit() {
+        assert!(matches!(
+            advance_cadence(Instant::now(), Duration::MAX, Instant::now()),
+            Err(ClientError::DurationOverflow)
+        ));
+    }
+}

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

@@ -25,6 +25,8 @@ use crate::{
     ClientConfig, ClientError, ClientEvent, OpenOutcome,
 };
 
+use super::schedule::{instant_abs_diff, ProbeSchedule};
+
 use super::{
     classify_client_error, ManagedClientConfig, ManagedCommandAcknowledgement,
     ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError,
@@ -504,6 +506,7 @@ impl TargetCounters {
 }
 
 struct TargetRuntime {
+    schedule: Option<ProbeSchedule>,
     instance: TargetInstance,
     config: ManagedTargetConfig,
     desired: bool,
@@ -991,6 +994,7 @@ impl ManagedClientTask {
             };
             created_instances.push(instance.clone());
             self.targets.push(TargetRuntime {
+                schedule: None,
                 instance,
                 server_addr: Arc::from(planned.target.server_addr.clone()),
                 config: planned.target,
@@ -1243,7 +1247,25 @@ impl ManagedClientTask {
                         self.targets[index].state = TargetState::Opening { client, open };
                         false
                     }
-                    Poll::Ready(Ok(OpenOutcome::Started { event, .. })) => {
+                    Poll::Ready(Ok(OpenOutcome::Started {
+                        event, negotiated, ..
+                    })) => {
+                        let ClientEvent::SessionStarted { at: opened_at, .. } = &event else {
+                            unreachable!("Started carries SessionStarted");
+                        };
+                        match ProbeSchedule::new(opened_at.mono, &negotiated) {
+                            Ok(schedule) => self.targets[index].schedule = Some(schedule),
+                            Err(error) => {
+                                return self.begin_open_session_failure(
+                                    index,
+                                    client,
+                                    ManagedTargetFailurePhase::Timing,
+                                    error,
+                                    now,
+                                    OpenSessionFailureCleanup::Close,
+                                )
+                            }
+                        }
                         self.publish_client_events(index, vec![event]);
                         if self.state == DriverState::Stopping
                             || self.effective_retirement(index).is_some()
@@ -1340,7 +1362,15 @@ impl ManagedClientTask {
                     self.finish_target(index, ManagedTargetEndReason::PeerClosed, None);
                     return false;
                 }
-                if client.is_run_complete() {
+                if self.targets[index]
+                    .schedule
+                    .as_mut()
+                    .is_some_and(|schedule| {
+                        schedule.permit_probe_at(now);
+                        schedule.is_finished()
+                    })
+                    && !client.has_pending_probes()
+                {
                     self.targets[index].sync_packets_sent(&client);
                     return self.begin_drain(
                         index,
@@ -1620,13 +1650,15 @@ impl ManagedClientTask {
             if !target.desired || target.retirement.is_some() {
                 continue;
             }
-            let TargetState::Active { client } = &target.state else {
+            let TargetState::Active { .. } = &target.state else {
                 continue;
             };
             active += 1;
-            let interval = client
-                .probe_interval()
-                .expect("active managed targets have a committed probe schedule");
+            let interval = target
+                .schedule
+                .as_ref()
+                .expect("active managed targets have a probe schedule")
+                .interval();
             minimum = minimum.into_iter().chain(Some(interval)).min();
         }
         minimum.map(|interval| (active, stagger_spacing(interval, active)))
@@ -1709,6 +1741,9 @@ impl ManagedClientTask {
             self.targets[index].state = state;
             return SendResult::NotAttempted;
         };
+        // A burst pass may span several slots; validate this target against
+        // fresh time rather than the pass's earlier scheduling snapshot.
+        let now = now.max(Instant::now());
         if client
             .next_probe_timeout_deadline()
             .is_some_and(|deadline| deadline <= now)
@@ -1717,7 +1752,21 @@ impl ManagedClientTask {
             self.targets[index].state = TargetState::Active { client };
             return SendResult::NotAttempted;
         }
-        if client
+        let schedule = self.targets[index]
+            .schedule
+            .as_mut()
+            .expect("active managed targets have a probe schedule");
+        if !schedule.permit_probe_at(now) {
+            self.targets[index].send_waiting = false;
+            if client.has_pending_probes() {
+                self.targets[index].state = TargetState::Active { client };
+                return SendResult::NotAttempted;
+            }
+            self.targets[index].sync_packets_sent(&client);
+            self.begin_drain(index, client, ManagedTargetEndReason::TestComplete, now);
+            return SendResult::Ready { accepted: false };
+        }
+        if schedule
             .next_send_deadline()
             .is_none_or(|deadline| deadline > now)
         {
@@ -1725,8 +1774,25 @@ impl ManagedClientTask {
             self.targets[index].state = TargetState::Active { client };
             return SendResult::NotAttempted;
         }
+        let schedule = self.targets[index].schedule.as_ref().unwrap();
+        let commit =
+            match schedule.preflight_managed_commit(schedule.next_send_deadline().unwrap(), now) {
+                Ok(commit) => commit,
+                Err(error) => {
+                    self.begin_open_session_failure(
+                        index,
+                        client,
+                        ManagedTargetFailurePhase::Timing,
+                        error,
+                        now,
+                        OpenSessionFailureCleanup::Drain,
+                    );
+                    return SendResult::Failed { accepted: false };
+                }
+            };
+        let scheduled_at = commit.scheduled_at;
         let before = client.packets_sent();
-        let result = client.poll_send_probe(cx);
+        let mut result = client.poll_send_probe(cx);
         let after = client.packets_sent();
         let accepted = after > before;
         let send_result = match &result {
@@ -1735,10 +1801,37 @@ impl ManagedClientTask {
             Poll::Ready(Err(_)) => SendResult::Failed { accepted },
         };
         self.record_stagger_acceptance(send_result, stagger_spacing, Instant::now());
-        let schedule_error = accepted
-            .then(|| client.skip_missed_probe_slots_at(Instant::now()))
-            .transpose()
-            .err();
+        if accepted {
+            self.targets[index]
+                .schedule
+                .as_mut()
+                .unwrap()
+                .commit(commit);
+            if let Poll::Ready(Ok(events)) = &mut result {
+                for event in events {
+                    if let ClientEvent::EchoSent {
+                        scheduled_at: intended,
+                        sent_at,
+                        timer_error,
+                        ..
+                    } = event
+                    {
+                        *intended = Some(scheduled_at);
+                        *timer_error = Some(instant_abs_diff(sent_at.mono, scheduled_at));
+                    }
+                }
+            }
+        }
+        let schedule_error = if accepted {
+            self.targets[index]
+                .schedule
+                .as_mut()
+                .unwrap()
+                .skip_missed_slots_at(Instant::now())
+                .err()
+        } else {
+            None
+        };
         self.targets[index].record_packets_sent(after);
         match result {
             Poll::Pending => {
@@ -1754,7 +1847,7 @@ impl ManagedClientTask {
                     let TargetState::Active { client } =
                         mem::replace(&mut self.targets[index].state, TargetState::Terminal)
                     else {
-                        unreachable!("probe sender remained active while publishing send events");
+                        unreachable!("sender remained active")
                     };
                     self.begin_open_session_failure(
                         index,
@@ -1769,6 +1862,7 @@ impl ManagedClientTask {
                     SendResult::Ready { accepted }
                 }
             }
+
             Poll::Ready(Err(error)) => {
                 self.targets[index].send_waiting = false;
                 self.begin_open_session_failure(
@@ -2032,7 +2126,12 @@ impl ManagedClientTask {
                     if target.desired && target.retirement.is_none() && !target.send_waiting {
                         send_deadline = send_deadline
                             .into_iter()
-                            .chain(client.next_send_deadline())
+                            .chain(
+                                target
+                                    .schedule
+                                    .as_ref()
+                                    .and_then(ProbeSchedule::next_send_deadline),
+                            )
                             .min();
                     }
                 }
@@ -2349,6 +2448,7 @@ fn build_task(
             }
         })?;
         runtimes.push(TargetRuntime {
+            schedule: None,
             instance: TargetInstance {
                 id: target.id.clone(),
                 generation,

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

@@ -6,7 +6,7 @@ use std::{
     sync::{Arc, Condvar, Mutex},
     task::{Context, Poll, Waker},
     thread::{self, JoinHandle},
-    time::{Duration, Instant, SystemTime},
+    time::{Duration, Instant},
 };
 
 use irtt_proto::{
@@ -16,10 +16,7 @@ use irtt_proto::{
 use tokio::runtime::{Builder, Runtime};
 
 use super::*;
-use crate::{
-    probe::PendingProbe, socket::resolution_call_counts, ClientAuthConfig, ClientTimestamp,
-    NegotiationPolicy, RunMode,
-};
+use crate::{socket::resolution_call_counts, ClientAuthConfig, NegotiationPolicy, RunMode};
 
 const TOKEN: u64 = 0x1234_5678_90ab_cdef;
 
@@ -337,110 +334,6 @@ fn wait_flag(flag: &Arc<(Mutex<bool>, Condvar)>) {
     assert!(*value && !timeout.timed_out());
 }
 
-fn client_events(
-    observations: &Mutex<Vec<(ManagedEvent, Arc<ManagedStatus>)>>,
-) -> Vec<ClientEvent> {
-    observations
-        .lock()
-        .unwrap()
-        .iter()
-        .filter_map(|(event, _)| match event {
-            ManagedEvent::Client { event, .. } => Some(event.clone()),
-            _ => None,
-        })
-        .collect()
-}
-
-fn timeout_probe(seq: u32, sent_at: Instant, timeout_at: Instant) -> PendingProbe {
-    PendingProbe {
-        wire_seq: seq,
-        sent_at: ClientTimestamp {
-            mono: sent_at,
-            wall: SystemTime::UNIX_EPOCH,
-        },
-        timeout_at,
-        tx_not_before_wall: SystemTime::UNIX_EPOCH,
-        kernel_tx_timestamp: None,
-    }
-}
-
-fn open_target_for_timeout_test(runtime: &Runtime, task: &mut ManagedClientTask, index: usize) {
-    let TargetState::Pending { client_config } =
-        mem::replace(&mut task.targets[index].state, TargetState::Terminal)
-    else {
-        panic!("timeout test target must start pending");
-    };
-    let client = runtime.block_on(async {
-        let mut client = AsyncClient::connect(client_config).await.unwrap();
-        client.open().await.unwrap();
-        client
-    });
-    task.targets[index].desired = false;
-    task.install_target_state(index, TargetState::Active { client });
-}
-
-fn replace_pending_for_timeout_test(
-    task: &mut ManagedClientTask,
-    index: usize,
-    probe: PendingProbe,
-) {
-    let client = match &mut task.targets[index].state {
-        TargetState::Active { client } | TargetState::Draining { client, .. } => client,
-        _ => panic!("timeout test target must be active or draining"),
-    };
-    client.replace_pending_for_test(probe);
-}
-
-fn remove_pending_for_timeout_test(
-    task: &mut ManagedClientTask,
-    index: usize,
-    wire_seq: u32,
-) -> PendingProbe {
-    let client = match &mut task.targets[index].state {
-        TargetState::Active { client } | TargetState::Draining { client, .. } => client,
-        _ => panic!("timeout test target must be active or draining"),
-    };
-    client
-        .remove_pending_for_test(wire_seq)
-        .expect("timeout test pending probe must exist")
-}
-
-fn close_timeout_test_target(runtime: &Runtime, task: &mut ManagedClientTask, index: usize) {
-    let (TargetState::Active { mut client } | TargetState::Draining { mut client, .. }) =
-        mem::replace(&mut task.targets[index].state, TargetState::Terminal)
-    else {
-        panic!("timeout test target did not remain active or draining");
-    };
-    runtime.block_on(client.close()).unwrap();
-}
-
-fn poll_task_once(runtime: &Runtime, task: &mut Pin<Box<ManagedClientTask>>) {
-    runtime.block_on(poll_fn(|cx| {
-        assert!(task.as_mut().poll(cx).is_pending());
-        Poll::Ready(())
-    }));
-}
-
-fn timeout_losses_for(
-    observations: &Mutex<Vec<(ManagedEvent, Arc<ManagedStatus>)>>,
-    id: &str,
-) -> usize {
-    observations
-        .lock()
-        .unwrap()
-        .iter()
-        .filter(|(event, _)| {
-            matches!(
-                event,
-                ManagedEvent::Client {
-                    target,
-                    event: ClientEvent::EchoLoss { .. },
-                } if target.id.as_ref() == id
-            )
-        })
-        .count()
-}
-
 fn probes(records: &[PacketRecord]) -> Vec<Instant> {
     records
         .iter()
@@ -471,7 +364,12 @@ fn echo_sends(
                         sent_at,
                         ..
                     },
-            } => Some((target.clone(), *seq, *scheduled_at, sent_at.mono)),
+            } => Some((
+                target.clone(),
+                *seq,
+                scheduled_at.expect("managed sends carry schedule metadata"),
+                sent_at.mono,
+            )),
             _ => None,
         })
         .collect()
@@ -1638,83 +1536,6 @@ fn peer_close_during_stop_drain_remains_authoritative() {
     assert!(!has_close(&records));
 }
 
-#[test]
-fn stop_during_opening_finishes_in_flight_work() {
-    let open_gate = Arc::new(PacketGate::default());
-    let server = start_server_with_gates(
-        ServerBehavior::Echo,
-        None,
-        None,
-        Some(Arc::clone(&open_gate)),
-        None,
-    );
-    let mut managed = config(ManagedPacing::Staggered);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    managed.client.open_timeouts = vec![Duration::from_secs(1), Duration::from_secs(1)];
-    let (task, handle) =
-        ManagedClient::task(managed, vec![target("success", server.addr)]).unwrap();
-    let runtime = runtime();
-    let mut task = Box::pin(task);
-    drive_task_until(&runtime, &mut task, |task| match &task.targets[0].state {
-        TargetState::Opening { open, .. } => open.has_in_flight_work(),
-        _ => false,
-    });
-    drop(handle.stop());
-    open_gate.release();
-    let outcome = runtime.block_on(task);
-    let records = server.finish();
-    assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested);
-    assert!(matches!(
-        outcome.recent_target_outcomes[0].end_reason,
-        ManagedTargetEndReason::Stopped
-    ));
-    assert_eq!(
-        records
-            .iter()
-            .filter(|record| matches!(record.kind, PacketKind::Open))
-            .count(),
-        1
-    );
-    assert!(probes(&records).is_empty() && has_close(&records));
-    let key = b"managed-retained-open-cleanup".to_vec();
-    let cleanup_server = start_server(ServerBehavior::Echo, Some(key.clone()));
-    let mut configured = target("cleanup", cleanup_server.addr);
-    configured.auth = Some(ClientAuthConfig {
-        hmac_key: Some(key),
-    });
-    let mut managed = config(ManagedPacing::Staggered);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    let (task, handle) = ManagedClient::task(managed, vec![configured]).unwrap();
-    let mut task = Box::pin(task);
-    drive_task_until(&runtime, &mut task, |task| {
-        matches!(task.targets[0].state, TargetState::Opening { .. })
-    });
-    let TargetState::Opening { client, .. } = &task.as_ref().get_ref().targets[0].state else {
-        unreachable!()
-    };
-    client.retain_open_cleanup_once();
-    drive_task_until(&runtime, &mut task, |task| match &task.targets[0].state {
-        TargetState::Opening { open, .. } => open.has_retained_cleanup(),
-        _ => false,
-    });
-    drop(handle.stop());
-    let outcome = runtime.block_on(task);
-    let records = cleanup_server.finish();
-    let target = &outcome.recent_target_outcomes[0];
-    assert!(matches!(target.end_reason, ManagedTargetEndReason::Stopped));
-    assert_eq!(outcome.failed_target_outcomes, 0);
-    assert!(matches!(
-        target.cleanup_failure,
-        Some(ManagedTargetFailure {
-            phase: ManagedTargetFailurePhase::Opening,
-            ..
-        })
-    ));
-    assert!(probes(&records).is_empty() && has_close(&records));
-}
-
 #[test]
 fn static_multi_target_completion() {
     let first = start_server(ServerBehavior::Echo, None);
@@ -1983,768 +1804,6 @@ fn timeout_discovery_inspects_at_most_one_budget_of_targets() {
     assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET);
 }
 
-#[test]
-fn sparse_timeout_backlog_persists_across_polls() {
-    let server = start_server(ServerBehavior::Echo, None);
-    let target_count = (2 * TIMEOUT_WORK_BUDGET) + 1;
-    let mut managed = config(ManagedPacing::Burst);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    managed.max_live_target_generations = target_count;
-    let targets = std::iter::once(target("sparse", server.addr))
-        .chain(
-            (1..target_count)
-                .map(|index| ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")),
-        )
-        .collect();
-    let (mut task, _) = ManagedClient::task(managed, targets).unwrap();
-    for target in &mut task.targets[1..] {
-        target.state = TargetState::Terminal;
-    }
-    let observations = Arc::new(Mutex::new(Vec::new()));
-    task.event_observations = Some(Arc::clone(&observations));
-    let runtime = runtime();
-    open_target_for_timeout_test(&runtime, &mut task, 0);
-    let now = Instant::now();
-    for seq in 0..u32::try_from((2 * TIMEOUT_WORK_BUDGET) + 64).unwrap() {
-        replace_pending_for_timeout_test(
-            &mut task,
-            0,
-            timeout_probe(
-                seq,
-                now - Duration::from_secs(2),
-                now - Duration::from_secs(1),
-            ),
-        );
-    }
-
-    assert!(task.poll_timeout_pass(now));
-    assert_eq!(
-        timeout_losses_for(&observations, "sparse"),
-        TIMEOUT_WORK_BUDGET
-    );
-    assert_eq!(task.timeout_backlog.len(), 1);
-
-    assert!(task.poll_timeout_pass(now));
-    assert_eq!(
-        timeout_losses_for(&observations, "sparse"),
-        2 * TIMEOUT_WORK_BUDGET
-    );
-    assert_eq!(task.timeout_backlog.len(), 1);
-    close_timeout_test_target(&runtime, &mut task, 0);
-    drop(task);
-    server.finish();
-}
-
-#[test]
-fn timeout_backlog_does_not_starve_discovery() {
-    let first = start_server(ServerBehavior::Echo, None);
-    let second = start_server(ServerBehavior::Echo, None);
-    let target_count = (2 * TIMEOUT_WORK_BUDGET) + 1;
-    let late_index = target_count - 1;
-    let mut managed = config(ManagedPacing::Burst);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    managed.max_live_target_generations = target_count;
-    let targets = (0..target_count)
-        .map(|index| match index {
-            0 => target("backlog", first.addr),
-            index if index == late_index => target("late", second.addr),
-            _ => ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9"),
-        })
-        .collect();
-    let (mut task, _) = ManagedClient::task(managed, targets).unwrap();
-    for target in &mut task.targets[1..late_index] {
-        target.state = TargetState::Terminal;
-    }
-    let observations = Arc::new(Mutex::new(Vec::new()));
-    task.event_observations = Some(Arc::clone(&observations));
-    let runtime = runtime();
-    open_target_for_timeout_test(&runtime, &mut task, 0);
-    open_target_for_timeout_test(&runtime, &mut task, late_index);
-    let now = Instant::now();
-    for seq in 0..u32::try_from(3 * TIMEOUT_WORK_BUDGET).unwrap() {
-        replace_pending_for_timeout_test(
-            &mut task,
-            0,
-            timeout_probe(
-                seq,
-                now - Duration::from_secs(2),
-                now - Duration::from_secs(1),
-            ),
-        );
-    }
-    replace_pending_for_timeout_test(
-        &mut task,
-        late_index,
-        timeout_probe(
-            0,
-            now - Duration::from_secs(2),
-            now - Duration::from_secs(1),
-        ),
-    );
-    task.timeout_backlog.push_back(TimeoutBacklogEntry {
-        index: 0,
-        generation: task.targets[0].instance.generation,
-    });
-    task.timeout_cursor = 1;
-
-    assert!(task.poll_timeout_pass(now));
-    assert_eq!(timeout_losses_for(&observations, "late"), 0);
-
-    assert!(task.poll_timeout_pass(now));
-    assert_eq!(timeout_losses_for(&observations, "late"), 1);
-    assert_eq!(task.timeout_backlog.len(), 1);
-    close_timeout_test_target(&runtime, &mut task, 0);
-    close_timeout_test_target(&runtime, &mut task, late_index);
-    drop(task);
-    first.finish();
-    second.finish();
-}
-
-#[test]
-fn stale_timeout_backlog_entry_is_dropped_after_pruning() {
-    let server = start_server(ServerBehavior::Echo, None);
-    let target_count = TIMEOUT_WORK_BUDGET + 2;
-    let mut managed = config(ManagedPacing::Burst);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    managed.max_live_target_generations = target_count;
-    let targets = vec![
-        ManagedTargetConfig::new("retired", "127.0.0.1:9"),
-        target("survivor", server.addr),
-    ]
-    .into_iter()
-    .chain(
-        (2..target_count)
-            .map(|index| ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")),
-    )
-    .collect();
-    let (mut task, _) = ManagedClient::task(managed, targets).unwrap();
-    let observations = Arc::new(Mutex::new(Vec::new()));
-    task.event_observations = Some(Arc::clone(&observations));
-    let runtime = runtime();
-    open_target_for_timeout_test(&runtime, &mut task, 1);
-    let now = Instant::now();
-    replace_pending_for_timeout_test(
-        &mut task,
-        1,
-        timeout_probe(
-            0,
-            now - Duration::from_secs(2),
-            now - Duration::from_secs(1),
-        ),
-    );
-    task.timeout_backlog.push_back(TimeoutBacklogEntry {
-        index: 0,
-        generation: task.targets[0].instance.generation,
-    });
-    task.targets[0].desired = false;
-    task.targets[0].state = TargetState::Terminal;
-    task.prune_undesired_terminal();
-    task.timeout_cursor = 1;
-
-    assert!(!task.poll_timeout_pass(now));
-    assert!(task.timeout_backlog.is_empty());
-    assert_eq!(timeout_losses_for(&observations, "survivor"), 0);
-
-    task.timeout_cursor = 0;
-    assert!(!task.poll_timeout_pass(now));
-    assert_eq!(timeout_losses_for(&observations, "survivor"), 1);
-    close_timeout_test_target(&runtime, &mut task, 0);
-    drop(task);
-    server.finish();
-}
-
-#[test]
-fn immediate_timeout_backlog_skips_global_deadline_scan() {
-    let server = start_server(ServerBehavior::Echo, None);
-    let mut managed = config(ManagedPacing::Burst);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    let (mut task, _) = ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap();
-    let runtime = runtime();
-    open_target_for_timeout_test(&runtime, &mut task, 0);
-    let now = Instant::now();
-    for seq in 0..u32::try_from(TIMEOUT_WORK_BUDGET + 1).unwrap() {
-        replace_pending_for_timeout_test(
-            &mut task,
-            0,
-            timeout_probe(
-                seq,
-                now - Duration::from_secs(2),
-                now - Duration::from_secs(1),
-            ),
-        );
-    }
-    task.timeout_backlog.push_back(TimeoutBacklogEntry {
-        index: 0,
-        generation: task.targets[0].instance.generation,
-    });
-    task.deadline_inspections = 0;
-
-    let mut task = Box::pin(task);
-    poll_task_once(&runtime, &mut task);
-    assert_eq!(task.deadline_inspections, 0);
-    assert_eq!(task.timeout_backlog.len(), 1);
-
-    assert!(!task.poll_timeout_pass(now));
-    assert!(task.timeout_backlog.is_empty());
-    task.deadline_inspections = 0;
-    poll_task_once(&runtime, &mut task);
-    assert_eq!(task.deadline_inspections, 1);
-    close_timeout_test_target(&runtime, &mut task, 0);
-    drop(task);
-    server.finish();
-}
-
-#[test]
-fn timeout_discovery_reaches_later_target_after_budget_slice() {
-    let server = start_server(ServerBehavior::Echo, None);
-    let target_count = TIMEOUT_WORK_BUDGET + 1;
-    let late_index = target_count - 1;
-    let mut managed = config(ManagedPacing::Burst);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    managed.max_live_target_generations = target_count;
-    let targets = (0..target_count)
-        .map(|index| {
-            if index == late_index {
-                target("late", server.addr)
-            } else {
-                ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")
-            }
-        })
-        .collect();
-    let (mut task, _) = ManagedClient::task(managed, targets).unwrap();
-    for target in &mut task.targets[..late_index] {
-        target.state = TargetState::Terminal;
-    }
-    let observations = Arc::new(Mutex::new(Vec::new()));
-    task.event_observations = Some(Arc::clone(&observations));
-    let runtime = runtime();
-    open_target_for_timeout_test(&runtime, &mut task, late_index);
-    let now = Instant::now();
-    replace_pending_for_timeout_test(
-        &mut task,
-        late_index,
-        timeout_probe(0, now - Duration::from_secs(1), now),
-    );
-
-    task.timeout_inspections = 0;
-    assert!(!task.poll_timeout_pass(now));
-    assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET);
-    assert_eq!(timeout_losses_for(&observations, "late"), 0);
-
-    task.timeout_inspections = 0;
-    assert!(!task.poll_timeout_pass(now));
-    assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET);
-    assert_eq!(timeout_losses_for(&observations, "late"), 1);
-
-    let TargetState::Active { mut client } =
-        mem::replace(&mut task.targets[late_index].state, TargetState::Terminal)
-    else {
-        panic!("later timeout target did not remain active");
-    };
-    runtime.block_on(client.close()).unwrap();
-    drop(task);
-    server.finish();
-}
-
-#[test]
-fn timeout_work_is_bounded_per_task_poll_and_eventually_drains() {
-    let server = start_server(ServerBehavior::Echo, None);
-    let mut managed = config(ManagedPacing::Staggered);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    let (mut task, handle) =
-        ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap();
-    let observations = Arc::new(Mutex::new(Vec::new()));
-    task.event_observations = Some(Arc::clone(&observations));
-    let runtime = runtime();
-    open_target_for_timeout_test(&runtime, &mut task, 0);
-    let now = Instant::now();
-    for seq in 0..u32::try_from(TIMEOUT_WORK_BUDGET + 1).unwrap() {
-        replace_pending_for_timeout_test(
-            &mut task,
-            0,
-            timeout_probe(
-                seq,
-                now - Duration::from_secs(2),
-                now - Duration::from_secs(1),
-            ),
-        );
-    }
-
-    let mut task = Box::pin(task);
-    poll_task_once(&runtime, &mut task);
-    assert_eq!(
-        timeout_losses_for(&observations, "one"),
-        TIMEOUT_WORK_BUDGET
-    );
-    assert!(matches!(
-        &task.targets[0].state,
-        TargetState::Active { client }
-            if client.next_probe_timeout_deadline().is_some()
-    ));
-
-    drive_task_until(&runtime, &mut task, |task| {
-        matches!(
-            &task.targets[0].state,
-            TargetState::Active { client }
-                if client.next_probe_timeout_deadline().is_none()
-        )
-    });
-    assert_eq!(
-        timeout_losses_for(&observations, "one"),
-        TIMEOUT_WORK_BUDGET + 1
-    );
-
-    drop(handle.stop());
-    runtime.block_on(task);
-    server.finish();
-}
-
-#[test]
-fn timeout_work_round_robins_simultaneous_target_backlogs() {
-    let first = start_server(ServerBehavior::Echo, None);
-    let second = start_server(ServerBehavior::Echo, None);
-    let mut managed = config(ManagedPacing::Staggered);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    let (mut task, handle) = ManagedClient::task(
-        managed,
-        vec![target("first", first.addr), target("second", second.addr)],
-    )
-    .unwrap();
-    let observations = Arc::new(Mutex::new(Vec::new()));
-    task.event_observations = Some(Arc::clone(&observations));
-    let runtime = runtime();
-    open_target_for_timeout_test(&runtime, &mut task, 0);
-    open_target_for_timeout_test(&runtime, &mut task, 1);
-    let now = Instant::now();
-    let per_target = TIMEOUT_WORK_BUDGET + 16;
-    for index in 0..2 {
-        for seq in 0..u32::try_from(per_target).unwrap() {
-            replace_pending_for_timeout_test(
-                &mut task,
-                index,
-                timeout_probe(
-                    seq,
-                    now - Duration::from_secs(2),
-                    now - Duration::from_secs(1),
-                ),
-            );
-        }
-    }
-
-    let mut task = Box::pin(task);
-    poll_task_once(&runtime, &mut task);
-    assert_eq!(
-        timeout_losses_for(&observations, "first"),
-        TIMEOUT_WORK_BUDGET / 2
-    );
-    assert_eq!(
-        timeout_losses_for(&observations, "second"),
-        TIMEOUT_WORK_BUDGET / 2
-    );
-    assert!(matches!(
-        &task.targets[0].state,
-        TargetState::Active { client }
-            if client.next_probe_timeout_deadline().is_some()
-    ));
-    assert!(matches!(
-        &task.targets[1].state,
-        TargetState::Active { client }
-            if client.next_probe_timeout_deadline().is_some()
-    ));
-
-    drive_task_until(&runtime, &mut task, |task| {
-        task.targets.iter().all(|target| {
-            matches!(
-                &target.state,
-                TargetState::Active { client }
-                    if client.next_probe_timeout_deadline().is_none()
-            )
-        })
-    });
-    assert_eq!(timeout_losses_for(&observations, "first"), per_target);
-    assert_eq!(timeout_losses_for(&observations, "second"), per_target);
-
-    drop(handle.stop());
-    runtime.block_on(task);
-    first.finish();
-    second.finish();
-}
-
-#[test]
-fn budget_deferred_timeout_blocks_send_until_timeout_processing() {
-    let reply_gate = Arc::new(PacketGate::default());
-    let server = start_server_with_gates(
-        ServerBehavior::Echo,
-        None,
-        None,
-        None,
-        Some(Arc::clone(&reply_gate)),
-    );
-    let target_count = TIMEOUT_WORK_BUDGET + 1;
-    let late_index = target_count - 1;
-    let mut managed = config(ManagedPacing::Burst);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    managed.client.interval = Duration::from_nanos(1);
-    managed.client.probe_timeout = Duration::from_secs(1);
-    managed.client.max_pending_probes = 1;
-    managed.max_live_target_generations = target_count;
-    let targets = (0..target_count)
-        .map(|index| {
-            if index == late_index {
-                target("late", server.addr)
-            } else {
-                ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")
-            }
-        })
-        .collect();
-    let (mut task, _) = ManagedClient::task(managed, targets).unwrap();
-    for target in &mut task.targets[..late_index] {
-        target.state = TargetState::Terminal;
-    }
-    let observations = Arc::new(Mutex::new(Vec::new()));
-    task.event_observations = Some(Arc::clone(&observations));
-    let runtime = runtime();
-    open_target_for_timeout_test(&runtime, &mut task, late_index);
-    task.targets[late_index].desired = true;
-    let sent = match &mut task.targets[late_index].state {
-        TargetState::Active { client } => runtime.block_on(client.send_probe()).unwrap(),
-        _ => panic!("deferred-send target did not open"),
-    };
-    assert!(matches!(
-        sent.as_slice(),
-        [ClientEvent::EchoSent { seq: 0, .. }]
-    ));
-    wait_flag(&server.probe_seen);
-
-    let overdue = Instant::now();
-    replace_pending_for_timeout_test(
-        &mut task,
-        late_index,
-        timeout_probe(0, overdue - Duration::from_secs(1), overdue),
-    );
-    task.state = DriverState::Running;
-    task.lifecycle = ManagedLifecycle::Running;
-    task.timeout_cursor = 0;
-    task.send_cursor = late_index;
-
-    let mut task = Box::pin(task);
-    poll_task_once(&runtime, &mut task);
-    assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET);
-    assert_eq!(timeout_losses_for(&observations, "late"), 0);
-    assert!(matches!(
-        &task.targets[late_index].state,
-        TargetState::Active { client }
-            if client.packets_sent() == 1
-                && client.next_probe_timeout_deadline() == Some(overdue)
-    ));
-    assert!(!task.targets[late_index].send_waiting);
-
-    poll_task_once(&runtime, &mut task);
-    assert_eq!(timeout_losses_for(&observations, "late"), 1);
-    assert!(matches!(
-        &task.targets[late_index].state,
-        TargetState::Active { client }
-            if client.next_probe_timeout_deadline().is_none()
-    ));
-
-    poll_task_once(&runtime, &mut task);
-    assert!(matches!(
-        &task.targets[late_index].state,
-        TargetState::Active { client } if client.packets_sent() == 2
-    ));
-
-    reply_gate.release();
-    let TargetState::Active { mut client } =
-        mem::replace(&mut task.targets[late_index].state, TargetState::Terminal)
-    else {
-        panic!("deferred-send target did not remain active");
-    };
-    runtime.block_on(client.close()).unwrap();
-    drop(task);
-    server.finish();
-}
-
-#[test]
-fn partial_timeout_batch_blocks_active_receive_until_loss_commits() {
-    let reply_gate = Arc::new(PacketGate::default());
-    let server = start_server_with_gates(
-        ServerBehavior::Echo,
-        None,
-        None,
-        None,
-        Some(Arc::clone(&reply_gate)),
-    );
-    let mut managed = config(ManagedPacing::Staggered);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    let (mut task, handle) =
-        ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap();
-    let observations = Arc::new(Mutex::new(Vec::new()));
-    task.event_observations = Some(Arc::clone(&observations));
-    let runtime = runtime();
-    open_target_for_timeout_test(&runtime, &mut task, 0);
-    let sent = match &mut task.targets[0].state {
-        TargetState::Active { client } => runtime.block_on(client.send_probe()).unwrap(),
-        _ => panic!("timeout test target did not open"),
-    };
-    assert!(matches!(
-        sent.as_slice(),
-        [ClientEvent::EchoSent { seq: 0, .. }]
-    ));
-    wait_flag(&server.probe_seen);
-
-    let now = Instant::now();
-    let _ = remove_pending_for_timeout_test(&mut task, 0, 0);
-    for seq in 1..=u32::try_from(TIMEOUT_WORK_BUDGET).unwrap() {
-        replace_pending_for_timeout_test(
-            &mut task,
-            0,
-            timeout_probe(
-                seq,
-                now - Duration::from_secs(2),
-                now - Duration::from_secs(1),
-            ),
-        );
-    }
-    replace_pending_for_timeout_test(
-        &mut task,
-        0,
-        timeout_probe(0, now - Duration::from_secs(1), now),
-    );
-    reply_gate.release();
-    wait_flag(&server.reply_sent);
-
-    let mut task = Box::pin(task);
-    poll_task_once(&runtime, &mut task);
-    assert_eq!(
-        timeout_losses_for(&observations, "one"),
-        TIMEOUT_WORK_BUDGET
-    );
-    assert!(!client_events(&observations).iter().any(|event| {
-        matches!(
-            event,
-            ClientEvent::EchoLoss { seq: 0, .. }
-                | ClientEvent::LateReply { seq: 0, .. }
-                | ClientEvent::EchoReply { seq: 0, .. }
-        )
-    }));
-
-    drive_task_until(&runtime, &mut task, |_| {
-        client_events(&observations)
-            .iter()
-            .any(|event| matches!(event, ClientEvent::LateReply { seq: 0, .. }))
-    });
-    let classified = client_events(&observations)
-        .into_iter()
-        .filter(|event| {
-            matches!(
-                event,
-                ClientEvent::EchoLoss { seq: 0, .. }
-                    | ClientEvent::LateReply { seq: 0, .. }
-                    | ClientEvent::EchoReply { seq: 0, .. }
-            )
-        })
-        .collect::<Vec<_>>();
-    assert!(matches!(
-        classified.as_slice(),
-        [
-            ClientEvent::EchoLoss { seq: 0, .. },
-            ClientEvent::LateReply { seq: 0, .. }
-        ]
-    ));
-
-    drop(handle.stop());
-    runtime.block_on(task);
-    server.finish();
-}
-
-#[test]
-fn partial_timeout_batch_blocks_draining_receive_until_loss_commits() {
-    let reply_gate = Arc::new(PacketGate::default());
-    let server = start_server_with_gates(
-        ServerBehavior::Echo,
-        None,
-        None,
-        None,
-        Some(Arc::clone(&reply_gate)),
-    );
-    let mut managed = config(ManagedPacing::Staggered);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    let (mut task, handle) =
-        ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap();
-    let observations = Arc::new(Mutex::new(Vec::new()));
-    task.event_observations = Some(Arc::clone(&observations));
-    let runtime = runtime();
-    open_target_for_timeout_test(&runtime, &mut task, 0);
-    let sent = match &mut task.targets[0].state {
-        TargetState::Active { client } => runtime.block_on(client.send_probe()).unwrap(),
-        _ => panic!("timeout test target did not open"),
-    };
-    assert!(matches!(
-        sent.as_slice(),
-        [ClientEvent::EchoSent { seq: 0, .. }]
-    ));
-    wait_flag(&server.probe_seen);
-
-    let now = Instant::now();
-    let _ = remove_pending_for_timeout_test(&mut task, 0, 0);
-    for seq in 1..=u32::try_from(TIMEOUT_WORK_BUDGET).unwrap() {
-        replace_pending_for_timeout_test(
-            &mut task,
-            0,
-            timeout_probe(
-                seq,
-                now - Duration::from_secs(2),
-                now - Duration::from_secs(1),
-            ),
-        );
-    }
-    replace_pending_for_timeout_test(
-        &mut task,
-        0,
-        timeout_probe(0, now - Duration::from_secs(1), now),
-    );
-    let TargetState::Active { client } =
-        mem::replace(&mut task.targets[0].state, TargetState::Terminal)
-    else {
-        panic!("timeout test target did not remain active");
-    };
-    assert!(task.begin_drain(0, client, ManagedTargetEndReason::Stopped, now));
-    reply_gate.release();
-    wait_flag(&server.reply_sent);
-
-    let mut task = Box::pin(task);
-    poll_task_once(&runtime, &mut task);
-    assert_eq!(
-        timeout_losses_for(&observations, "one"),
-        TIMEOUT_WORK_BUDGET
-    );
-    assert!(matches!(
-        &task.targets[0].state,
-        TargetState::Draining {
-            primary_end: ManagedTargetEndReason::Stopped,
-            cleanup_failure: None,
-            ..
-        }
-    ));
-    assert!(!client_events(&observations).iter().any(|event| {
-        matches!(
-            event,
-            ClientEvent::EchoLoss { seq: 0, .. }
-                | ClientEvent::LateReply { seq: 0, .. }
-                | ClientEvent::EchoReply { seq: 0, .. }
-        )
-    }));
-
-    drive_task_until(&runtime, &mut task, |_| {
-        client_events(&observations)
-            .iter()
-            .any(|event| matches!(event, ClientEvent::LateReply { seq: 0, .. }))
-    });
-    let classified = client_events(&observations)
-        .into_iter()
-        .filter(|event| {
-            matches!(
-                event,
-                ClientEvent::EchoLoss { seq: 0, .. }
-                    | ClientEvent::LateReply { seq: 0, .. }
-                    | ClientEvent::EchoReply { seq: 0, .. }
-            )
-        })
-        .collect::<Vec<_>>();
-    assert!(matches!(
-        classified.as_slice(),
-        [
-            ClientEvent::EchoLoss { seq: 0, .. },
-            ClientEvent::LateReply { seq: 0, .. }
-        ]
-    ));
-
-    drop(handle.stop());
-    let outcome = runtime.block_on(task);
-    let target = &outcome.recent_target_outcomes[0];
-    assert!(matches!(target.end_reason, ManagedTargetEndReason::Stopped));
-    assert!(target.cleanup_failure.is_none());
-    server.finish();
-}
-
-#[test]
-fn overdue_queued_reply_is_lost_then_late() {
-    let reply_gate = Arc::new(PacketGate::default());
-    let server = start_server_with_gates(
-        ServerBehavior::Echo,
-        None,
-        None,
-        None,
-        Some(Arc::clone(&reply_gate)),
-    );
-    let mut managed = config(ManagedPacing::Staggered);
-    managed.completion = ManagedCompletionPolicy::ExplicitStop;
-    managed.client.duration = None;
-    managed.client.interval = Duration::from_secs(1);
-    managed.client.probe_timeout = Duration::from_millis(30);
-    let (mut task, handle) =
-        ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap();
-    let observations = Arc::new(Mutex::new(Vec::new()));
-    task.event_observations = Some(Arc::clone(&observations));
-    let runtime = runtime();
-    let mut task = Box::pin(task);
-    drive_task_until(&runtime, &mut task, |task| {
-        task.targets[0].counters.packets_sent == 1
-    });
-    let timeout = match &task.targets[0].state {
-        TargetState::Active { client } => client.next_probe_timeout_deadline().unwrap(),
-        _ => panic!("probe sender did not remain active"),
-    };
-    thread::sleep(timeout.saturating_duration_since(Instant::now()));
-    while Instant::now() <= timeout {
-        thread::yield_now();
-    }
-    reply_gate.release();
-    wait_flag(&server.reply_sent);
-    runtime.block_on(poll_fn(|cx| {
-        let ready = match &task.as_ref().get_ref().targets[0].state {
-            TargetState::Active { client } => client.poll_recv_ready_for_test(cx),
-            _ => panic!("probe sender did not remain active"),
-        };
-        ready.map(|result| result.unwrap())
-    }));
-    runtime.block_on(poll_fn(|cx| {
-        assert!(task.as_mut().poll(cx).is_pending());
-        Poll::Ready(())
-    }));
-    let classified = client_events(&observations)
-        .into_iter()
-        .filter(|event| {
-            matches!(
-                event,
-                ClientEvent::EchoLoss { .. }
-                    | ClientEvent::LateReply { .. }
-                    | ClientEvent::EchoReply { .. }
-            )
-        })
-        .collect::<Vec<_>>();
-    assert!(matches!(
-        classified.as_slice(),
-        [ClientEvent::EchoLoss { .. }, ClientEvent::LateReply { .. }]
-    ));
-    drop(handle.stop());
-    let outcome = runtime.block_on(task);
-    let target = &outcome.recent_target_outcomes[0];
-    assert_eq!(target.replies_received, 0);
-    assert_eq!(target.late, 1);
-    server.finish();
-}
-
 #[test]
 fn retained_probe_deadline_shortens_after_late_reply() {
     let reply_gate = Arc::new(PacketGate::default());