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

@@ -591,6 +591,8 @@ pub struct ManagedClientTask {
     stagger_observations: Option<StaggerObservations>,
     #[cfg(test)]
     drain_test_hook: DrainTestHook,
+    #[cfg(test)]
+    timeout_inspections: usize,
 }
 
 #[cfg(test)]
@@ -1730,6 +1732,10 @@ impl ManagedClientTask {
         for _ in 0..TIMEOUT_WORK_BUDGET {
             let index = self.timeout_cursor;
             self.timeout_cursor = (self.timeout_cursor + 1) % target_len;
+            #[cfg(test)]
+            {
+                self.timeout_inspections += 1;
+            }
             if self.target_has_due_timeout(index, now) {
                 let step = self.poll_target_timeout(index, now);
                 more_due |= step.more_due;
@@ -2263,6 +2269,8 @@ fn build_task(
         stagger_observations: None,
         #[cfg(test)]
         drain_test_hook: DrainTestHook::default(),
+        #[cfg(test)]
+        timeout_inspections: 0,
     };
     let handle = ManagedClientHandle {
         stop,

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

@@ -1738,6 +1738,77 @@ fn stagger_gate_tracks_active_membership() {
     second.finish();
 }
 
+#[test]
+fn timeout_discovery_inspects_at_most_one_budget_of_targets() {
+    let target_count = TIMEOUT_WORK_BUDGET + 1;
+    let mut managed = config(ManagedPacing::Burst);
+    managed.max_live_target_generations = target_count;
+    let targets = (0..target_count)
+        .map(|index| ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9"))
+        .collect();
+    let (mut task, _) = ManagedClient::task(managed, targets).unwrap();
+
+    task.timeout_inspections = 0;
+    assert!(!task.poll_timeout_pass(Instant::now()));
+
+    assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET);
+    assert_eq!(task.timeout_cursor, TIMEOUT_WORK_BUDGET);
+}
+
+#[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!(task.timeout_cursor, late_index);
+    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);
@@ -1864,6 +1935,101 @@ fn timeout_work_round_robins_simultaneous_target_backlogs() {
     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());