crates/irtt-client/src/async_client.rs

@@ -86,6 +86,8 @@ pub(crate) struct AsyncOpenState {
     deadline: Option<Instant>,
     deadline_timer: Option<Pin<Box<time::Sleep>>>,
     request_submitted: bool,
+    stop_after_current_attempt: bool,
+    stopped_after_current_attempt: bool,
     buffer: [u8; MAX_OPEN_PACKET_SIZE],
     cleanup: Option<AsyncOpenCleanup>,
 }
@@ -97,6 +99,8 @@ impl AsyncOpenState {
             deadline: None,
             deadline_timer: None,
             request_submitted: false,
+            stop_after_current_attempt: false,
+            stopped_after_current_attempt: false,
             buffer: [0_u8; MAX_OPEN_PACKET_SIZE],
             cleanup: None,
         }
@@ -117,6 +121,24 @@ impl AsyncOpenState {
         self.deadline = None;
         self.deadline_timer = None;
         self.request_submitted = false;
+        self.stopped_after_current_attempt = self.stop_after_current_attempt;
+    }
+
+    pub(crate) fn has_in_flight_work(&self) -> bool {
+        self.request_submitted || self.cleanup.is_some()
+    }
+
+    pub(crate) fn request_stop_after_current_attempt(&mut self) {
+        self.stop_after_current_attempt = true;
+    }
+
+    pub(crate) fn stopped_after_current_attempt(&self) -> bool {
+        self.stopped_after_current_attempt
+    }
+
+    #[cfg(test)]
+    pub(crate) fn has_retained_cleanup(&self) -> bool {
+        self.cleanup.is_some()
     }
 
     fn deadline(&self) -> Instant {
@@ -149,6 +171,7 @@ struct AsyncClientTestHooks {
     fail_cleanup_send: Cell<bool>,
     fail_peer_close_dscp: Cell<bool>,
     fail_dscp_restore: Cell<bool>,
+    pause_open_cleanup_before_send: Cell<bool>,
     pause_open_before_writable: Cell<bool>,
     pause_open_before_readable: Cell<bool>,
     pause_probe_before_writable: Cell<bool>,
@@ -301,6 +324,10 @@ impl AsyncClient {
             if state.attempt >= attempt_count {
                 return Poll::Ready(Err(ClientError::OpenTimeout));
             }
+            if state.stop_after_current_attempt && state.deadline.is_none() {
+                state.stopped_after_current_attempt = true;
+                return Poll::Ready(Err(ClientError::OpenTimeout));
+            }
             if state.deadline.is_none() {
                 let timeout = self.machine.config().open_timeouts[state.attempt];
                 if let Err(error) = state.start_attempt(timeout) {
@@ -333,10 +360,10 @@ impl AsyncClient {
                 let send_result = self.socket.try_send(&request.bytes);
                 match send_result {
                     Ok(bytes) => {
+                        state.request_submitted = true;
                         if let Err(error) = validate_datagram_length(request.bytes.len(), bytes) {
                             return Poll::Ready(Err(error));
                         }
-                        state.request_submitted = true;
                     }
                     Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue,
                     Err(error) => return Poll::Ready(Err(ClientError::Socket(error))),
@@ -352,6 +379,9 @@ impl AsyncClient {
                 Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
                 Poll::Ready(Ok(false)) => {
                     state.finish_attempt();
+                    if state.stopped_after_current_attempt() {
+                        return Poll::Ready(Err(ClientError::OpenTimeout));
+                    }
                     continue;
                 }
                 Poll::Ready(Ok(true)) => {}
@@ -364,6 +394,9 @@ impl AsyncClient {
             };
             if datagram.received_at.mono > state.deadline() {
                 state.finish_attempt();
+                if state.stopped_after_current_attempt() {
+                    return Poll::Ready(Err(ClientError::OpenTimeout));
+                }
                 continue;
             }
 
@@ -663,6 +696,13 @@ impl AsyncClient {
         self.schedule.as_ref()?.next_send_deadline()
     }
 
+    pub(crate) fn probe_interval(&self) -> Option<Duration> {
+        if !self.machine.is_open() {
+            return None;
+        }
+        self.schedule.as_ref().map(ProbeSchedule::interval)
+    }
+
     /// Return the configured local probe timeout.
     pub fn probe_timeout(&self) -> Duration {
         self.machine.probe_timeout()
@@ -682,6 +722,52 @@ impl AsyncClient {
         self.machine.is_peer_closed()
     }
 
+    pub(crate) fn remote_addr(&self) -> SocketAddr {
+        self.remote
+    }
+
+    pub(crate) fn packets_sent(&self) -> u64 {
+        self.machine.packets_sent()
+    }
+
+    pub(crate) fn next_probe_timeout_deadline(&self) -> Option<Instant> {
+        self.machine.next_probe_timeout_deadline()
+    }
+
+    pub(crate) fn latest_probe_timeout_deadline(&self) -> Option<Instant> {
+        self.machine.latest_probe_timeout_deadline()
+    }
+
+    pub(crate) fn discard_prepared_probe(&mut self) {
+        self.prepared_probe = None;
+    }
+
+    #[cfg(test)]
+    pub(crate) fn retain_open_cleanup_once(&self) {
+        self.test_hooks.fail_open_dscp.set(true);
+        self.test_hooks.pause_open_cleanup_before_send.set(true);
+    }
+
+    #[cfg(test)]
+    pub(crate) fn poll_recv_ready_for_test(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.socket.poll_recv_ready(cx)
+    }
+
+    pub(crate) fn skip_missed_probe_slots_at(&mut self, now: Instant) -> Result<(), ClientError> {
+        let Some(schedule) = self.schedule.as_mut() else {
+            return Ok(());
+        };
+        let Some(deadline) = schedule.next_send_deadline() else {
+            return Ok(());
+        };
+        if deadline > now {
+            return Ok(());
+        }
+        let commit = schedule.preflight_managed_commit(deadline, now)?;
+        schedule.commit(commit);
+        Ok(())
+    }
+
     fn prepare_async_open(
         &mut self,
         machine: PreparedOpenAcceptance,
@@ -779,6 +865,15 @@ impl AsyncClient {
                 return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary()));
             };
 
+            #[cfg(test)]
+            if self
+                .test_hooks
+                .pause_open_cleanup_before_send
+                .replace(false)
+            {
+                return Poll::Pending;
+            }
+
             #[cfg(test)]
             if self.test_hooks.fail_cleanup_send.replace(false) {
                 return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary()));

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

@@ -50,6 +50,11 @@ impl ProbeSchedule {
         self.next_send_at
     }
 
+    #[cfg(feature = "tokio")]
+    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;

crates/irtt-client/src/lib.rs

@@ -59,7 +59,7 @@ mod client;
 mod config;
 mod error;
 mod event;
-mod managed;
+pub mod managed;
 mod metadata;
 mod probe;
 mod receive;
@@ -80,14 +80,15 @@ pub use event::{
     ClientEvent, OneWayDelaySample, OpenOutcome, PacketMeta, ReceivedStatsSample, RttSample,
     ServerTiming, SignedDuration, WarningKind,
 };
-pub use managed::{
+pub use managed::legacy::{
     CancellationToken, EventHub, EventSubscription, ManagedClient, ManagedClientGroup,
     ManagedClientGroupConfig, ManagedClientGroupSession, ManagedClientSession,
     ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, ManagedGroupOutcome,
     ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure,
     ManagedTargetFailureKind, ManagedTargetOutcome, SessionEndReason, SessionOutcome,
-    SubscriberConfig, SubscriberOverflow, TargetEvent, TargetEventSubscription, TargetId,
+    SubscriberConfig, SubscriberOverflow, TargetEvent, TargetEventSubscription,
     MANAGED_GROUP_OUTCOME_HISTORY_LIMIT,
 };
+pub use managed::TargetId;
 pub use session::{NegotiatedParams, NegotiationRestriction};
 pub use timing::ClientTimestamp;

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

@@ -1,7 +1,6 @@
 use std::{
     collections::{HashMap, HashSet, VecDeque},
     fmt,
-    hash::{Hash, Hasher},
     net::{SocketAddr, UdpSocket},
     sync::{
         atomic::{AtomicU64, Ordering},
@@ -37,6 +36,7 @@ use crate::client::ProbeSendTimestamps;
 use super::{
     cancellation::CancellationToken,
     hub::{EventHub, EventSubscription, SubscriberConfig},
+    TargetId,
 };
 
 const GROUP_RECV_TIMEOUT: Duration = Duration::from_millis(20);
@@ -49,59 +49,6 @@ const RECV_BUFFER_SIZE: usize = 65_536;
 /// Maximum number of recent target outcomes retained in [`ManagedGroupOutcome`].
 pub const MANAGED_GROUP_OUTCOME_HISTORY_LIMIT: usize = 256;
 
-/// Caller-owned target identity for managed multi-target probing.
-#[derive(Clone, Eq)]
-pub struct TargetId(Arc<str>);
-
-impl TargetId {
-    /// Borrow the identifier as a string slice.
-    pub fn as_str(&self) -> &str {
-        &self.0
-    }
-}
-
-impl fmt::Debug for TargetId {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_tuple("TargetId").field(&self.0).finish()
-    }
-}
-
-impl fmt::Display for TargetId {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str(&self.0)
-    }
-}
-
-impl PartialEq for TargetId {
-    fn eq(&self, other: &Self) -> bool {
-        self.0 == other.0
-    }
-}
-
-impl Hash for TargetId {
-    fn hash<H: Hasher>(&self, state: &mut H) {
-        self.0.hash(state);
-    }
-}
-
-impl From<&str> for TargetId {
-    fn from(value: &str) -> Self {
-        Self(Arc::from(value))
-    }
-}
-
-impl From<String> for TargetId {
-    fn from(value: String) -> Self {
-        Self(Arc::from(value))
-    }
-}
-
-impl AsRef<str> for TargetId {
-    fn as_ref(&self) -> &str {
-        self.as_str()
-    }
-}
-
 /// Per-target configuration for [`ManagedClientGroup`].
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ManagedTargetConfig {
@@ -2283,7 +2230,7 @@ fn divide_duration(duration: Duration, divisor: usize) -> Duration {
 mod tests {
     use super::*;
     use crate::{
-        config::NegotiationPolicy, managed::SubscriberOverflow, EventSubscriptionError, WarningKind,
+        config::NegotiationPolicy, EventSubscriptionError, SubscriberOverflow, WarningKind,
     };
     use irtt_proto::{
         compute_hmac_in_place, echo_packet_len,

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

@@ -2,14 +2,34 @@ mod cancellation;
 mod group;
 mod hub;
 mod runner;
+mod target_id;
 
-pub use cancellation::CancellationToken;
-pub use group::{
+pub use target_id::TargetId;
+
+/// Temporary exports for the obsolete thread-based managed implementations.
+#[doc(hidden)]
+pub mod legacy {
+    pub use super::cancellation::CancellationToken;
+    pub use super::group::{
         ManagedClientGroup, ManagedClientGroupConfig, ManagedClientGroupSession,
-    ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, ManagedGroupOutcome,
-    ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure,
-    ManagedTargetFailureKind, ManagedTargetOutcome, TargetEvent, TargetEventSubscription, TargetId,
-    MANAGED_GROUP_OUTCOME_HISTORY_LIMIT,
-};
-pub use hub::{EventHub, EventSubscription, SubscriberConfig, SubscriberOverflow};
-pub use runner::{ManagedClient, ManagedClientSession, SessionEndReason, SessionOutcome};
+        ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent,
+        ManagedGroupOutcome, ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason,
+        ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetOutcome, TargetEvent,
+        TargetEventSubscription, MANAGED_GROUP_OUTCOME_HISTORY_LIMIT,
+    };
+    pub use super::hub::{EventHub, EventSubscription, SubscriberConfig, SubscriberOverflow};
+    pub use super::runner::{
+        ManagedClient, ManagedClientSession, SessionEndReason, SessionOutcome,
+    };
+    pub use super::TargetId;
+}
+
+#[cfg(feature = "tokio")]
+mod model;
+#[cfg(feature = "tokio")]
+mod task;
+
+#[cfg(feature = "tokio")]
+pub use model::*;
+#[cfg(feature = "tokio")]
+pub use task::*;

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

@@ -0,0 +1,331 @@
+use std::{net::SocketAddr, sync::Arc, time::Duration};
+
+use thiserror::Error;
+
+use crate::{ClientAuthConfig, ClientConfig, ClientError, ClientEvent};
+
+use super::TargetId;
+
+/// Default capacity of the lossy managed event channel.
+pub const DEFAULT_MANAGED_EVENT_CAPACITY: usize = 256;
+/// Default number of recent target outcomes retained in status and output.
+pub const DEFAULT_MANAGED_OUTCOME_HISTORY_LIMIT: usize = 256;
+/// Default maximum number of simultaneously live target generations.
+pub const DEFAULT_MANAGED_MAX_LIVE_TARGET_GENERATIONS: usize = 256;
+/// Default time retained for late replies after the final committed timeout.
+pub const DEFAULT_MANAGED_FINAL_DRAIN: Duration = Duration::from_millis(100);
+
+/// One globally allocated incarnation of a target within a managed task.
+///
+/// Generations increase globally across the task, not independently per ID.
+#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub struct TargetInstance {
+    pub id: TargetId,
+    pub generation: u64,
+}
+
+/// Endpoint and optional authentication override for one managed target.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ManagedTargetConfig {
+    pub id: TargetId,
+    pub server_addr: String,
+    pub auth: Option<ClientAuthConfig>,
+}
+
+impl ManagedTargetConfig {
+    pub fn new(id: impl Into<TargetId>, server_addr: impl Into<String>) -> Self {
+        Self {
+            id: id.into(),
+            server_addr: server_addr.into(),
+            auth: None,
+        }
+    }
+}
+
+/// Coordination mode for sends across active targets.
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
+pub enum ManagedPacing {
+    #[default]
+    Staggered,
+    Burst,
+}
+
+/// Policy controlling top-level natural completion.
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
+pub enum ManagedCompletionPolicy {
+    #[default]
+    FinishWhenQuiescent,
+    ExplicitStop,
+}
+
+/// Shared configuration for one managed task.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ManagedClientConfig {
+    pub client: ClientConfig,
+    pub pacing: ManagedPacing,
+    pub completion: ManagedCompletionPolicy,
+    pub event_capacity: usize,
+    pub outcome_history_limit: usize,
+    pub max_live_target_generations: usize,
+    pub final_drain: Duration,
+}
+
+impl Default for ManagedClientConfig {
+    fn default() -> Self {
+        Self {
+            client: ClientConfig::default(),
+            pacing: ManagedPacing::default(),
+            completion: ManagedCompletionPolicy::default(),
+            event_capacity: DEFAULT_MANAGED_EVENT_CAPACITY,
+            outcome_history_limit: DEFAULT_MANAGED_OUTCOME_HISTORY_LIMIT,
+            max_live_target_generations: DEFAULT_MANAGED_MAX_LIVE_TARGET_GENERATIONS,
+            final_drain: DEFAULT_MANAGED_FINAL_DRAIN,
+        }
+    }
+}
+
+/// Durable top-level lifecycle.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ManagedLifecycle {
+    NotStarted,
+    Running,
+    Stopping,
+    Completed,
+    Failed,
+    Abandoned,
+}
+
+/// Durable lifecycle for one target generation.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ManagedTargetLifecycle {
+    Pending,
+    Connecting,
+    Opening,
+    Active,
+    Draining,
+    Closing,
+    Terminal,
+}
+
+/// Status for one target generation.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ManagedTargetStatus {
+    pub target: TargetInstance,
+    pub lifecycle: ManagedTargetLifecycle,
+    pub server_addr: Arc<str>,
+    pub remote: Option<SocketAddr>,
+}
+
+/// Authoritative immutable managed status snapshot.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ManagedStatus {
+    pub lifecycle: ManagedLifecycle,
+    pub stop_requested: bool,
+    pub desired_target_count: usize,
+    pub connecting_target_count: usize,
+    pub opening_target_count: usize,
+    pub active_target_count: usize,
+    pub draining_target_count: usize,
+    pub closing_target_count: usize,
+    pub terminal_target_count: usize,
+    pub total_target_outcomes: u64,
+    pub successful_target_outcomes: u64,
+    pub failed_target_outcomes: u64,
+    pub peer_closed_target_outcomes: u64,
+    pub discarded_target_outcomes: u64,
+    pub targets: Arc<[ManagedTargetStatus]>,
+    pub recent_target_outcomes: Arc<[ManagedTargetOutcome]>,
+    pub final_outcome: Option<Arc<ManagedOutcome>>,
+}
+
+/// Lossy presentation event emitted by the managed task.
+#[allow(clippy::large_enum_variant)]
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum ManagedEvent {
+    Started,
+    TargetStateChanged {
+        target: TargetInstance,
+        lifecycle: ManagedTargetLifecycle,
+    },
+    Client {
+        target: TargetInstance,
+        event: ClientEvent,
+    },
+    TargetFinished {
+        outcome: Arc<ManagedTargetOutcome>,
+    },
+    Stopping,
+    Completed {
+        outcome: Arc<ManagedOutcome>,
+    },
+    Failed {
+        outcome: Arc<ManagedOutcome>,
+    },
+    Abandoned,
+}
+
+/// Receiving half of the lossy managed event stream.
+pub type ManagedEventSubscription = tokio::sync::broadcast::Receiver<ManagedEvent>;
+
+/// Final authoritative outcome returned by the task.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ManagedOutcome {
+    pub end_reason: ManagedEndReason,
+    pub total_target_outcomes: u64,
+    pub successful_target_outcomes: u64,
+    pub failed_target_outcomes: u64,
+    pub peer_closed_target_outcomes: u64,
+    pub discarded_target_outcomes: u64,
+    pub recent_target_outcomes: Arc<[ManagedTargetOutcome]>,
+}
+
+/// Why the managed task ended.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum ManagedEndReason {
+    TargetsComplete,
+    StopRequested,
+    DriverFailed(ManagedDriverFailure),
+}
+
+/// Durable outcome for one target generation.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ManagedTargetOutcome {
+    pub target: TargetInstance,
+    pub server_addr: Arc<str>,
+    pub remote: Option<SocketAddr>,
+    pub end_reason: ManagedTargetEndReason,
+    pub packets_sent: u64,
+    pub replies_received: u64,
+    pub duplicates: u64,
+    pub late: u64,
+    pub warning_events: u64,
+    pub cleanup_failure: Option<ManagedTargetFailure>,
+}
+
+/// Why one target generation ended.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum ManagedTargetEndReason {
+    TestComplete,
+    NoTestComplete,
+    PeerClosed,
+    Removed,
+    Replaced,
+    Stopped,
+    Failed(ManagedTargetFailure),
+}
+
+/// Phase in which a target-local failure occurred.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ManagedTargetFailurePhase {
+    Connecting,
+    Opening,
+    Sending,
+    Receiving,
+    Timing,
+    Closing,
+}
+
+/// Stable category for a target-local failure.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ManagedTargetFailureKind {
+    Resolve,
+    Socket,
+    SocketOption,
+    Protocol,
+    Timeout,
+    Configuration,
+    ResourceExhausted,
+    InvalidState,
+    Other,
+}
+
+/// Durable target-local failure details.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ManagedTargetFailure {
+    pub phase: ManagedTargetFailurePhase,
+    pub kind: ManagedTargetFailureKind,
+    pub message: Arc<str>,
+}
+
+/// Configuration failure detected before any runtime work begins.
+#[derive(Debug, Error)]
+pub enum ManagedConfigError {
+    #[error("FinishWhenQuiescent requires at least one initial target")]
+    EmptyInitialTargets,
+    #[error("duplicate managed target id {id}")]
+    DuplicateTargetId { id: TargetId },
+    #[error("configured {configured} targets exceeds live-generation limit {limit}")]
+    TooManyTargets { configured: usize, limit: usize },
+    #[error("managed event capacity must be greater than zero")]
+    ZeroEventCapacity,
+    #[error("managed live-generation limit must be greater than zero")]
+    ZeroLiveGenerationLimit,
+    #[error("managed final drain {duration:?} cannot be scheduled from the current instant")]
+    UnschedulableFinalDrain { duration: Duration },
+    #[error("invalid managed target {id}: {source}")]
+    InvalidTarget {
+        id: TargetId,
+        #[source]
+        source: ClientError,
+    },
+    #[error("managed target generation space is exhausted")]
+    GenerationExhausted,
+}
+
+/// Failure to create a new event subscription.
+#[derive(Clone, Debug, Eq, Error, PartialEq)]
+pub enum ManagedSubscribeError {
+    #[error("managed event stream is closed")]
+    Closed,
+}
+
+/// Whole-driver failure.
+#[derive(Clone, Debug, Eq, Error, PartialEq)]
+pub enum ManagedDriverFailure {
+    #[error("ManagedClientTask requires a current Tokio runtime")]
+    NoTokioRuntime,
+    #[error("managed driver exhausted resources for {operation}")]
+    ResourceExhausted { operation: &'static str },
+    #[error("managed driver invariant failed: {message}")]
+    Internal { message: Arc<str> },
+}
+
+pub(crate) fn classify_client_error(
+    phase: ManagedTargetFailurePhase,
+    error: &ClientError,
+) -> ManagedTargetFailure {
+    let kind = match error {
+        ClientError::Resolve { .. } => ManagedTargetFailureKind::Resolve,
+        ClientError::Socket(_) => ManagedTargetFailureKind::Socket,
+        ClientError::SocketOption { .. } | ClientError::ReadTimeoutRestore { .. } => {
+            ManagedTargetFailureKind::SocketOption
+        }
+        ClientError::Protocol(_)
+        | ClientError::ProtocolVersionMismatch { .. }
+        | ClientError::ZeroToken
+        | ClientError::UnexpectedNoTestReply
+        | ClientError::NonZeroNoTestToken { .. }
+        | ClientError::ServerRejected
+        | ClientError::NegotiationRejected { .. } => ManagedTargetFailureKind::Protocol,
+        ClientError::OpenTimeout => ManagedTargetFailureKind::Timeout,
+        ClientError::InvalidConfig { .. }
+        | ClientError::OpenTimeoutTooSmall { .. }
+        | ClientError::NoOpenTimeouts => ManagedTargetFailureKind::Configuration,
+        ClientError::AllocationFailed { .. }
+        | ClientError::CounterOverflow { .. }
+        | ClientError::DurationOverflow
+        | ClientError::PendingLimitExceeded { .. } => ManagedTargetFailureKind::ResourceExhausted,
+        ClientError::NotOpen
+        | ClientError::AlreadyOpen
+        | ClientError::AlreadyCompleted
+        | ClientError::AlreadyClosed
+        | ClientError::StalePreparedProbe { .. }
+        | ClientError::PendingSequenceCollision { .. } => ManagedTargetFailureKind::InvalidState,
+        _ => ManagedTargetFailureKind::Other,
+    };
+    ManagedTargetFailure {
+        phase,
+        kind,
+        message: Arc::from(error.to_string()),
+    }
+}

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

@@ -0,0 +1,47 @@
+use std::{fmt, sync::Arc};
+
+/// Caller-owned stable target identity.
+#[derive(Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub struct TargetId(Arc<str>);
+
+impl TargetId {
+    /// Construct a target identifier.
+    pub fn new(value: impl Into<Arc<str>>) -> Self {
+        Self(value.into())
+    }
+
+    /// Borrow the identifier as a string slice.
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl fmt::Debug for TargetId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_tuple("TargetId").field(&self.0).finish()
+    }
+}
+
+impl fmt::Display for TargetId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(&self.0)
+    }
+}
+
+impl From<&str> for TargetId {
+    fn from(value: &str) -> Self {
+        Self::new(value)
+    }
+}
+
+impl From<String> for TargetId {
+    fn from(value: String) -> Self {
+        Self::new(value)
+    }
+}
+
+impl AsRef<str> for TargetId {
+    fn as_ref(&self) -> &str {
+        self.as_str()
+    }
+}

crates/irtt-client/src/probe.rs

@@ -71,6 +71,16 @@ impl PendingMap {
         self.map.len()
     }
 
+    #[cfg(feature = "tokio")]
+    pub fn next_timeout_deadline(&self) -> Option<Instant> {
+        self.map.values().map(|probe| probe.timeout_at).min()
+    }
+
+    #[cfg(feature = "tokio")]
+    pub fn latest_timeout_deadline(&self) -> Option<Instant> {
+        self.map.values().map(|probe| probe.timeout_at).max()
+    }
+
     #[cfg(test)]
     pub fn contains(&self, wire_seq: u32) -> bool {
         self.map.contains_key(&wire_seq)
@@ -139,6 +149,11 @@ impl TimedOutMap {
         self.map.len()
     }
 
+    #[cfg(feature = "tokio")]
+    pub fn latest_timeout_deadline(&self) -> Option<Instant> {
+        self.map.values().map(|probe| probe.timeout_at).max()
+    }
+
     #[cfg(test)]
     pub fn contains(&self, wire_seq: u32) -> bool {
         self.map.contains_key(&wire_seq)

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

@@ -169,6 +169,18 @@ impl SessionMachine {
         config: ClientConfig,
         remote: std::net::SocketAddr,
     ) -> Result<Self, ClientError> {
+        Self::validate_config(&config)?;
+        let requested = params_from_config(&config)?;
+
+        Ok(Self {
+            config,
+            remote,
+            requested,
+            state: MachineState::Connected,
+        })
+    }
+
+    pub(crate) fn validate_config(config: &ClientConfig) -> Result<(), ClientError> {
         if config.max_pending_probes == 0 {
             return Err(ClientError::InvalidConfig {
                 reason: "max_pending_probes must be greater than zero".to_owned(),
@@ -179,14 +191,7 @@ impl SessionMachine {
                 reason: "probe_timeout must be greater than zero".to_owned(),
             });
         }
-        let requested = params_from_config(&config)?;
-
-        Ok(Self {
-            config,
-            remote,
-            requested,
-            state: MachineState::Connected,
-        })
+        params_from_config(config).map(|_| ())
     }
 
     pub(crate) fn config(&self) -> &ClientConfig {
@@ -490,6 +495,31 @@ impl SessionMachine {
         }
     }
 
+    #[cfg(feature = "tokio")]
+    pub(crate) fn next_probe_timeout_deadline(&self) -> Option<Instant> {
+        match &self.state {
+            MachineState::Open(session) => session.pending.next_timeout_deadline(),
+            MachineState::Connected
+            | MachineState::NoTestCompleted
+            | MachineState::Closed { .. } => None,
+        }
+    }
+
+    #[cfg(feature = "tokio")]
+    pub(crate) fn latest_probe_timeout_deadline(&self) -> Option<Instant> {
+        match &self.state {
+            MachineState::Open(session) => session
+                .pending
+                .latest_timeout_deadline()
+                .into_iter()
+                .chain(session.timed_out.latest_timeout_deadline())
+                .max(),
+            MachineState::Connected
+            | MachineState::NoTestCompleted
+            | MachineState::Closed { .. } => None,
+        }
+    }
+
     pub(crate) fn is_open(&self) -> bool {
         matches!(self.state, MachineState::Open(_))
     }