From 543f0e57c80d292ded0f04cb3e2a12d1fa25af26 Mon Sep 17 00:00:00 2001
From: Nils Andreas Svee <me@lochnair.net>
Date: Tue, 23 Jun 2026 21:12:01 +0700
Subject: [PATCH] fix(client): prevent premature task rescheduling in staggered
 pacing

Ensure targets are only scheduled when they are actually due to prevent unchanged targets from resending too early when the active set changes.
---
 crates/irtt-client/src/managed/group.rs | 78 +++++++++++++++++++++++--
 1 file changed, 74 insertions(+), 4 deletions(-)

diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs
index e323a74..5dfb7fd 100644
--- a/crates/irtt-client/src/managed/group.rs
+++ b/crates/irtt-client/src/managed/group.rs
@@ -1273,10 +1273,19 @@ impl PacingRuntime {
             self.next_slot_at = None;
             return None;
         }
-        let target = active[self.slot_index % active.len()].clone();
-        self.slot_index = (self.slot_index + 1) % active.len();
-        self.next_slot_at = Some(scheduled_at + divide_duration(interval, active.len()));
-        Some((target, scheduled_at))
+        for offset in 0..active.len() {
+            let index = (self.slot_index + offset) % active.len();
+            if !target_is_due_for_slot(&active[index], scheduled_at) {
+                continue;
+            }
+            let target = active[index].clone();
+            self.slot_index = (index + 1) % active.len();
+            self.next_slot_at = Some(scheduled_at + divide_duration(interval, active.len()));
+            return Some((target, scheduled_at));
+        }
+
+        self.next_slot_at = active.iter().filter_map(target_next_send_deadline).min();
+        None
     }
 
     fn next_burst(&mut self, interval: Duration) -> Option<Instant> {
@@ -1293,6 +1302,19 @@ impl PacingRuntime {
     }
 }
 
+fn target_is_due_for_slot(target: &Arc<Mutex<TargetState>>, scheduled_at: Instant) -> bool {
+    target_next_send_deadline(target).is_some_and(|deadline| deadline <= scheduled_at)
+}
+
+fn target_next_send_deadline(target: &Arc<Mutex<TargetState>>) -> Option<Instant> {
+    let target = target.lock().expect("target mutex poisoned");
+    if matches!(target.status, TargetStatus::Active) {
+        target.runtime.next_send_deadline()
+    } else {
+        None
+    }
+}
+
 fn wait_for_next_scheduler_wakeup(
     control_rx: &mpsc::Receiver<ControlMessage>,
     shared: &GroupShared,
@@ -1815,6 +1837,54 @@ mod tests {
         }));
     }
 
+    #[test]
+    fn staggered_active_set_change_does_not_resend_unchanged_target_early() {
+        let duration = Duration::from_millis(360);
+        let interval = Duration::from_millis(100);
+        let params = test_params(Some(duration), interval);
+        let a = start_echo_server(params.clone(), Duration::ZERO);
+        let b = start_echo_server(params, Duration::from_millis(150));
+
+        let (session, sub) = ManagedClientGroup::start_with_subscription(
+            group_config(Some(duration), interval, ManagedGroupPacing::Staggered),
+            vec![target("a", a.addr), target("b", b.addr)],
+            SubscriberConfig {
+                capacity: 512,
+                overflow: SubscriberOverflow::DropNewest,
+            },
+        )
+        .unwrap();
+
+        let _ = session.join().unwrap();
+        let events = drain_after_join(&sub);
+        a.join();
+        b.join();
+
+        let a_sends: Vec<Instant> = events
+            .iter()
+            .filter_map(|event| match (&event.target, &event.event) {
+                (id, ClientEvent::EchoSent { sent_at, .. }) if id.as_str() == "a" => {
+                    Some(sent_at.mono)
+                }
+                _ => None,
+            })
+            .collect();
+        assert!(
+            a_sends.len() >= 3,
+            "expected enough target a sends to verify spacing, got {a_sends:?}"
+        );
+
+        let min_delta = a_sends
+            .windows(2)
+            .map(|window| window[1].duration_since(window[0]))
+            .min()
+            .unwrap();
+        assert!(
+            min_delta >= Duration::from_millis(75),
+            "target a was resent too early after active set changed: {min_delta:?}"
+        );
+    }
+
     #[test]
     fn update_targets_removes_and_adds_without_restarting_unchanged_target() {
         let interval = Duration::from_millis(30);