diff --git a/crates/irtt-client/src/managed/cancellation.rs b/crates/irtt-client/src/managed/cancellation.rs index b477290..4eca118 100644 --- a/crates/irtt-client/src/managed/cancellation.rs +++ b/crates/irtt-client/src/managed/cancellation.rs @@ -17,6 +17,7 @@ impl CancellationToken { self.cancelled.store(true, Ordering::SeqCst); } + #[must_use] pub fn is_cancelled(&self) -> bool { self.cancelled.load(Ordering::SeqCst) } @@ -32,12 +33,14 @@ mod tests { } #[test] - fn cancel_is_visible_through_clones() { + fn cancel_is_idempotent_and_visible_through_clones() { let token = CancellationToken::new(); let clone = token.clone(); + token.cancel(); token.cancel(); + assert!(token.is_cancelled()); assert!(clone.is_cancelled()); } } diff --git a/crates/irtt-client/src/managed/hub.rs b/crates/irtt-client/src/managed/hub.rs index 1aec013..3e9a6b7 100644 --- a/crates/irtt-client/src/managed/hub.rs +++ b/crates/irtt-client/src/managed/hub.rs @@ -23,10 +23,17 @@ impl Default for SubscriberConfig { } } +/// Overflow behavior for a bounded event subscription queue. +/// +/// These policies are applied independently per subscriber when that +/// subscriber's queue is full. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SubscriberOverflow { + /// Leave the existing queue unchanged and discard the newly published event. DropNewest, + /// Remove the oldest queued event and enqueue the newly published event. DropOldest, + /// Disconnect the subscriber and clear its queue. Disconnect, } @@ -41,6 +48,12 @@ struct HubInner { subscribers: Mutex>>, } +/// Handle for receiving managed client events. +/// +/// Each subscription has its own bounded queue configured by +/// [`SubscriberConfig`]. The configured [`SubscriberOverflow`] policy applies +/// when that queue is full. Dropping the handle unregisters the subscriber. +#[must_use = "dropping the subscription unregisters it"] #[derive(Debug)] pub struct EventSubscription { id: u64, @@ -115,6 +128,8 @@ impl EventHub { let mut disconnected = Vec::new(); for (id, subscriber) in subscribers { + // Events are cloned per subscriber; bounded queues keep slow + // consumers from causing unbounded memory growth. if !subscriber.publish(event.clone()) { disconnected.push(id); } @@ -166,6 +181,11 @@ impl Default for EventHub { } impl EventSubscription { + /// Block until the next queued event is available. + /// + /// If the subscription is disconnected, this returns + /// [`EventSubscriptionError::Disconnected`] after any already queued events + /// have been drained. pub fn recv(&self) -> Result { let mut state = self.inner.state.lock().expect("subscriber mutex poisoned"); loop { @@ -183,6 +203,12 @@ impl EventSubscription { } } + /// Try to receive one queued event without blocking. + /// + /// Returns `Ok(None)` when the subscription is still connected but no event + /// is queued. If the subscription is disconnected, this returns + /// [`EventSubscriptionError::Disconnected`] after any already queued events + /// have been drained. pub fn try_recv(&self) -> Result, EventSubscriptionError> { let mut state = self.inner.state.lock().expect("subscriber mutex poisoned"); if let Some(event) = state.queue.pop_front() { diff --git a/crates/irtt-client/src/managed/runner.rs b/crates/irtt-client/src/managed/runner.rs index 123baa0..b79014d 100644 --- a/crates/irtt-client/src/managed/runner.rs +++ b/crates/irtt-client/src/managed/runner.rs @@ -25,6 +25,13 @@ const MAX_SLEEP: Duration = Duration::from_millis(20); #[derive(Debug)] pub struct ManagedClient; +/// Running managed client session. +/// +/// [`ManagedClientSession::join`] waits for the worker and returns the session +/// outcome or client error. Dropping the session requests cooperative +/// cancellation; callers that need the outcome should explicitly join instead +/// of relying on drop. +#[must_use = "dropping the session cancels the managed client; call join() to wait for completion"] #[derive(Debug)] pub struct ManagedClientSession { hub: EventHub, @@ -32,6 +39,12 @@ pub struct ManagedClientSession { worker: Option>>, } +/// Outcome returned by a completed managed client session. +/// +/// These are runner-level lifecycle counters, not statistical summaries. Use +/// `irtt-stats` with emitted `ClientEvent` values for RTT, loss, IPDV, and +/// related summaries. +#[must_use = "managed session outcomes contain completion status and counters"] #[derive(Debug, Clone, PartialEq, Eq)] pub struct SessionOutcome { pub end_reason: SessionEndReason, @@ -111,7 +124,10 @@ impl ManagedClientSession { } pub fn join(mut self) -> Result { - let worker = self.worker.take().expect("worker handle must be present"); + let worker = self + .worker + .take() + .expect("ManagedClientSession invariant violated: worker handle missing before join"); match worker.join() { Ok(outcome) => { self.hub.disconnect_all();