fix(managed): skip missed probe slots

Advance delayed managed schedules to the latest due cadence slot and then directly to the first future deadline. Apply the same absolute-cadence behavior to the single runner and group burst/staggered pacing, while leaving the manual Client send API caller-paced.

Bound receive waits by the next send deadline so 10 ms intervals are not hidden behind the managed socket timeout. Add deterministic delayed-slot, finite-end, burst, staggered, and receive-bound regressions.

crates/irtt-client/src/client.rs

@@ -163,6 +163,29 @@ impl Client {
     /// Returns an empty event list when the socket read would block or times
     /// out. Malformed or unrelated datagrams are reported as warning events.
     pub fn recv_once(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
+        self.socket
+            .set_read_timeout(self.runtime.config().socket_config.recv_timeout)?;
+        self.recv_once_inner()
+    }
+
+    pub(crate) fn recv_once_until(
+        &mut self,
+        deadline: Option<Instant>,
+        max_wait: Duration,
+    ) -> Result<Vec<ClientEvent>, ClientError> {
+        let Some(timeout) = bounded_receive_timeout(
+            deadline,
+            self.runtime.config().socket_config.recv_timeout,
+            max_wait,
+            Instant::now(),
+        ) else {
+            return Ok(Vec::new());
+        };
+        self.socket.set_read_timeout(Some(timeout))?;
+        self.recv_once_inner()
+    }
+
+    fn recv_once_inner(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
         let datagram = match recv_datagram(&self.socket, &mut self.recv_buffer) {
             Ok(datagram) => datagram,
             Err(err)
@@ -186,9 +209,11 @@ impl Client {
     /// Receive and classify datagrams until a receive produces no events or the
     /// receive budget is exhausted.
     pub fn recv_available(&mut self, budget: RecvBudget) -> Result<Vec<ClientEvent>, ClientError> {
+        self.socket
+            .set_read_timeout(self.runtime.config().socket_config.recv_timeout)?;
         let mut all_events = Vec::new();
         for _ in 0..budget.max_packets {
-            let events = self.recv_once()?;
+            let events = self.recv_once_inner()?;
             if events.is_empty() {
                 break;
             }
@@ -243,6 +268,13 @@ impl Client {
         self.runtime.packets_sent()
     }
 
+    pub(crate) fn send_managed_probe(
+        &mut self,
+        scheduled_at: Instant,
+    ) -> Result<Vec<ClientEvent>, ClientError> {
+        self.send_managed_probe_inner(scheduled_at, None)
+    }
+
     fn send_probe_inner(
         &mut self,
         override_ts: Option<ClientTimestamp>,
@@ -260,6 +292,26 @@ impl Client {
             })
         })
     }
+
+    fn send_managed_probe_inner(
+        &mut self,
+        scheduled_at: Instant,
+        override_ts: Option<ClientTimestamp>,
+    ) -> Result<Vec<ClientEvent>, ClientError> {
+        let socket = &self.socket;
+        self.runtime
+            .send_probe_for_deadline(override_ts, scheduled_at, |packet| {
+                let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now);
+                let send_call_start = Instant::now();
+                let bytes = socket.send(packet)?;
+                let send_call = send_call_start.elapsed();
+                Ok(SendProbeResult {
+                    sent_at,
+                    bytes,
+                    send_call,
+                })
+            })
+    }
 }
 
 #[cfg(test)]
@@ -267,6 +319,34 @@ impl Client {
     fn send_probe_at(&mut self, ts: ClientTimestamp) -> Result<Vec<ClientEvent>, ClientError> {
         self.send_probe_inner(Some(ts))
     }
+
+    fn send_managed_probe_at(
+        &mut self,
+        scheduled_at: Instant,
+        ts: ClientTimestamp,
+    ) -> Result<Vec<ClientEvent>, ClientError> {
+        self.send_managed_probe_inner(scheduled_at, Some(ts))
+    }
+}
+
+fn bounded_receive_timeout(
+    deadline: Option<Instant>,
+    configured_timeout: Option<Duration>,
+    max_wait: Duration,
+    now: Instant,
+) -> Option<Duration> {
+    if max_wait.is_zero() {
+        return None;
+    }
+    let mut timeout = configured_timeout.unwrap_or(max_wait).min(max_wait);
+    if let Some(deadline) = deadline {
+        let remaining = deadline.checked_duration_since(now)?;
+        if remaining.is_zero() {
+            return None;
+        }
+        timeout = timeout.min(remaining);
+    }
+    (!timeout.is_zero()).then_some(timeout)
 }
 
 #[cfg(test)]

crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs

@@ -101,6 +101,129 @@ fn send_probe_respects_finite_duration_exclusive_end() {
     server.join();
 }
 
+#[test]
+fn managed_probe_skips_missed_schedule_slots() {
+    let interval = Duration::from_millis(10);
+    let params = Params {
+        protocol_version: 1,
+        duration_ns: 0,
+        interval_ns: 10_000_000,
+        received_stats: ReceivedStats::Both,
+        stamp_at: StampAt::Both,
+        clock: Clock::Both,
+        ..Params::default()
+    };
+    let server = echo_server(params);
+    let config = ClientConfig {
+        duration: None,
+        interval,
+        ..default_test_config(server.addr)
+    };
+    let mut client = Client::connect(config).unwrap();
+    assert_open_started(client.open().unwrap());
+
+    let first_deadline = client.next_send_deadline().unwrap();
+    let delayed_send = ClientTimestamp {
+        mono: first_deadline + Duration::from_millis(45),
+        wall: SystemTime::now(),
+    };
+    let events = client
+        .send_managed_probe_at(first_deadline, delayed_send)
+        .unwrap();
+
+    assert_eq!(events.len(), 1);
+    assert!(matches!(
+        events[0],
+        ClientEvent::EchoSent {
+            seq: 0,
+            scheduled_at,
+            sent_at,
+            ..
+        } if scheduled_at == first_deadline + Duration::from_millis(40)
+            && sent_at == delayed_send
+    ));
+    assert_eq!(
+        client.next_send_deadline(),
+        Some(first_deadline + Duration::from_millis(50))
+    );
+
+    client.close().unwrap();
+    server.join();
+}
+
+#[test]
+fn managed_probe_skip_preserves_finite_run_end() {
+    let interval = Duration::from_millis(10);
+    let duration = Duration::from_millis(50);
+    let params = Params {
+        protocol_version: 1,
+        duration_ns: 50_000_000,
+        interval_ns: 10_000_000,
+        received_stats: ReceivedStats::Both,
+        stamp_at: StampAt::Both,
+        clock: Clock::Both,
+        ..Params::default()
+    };
+    let server = echo_server(params);
+    let config = ClientConfig {
+        duration: Some(duration),
+        interval,
+        ..default_test_config(server.addr)
+    };
+    let mut client = Client::connect(config).unwrap();
+    assert_open_started(client.open().unwrap());
+
+    let first_deadline = client.next_send_deadline().unwrap();
+    let delayed_send = ClientTimestamp {
+        mono: first_deadline + Duration::from_millis(45),
+        wall: SystemTime::now(),
+    };
+    let first = client
+        .send_managed_probe_at(first_deadline, delayed_send)
+        .unwrap();
+    let second = client
+        .send_managed_probe_at(
+            first_deadline + duration,
+            ClientTimestamp {
+                mono: first_deadline + Duration::from_millis(46),
+                wall: SystemTime::now(),
+            },
+        )
+        .unwrap();
+
+    assert_eq!(first.len(), 1);
+    assert!(second.is_empty());
+    assert!(client.next_send_deadline().is_none());
+
+    client.close().unwrap();
+    server.join();
+}
+
+#[test]
+fn receive_wait_is_bounded_by_next_ten_millisecond_deadline() {
+    let now = Instant::now();
+    let deadline = now + Duration::from_millis(10);
+
+    assert_eq!(
+        bounded_receive_timeout(
+            Some(deadline),
+            Some(Duration::from_millis(20)),
+            Duration::from_millis(20),
+            now,
+        ),
+        Some(Duration::from_millis(10))
+    );
+    assert_eq!(
+        bounded_receive_timeout(
+            Some(now),
+            Some(Duration::from_millis(20)),
+            Duration::from_millis(20),
+            now,
+        ),
+        None
+    );
+}
+
 #[test]
 fn connect_rejects_invalid_probe_limits() {
     for config in [

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

@@ -13,7 +13,7 @@ use crate::{
     error::ClientError,
     event::{ClientEvent, OpenOutcome},
     receive::recv_datagram_from,
-    runtime::{params_from_config, SendProbeResult, SessionRuntime},
+    runtime::{advance_cadence, params_from_config, SendProbeResult, SessionRuntime},
     socket::{bind_unconnected_udp_socket, validate_open_timeouts},
     socket_options::apply_dscp_to_socket,
     timing::ClientTimestamp,
@@ -149,6 +149,10 @@ pub struct TargetEvent {
 ///
 /// Per-session events are followed by exactly one [`TargetFinished`](Self::TargetFinished)
 /// event when that target incarnation reaches a terminal state.
+// ClientEvent intentionally owns its structured measurement payload. EventHub
+// queues are explicitly bounded, so preserving the direct event API does not
+// create unbounded retention.
+#[allow(clippy::large_enum_variant)]
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum ManagedGroupEvent {
     /// A protocol lifecycle, measurement, or warning event from one target.
@@ -589,6 +593,8 @@ struct TargetState {
     final_reason: Option<ManagedTargetEndReason>,
 }
 
+type ScheduledTarget = (Arc<Mutex<TargetState>>, Instant);
+
 #[derive(Debug)]
 enum TargetStatus {
     Opening {
@@ -751,17 +757,17 @@ fn run_group_scheduler(
 
         let active = active_targets(&shared);
         pacing.reconcile(&active, now);
-        if pacing.send_due(&active, now, config.client.interval) {
+        if pacing.send_due(&active, now) {
             match config.pacing {
                 ManagedGroupPacing::Staggered => {
                     if let Some((target, scheduled_at)) =
-                        pacing.next_staggered(&active, config.client.interval)
+                        pacing.next_staggered(&active, config.client.interval, now)?
                     {
                         send_echo_to_target(&socket, &shared, target, scheduled_at);
                     }
                 }
                 ManagedGroupPacing::Burst => {
-                    if let Some(scheduled_at) = pacing.next_burst(config.client.interval) {
+                    if let Some(scheduled_at) = pacing.next_burst(config.client.interval, now)? {
                         for target in active {
                             send_echo_to_target(&socket, &shared, target, scheduled_at);
                         }
@@ -1227,7 +1233,7 @@ fn send_echo_to_target(
     let remote = target.remote;
     let result = target
         .runtime
-        .send_probe_for_deadline(scheduled_at, |packet| {
+        .send_probe_for_deadline(None, scheduled_at, |packet| {
             let sent_at = ClientTimestamp::now();
             let send_call_start = Instant::now();
             let bytes = socket.send_to(packet, remote)?;
@@ -1449,12 +1455,7 @@ impl PacingRuntime {
         }
     }
 
-    fn send_due(
+    fn send_due(&self, active: &[Arc<Mutex<TargetState>>], now: Instant) -> bool {
-        &self,
-        active: &[Arc<Mutex<TargetState>>],
-        now: Instant,
-        _interval: Duration,
-    ) -> bool {
         if active.is_empty() {
             return false;
         }
@@ -1470,31 +1471,49 @@ impl PacingRuntime {
         &mut self,
         active: &[Arc<Mutex<TargetState>>],
         interval: Duration,
-    ) -> Option<(Arc<Mutex<TargetState>>, Instant)> {
+        now: Instant,
-        let scheduled_at = self.next_slot_at?;
+    ) -> Result<Option<ScheduledTarget>, ClientError> {
+        let Some(next_slot_at) = self.next_slot_at else {
+            return Ok(None);
+        };
         if active.is_empty() {
             self.next_slot_at = None;
-            return None;
+            return Ok(None);
         }
+        let (scheduled_at, next_slot_at, first_index) =
+            advance_staggered_slot(next_slot_at, interval, active.len(), self.slot_index, now)?;
+        self.next_slot_at = Some(next_slot_at);
+
         for offset in 0..active.len() {
-            let index = (self.slot_index + offset) % active.len();
+            let index = (first_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 Ok(Some((target, scheduled_at)));
-            return Some((target, scheduled_at));
         }
 
-        self.next_slot_at = active.iter().filter_map(target_next_send_deadline).min();
+        self.slot_index = (first_index + 1) % active.len();
-        None
+        if let Some(target_deadline) = active.iter().filter_map(target_next_send_deadline).min() {
+            self.next_slot_at = Some(self.next_slot_at.map_or(target_deadline, |group_deadline| {
+                group_deadline.min(target_deadline)
+            }));
+        }
+        Ok(None)
     }
 
-    fn next_burst(&mut self, interval: Duration) -> Option<Instant> {
+    fn next_burst(
-        let scheduled_at = self.next_burst_at?;
+        &mut self,
-        self.next_burst_at = Some(scheduled_at + interval);
+        interval: Duration,
-        Some(scheduled_at)
+        now: Instant,
+    ) -> Result<Option<Instant>, ClientError> {
+        let Some(next_burst_at) = self.next_burst_at else {
+            return Ok(None);
+        };
+        let (scheduled_at, next_burst_at, _) = advance_cadence(next_burst_at, interval, now)?;
+        self.next_burst_at = Some(next_burst_at);
+        Ok(Some(scheduled_at))
     }
 
     fn next_wakeup(&self) -> Option<Instant> {
@@ -1505,6 +1524,23 @@ impl PacingRuntime {
     }
 }
 
+fn advance_staggered_slot(
+    next_slot_at: Instant,
+    interval: Duration,
+    active_len: usize,
+    slot_index: usize,
+    now: Instant,
+) -> Result<(Instant, Instant, usize), ClientError> {
+    debug_assert!(active_len > 0);
+    let spacing = divide_duration(interval, active_len);
+    let (scheduled_at, following_slot_at, skipped_slots) =
+        advance_cadence(next_slot_at, spacing, now)?;
+    let skipped_mod = usize::try_from(skipped_slots % active_len as u128)
+        .expect("staggered slot remainder fits usize");
+    let first_index = (slot_index + skipped_mod) % active_len;
+    Ok((scheduled_at, following_slot_at, first_index))
+}
+
 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)
 }
@@ -2262,6 +2298,39 @@ mod tests {
         );
     }
 
+    #[test]
+    fn delayed_burst_pacing_emits_one_current_slot_and_advances_to_future() {
+        let interval = Duration::from_millis(10);
+        let first_slot = Instant::now();
+        let delayed_now = first_slot + Duration::from_millis(45);
+        let mut pacing = PacingRuntime::new(ManagedGroupPacing::Burst);
+        pacing.next_burst_at = Some(first_slot);
+
+        let scheduled_at = pacing.next_burst(interval, delayed_now).unwrap().unwrap();
+
+        assert_eq!(scheduled_at, first_slot + Duration::from_millis(40));
+        assert_eq!(
+            pacing.next_wakeup(),
+            Some(first_slot + Duration::from_millis(50))
+        );
+        assert!(pacing.next_wakeup().unwrap() > delayed_now);
+    }
+
+    #[test]
+    fn delayed_staggered_pacing_skips_historical_slots() {
+        let first_slot = Instant::now();
+        let delayed_now = first_slot + Duration::from_millis(35);
+
+        let (scheduled_at, next_slot_at, target_index) =
+            advance_staggered_slot(first_slot, Duration::from_millis(30), 3, 0, delayed_now)
+                .unwrap();
+
+        assert_eq!(scheduled_at, first_slot + Duration::from_millis(30));
+        assert_eq!(next_slot_at, first_slot + Duration::from_millis(40));
+        assert_eq!(target_index, 0);
+        assert!(next_slot_at > delayed_now);
+    }
+
     #[test]
     fn staggered_pacing_does_not_burst_active_targets() {
         let duration = Duration::from_millis(170);

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

@@ -19,7 +19,6 @@ const MANAGED_RECV_TIMEOUT: Duration = Duration::from_millis(20);
 const MANAGED_RECV_BUDGET: RecvBudget = RecvBudget { max_packets: 64 };
 const MANAGED_FINAL_DRAIN: Duration = Duration::from_millis(100);
 const IDLE_SLEEP: Duration = Duration::from_millis(1);
-const MAX_SLEEP: Duration = Duration::from_millis(20);
 
 /// Entry point for running a client session on a worker thread.
 ///
@@ -232,12 +231,9 @@ fn run_client(
             break;
         }
 
-        let now = Instant::now();
+        let send_deadline = client.next_send_deadline();
-        if client
+        if let Some(deadline) = send_deadline.filter(|deadline| *deadline <= Instant::now()) {
-            .next_send_deadline()
+            let events = client.send_managed_probe(deadline)?;
-            .is_some_and(|deadline| deadline <= now)
-        {
-            let events = client.send_probe()?;
             publish_events(&hub, &mut counters, events);
             continue;
         }
@@ -245,7 +241,7 @@ fn run_client(
         publish_events(
             &hub,
             &mut counters,
-            client.recv_available(MANAGED_RECV_BUDGET)?,
+            client.recv_once_until(send_deadline, MANAGED_RECV_TIMEOUT)?,
         );
         if client.is_peer_closed() {
             break;
@@ -255,8 +251,6 @@ fn run_client(
         if client.is_run_complete() {
             break;
         }
-
-        sleep_until_next_wakeup(client.next_send_deadline());
     }
 
     if !cancelled {
@@ -340,16 +334,6 @@ fn publish_events(hub: &EventHub, counters: &mut OutcomeCounters, events: Vec<Cl
     }
 }
 
-fn sleep_until_next_wakeup(deadline: Option<Instant>) {
-    let sleep_for = deadline
-        .and_then(|deadline| deadline.checked_duration_since(Instant::now()))
-        .map(|duration| duration.min(MAX_SLEEP))
-        .unwrap_or(IDLE_SLEEP);
-    if sleep_for > Duration::ZERO {
-        thread::sleep(sleep_for);
-    }
-}
-
 #[derive(Debug, Default)]
 struct OutcomeCounters {
     replies_received: u64,

crates/irtt-client/src/runtime.rs

@@ -152,24 +152,26 @@ impl SessionRuntime {
     where
         F: FnOnce(&[u8]) -> Result<SendProbeResult, ClientError>,
     {
-        self.send_probe_inner(override_ts, None, send)
+        self.send_probe_inner(override_ts, None, false, send)
     }
 
     pub(crate) fn send_probe_for_deadline<F>(
         &mut self,
+        override_ts: Option<ClientTimestamp>,
         scheduled_at: Instant,
         send: F,
     ) -> Result<Vec<ClientEvent>, ClientError>
     where
         F: FnOnce(&[u8]) -> Result<SendProbeResult, ClientError>,
     {
-        self.send_probe_inner(None, Some(scheduled_at), send)
+        self.send_probe_inner(override_ts, Some(scheduled_at), true, send)
     }
 
     fn send_probe_inner<F>(
         &mut self,
         override_ts: Option<ClientTimestamp>,
         scheduled_at_override: Option<Instant>,
+        skip_missed_slots: bool,
         send: F,
     ) -> Result<Vec<ClientEvent>, ClientError>
     where
@@ -218,7 +220,6 @@ impl SessionRuntime {
         };
         let packet = encode_echo_request(&request, self.config.hmac_key.as_deref())?;
         let send_result = send(&packet)?;
-        let timer_error = instant_abs_diff(send_result.sent_at.mono, scheduled_at);
 
         let session = self
             .session
@@ -251,17 +252,23 @@ impl SessionRuntime {
             .expect("negotiated must exist when Open");
         let interval_ns = u64::try_from(negotiated.params.interval_ns)
             .expect("validated positive negotiated interval");
+        let interval = Duration::from_nanos(interval_ns);
+        let (scheduled_at, next_send_at) = if skip_missed_slots {
+            let (scheduled_at, next_send_at, _) =
+                advance_cadence(scheduled_at, interval, send_result.sent_at.mono)?;
+            (scheduled_at, next_send_at)
+        } else {
+            (
+                scheduled_at,
+                next_probe_deadline(session.start_mono, interval_ns, session.packets_sent)?,
+            )
+        };
+        let timer_error = instant_abs_diff(send_result.sent_at.mono, scheduled_at);
         let session = self
             .session
             .as_mut()
             .expect("session must exist when phase is Open");
-        session.next_send_at = if let Some(scheduled_at) = scheduled_at_override {
+        session.next_send_at = next_send_at;
-            scheduled_at
-                .checked_add(Duration::from_nanos(interval_ns))
-                .ok_or(ClientError::DurationOverflow)?
-        } else {
-            next_probe_deadline(session.start_mono, interval_ns, session.packets_sent)?
-        };
 
         if let Some(end) = session.end_mono {
             if session.next_send_at >= end {
@@ -684,6 +691,54 @@ fn next_probe_deadline(
         .ok_or(ClientError::DurationOverflow)
 }
 
+pub(crate) fn advance_cadence(
+    deadline: Instant,
+    interval: Duration,
+    now: Instant,
+) -> Result<(Instant, Instant, u128), 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, elapsed_slots))
+}
+
+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 sequence_is_after(candidate: u32, current: u32) -> bool {
     candidate != current && candidate.wrapping_sub(current) < (1 << 31)
 }