irtt_client/managed/
model.rs

1use std::{fmt, net::SocketAddr, sync::Arc, time::Duration};
2
3use thiserror::Error;
4
5use crate::{AddressFamily, Authentication, ClientConfig, ClientError, ClientEvent};
6
7use super::TargetId;
8
9/// Default capacity of the lossy managed event channel.
10pub const DEFAULT_MANAGED_EVENT_CAPACITY: usize = 256;
11/// Default capacity of the bounded managed control queue.
12pub const DEFAULT_MANAGED_COMMAND_CAPACITY: usize = 64;
13/// Default number of recent target outcomes retained in status and output.
14pub const DEFAULT_MANAGED_OUTCOME_HISTORY_LIMIT: usize = 256;
15/// Default maximum number of simultaneously live target generations.
16pub const DEFAULT_MANAGED_MAX_LIVE_TARGET_GENERATIONS: usize = 256;
17/// Default time retained for late replies after the final committed timeout.
18pub const DEFAULT_MANAGED_FINAL_DRAIN: Duration = Duration::from_millis(100);
19
20/// One globally allocated incarnation of a target within a managed task.
21///
22/// Generations increase globally across the task, not independently per ID.
23#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
24pub struct TargetInstance {
25    pub id: TargetId,
26    pub generation: u64,
27}
28
29/// Authentication specification for a managed target.
30///
31/// Equality preserves the specification: inheritance and an explicit override
32/// remain distinct even when they currently resolve to the same authentication.
33#[derive(Clone, Debug, Default, Eq, PartialEq)]
34pub enum TargetAuth {
35    /// Use the shared client configuration's authentication.
36    #[default]
37    Inherit,
38    /// Replace shared authentication. `Unauthenticated` explicitly disables it.
39    Override(Authentication),
40}
41
42impl TargetAuth {
43    pub(crate) fn resolve(&self, shared: &Authentication) -> Authentication {
44        match self {
45            Self::Inherit => shared.clone(),
46            Self::Override(auth) => auth.clone(),
47        }
48    }
49}
50
51/// Endpoint, address-family and authentication specification for one managed target.
52#[derive(Clone, Debug, Eq, PartialEq)]
53pub struct ManagedTargetConfig {
54    pub id: TargetId,
55    pub server_addr: String,
56    /// Override the shared address-family policy. `None` inherits it.
57    ///
58    /// The endpoint remains a hostname when supplied as one, so every new
59    /// generation resolves it again using the effective family policy.
60    pub address_family: Option<AddressFamily>,
61    pub auth: TargetAuth,
62}
63
64impl ManagedTargetConfig {
65    pub fn new(id: impl Into<TargetId>, server_addr: impl Into<String>) -> Self {
66        Self {
67            id: id.into(),
68            server_addr: server_addr.into(),
69            address_family: None,
70            auth: TargetAuth::Inherit,
71        }
72    }
73}
74
75/// Coordination mode for sends across active targets.
76#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
77pub enum ManagedPacing {
78    #[default]
79    Staggered,
80    Burst,
81}
82
83/// Policy controlling top-level natural completion.
84#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
85pub enum ManagedCompletionPolicy {
86    #[default]
87    FinishWhenQuiescent,
88    ExplicitStop,
89}
90
91/// Shared configuration for one managed task.
92#[derive(Clone, Debug, Eq, PartialEq)]
93pub struct ManagedClientConfig {
94    /// Reusable session/socket configuration shared across target generations.
95    /// Each target supplies its endpoint and resolves its family and authentication policies.
96    pub client: ClientConfig,
97    pub pacing: ManagedPacing,
98    pub completion: ManagedCompletionPolicy,
99    pub event_capacity: usize,
100    pub command_capacity: usize,
101    pub outcome_history_limit: usize,
102    pub max_live_target_generations: usize,
103    pub final_drain: Duration,
104}
105
106impl Default for ManagedClientConfig {
107    fn default() -> Self {
108        Self {
109            client: ClientConfig::default(),
110            pacing: ManagedPacing::default(),
111            completion: ManagedCompletionPolicy::default(),
112            event_capacity: DEFAULT_MANAGED_EVENT_CAPACITY,
113            command_capacity: DEFAULT_MANAGED_COMMAND_CAPACITY,
114            outcome_history_limit: DEFAULT_MANAGED_OUTCOME_HISTORY_LIMIT,
115            max_live_target_generations: DEFAULT_MANAGED_MAX_LIVE_TARGET_GENERATIONS,
116            final_drain: DEFAULT_MANAGED_FINAL_DRAIN,
117        }
118    }
119}
120
121/// Durable top-level lifecycle.
122#[derive(Clone, Copy, Debug, Eq, PartialEq)]
123pub enum ManagedLifecycle {
124    NotStarted,
125    Running,
126    Stopping,
127    Completed,
128    Failed,
129    Abandoned,
130}
131
132/// Durable lifecycle for one target generation.
133#[derive(Clone, Copy, Debug, Eq, PartialEq)]
134pub enum ManagedTargetLifecycle {
135    Pending,
136    Connecting,
137    Opening,
138    Active,
139    Draining,
140    Closing,
141    Terminal,
142}
143
144/// Status for one target generation.
145#[derive(Clone, Debug, Eq, PartialEq)]
146pub struct ManagedTargetStatus {
147    pub target: TargetInstance,
148    pub desired: bool,
149    pub lifecycle: ManagedTargetLifecycle,
150    /// Authoritative completed accounting for this target generation.
151    ///
152    /// Present if and only if `lifecycle` is [`ManagedTargetLifecycle::Terminal`].
153    /// This shares the terminal phase's outcome with [`ManagedEvent::TargetFinished`],
154    /// so terminal details remain durable even when presentation events are lost.
155    pub outcome: Option<Arc<ManagedTargetOutcome>>,
156    pub server_addr: Arc<str>,
157    pub remote: Option<SocketAddr>,
158}
159
160/// Authoritative immutable managed status snapshot.
161#[derive(Clone, Debug, Eq, PartialEq)]
162pub struct ManagedStatus {
163    pub lifecycle: ManagedLifecycle,
164    /// Whether the task durably observed an explicit stop request.
165    ///
166    /// This remains false until the task observes the request latch. Natural
167    /// stopping and terminality alone do not imply a stop request was observed.
168    pub stop_requested: bool,
169    pub applied_command_sequence: u64,
170    pub desired_target_count: usize,
171    pub connecting_target_count: usize,
172    pub opening_target_count: usize,
173    pub active_target_count: usize,
174    pub draining_target_count: usize,
175    pub closing_target_count: usize,
176    pub terminal_target_count: usize,
177    pub total_target_outcomes: u64,
178    pub successful_target_outcomes: u64,
179    pub failed_target_outcomes: u64,
180    pub peer_closed_target_outcomes: u64,
181    pub discarded_target_outcomes: u64,
182    pub targets: Arc<[ManagedTargetStatus]>,
183    pub recent_target_outcomes: Arc<[ManagedTargetOutcome]>,
184    /// Authoritative final accounting, present if and only if `lifecycle` is
185    /// [`ManagedLifecycle::Completed`] or [`ManagedLifecycle::Failed`].
186    ///
187    /// Shares the terminal task phase's outcome with [`ManagedEvent::Completed`]
188    /// or [`ManagedEvent::Failed`]. Abandoned deliberately has no outcome.
189    pub final_outcome: Option<Arc<ManagedOutcome>>,
190}
191
192/// Receiving half of authoritative durable latest-state observation.
193///
194/// The current snapshot is immediately readable. Updates may coalesce, and the
195/// final snapshot remains available after the sender closes at task termination.
196/// Use [`ManagedEventSubscription`] for lossy presentation events.
197pub type ManagedStatusSubscription = tokio::sync::watch::Receiver<Arc<ManagedStatus>>;
198
199/// Lossy presentation event emitted by the managed task.
200#[allow(clippy::large_enum_variant)]
201#[derive(Clone, Debug, Eq, PartialEq)]
202pub enum ManagedEvent {
203    Started,
204    TargetStateChanged {
205        target: TargetInstance,
206        lifecycle: ManagedTargetLifecycle,
207    },
208    Client {
209        target: TargetInstance,
210        event: ClientEvent,
211    },
212    TargetFinished {
213        outcome: Arc<ManagedTargetOutcome>,
214    },
215    Stopping,
216    Completed {
217        outcome: Arc<ManagedOutcome>,
218    },
219    Failed {
220        outcome: Arc<ManagedOutcome>,
221    },
222    Abandoned,
223}
224
225/// Receiving half of the lossy managed event stream.
226pub type ManagedEventSubscription = tokio::sync::broadcast::Receiver<ManagedEvent>;
227
228/// Immediate result of attempting to receive a managed presentation event.
229pub type ManagedEventTryRecvError = tokio::sync::broadcast::error::TryRecvError;
230
231/// Final authoritative outcome returned by the task.
232#[derive(Clone, Debug, Eq, PartialEq)]
233pub struct ManagedOutcome {
234    pub end_reason: ManagedEndReason,
235    pub applied_command_sequence: u64,
236    pub total_target_outcomes: u64,
237    pub successful_target_outcomes: u64,
238    pub failed_target_outcomes: u64,
239    pub peer_closed_target_outcomes: u64,
240    pub discarded_target_outcomes: u64,
241    pub recent_target_outcomes: Arc<[ManagedTargetOutcome]>,
242}
243
244/// Why the managed task ended.
245#[derive(Clone, Debug, Eq, PartialEq)]
246pub enum ManagedEndReason {
247    TargetsComplete,
248    StopRequested,
249    DriverFailed(ManagedDriverFailure),
250}
251
252/// Durable outcome for one target generation.
253///
254/// These counters are durable: unlike [`ManagedEvent`], which is a lossy
255/// presentation stream, they account for everything the target observed.
256#[derive(Clone, Debug, Eq, PartialEq)]
257pub struct ManagedTargetOutcome {
258    pub target: TargetInstance,
259    pub server_addr: Arc<str>,
260    pub remote: Option<SocketAddr>,
261    pub end_reason: ManagedTargetEndReason,
262    /// Probes the underlying client reported as sent.
263    pub packets_sent: u64,
264    /// Unique echo replies that arrived in order and before their probe timed
265    /// out.
266    ///
267    /// A unique reply that arrives behind an already-received sequence, or
268    /// after its own probe timed out, is counted by `late` instead.
269    pub replies_received: u64,
270    /// Replies for a sequence that had already been answered.
271    pub duplicates: u64,
272    /// Late replies: those arriving behind an already-received sequence
273    /// (reordering) or after their own probe timed out.
274    ///
275    /// This counts every late reply, whether or not the client could still
276    /// match it to retained send state.
277    pub late: u64,
278    /// Warning events observed for this target.
279    pub warning_events: u64,
280    pub cleanup_failure: Option<ManagedTargetFailure>,
281}
282
283/// Why one target generation ended.
284#[derive(Clone, Debug, Eq, PartialEq)]
285pub enum ManagedTargetEndReason {
286    TestComplete,
287    NoTestComplete,
288    PeerClosed,
289    Removed,
290    Replaced,
291    Stopped,
292    Failed(ManagedTargetFailure),
293}
294
295/// Phase in which a target-local failure occurred.
296#[derive(Clone, Copy, Debug, Eq, PartialEq)]
297pub enum ManagedTargetFailurePhase {
298    Connecting,
299    Opening,
300    Sending,
301    Receiving,
302    Timing,
303    Closing,
304}
305
306impl fmt::Display for ManagedTargetFailurePhase {
307    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
308        let text = match self {
309            Self::Connecting => "connecting",
310            Self::Opening => "opening",
311            Self::Sending => "sending",
312            Self::Receiving => "receiving",
313            Self::Timing => "timing",
314            Self::Closing => "closing",
315        };
316        f.write_str(text)
317    }
318}
319
320/// Stable category for a target-local failure.
321#[derive(Clone, Copy, Debug, Eq, PartialEq)]
322pub enum ManagedTargetFailureKind {
323    Resolve,
324    Socket,
325    SocketOption,
326    Protocol,
327    Timeout,
328    Configuration,
329    ResourceExhausted,
330    InvalidState,
331    /// A failure that deliberately has no more specific category.
332    ///
333    /// This is a classification decision, not a fallback: every current
334    /// [`ClientError`] variant is classified explicitly, so nothing reaches
335    /// this category by default.
336    Other,
337}
338
339impl fmt::Display for ManagedTargetFailureKind {
340    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
341        let text = match self {
342            Self::Resolve => "resolve",
343            Self::Socket => "socket",
344            Self::SocketOption => "socket option",
345            Self::Protocol => "protocol",
346            Self::Timeout => "timeout",
347            Self::Configuration => "configuration",
348            Self::ResourceExhausted => "resource exhausted",
349            Self::InvalidState => "invalid state",
350            Self::Other => "other",
351        };
352        f.write_str(text)
353    }
354}
355
356/// Durable target-local failure details.
357#[derive(Clone, Debug, Eq, PartialEq)]
358pub struct ManagedTargetFailure {
359    pub phase: ManagedTargetFailurePhase,
360    pub kind: ManagedTargetFailureKind,
361    pub message: Arc<str>,
362}
363
364/// Configuration failure detected before any runtime work begins.
365#[derive(Debug, Error)]
366pub enum ManagedConfigError {
367    #[error("FinishWhenQuiescent requires at least one initial target")]
368    EmptyInitialTargets,
369    #[error("duplicate managed target id {id}")]
370    DuplicateTargetId { id: TargetId },
371    #[error("configured {configured} targets exceeds live-generation limit {limit}")]
372    TooManyTargets { configured: usize, limit: usize },
373    #[error("managed event capacity must be greater than zero")]
374    ZeroEventCapacity,
375    #[error("managed event capacity {configured} exceeds Tokio's maximum {maximum}")]
376    EventCapacityTooLarge { configured: usize, maximum: usize },
377    #[error("managed command capacity must be greater than zero")]
378    ZeroCommandCapacity,
379    #[error("managed command capacity {configured} exceeds Tokio's maximum {maximum}")]
380    CommandCapacityTooLarge { configured: usize, maximum: usize },
381    #[error("managed live-generation limit must be greater than zero")]
382    ZeroLiveGenerationLimit,
383    #[error("managed final drain {duration:?} cannot be scheduled from the current instant")]
384    UnschedulableFinalDrain { duration: Duration },
385    #[error("invalid managed target {id}: {source}")]
386    InvalidTarget {
387        id: TargetId,
388        #[source]
389        source: ClientError,
390    },
391    #[error("managed target generation space is exhausted")]
392    GenerationExhausted,
393}
394
395/// Immediate failure to submit a managed target update.
396#[derive(Clone, Debug, Eq, Error, PartialEq)]
397pub enum ManagedCommandError {
398    #[error("managed target updates are no longer accepted")]
399    Stopping,
400    #[error("managed target update contains {configured} targets but the limit is {limit}")]
401    TooManyTargets { configured: usize, limit: usize },
402    #[error("managed command queue is full")]
403    QueueFull,
404    #[error("managed command receiver is closed")]
405    DriverClosed,
406}
407
408/// Failure while applying an accepted managed target update.
409#[derive(Debug, Error)]
410pub enum ManagedCommandApplyError {
411    #[error("managed target updates are no longer accepted")]
412    Stopping,
413    #[error("duplicate target id {id}")]
414    DuplicateTargetId { id: TargetId },
415    #[error("invalid configuration for target {id}: {source}")]
416    InvalidTarget {
417        id: TargetId,
418        #[source]
419        source: ClientError,
420    },
421    #[error("target update requires {required} live generations but limit is {limit}")]
422    LiveGenerationLimitExceeded { required: usize, limit: usize },
423    #[error("managed target generation space is exhausted")]
424    GenerationExhausted,
425    #[error("managed command sequence is exhausted")]
426    CommandSequenceExhausted,
427    #[error("managed task ended before command acknowledgement")]
428    AcknowledgementDisconnected,
429    #[error("managed driver failed before command application: {0}")]
430    DriverFailed(ManagedDriverFailure),
431}
432
433#[derive(Clone, Debug, Eq, PartialEq)]
434pub struct ManagedCommandAcknowledgement {
435    pub sequence: u64,
436    pub status: Arc<ManagedStatus>,
437}
438
439/// Failure to create a new event subscription.
440#[derive(Clone, Debug, Eq, Error, PartialEq)]
441pub enum ManagedSubscribeError {
442    #[error("managed event stream is closed")]
443    Closed,
444}
445
446/// Whole-driver failure.
447#[derive(Clone, Debug, Eq, Error, PartialEq)]
448pub enum ManagedDriverFailure {
449    #[error("ManagedClientTask requires a current Tokio runtime")]
450    NoTokioRuntime,
451    #[error("managed driver exhausted resources for {operation}")]
452    ResourceExhausted { operation: &'static str },
453    #[error("managed driver invariant failed: {message}")]
454    Internal { message: Arc<str> },
455}
456
457/// Map a [`ClientError`] onto its durable managed failure classification.
458///
459/// The match is deliberately exhaustive over every [`ClientError`] variant. A
460/// new variant must fail to compile here until its real classification is
461/// chosen, rather than silently degrading to
462/// [`ManagedTargetFailureKind::Other`].
463pub(crate) fn classify_client_error(
464    phase: ManagedTargetFailurePhase,
465    error: &ClientError,
466) -> ManagedTargetFailure {
467    let kind = match error {
468        ClientError::Resolve { .. } => ManagedTargetFailureKind::Resolve,
469        ClientError::Socket(_) => ManagedTargetFailureKind::Socket,
470        ClientError::SocketOption { .. } | ClientError::ReadTimeoutRestore { .. } => {
471            ManagedTargetFailureKind::SocketOption
472        }
473        ClientError::Protocol(_)
474        | ClientError::ProtocolVersionMismatch { .. }
475        | ClientError::ZeroToken
476        | ClientError::UnexpectedNoTestReply
477        | ClientError::NonZeroNoTestToken { .. }
478        | ClientError::ServerRejected
479        | ClientError::NegotiationRejected { .. } => ManagedTargetFailureKind::Protocol,
480        ClientError::OpenTimeout => ManagedTargetFailureKind::Timeout,
481        ClientError::InvalidConfig { .. }
482        | ClientError::OpenTimeoutTooSmall { .. }
483        | ClientError::NoOpenTimeouts => ManagedTargetFailureKind::Configuration,
484        ClientError::AllocationFailed { .. }
485        | ClientError::CounterOverflow { .. }
486        | ClientError::DurationOverflow
487        | ClientError::PendingLimitExceeded { .. } => ManagedTargetFailureKind::ResourceExhausted,
488        ClientError::DatagramLengthMismatch { .. } => ManagedTargetFailureKind::Socket,
489        #[cfg(feature = "tokio")]
490        ClientError::NoTokioRuntime => ManagedTargetFailureKind::InvalidState,
491        ClientError::NotOpen
492        | ClientError::AlreadyOpen
493        | ClientError::AlreadyCompleted
494        | ClientError::AlreadyClosed
495        | ClientError::StalePreparedProbe { .. }
496        | ClientError::PendingSequenceCollision { .. } => ManagedTargetFailureKind::InvalidState,
497    };
498    ManagedTargetFailure {
499        phase,
500        kind,
501        message: Arc::from(error.to_string()),
502    }
503}
irtt_client/managed/model.rs

irtt_client/managed/
model.rs
