refactor(client): separate probe scheduling from session state

crates/irtt-client/src/client.rs

@@ -18,6 +18,10 @@ use crate::{
     timing::ClientTimestamp,
 };
 
+pub(crate) mod schedule;
+
+use schedule::ProbeSchedule;
+
 #[cfg(test)]
 use crate::{
     session::machine::{
@@ -37,6 +41,7 @@ use crate::{
 #[derive(Debug)]
 pub struct Client {
     runtime: SessionMachine,
+    schedule: Option<ProbeSchedule>,
     socket: UdpSocket,
     remote: SocketAddr,
     recv_buffer: Vec<u8>,
@@ -56,6 +61,7 @@ impl Client {
 
         Ok(Self {
             runtime,
+            schedule: None,
             socket,
             remote,
             recv_buffer: vec![0_u8; recv_buffer_size(false, None)?],
@@ -79,26 +85,30 @@ impl Client {
                 match self.socket.recv(&mut buf) {
                     Ok(size) => {
                         let reply = self.runtime.decode_open_reply(&buf[..size])?;
+                        let opened_at = ClientTimestamp::now();
                         let remote = self.remote;
                         let has_hmac = self.runtime.has_hmac();
                         let socket = &self.socket;
                         let recv_buffer = &mut self.recv_buffer;
+                        let mut schedule = None;
 
-                        return self.runtime.accept_open_reply(
+                        let outcome =
-                            reply,
+                            self.runtime
-                            ClientTimestamp::now(),
+                                .accept_open_reply(reply, opened_at, |negotiated| {
-                            |negotiated| {
+                                    let size = recv_buffer_size(has_hmac, Some(negotiated))?;
-                                let size = recv_buffer_size(has_hmac, Some(negotiated))?;
+                                    let negotiated_dscp = u8::try_from(negotiated.params.dscp)
-                                let negotiated_dscp = u8::try_from(negotiated.params.dscp)
+                                        .map_err(|_| ClientError::InvalidConfig {
-                                    .map_err(|_| ClientError::InvalidConfig {
+                                            reason: "negotiated dscp must be in range 0..=63"
-                                        reason: "negotiated dscp must be in range 0..=63"
+                                                .to_owned(),
-                                            .to_owned(),
+                                        })?;
-                                    })?;
+                                    apply_dscp_to_socket(socket, remote, negotiated_dscp)?;
-                                apply_dscp_to_socket(socket, remote, negotiated_dscp)?;
+                                    recv_buffer.resize(size, 0);
-                                recv_buffer.resize(size, 0);
+                                    schedule =
-                                Ok(())
+                                        Some(ProbeSchedule::new(opened_at.mono, negotiated)?);
-                            },
+                                    Ok(())
-                        );
+                                })?;
+                        self.schedule = schedule;
+                        return Ok(outcome);
                     }
                     Err(err)
                         if matches!(
@@ -131,17 +141,22 @@ impl Client {
     pub fn close(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
         let socket = &self.socket;
         let remote = self.remote;
-        self.runtime.close_with(|packet| {
+        let events = self.runtime.close_with(|packet| {
             clear_dscp_on_socket(socket, remote)?;
             socket.send(packet)?;
             Ok(())
-        })
+        })?;
+        self.schedule = None;
+        Ok(events)
     }
 
     /// Return the monotonic deadline for the next probe send, if another probe
     /// is scheduled.
     pub fn next_send_deadline(&self) -> Option<Instant> {
-        self.runtime.next_send_deadline()
+        if !self.runtime.is_open() {
+            return None;
+        }
+        self.schedule.as_ref()?.next_send_deadline()
     }
 
     /// Return the local timeout used to classify pending probes as lost.
@@ -199,11 +214,15 @@ impl Client {
             Err(err) => return Err(ClientError::Socket(err)),
         };
 
-        self.runtime.process_received_echo_packet(
+        let events = self.runtime.process_received_echo_packet(
             &self.recv_buffer[..datagram.len],
             datagram.received_at,
             datagram.meta,
-        )
+        )?;
+        if self.runtime.is_terminal() {
+            self.schedule = None;
+        }
+        Ok(events)
     }
 
     /// Receive and classify datagrams until a receive produces no events or the
@@ -247,7 +266,11 @@ impl Client {
     /// pending probes have either replied or timed out. No-test and closed
     /// sessions are also considered complete.
     pub fn is_run_complete(&self) -> bool {
-        self.runtime.is_run_complete()
+        self.runtime.is_terminal()
+            || self
+                .schedule
+                .as_ref()
+                .is_some_and(|schedule| schedule.is_finished() && self.runtime.pending_is_empty())
     }
 
     /// Return whether the session was closed by a peer close-flagged reply.
@@ -280,17 +303,23 @@ impl Client {
         override_ts: Option<ClientTimestamp>,
     ) -> Result<Vec<ClientEvent>, ClientError> {
         let socket = &self.socket;
-        self.runtime.send_probe_with(override_ts, |packet| {
+        self.runtime.ensure_open()?;
-            let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now);
+        let schedule = self
-            let send_call_start = Instant::now();
+            .schedule
-            let bytes = socket.send(packet)?;
+            .as_mut()
-            let send_call = send_call_start.elapsed();
+            .expect("open sessions always have a probe schedule");
-            Ok(SendProbeResult {
+        self.runtime
-                sent_at,
+            .send_probe_with(schedule, override_ts, |packet| {
-                bytes,
+                let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now);
-                send_call,
+                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,
+                })
             })
-        })
     }
 
     fn send_managed_probe_inner(
@@ -299,8 +328,13 @@ impl Client {
         override_ts: Option<ClientTimestamp>,
     ) -> Result<Vec<ClientEvent>, ClientError> {
         let socket = &self.socket;
+        self.runtime.ensure_open()?;
+        let schedule = self
+            .schedule
+            .as_mut()
+            .expect("open sessions always have a probe schedule");
         self.runtime
-            .send_probe_for_deadline(override_ts, scheduled_at, |packet| {
+            .send_probe_for_deadline(schedule, 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)?;

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

@@ -0,0 +1,312 @@
+use std::time::{Duration, Instant};
+
+use crate::{error::ClientError, session::NegotiatedParams};
+
+#[derive(Debug)]
+pub(crate) struct ProbeSchedule {
+    start_at: Instant,
+    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>,
+    pub(crate) timer_error: Duration,
+}
+
+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 {
+            start_at,
+            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 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_caller_commit(
+        &self,
+        sent_at: Instant,
+        next_packets_sent: u64,
+    ) -> Result<ScheduleCommit, ClientError> {
+        let scheduled_at = self
+            .next_send_at
+            .expect("a caller schedule commit requires a permitted probe");
+        let interval_ns =
+            u64::try_from(self.interval.as_nanos()).map_err(|_| ClientError::DurationOverflow)?;
+        let next_send_at = next_probe_deadline(self.start_at, interval_ns, next_packets_sent)?;
+        Ok(self.finish_commit(scheduled_at, next_send_at, sent_at))
+    }
+
+    pub(crate) fn preflight_managed_commit(
+        &self,
+        scheduled_at: Instant,
+        sent_at: Instant,
+    ) -> Result<ScheduleCommit, ClientError> {
+        let (scheduled_at, next_send_at, _) =
+            advance_cadence(scheduled_at, self.interval, sent_at)?;
+        Ok(self.finish_commit(scheduled_at, next_send_at, sent_at))
+    }
+
+    pub(crate) fn commit(&mut self, commit: ScheduleCommit) {
+        self.next_send_at = commit.next_send_at;
+    }
+
+    pub(crate) fn is_finished(&self) -> bool {
+        self.next_send_at.is_none()
+    }
+
+    fn finish_commit(
+        &self,
+        scheduled_at: Instant,
+        next_send_at: Instant,
+        sent_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)
+            },
+            timer_error: instant_abs_diff(sent_at, scheduled_at),
+        }
+    }
+}
+
+fn next_probe_deadline(
+    start: Instant,
+    interval_ns: u64,
+    packets_sent: u64,
+) -> Result<Instant, ClientError> {
+    let offset_ns = interval_ns
+        .checked_mul(packets_sent)
+        .ok_or(ClientError::DurationOverflow)?;
+    start
+        .checked_add(Duration::from_nanos(offset_ns))
+        .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))
+}
+
+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(interval_ns: i64, duration_ns: i64) -> NegotiatedParams {
+        NegotiatedParams {
+            params: Params {
+                interval_ns,
+                duration_ns,
+                ..Params::default()
+            },
+            restrictions: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn initial_deadline_is_the_opening_time() {
+        let opened_at = Instant::now();
+
+        let schedule = ProbeSchedule::new(opened_at, &negotiated(10_000_000, 0)).unwrap();
+
+        assert_eq!(schedule.next_send_deadline(), Some(opened_at));
+        assert!(!schedule.is_finished());
+    }
+
+    #[test]
+    fn caller_preflight_keeps_absolute_cadence_without_mutation() {
+        let start = Instant::now();
+        let schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 0)).unwrap();
+        let sent_at = start + Duration::from_millis(4);
+
+        let commit = schedule.preflight_caller_commit(sent_at, 3).unwrap();
+
+        assert_eq!(commit.scheduled_at, start);
+        assert_eq!(commit.next_send_at, Some(start + Duration::from_millis(30)));
+        assert_eq!(commit.timer_error, Duration::from_millis(4));
+        assert_eq!(schedule.next_send_deadline(), Some(start));
+    }
+
+    #[test]
+    fn caller_commit_advances_exactly_one_cadence_slot() {
+        let start = Instant::now();
+        let mut schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 0)).unwrap();
+
+        let commit = schedule.preflight_caller_commit(start, 1).unwrap();
+        assert_eq!(schedule.next_send_deadline(), Some(start));
+
+        schedule.commit(commit);
+        assert_eq!(
+            schedule.next_send_deadline(),
+            Some(start + Duration::from_millis(10))
+        );
+    }
+
+    #[test]
+    fn finite_duration_uses_an_exclusive_end_boundary() {
+        let start = Instant::now();
+        let mut schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 20_000_000)).unwrap();
+
+        let first = schedule.preflight_caller_commit(start, 1).unwrap();
+        schedule.commit(first);
+        assert_eq!(
+            schedule.next_send_deadline(),
+            Some(start + Duration::from_millis(10))
+        );
+
+        let second = schedule
+            .preflight_caller_commit(start + Duration::from_millis(10), 2)
+            .unwrap();
+        schedule.commit(second);
+
+        assert!(schedule.is_finished());
+        assert_eq!(schedule.next_send_deadline(), None);
+        assert!(!schedule.permit_probe_at(start + Duration::from_millis(20)));
+    }
+
+    #[test]
+    fn permit_at_exclusive_end_finishes_schedule_without_a_send() {
+        let start = Instant::now();
+        let mut schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 20_000_000)).unwrap();
+
+        assert!(!schedule.permit_probe_at(start + Duration::from_millis(20)));
+        assert!(schedule.is_finished());
+        assert_eq!(schedule.next_send_deadline(), None);
+    }
+
+    #[test]
+    fn managed_preflight_skips_missed_slots_without_mutation() {
+        let start = Instant::now();
+        let schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 0)).unwrap();
+        let sent_at = start + Duration::from_millis(45);
+
+        let commit = schedule.preflight_managed_commit(start, sent_at).unwrap();
+
+        assert_eq!(commit.scheduled_at, start + Duration::from_millis(40));
+        assert_eq!(commit.next_send_at, Some(start + Duration::from_millis(50)));
+        assert_eq!(commit.timer_error, Duration::from_millis(5));
+        assert_eq!(schedule.next_send_deadline(), Some(start));
+    }
+
+    #[test]
+    fn delayed_managed_commit_does_not_replay_catch_up_slots() {
+        let start = Instant::now();
+        let mut schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 0)).unwrap();
+        let delayed_send = start + Duration::from_millis(45);
+
+        let commit = schedule
+            .preflight_managed_commit(start, delayed_send)
+            .unwrap();
+        schedule.commit(commit);
+
+        assert_eq!(
+            schedule.next_send_deadline(),
+            Some(start + Duration::from_millis(50))
+        );
+        assert!(schedule.next_send_deadline().unwrap() > delayed_send);
+    }
+
+    #[test]
+    fn deadline_arithmetic_overflow_is_detected_before_commit() {
+        let start = Instant::now();
+        let schedule = ProbeSchedule::new(start, &negotiated(2, 0)).unwrap();
+
+        assert!(matches!(
+            schedule.preflight_caller_commit(start, u64::MAX),
+            Err(ClientError::DurationOverflow)
+        ));
+        assert_eq!(schedule.next_send_deadline(), Some(start));
+    }
+}

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

@@ -118,6 +118,7 @@ fn close_flagged_echo_reply_emits_reply_then_closes_without_sending_close() {
     ));
     assert_eq!(events.len(), 2);
     assert!(client.next_send_deadline().is_none());
+    assert!(client.schedule.is_none());
     assert!(matches!(
         client.send_probe(),
         Err(ClientError::AlreadyClosed)

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

@@ -8,8 +8,11 @@ fn no_test_success_validates_params() {
     let server = no_test_server(params.clone(), 0);
     config.server_addr = server.addr.to_string();
     let mut client = Client::connect(config).unwrap();
+    assert!(client.next_send_deadline().is_none());
     let negotiated = assert_no_test_completed(client.open().unwrap());
     assert_eq!(negotiated.params, params);
+    assert!(client.next_send_deadline().is_none());
+    assert!(client.schedule.is_none());
     server.join();
 }
 

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

@@ -20,6 +20,7 @@ fn send_probe_sends_valid_echo_request() {
         ..default_test_config(server.addr)
     };
     let mut client = Client::connect(config).unwrap();
+    assert!(client.next_send_deadline().is_none());
     assert_open_started(client.open().unwrap());
     let events = client.send_probe().unwrap();
     assert_eq!(events.len(), 1);

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

@@ -12,12 +12,13 @@ use std::{
 };
 
 use crate::{
+    client::schedule::{advance_cadence, ProbeSchedule},
     config::{ClientAuthConfig, ClientConfig},
     error::ClientError,
     event::{ClientEvent, OpenOutcome},
     metadata::ReceiveMeta,
     receive::recv_datagram_from,
-    session::machine::{advance_cadence, params_from_config, SendProbeResult, SessionMachine},
+    session::machine::{params_from_config, SendProbeResult, SessionMachine},
     socket::{bind_unconnected_udp_socket, validate_open_timeouts},
     socket_options::apply_dscp_to_socket,
     timing::ClientTimestamp,
@@ -626,6 +627,7 @@ struct TargetState {
     remote: SocketAddr,
     configured_auth: Option<ClientAuthConfig>,
     runtime: SessionMachine,
+    schedule: Option<ProbeSchedule>,
     status: TargetStatus,
     open_packet: Vec<u8>,
     counters: TargetCounters,
@@ -716,6 +718,7 @@ impl TargetState {
             remote: target.remote,
             configured_auth: target.auth,
             runtime,
+            schedule: None,
             status: TargetStatus::Opening {
                 attempt: 0,
                 next_send_at: now,
@@ -749,6 +752,14 @@ impl TargetState {
         self.final_reason = Some(reason);
     }
 
+    fn is_run_complete(&self) -> bool {
+        self.runtime.is_terminal()
+            || self
+                .schedule
+                .as_ref()
+                .is_some_and(|schedule| schedule.is_finished() && self.runtime.pending_is_empty())
+    }
+
     fn outcome(&self, reason: ManagedTargetEndReason) -> ManagedTargetOutcome {
         ManagedTargetOutcome {
             id: self.id.clone(),
@@ -872,16 +883,21 @@ fn run_group_receiver(socket: UdpSocket, shared: Arc<GroupShared>) -> Result<(),
             let mut target = target.lock().expect("target mutex poisoned");
             match target.status {
                 TargetStatus::Opening { .. } => {
+                    let opened_at = datagram.received_at;
+                    let mut schedule = None;
                     let result = target.runtime.decode_open_reply(packet).and_then(|reply| {
-                        target.runtime.accept_open_reply(
+                        target
-                            reply,
+                            .runtime
-                            datagram.received_at,
+                            .accept_open_reply(reply, opened_at, |negotiated| {
-                            |negotiated| validate_group_negotiation(&shared, negotiated),
+                                validate_group_negotiation(&shared, negotiated)?;
-                        )
+                                schedule = Some(ProbeSchedule::new(opened_at.mono, negotiated)?);
+                                Ok(())
+                            })
                     });
 
                     match result {
                         Ok(outcome) => {
+                            target.schedule = schedule;
                             publish_open_outcome(&shared.hub, &mut target, outcome);
                             wake_scheduler = true;
                         }
@@ -985,6 +1001,7 @@ fn process_active_target_packet(
         Ok(events) => {
             publish_events(hub, target, events);
             if target.runtime.is_peer_closed() {
+                target.schedule = None;
                 target.mark_finished(ManagedTargetEndReason::PeerClosed);
                 true
             } else {
@@ -1251,7 +1268,7 @@ fn finish_completed_targets(socket: &UdpSocket, shared: &GroupShared, now: Insta
     for target in all_targets(shared) {
         let mut target = target.lock().expect("target mutex poisoned");
         match target.status {
-            TargetStatus::Active if target.runtime.is_run_complete() => {
+            TargetStatus::Active if target.is_run_complete() => {
                 if target.runtime.is_peer_closed() {
                     target.mark_finished(ManagedTargetEndReason::PeerClosed);
                 } else if target.runtime.has_timed_out_metadata() {
@@ -1295,19 +1312,27 @@ fn send_echo_to_target(
         return;
     }
     let remote = target.remote;
-    let result = target
+    target
         .runtime
-        .send_probe_for_deadline(None, scheduled_at, |packet| {
+        .ensure_open()
-            let sent_at = ClientTimestamp::now();
+        .expect("active targets are open");
-            let send_call_start = Instant::now();
+    let TargetState {
-            let bytes = socket.send_to(packet, remote)?;
+        runtime, schedule, ..
-            let send_call = send_call_start.elapsed();
+    } = &mut *target;
-            Ok(SendProbeResult {
+    let schedule = schedule
-                sent_at,
+        .as_mut()
-                bytes,
+        .expect("active targets always have a probe schedule");
-                send_call,
+    let result = runtime.send_probe_for_deadline(schedule, None, scheduled_at, |packet| {
-            })
+        let sent_at = ClientTimestamp::now();
-        });
+        let send_call_start = Instant::now();
+        let bytes = socket.send_to(packet, remote)?;
+        let send_call = send_call_start.elapsed();
+        Ok(SendProbeResult {
+            sent_at,
+            bytes,
+            send_call,
+        })
+    });
 
     match result {
         Ok(events) => {
@@ -1437,7 +1462,10 @@ fn close_locked_target(
             socket.send_to(packet, remote)?;
             Ok(())
         }) {
-            Ok(events) => publish_events(hub, target, events),
+            Ok(events) => {
+                publish_events(hub, target, events);
+                target.schedule = None;
+            }
             Err(err) => {
                 target.mark_finished(ManagedTargetEndReason::Failed(
                     ManagedTargetFailure::runtime(&err),
@@ -1619,7 +1647,7 @@ fn target_is_due_for_slot(target: &Arc<Mutex<TargetState>>, scheduled_at: Instan
 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()
+        target.schedule.as_ref()?.next_send_deadline()
     } else {
         None
     }
@@ -1810,24 +1838,38 @@ mod tests {
 
         let open_packet = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params);
         let open_reply = target.runtime.decode_open_reply(&open_packet).unwrap();
+        let opened_at = ClientTimestamp::now();
+        let mut opened_schedule = None;
         let open_outcome = target
             .runtime
-            .accept_open_reply(open_reply, ClientTimestamp::now(), |_| Ok(()))
+            .accept_open_reply(open_reply, opened_at, |negotiated| {
+                opened_schedule = Some(ProbeSchedule::new(opened_at.mono, negotiated)?);
+                Ok(())
+            })
             .unwrap();
+        target.schedule = opened_schedule;
         publish_open_outcome(&hub, &mut target, open_outcome);
 
         let scheduled_at = Instant::now();
         let sent_at = ClientTimestamp::now();
-        let sent_events = target
+        let sent_events = {
-            .runtime
+            let TargetState {
-            .send_probe_for_deadline(Some(sent_at), scheduled_at, |packet| {
+                runtime, schedule, ..
-                Ok(SendProbeResult {
+            } = &mut target;
-                    sent_at,
+            runtime.send_probe_for_deadline(
-                    bytes: packet.len(),
+                schedule.as_mut().unwrap(),
-                    send_call: Duration::ZERO,
+                Some(sent_at),
-                })
+                scheduled_at,
-            })
+                |packet| {
-            .unwrap();
+                    Ok(SendProbeResult {
+                        sent_at,
+                        bytes: packet.len(),
+                        send_call: Duration::ZERO,
+                    })
+                },
+            )
+        }
+        .unwrap();
         publish_events(&hub, &mut target, sent_events);
 
         ActiveTargetFixture {

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

@@ -7,6 +7,7 @@ use irtt_proto::{
 };
 
 use crate::{
+    client::schedule::ProbeSchedule,
     config::{
         ClientConfig, RunMode, MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_UDP_PAYLOAD_LENGTH,
     },
@@ -56,13 +57,9 @@ struct ActiveSession {
     next_wire_seq: u32,
     highest_received_seq: Option<u32>,
     packets_sent: u64,
-    start_mono: Instant,
-    end_mono: Option<Instant>,
-    next_send_at: Instant,
     pending: PendingMap,
     timed_out: TimedOutMap,
     completed: CompletedSet,
-    sending_done: bool,
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -159,33 +156,25 @@ impl SessionMachine {
         }
     }
 
-    pub(crate) fn next_send_deadline(&self) -> Option<Instant> {
-        let MachineState::Open(session) = &self.state else {
-            return None;
-        };
-        if session.sending_done {
-            return None;
-        }
-        Some(session.next_send_at)
-    }
-
     pub(crate) fn probe_timeout(&self) -> Duration {
         self.config.probe_timeout
     }
 
     pub(crate) fn send_probe_with<F>(
         &mut self,
+        schedule: &mut ProbeSchedule,
         override_ts: Option<ClientTimestamp>,
         send: F,
     ) -> Result<Vec<ClientEvent>, ClientError>
     where
         F: FnOnce(&[u8]) -> Result<SendProbeResult, ClientError>,
     {
-        self.send_probe_inner(override_ts, None, false, send)
+        self.send_probe_inner(schedule, override_ts, None, false, send)
     }
 
     pub(crate) fn send_probe_for_deadline<F>(
         &mut self,
+        schedule: &mut ProbeSchedule,
         override_ts: Option<ClientTimestamp>,
         scheduled_at: Instant,
         send: F,
@@ -193,11 +182,12 @@ impl SessionMachine {
     where
         F: FnOnce(&[u8]) -> Result<SendProbeResult, ClientError>,
     {
-        self.send_probe_inner(override_ts, Some(scheduled_at), true, send)
+        self.send_probe_inner(schedule, override_ts, Some(scheduled_at), true, send)
     }
 
     fn send_probe_inner<F>(
         &mut self,
+        schedule: &mut ProbeSchedule,
         override_ts: Option<ClientTimestamp>,
         scheduled_at_override: Option<Instant>,
         skip_missed_slots: bool,
@@ -206,6 +196,11 @@ impl SessionMachine {
     where
         F: FnOnce(&[u8]) -> Result<SendProbeResult, ClientError>,
     {
+        let now = override_ts.unwrap_or_else(ClientTimestamp::now);
+        if !schedule.permit_probe_at(now.mono) {
+            return Ok(vec![]);
+        }
+
         let (config, state) = (&self.config, &mut self.state);
         let probe_timeout = config.probe_timeout;
         let hmac_key = config.hmac_key.as_deref();
@@ -216,23 +211,9 @@ impl SessionMachine {
             MachineState::NoTestCompleted => return Err(ClientError::AlreadyCompleted),
         };
 
-        if session.sending_done {
-            return Ok(vec![]);
-        }
-
-        let now = override_ts.unwrap_or_else(ClientTimestamp::now);
-
-        if let Some(end) = session.end_mono {
-            if now.mono >= end {
-                session.sending_done = true;
-                return Ok(vec![]);
-            }
-        }
-
         session.pending.check_capacity()?;
 
         let wire_seq = session.next_wire_seq;
-        let scheduled_at = scheduled_at_override.unwrap_or(session.next_send_at);
 
         let request = EchoRequest {
             token: session.token,
@@ -263,27 +244,17 @@ impl SessionMachine {
                     counter: "packets_sent",
                 })?;
 
-        let interval_ns = u64::try_from(session.negotiated.params.interval_ns)
+        let schedule_commit = if skip_missed_slots {
-            .expect("validated positive negotiated interval");
+            schedule.preflight_managed_commit(
-        let interval = Duration::from_nanos(interval_ns);
+                scheduled_at_override.expect("managed sends include a scheduled deadline"),
-        let (scheduled_at, next_send_at) = if skip_missed_slots {
+                send_result.sent_at.mono,
-            let (scheduled_at, next_send_at, _) =
+            )?
-                advance_cadence(scheduled_at, interval, send_result.sent_at.mono)?;
-            (scheduled_at, next_send_at)
         } else {
-            (
+            schedule.preflight_caller_commit(send_result.sent_at.mono, session.packets_sent)?
-                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 scheduled_at = schedule_commit.scheduled_at;
-        session.next_send_at = next_send_at;
+        let timer_error = schedule_commit.timer_error;
-
+        schedule.commit(schedule_commit);
-        if let Some(end) = session.end_mono {
-            if session.next_send_at >= end {
-                session.sending_done = true;
-            }
-        }
 
         Ok(vec![ClientEvent::EchoSent {
             seq: wire_seq,
@@ -358,14 +329,21 @@ impl SessionMachine {
         }])
     }
 
-    pub(crate) fn is_run_complete(&self) -> bool {
+    pub(crate) fn pending_is_empty(&self) -> bool {
         match &self.state {
-            MachineState::Open(session) => session.sending_done && session.pending.len() == 0,
+            MachineState::Open(session) => session.pending.len() == 0,
             MachineState::NoTestCompleted | MachineState::Closed { .. } => true,
             MachineState::Connected => false,
         }
     }
 
+    pub(crate) fn is_terminal(&self) -> bool {
+        matches!(
+            self.state,
+            MachineState::NoTestCompleted | MachineState::Closed { .. }
+        )
+    }
+
     pub(crate) fn is_peer_closed(&self) -> bool {
         matches!(
             self.state,
@@ -395,6 +373,10 @@ impl SessionMachine {
         matches!(self.state, MachineState::Open(_))
     }
 
+    pub(crate) fn ensure_open(&self) -> Result<(), ClientError> {
+        self.open_session().map(|_| ())
+    }
+
     fn accept_normal_open<F>(
         &mut self,
         reply: OpenReply,
@@ -412,28 +394,15 @@ impl SessionMachine {
         )?;
         before_normal_open(&negotiated)?;
 
-        let end_mono = if negotiated.params.duration_ns > 0 {
-            Some(negotiated_end_mono(
-                now.mono,
-                negotiated.params.duration_ns,
-            )?)
-        } else {
-            None
-        };
-
         self.state = MachineState::Open(Box::new(ActiveSession {
             token,
             negotiated: negotiated.clone(),
             next_wire_seq: 0,
             highest_received_seq: None,
             packets_sent: 0,
-            start_mono: now.mono,
-            end_mono,
-            next_send_at: now.mono,
             pending: PendingMap::new(self.config.max_pending_probes),
             timed_out: TimedOutMap::new(self.config.max_pending_probes),
             completed: CompletedSet::new(self.config.max_pending_probes),
-            sending_done: false,
         }));
 
         let event = ClientEvent::SessionStarted {
@@ -704,67 +673,6 @@ pub(crate) fn update_highest_received(highest_received_seq: &mut Option<u32>, wi
     }));
 }
 
-fn next_probe_deadline(
-    start: Instant,
-    interval_ns: u64,
-    packets_sent: u64,
-) -> Result<Instant, ClientError> {
-    let offset_ns = interval_ns
-        .checked_mul(packets_sent)
-        .ok_or(ClientError::DurationOverflow)?;
-    start
-        .checked_add(Duration::from_nanos(offset_ns))
-        .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)
 }
@@ -773,12 +681,6 @@ pub(crate) fn sequence_is_before(candidate: u32, current: u32) -> bool {
     current != candidate && current.wrapping_sub(candidate) < (1 << 31)
 }
 
-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)
-}
-
 pub(crate) fn compute_rtt(
     sent_at: &ClientTimestamp,
     received_at: &ClientTimestamp,
@@ -929,13 +831,3 @@ fn config_duration_to_ns(field: &str, duration: Duration) -> Result<i64, ClientE
         reason: format!("{field} is too large to encode as nanoseconds"),
     })
 }
-
-fn negotiated_end_mono(start: Instant, duration_ns: i64) -> Result<Instant, ClientError> {
-    debug_assert!(duration_ns > 0);
-    let duration_ns = u64::try_from(duration_ns).expect("validated positive negotiated duration");
-    start
-        .checked_add(Duration::from_nanos(duration_ns))
-        .ok_or_else(|| ClientError::NegotiationRejected {
-            reason: "duration is too large to schedule".to_owned(),
-        })
-}