fix(cli): harden managed frontend cutover

crates/irtt-cli/src/cmd/client/run.rs

@@ -26,8 +26,8 @@ use crate::shared::client::expected_probe_count;
 use crate::shared::client::{
     is_shutdown_requested,
     session::{
-        peer_close_run_error, request_managed_stop_for_peer_close, request_managed_stop_once,
-        should_print_final_summary,
+        drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close,
+        request_managed_stop_once, should_print_final_summary, ManagedDrainState,
     },
 };
 #[cfg(feature = "stats")]
@@ -116,7 +116,6 @@ pub fn run_stream(
     let mut dropped_events = 0_u64;
     let mut interrupted = false;
     let mut stop_requested = false;
-    let mut peer_close_requested_stop = false;
     let mut subscription_closed = false;
     loop {
         if is_shutdown_requested(shutdown_requested) {
@@ -129,12 +128,11 @@ pub fn run_stream(
             continuous,
             interrupted,
             handle.status().peer_closed_target_outcomes,
-            &mut peer_close_requested_stop,
             &mut stop_requested,
         ) {
             drop(handle.stop());
         }
-        drain_events(
+        let drain_state = drain_events(
             &mut events,
             &mut stream_output,
             #[cfg(feature = "stats")]
@@ -142,35 +140,20 @@ pub fn run_stream(
             &mut terminal_targets,
             &mut dropped_events,
         )?;
+        match drain_state {
+            ManagedDrainState::Empty => thread::sleep(MANAGED_EVENT_WAIT_SLICE),
+            ManagedDrainState::Closed => subscription_closed = true,
+            ManagedDrainState::BudgetExhausted => {}
+        }
         if handle.status().final_outcome.is_some() || subscription_closed {
             break;
         }
-        match events.try_recv() {
-            Ok(event) => process_event(
-                event,
-                &mut stream_output,
-                #[cfg(feature = "stats")]
-                &mut stats,
-                &mut terminal_targets,
-            )?,
-            Err(ManagedEventTryRecvError::Empty) => thread::sleep(MANAGED_EVENT_WAIT_SLICE),
-            Err(ManagedEventTryRecvError::Lagged(count)) => dropped_events += count,
-            Err(ManagedEventTryRecvError::Closed) => subscription_closed = true,
-        }
     }
     if interrupted {
         eprintln!("interrupted, closing managed run...");
     }
-    drain_events(
-        &mut events,
-        &mut stream_output,
-        #[cfg(feature = "stats")]
-        &mut stats,
-        &mut terminal_targets,
-        &mut dropped_events,
-    )?;
     let outcome = owner.join()?;
-    drain_events(
+    drain_final_events(
         &mut events,
         &mut stream_output,
         #[cfg(feature = "stats")]
@@ -275,6 +258,24 @@ fn drain_events<W: Write>(
     #[cfg(feature = "stats")] stats: &mut BTreeMap<String, StatsCollector>,
     terminal_targets: &mut HashSet<TargetInstance>,
     dropped_events: &mut u64,
+) -> io::Result<ManagedDrainState> {
+    drain_managed_events(events, dropped_events, |event| {
+        process_event(
+            event,
+            stream_output,
+            #[cfg(feature = "stats")]
+            stats,
+            terminal_targets,
+        )
+    })
+}
+
+fn drain_final_events<W: Write>(
+    events: &mut ManagedEventSubscription,
+    stream_output: &mut StreamOutput<'_, W>,
+    #[cfg(feature = "stats")] stats: &mut BTreeMap<String, StatsCollector>,
+    terminal_targets: &mut HashSet<TargetInstance>,
+    dropped_events: &mut u64,
 ) -> io::Result<()> {
     loop {
         match events.try_recv() {
@@ -286,19 +287,35 @@ fn drain_events<W: Write>(
                 terminal_targets,
             )?,
             Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break,
-            Err(ManagedEventTryRecvError::Lagged(count)) => *dropped_events += count,
+            Err(ManagedEventTryRecvError::Lagged(count)) => {
+                *dropped_events = dropped_events.saturating_add(count);
+            }
         }
     }
     Ok(())
 }
 
 fn report_target_failure(target: &ManagedTargetOutcome) {
+    for message in target_failure_messages(target) {
+        eprintln!("{message}");
+    }
+}
+
+fn target_failure_messages(target: &ManagedTargetOutcome) -> Vec<String> {
+    let mut messages = Vec::with_capacity(2);
     if let ManagedTargetEndReason::Failed(failure) = &target.end_reason {
-        eprintln!(
+        messages.push(format!(
             "irtt-rs: target {} failed ({:?} {:?}): {}",
             target.target.id, failure.phase, failure.kind, failure.message
-        );
+        ));
     }
+    if let Some(failure) = &target.cleanup_failure {
+        messages.push(format!(
+            "irtt-rs: target {} cleanup failed ({:?} {:?}): {}",
+            target.target.id, failure.phase, failure.kind, failure.message
+        ));
+    }
+    messages
 }
 fn dropped_event_warning(dropped_events: u64) -> Option<String> {
     (dropped_events > 0).then(|| format!("irtt-rs: warning: dropped {dropped_events} managed run event{}; output and statistics may be incomplete", if dropped_events == 1 { "" } else { "s" }))
@@ -419,14 +436,55 @@ fn finite_stats_memory_warning(args: &ClientArgs) -> Option<String> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use irtt_client::managed::{
+        ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetFailurePhase, TargetId,
+    };
+    use std::sync::Arc;
+
     #[test]
-    fn lagged_events_are_counted_exactly() {
-        let mut dropped = 0;
-        dropped += 7;
-        dropped += 3;
+    fn lagged_events_are_counted_saturatingly() {
+        let mut dropped = 0_u64;
+        dropped = dropped.saturating_add(7);
+        dropped = dropped.saturating_add(3);
         assert_eq!(dropped, 10);
+        assert_eq!(u64::MAX.saturating_add(1), u64::MAX);
         assert!(dropped_event_warning(dropped)
             .unwrap()
             .contains("dropped 10 managed run events"));
     }
+
+    #[test]
+    fn target_failure_diagnostics_include_primary_and_cleanup_failures() {
+        let target = ManagedTargetOutcome {
+            target: TargetInstance {
+                id: TargetId::from("edge"),
+                generation: 1,
+            },
+            server_addr: Arc::from("127.0.0.1:2112"),
+            remote: None,
+            end_reason: ManagedTargetEndReason::Failed(ManagedTargetFailure {
+                phase: ManagedTargetFailurePhase::Receiving,
+                kind: ManagedTargetFailureKind::Protocol,
+                message: "primary failure".into(),
+            }),
+            packets_sent: 0,
+            replies_received: 0,
+            duplicates: 0,
+            late: 0,
+            warning_events: 0,
+            cleanup_failure: Some(ManagedTargetFailure {
+                phase: ManagedTargetFailurePhase::Closing,
+                kind: ManagedTargetFailureKind::Socket,
+                message: "cleanup failure".into(),
+            }),
+        };
+
+        assert_eq!(
+            target_failure_messages(&target),
+            [
+                "irtt-rs: target edge failed (Receiving Protocol): primary failure",
+                "irtt-rs: target edge cleanup failed (Closing Socket): cleanup failure",
+            ]
+        );
+    }
 }

crates/irtt-cli/src/cmd/tui/run.rs

@@ -12,7 +12,8 @@ use crate::{
     shared::client::{
         is_shutdown_requested,
         session::{
-            peer_close_run_error, request_managed_stop_for_peer_close, request_managed_stop_once,
+            drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close,
+            request_managed_stop_once, ManagedDrainState,
         },
     },
 };
@@ -72,7 +73,6 @@ pub fn run_tui(
     let handle = owner.handle();
     let mut interrupted = false;
     let mut stop_requested = false;
-    let mut peer_close_requested_stop = false;
     let mut terminal_targets = HashSet::new();
     let mut dropped_events = 0;
     let mut subscription_closed = false;
@@ -96,28 +96,26 @@ pub fn run_tui(
             continuous,
             interrupted,
             handle.status().peer_closed_target_outcomes,
-            &mut peer_close_requested_stop,
             &mut stop_requested,
         ) {
             drop(handle.stop());
         }
-        drain_tui_events(
+        let drain_state = drain_tui_events(
             &mut events,
             &mut state,
             &mut terminal_targets,
             &mut dropped_events,
         );
+        match drain_state {
+            ManagedDrainState::Empty => {
+                thread::sleep(managed_tui_wait_duration(&next_render, state.paused));
+            }
+            ManagedDrainState::Closed => subscription_closed = true,
+            ManagedDrainState::BudgetExhausted => {}
+        }
         if handle.status().final_outcome.is_some() || subscription_closed {
             break;
         }
-        match events.try_recv() {
-            Ok(event) => process_tui_event(event, &mut state, &mut terminal_targets),
-            Err(ManagedEventTryRecvError::Empty) => {
-                thread::sleep(managed_tui_wait_duration(&next_render, state.paused))
-            }
-            Err(ManagedEventTryRecvError::Lagged(count)) => dropped_events += count,
-            Err(ManagedEventTryRecvError::Closed) => subscription_closed = true,
-        }
         render_if_due(&mut terminal, &state, &mut next_render, false)?;
     }
     if interrupted {
@@ -126,14 +124,8 @@ pub fn run_tui(
     }
     state.set_status(TuiStatus::Closing);
     render_if_due(&mut terminal, &state, &mut next_render, true)?;
-    drain_tui_events(
-        &mut events,
-        &mut state,
-        &mut terminal_targets,
-        &mut dropped_events,
-    );
     let outcome = owner.join()?;
-    drain_tui_events(
+    drain_final_tui_events(
         &mut events,
         &mut state,
         &mut terminal_targets,
@@ -198,12 +190,27 @@ fn drain_tui_events(
     state: &mut TuiState,
     terminal_targets: &mut HashSet<TargetInstance>,
     dropped_events: &mut u64,
+) -> ManagedDrainState {
+    drain_managed_events(events, dropped_events, |event| {
+        process_tui_event(event, state, terminal_targets);
+        Ok::<(), std::convert::Infallible>(())
+    })
+    .expect("processing TUI managed events is infallible")
+}
+
+fn drain_final_tui_events(
+    events: &mut ManagedEventSubscription,
+    state: &mut TuiState,
+    terminal_targets: &mut HashSet<TargetInstance>,
+    dropped_events: &mut u64,
 ) {
     loop {
         match events.try_recv() {
             Ok(event) => process_tui_event(event, state, terminal_targets),
             Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break,
-            Err(ManagedEventTryRecvError::Lagged(count)) => *dropped_events += count,
+            Err(ManagedEventTryRecvError::Lagged(count)) => {
+                *dropped_events = dropped_events.saturating_add(count);
+            }
         }
     }
 }

crates/irtt-cli/src/cmd/tui/ui.rs

@@ -170,7 +170,7 @@ impl TuiState {
             idx
         };
 
-        let (status, recent, warning) = match &outcome.end_reason {
+        let (status, recent, primary_warning) = match &outcome.end_reason {
             ManagedTargetEndReason::TestComplete => {
                 (TargetStatus::Closed, "test completed".to_owned(), None)
             }
@@ -203,6 +203,19 @@ impl TuiState {
             }
         };
 
+        let cleanup_warning = outcome.cleanup_failure.as_ref().map(|failure| {
+            format!(
+                "cleanup failed ({:?} {:?}): {}",
+                failure.phase, failure.kind, failure.message
+            )
+        });
+        let warning = match (primary_warning, cleanup_warning.as_ref()) {
+            (Some(primary), Some(cleanup)) => Some(format!("{primary}; {cleanup}")),
+            (Some(primary), None) => Some(primary),
+            (None, Some(cleanup)) => Some(cleanup.clone()),
+            (None, None) => None,
+        };
+
         if let Some(target) = self.targets.get_mut(idx) {
             target.remote = outcome.remote.map(|remote| remote.to_string());
             target.status = status;
@@ -214,6 +227,9 @@ impl TuiState {
             self.last_warning = Some(warning);
         }
         self.push_event(format!("{label}: {recent}"));
+        if let Some(cleanup_warning) = cleanup_warning {
+            self.push_event(format!("{label}: {cleanup_warning}"));
+        }
 
         if self
             .targets
@@ -2019,15 +2035,18 @@ fn duration_ns(value: Duration) -> i128 {
     i128::try_from(value.as_nanos()).unwrap_or(i128::MAX)
 }
 
-#[cfg(any())]
+#[cfg(test)]
 mod tests {
     use super::*;
+    use irtt_client::managed::{
+        ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetFailurePhase, TargetId,
+    };
     use irtt_client::{
-        ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, ServerTiming, TargetId,
-        WarningKind,
+        ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, ServerTiming, WarningKind,
     };
     use std::{
         net::{IpAddr, Ipv4Addr, SocketAddr},
+        sync::Arc,
         time::UNIX_EPOCH,
     };
 
@@ -2131,23 +2150,25 @@ mod tests {
         &state.targets[0]
     }
 
-    fn target_event(label: &str, event: ClientEvent) -> UiEventScope {
-        UiEventScope {
-            target: TargetId::from(label),
-            event,
+    fn target_instance(label: &str) -> TargetInstance {
+        TargetInstance {
+            id: TargetId::from(label),
+            generation: 1,
         }
     }
 
     fn target_outcome(label: &str, end_reason: ManagedTargetEndReason) -> ManagedTargetOutcome {
         ManagedTargetOutcome {
-            id: TargetId::from(label),
-            remote: remote(),
+            target: target_instance(label),
+            server_addr: Arc::from("127.0.0.1:2112"),
+            remote: Some(remote()),
             end_reason,
             packets_sent: 0,
             replies_received: 0,
             duplicates: 0,
             late: 0,
             warning_events: 0,
+            cleanup_failure: None,
         }
     }
 
@@ -2165,7 +2186,8 @@ mod tests {
     #[test]
     fn session_started_sets_remote_session_and_running_status() {
         let mut state = TuiState::default();
-        state.process_event(&ClientEvent::SessionStarted {
+        let target = target_instance("target");
+        let event = ClientEvent::SessionStarted {
             remote: remote(),
             token: 0xabc,
             negotiated: NegotiatedParams {
@@ -2173,7 +2195,8 @@ mod tests {
                 restrictions: Vec::new(),
             },
             at: ts(Duration::ZERO),
-        });
+        };
+        state.process_target_event(&target, &event);
 
         let target = primary_target(&state);
         assert_eq!(target.remote.as_deref(), Some("127.0.0.1:2112"));
@@ -2185,8 +2208,10 @@ mod tests {
     #[test]
     fn echo_reply_appends_bounded_primary_history() {
         let mut state = TuiState::default();
+        let target = target_instance("target");
         for seq in 0..(HISTORY_LIMIT as u32 + 3) {
-            state.process_event(&reply(seq, i128::from(seq) * 1_000));
+            let event = reply(seq, i128::from(seq) * 1_000);
+            state.process_target_event(&target, &event);
         }
 
         let target = primary_target(&state);
@@ -2205,13 +2230,15 @@ mod tests {
     #[test]
     fn duplicate_and_late_replies_do_not_append_primary_history() {
         let mut state = TuiState::default();
-        state.process_event(&ClientEvent::DuplicateReply {
+        let target = target_instance("target");
+        let duplicate = ClientEvent::DuplicateReply {
             seq: 7,
             remote: remote(),
             received_at: ts(Duration::from_secs(1)),
             bytes: 64,
-        });
-        state.process_event(&ClientEvent::LateReply {
+        };
+        state.process_target_event(&target, &duplicate);
+        let late = ClientEvent::LateReply {
             seq: 8,
             highest_seen: 9,
             remote: remote(),
@@ -2223,7 +2250,8 @@ mod tests {
             received_stats: None,
             bytes: 64,
             packet_meta: PacketMeta::default(),
-        });
+        };
+        state.process_target_event(&target, &late);
 
         let target = primary_target(&state);
         assert!(target.graph_history.is_empty());
@@ -2250,7 +2278,9 @@ mod tests {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
 
-        state.process_target_event(&target_event("b", reply(11, 2_500_000)));
+        let target = target_instance("b");
+        let event = reply(11, 2_500_000);
+        state.process_target_event(&target, &event);
 
         assert!(state.targets[0].graph_history.is_empty());
         assert_eq!(state.targets[0].stats.snapshot().packets.unique_replies, 0);
@@ -2268,31 +2298,29 @@ mod tests {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
 
-        state.process_target_event(&target_event(
-            "a",
-            ClientEvent::DuplicateReply {
-                seq: 7,
-                remote: remote(),
-                received_at: ts(Duration::from_secs(1)),
-                bytes: 64,
-            },
-        ));
-        state.process_target_event(&target_event(
-            "b",
-            ClientEvent::LateReply {
-                seq: 8,
-                highest_seen: 9,
-                remote: remote(),
-                sent_at: Some(ts(Duration::from_secs(1))),
-                received_at: ts(Duration::from_secs(2)),
-                rtt: Some(rtt(2_000_000)),
-                server_timing: None,
-                one_way: None,
-                received_stats: None,
-                bytes: 64,
-                packet_meta: PacketMeta::default(),
-            },
-        ));
+        let a = target_instance("a");
+        let duplicate = ClientEvent::DuplicateReply {
+            seq: 7,
+            remote: remote(),
+            received_at: ts(Duration::from_secs(1)),
+            bytes: 64,
+        };
+        state.process_target_event(&a, &duplicate);
+        let b = target_instance("b");
+        let late = ClientEvent::LateReply {
+            seq: 8,
+            highest_seen: 9,
+            remote: remote(),
+            sent_at: Some(ts(Duration::from_secs(1))),
+            received_at: ts(Duration::from_secs(2)),
+            rtt: Some(rtt(2_000_000)),
+            server_timing: None,
+            one_way: None,
+            received_stats: None,
+            bytes: 64,
+            packet_meta: PacketMeta::default(),
+        };
+        state.process_target_event(&b, &late);
 
         assert!(state.targets[0].graph_history.is_empty());
         assert!(state.targets[1].graph_history.is_empty());
@@ -2306,33 +2334,29 @@ mod tests {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
 
-        state.process_target_event(&target_event(
-            "a",
-            ClientEvent::EchoLoss {
-                seq: 3,
-                sent_at: ts(Duration::from_millis(3)),
-                timeout_at: Instant::now(),
+        let a = target_instance("a");
+        let loss = ClientEvent::EchoLoss {
+            seq: 3,
+            sent_at: ts(Duration::from_millis(3)),
+            timeout_at: Instant::now(),
+        };
+        state.process_target_event(&a, &loss);
+        let b = target_instance("b");
+        let warning = ClientEvent::Warning {
+            kind: WarningKind::WrongToken,
+            message: "wrong token".to_owned(),
+            at: ts(Duration::ZERO),
+        };
+        state.process_target_event(&b, &warning);
+        let no_test = ClientEvent::NoTestCompleted {
+            remote: remote(),
+            negotiated: NegotiatedParams {
+                params: irtt_proto::Params::default(),
+                restrictions: Vec::new(),
             },
-        ));
-        state.process_target_event(&target_event(
-            "b",
-            ClientEvent::Warning {
-                kind: WarningKind::WrongToken,
-                message: "wrong token".to_owned(),
-                at: ts(Duration::ZERO),
-            },
-        ));
-        state.process_target_event(&target_event(
-            "a",
-            ClientEvent::NoTestCompleted {
-                remote: remote(),
-                negotiated: NegotiatedParams {
-                    params: irtt_proto::Params::default(),
-                    restrictions: Vec::new(),
-                },
-                at: ts(Duration::ZERO),
-            },
-        ));
+            at: ts(Duration::ZERO),
+        };
+        state.process_target_event(&a, &no_test);
 
         assert_eq!(state.targets[0].stats.snapshot().events.loss_events, 1);
         assert_eq!(state.targets[0].status, TargetStatus::NoTest);
@@ -2349,30 +2373,28 @@ mod tests {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
 
-        state.process_target_event(&target_event(
-            "a",
-            ClientEvent::NoTestCompleted {
-                remote: remote(),
-                negotiated: NegotiatedParams {
-                    params: irtt_proto::Params::default(),
-                    restrictions: Vec::new(),
-                },
-                at: ts(Duration::ZERO),
+        let a = target_instance("a");
+        let no_test = ClientEvent::NoTestCompleted {
+            remote: remote(),
+            negotiated: NegotiatedParams {
+                params: irtt_proto::Params::default(),
+                restrictions: Vec::new(),
             },
-        ));
+            at: ts(Duration::ZERO),
+        };
+        state.process_target_event(&a, &no_test);
 
         assert_eq!(state.targets[0].status, TargetStatus::NoTest);
         assert_eq!(state.targets[1].status, TargetStatus::Opening);
         assert_eq!(state.status, TuiStatus::Running);
 
-        state.process_target_event(&target_event(
-            "b",
-            ClientEvent::SessionClosed {
-                remote: remote(),
-                token: 0xabc,
-                at: ts(Duration::ZERO),
-            },
-        ));
+        let b = target_instance("b");
+        let closed = ClientEvent::SessionClosed {
+            remote: remote(),
+            token: 0xabc,
+            at: ts(Duration::ZERO),
+        };
+        state.process_target_event(&b, &closed);
 
         assert_eq!(state.targets[1].status, TargetStatus::Closed);
         assert_eq!(state.status, TuiStatus::Complete);
@@ -2382,14 +2404,15 @@ mod tests {
     fn terminal_failure_marks_the_correct_target_failed() {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
-        let failure = irtt_client::ManagedTargetFailure {
-            kind: irtt_client::ManagedTargetFailureKind::OpeningTimeout,
-            message: "all open requests timed out".to_owned(),
+        let failure = ManagedTargetFailure {
+            phase: ManagedTargetFailurePhase::Opening,
+            kind: ManagedTargetFailureKind::Timeout,
+            message: "all open requests timed out".into(),
         };
 
         state.process_target_outcome(&target_outcome(
             "b",
-            ManagedTargetEndReason::OpenFailed(failure),
+            ManagedTargetEndReason::Failed(failure),
         ));
 
         assert_eq!(state.targets[0].status, TargetStatus::Opening);
@@ -2397,7 +2420,7 @@ mod tests {
         assert!(state.targets[1]
             .last_warning
             .as_deref()
-            .is_some_and(|warning| warning.contains("opening timeout")));
+            .is_some_and(|warning| warning.contains("Timeout")));
     }
 
     #[test]
@@ -2407,9 +2430,10 @@ mod tests {
         state.process_target_outcome(&target_outcome("a", ManagedTargetEndReason::PeerClosed));
         state.process_target_outcome(&target_outcome(
             "b",
-            ManagedTargetEndReason::Failed(irtt_client::ManagedTargetFailure {
-                kind: irtt_client::ManagedTargetFailureKind::RuntimeProtocol,
-                message: "pending probe limit exceeded".to_owned(),
+            ManagedTargetEndReason::Failed(ManagedTargetFailure {
+                phase: ManagedTargetFailurePhase::Receiving,
+                kind: ManagedTargetFailureKind::Protocol,
+                message: "pending probe limit exceeded".into(),
             }),
         ));
 
@@ -2422,6 +2446,79 @@ mod tests {
             .is_some_and(|warning| warning.contains("pending probe limit")));
     }
 
+    #[test]
+    fn cleanup_failure_warns_without_changing_successful_target_status() {
+        let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]);
+        let mut outcome = target_outcome("a", ManagedTargetEndReason::TestComplete);
+        outcome.remote = None;
+        outcome.cleanup_failure = Some(ManagedTargetFailure {
+            phase: ManagedTargetFailurePhase::Closing,
+            kind: ManagedTargetFailureKind::Socket,
+            message: "close datagram failed".into(),
+        });
+
+        state.process_target_outcome(&outcome);
+
+        assert_eq!(state.targets[0].status, TargetStatus::Closed);
+        assert_eq!(state.targets[0].remote, None);
+        assert!(state.targets[0]
+            .last_warning
+            .as_deref()
+            .is_some_and(|warning| warning.contains("cleanup failed (Closing Socket)")));
+        assert!(state
+            .recent_events
+            .iter()
+            .any(|event| event.contains("cleanup failed (Closing Socket)")));
+    }
+
+    #[test]
+    fn primary_and_cleanup_failures_are_both_visible() {
+        let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]);
+        let mut outcome = target_outcome(
+            "a",
+            ManagedTargetEndReason::Failed(ManagedTargetFailure {
+                phase: ManagedTargetFailurePhase::Receiving,
+                kind: ManagedTargetFailureKind::Protocol,
+                message: "primary failure".into(),
+            }),
+        );
+        outcome.cleanup_failure = Some(ManagedTargetFailure {
+            phase: ManagedTargetFailurePhase::Closing,
+            kind: ManagedTargetFailureKind::Socket,
+            message: "cleanup failure".into(),
+        });
+
+        state.process_target_outcome(&outcome);
+
+        assert_eq!(state.targets[0].status, TargetStatus::Failed);
+        assert!(state.targets[0]
+            .last_warning
+            .as_deref()
+            .is_some_and(|warning| warning.contains("primary failure")
+                && warning.contains("cleanup failure")));
+        assert!(state
+            .recent_events
+            .iter()
+            .any(|event| event.contains("target failed: Protocol: primary failure")));
+        assert!(state
+            .recent_events
+            .iter()
+            .any(|event| event.contains("cleanup failed (Closing Socket): cleanup failure")));
+    }
+
+    #[test]
+    fn managed_outcomes_map_stopped_removed_and_replaced_statuses() {
+        for (reason, expected) in [
+            (ManagedTargetEndReason::Stopped, TargetStatus::Stopped),
+            (ManagedTargetEndReason::Removed, TargetStatus::Removed),
+            (ManagedTargetEndReason::Replaced, TargetStatus::Replaced),
+        ] {
+            let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]);
+            state.process_target_outcome(&target_outcome("a", reason));
+            assert_eq!(state.targets[0].status, expected);
+        }
+    }
+
     #[test]
     fn run_error_preserves_peer_closed_target_status() {
         let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]);
@@ -2446,9 +2543,10 @@ mod tests {
         for label in ["a", "b"] {
             state.process_target_outcome(&target_outcome(
                 label,
-                ManagedTargetEndReason::OpenFailed(irtt_client::ManagedTargetFailure {
-                    kind: irtt_client::ManagedTargetFailureKind::OpeningProtocol,
-                    message: "server returned a zero connection token".to_owned(),
+                ManagedTargetEndReason::Failed(ManagedTargetFailure {
+                    phase: ManagedTargetFailurePhase::Opening,
+                    kind: ManagedTargetFailureKind::Protocol,
+                    message: "server returned a zero connection token".into(),
                 }),
             ));
         }
@@ -2465,8 +2563,12 @@ mod tests {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
 
-        state.process_target_event(&target_event("a", reply(1, 1_000_000)));
-        state.process_target_event(&target_event("b", reply(2, 2_000_000)));
+        let a = target_instance("a");
+        let a_reply = reply(1, 1_000_000);
+        state.process_target_event(&a, &a_reply);
+        let b = target_instance("b");
+        let b_reply = reply(2, 2_000_000);
+        state.process_target_event(&b, &b_reply);
         state.clear_visible_history();
 
         assert!(state.targets[0].graph_history.is_empty());
@@ -2480,11 +2582,13 @@ mod tests {
     #[test]
     fn warning_updates_recent_events_and_last_warning() {
         let mut state = TuiState::default();
-        state.process_event(&ClientEvent::Warning {
+        let target = target_instance("target");
+        let event = ClientEvent::Warning {
             kind: WarningKind::WrongToken,
             message: "wrong token".to_owned(),
             at: ts(Duration::ZERO),
-        });
+        };
+        state.process_target_event(&target, &event);
 
         assert_eq!(
             state.last_warning.as_deref(),
@@ -2501,7 +2605,7 @@ mod tests {
     fn dropped_events_mark_tui_statistics_incomplete() {
         let mut state = TuiState::default();
 
-        state.mark_dropped_group_events(9);
+        state.mark_dropped_managed_events(9);
 
         assert_eq!(state.dropped_events, 9);
         assert!(state
@@ -2511,20 +2615,20 @@ mod tests {
         assert!(state
             .recent_events
             .back()
-            .is_some_and(|event| event.contains("dropped 9 managed group events")));
+            .is_some_and(|event| event.contains("dropped 9 managed run events")));
     }
 
     #[test]
     fn dropped_single_client_events_are_visible() {
         let mut state = TuiState::default();
 
-        state.mark_dropped_client_events(3);
+        state.mark_dropped_managed_events(3);
 
         assert_eq!(state.dropped_events, 3);
         assert!(state
             .recent_events
             .back()
-            .is_some_and(|event| event.contains("dropped 3 managed client events")));
+            .is_some_and(|event| event.contains("dropped 3 managed run events")));
     }
 
     #[test]
@@ -2803,7 +2907,9 @@ mod tests {
     #[test]
     fn sample_details_preserve_signed_values() {
         let mut state = TuiState::default();
-        state.process_event(&reply(4, -1_250_000));
+        let target = target_instance("target");
+        let event = reply(4, -1_250_000);
+        state.process_target_event(&target, &event);
 
         let target = primary_target(&state);
         let sample = target.last_sample.unwrap();

crates/irtt-cli/src/shared/client/session.rs

@@ -1,5 +1,40 @@
 use std::sync::atomic::{AtomicBool, Ordering};
 
+#[cfg(feature = "client-runtime")]
+use irtt_client::managed::{ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError};
+
+#[cfg(feature = "client-runtime")]
+pub(crate) const MANAGED_EVENT_WORK_BUDGET: usize = 128;
+
+#[cfg(feature = "client-runtime")]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum ManagedDrainState {
+    Empty,
+    Closed,
+    BudgetExhausted,
+}
+
+/// Drain a bounded amount of lossy presentation work before returning to the
+/// frontend control loop.
+#[cfg(feature = "client-runtime")]
+pub(crate) fn drain_managed_events<E>(
+    events: &mut ManagedEventSubscription,
+    dropped_events: &mut u64,
+    mut process: impl FnMut(ManagedEvent) -> Result<(), E>,
+) -> Result<ManagedDrainState, E> {
+    for _ in 0..MANAGED_EVENT_WORK_BUDGET {
+        match events.try_recv() {
+            Ok(event) => process(event)?,
+            Err(ManagedEventTryRecvError::Lagged(count)) => {
+                *dropped_events = dropped_events.saturating_add(count);
+            }
+            Err(ManagedEventTryRecvError::Empty) => return Ok(ManagedDrainState::Empty),
+            Err(ManagedEventTryRecvError::Closed) => return Ok(ManagedDrainState::Closed),
+        }
+    }
+    Ok(ManagedDrainState::BudgetExhausted)
+}
+
 pub fn is_shutdown_requested(shutdown_requested: &AtomicBool) -> bool {
     shutdown_requested.load(Ordering::Relaxed)
 }
@@ -31,13 +66,11 @@ pub fn request_managed_stop_for_peer_close(
     continuous: bool,
     interrupted: bool,
     peer_closed_target_outcomes: u64,
-    peer_close_requested_stop: &mut bool,
     stop_requested: &mut bool,
 ) -> bool {
     if !continuous || interrupted || peer_closed_target_outcomes == 0 {
         return false;
     }
-    *peer_close_requested_stop = true;
     request_managed_stop_once(stop_requested)
 }
 
@@ -70,46 +103,38 @@ mod tests {
 
     #[test]
     fn grouped_peer_close_stop_policy_is_run_mode_and_one_shot() {
-        let mut peer_close_requested_stop = false;
         let mut stop_requested = false;
         assert!(!request_managed_stop_for_peer_close(
             false,
             false,
             1,
-            &mut peer_close_requested_stop,
             &mut stop_requested,
         ));
         assert!(!request_managed_stop_for_peer_close(
             true,
             true,
             1,
-            &mut peer_close_requested_stop,
             &mut stop_requested,
         ));
         assert!(!request_managed_stop_for_peer_close(
             true,
             false,
             0,
-            &mut peer_close_requested_stop,
             &mut stop_requested,
         ));
-        assert!(!peer_close_requested_stop);
         assert!(!stop_requested);
 
         assert!(request_managed_stop_for_peer_close(
             true,
             false,
             1,
-            &mut peer_close_requested_stop,
             &mut stop_requested,
         ));
-        assert!(peer_close_requested_stop);
         assert!(stop_requested);
         assert!(!request_managed_stop_for_peer_close(
             true,
             false,
             1,
-            &mut peer_close_requested_stop,
             &mut stop_requested,
         ));
     }

crates/irtt-client/src/client.rs

@@ -94,8 +94,8 @@ use crate::{
 ///
 /// `Client` exposes the protocol steps directly: connect a UDP socket, open a
 /// session, send probes, receive replies, poll timeouts, and close. Callers
-/// that do not need to own this loop can use [`ManagedClient`](crate::ManagedClient)
-/// instead.
+/// that do not need to own this loop can use the unified managed API in
+/// [`crate::managed`] when the `tokio` feature is enabled.
 #[derive(Debug)]
 pub struct Client {
     runtime: SessionMachine,

crates/irtt-client/src/lib.rs

@@ -14,10 +14,9 @@
 //! values usually indicate client/server clock skew.
 //!
 //! [`Client`] is the runtime-free low-level blocking adapter. With the
-//! `tokio` feature, [`AsyncClient`] provides the corresponding Tokio adapter;
-//! [`managed`] contains the unified managed task/handle implementation and
-//! [`managed::BlockingManagedClient`] owns its dedicated runtime thread for
-//! synchronous managed callers.
+//! `tokio` feature, `AsyncClient` provides the corresponding Tokio adapter;
+//! [`managed`] contains the unified managed task/handle implementation,
+//! including the dedicated-runtime blocking frontend for synchronous callers.
 //!
 #![cfg_attr(
     not(all(target_os = "linux", feature = "ancillary")),

crates/irtt-client/src/receive/linux.rs

@@ -105,62 +105,6 @@ fn recv_datagram_fd(
     })
 }
 
-#[cfg(any())]
-pub(crate) fn recv_datagram_from(
-    socket: &UdpSocket,
-    buf: &mut [u8],
-) -> Result<ReceivedDatagramFrom, io::Error> {
-    let mut control = ControlBuffer::new();
-    let mut name: libc::sockaddr_storage = unsafe { mem::zeroed() };
-    let mut iov = libc::iovec {
-        iov_base: buf.as_mut_ptr().cast(),
-        iov_len: buf.len(),
-    };
-    let mut msg: libc::msghdr = unsafe { mem::zeroed() };
-    msg.msg_name = (&mut name as *mut libc::sockaddr_storage).cast();
-    msg.msg_namelen = mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
-    msg.msg_iov = &mut iov;
-    msg.msg_iovlen = 1;
-    msg.msg_control = control.as_mut_ptr().cast();
-    msg.msg_controllen = control.len();
-
-    let len = unsafe {
-        // SAFETY: `msg` points to one writable iovec backed by `buf`, a
-        // writable source-address buffer, and a writable control buffer. All
-        // buffers live until `recvmsg` returns. The socket file descriptor is
-        // borrowed from a valid `UdpSocket`.
-        libc::recvmsg(socket.as_raw_fd(), &mut msg, 0)
-    };
-    if len < 0 {
-        return Err(io::Error::last_os_error());
-    }
-    let len = usize::try_from(len).map_err(|_| {
-        io::Error::new(
-            io::ErrorKind::InvalidData,
-            "recvmsg returned an unrepresentable datagram length",
-        )
-    })?;
-    let received_at = ClientTimestamp::now();
-    let source = unsafe {
-        // SAFETY: `name` and `msg_namelen` were initialized by a successful
-        // `recvmsg` call above.
-        socket_addr_from_storage(&name, msg.msg_namelen)?
-    };
-    let meta = unsafe {
-        // SAFETY: `msg` was initialized by a successful `recvmsg` call. The
-        // parser only reads cmsghdr entries within `msg_controllen` and ignores
-        // short or unrelated control messages.
-        parse_receive_meta(&msg)
-    };
-
-    Ok(ReceivedDatagramFrom {
-        len,
-        source,
-        received_at,
-        meta,
-    })
-}
-
 unsafe fn parse_receive_meta(msg: &libc::msghdr) -> ReceiveMeta {
     let mut meta = ReceiveMeta::default();
     let mut cmsg = libc::CMSG_FIRSTHDR(msg);
@@ -182,53 +126,6 @@ unsafe fn parse_receive_meta(msg: &libc::msghdr) -> ReceiveMeta {
     meta
 }
 
-#[cfg(any())]
-unsafe fn socket_addr_from_storage(
-    storage: &libc::sockaddr_storage,
-    len: libc::socklen_t,
-) -> io::Result<SocketAddr> {
-    match i32::from(storage.ss_family) {
-        libc::AF_INET => {
-            if usize::try_from(len).unwrap_or(0) < mem::size_of::<libc::sockaddr_in>() {
-                return Err(io::Error::new(
-                    io::ErrorKind::InvalidData,
-                    "short IPv4 source address from recvmsg",
-                ));
-            }
-            let addr = ptr::read_unaligned(
-                (storage as *const libc::sockaddr_storage).cast::<libc::sockaddr_in>(),
-            );
-            let ip = Ipv4Addr::from(u32::from_be(addr.sin_addr.s_addr));
-            let port = u16::from_be(addr.sin_port);
-            Ok(SocketAddr::V4(SocketAddrV4::new(ip, port)))
-        }
-        libc::AF_INET6 => {
-            if usize::try_from(len).unwrap_or(0) < mem::size_of::<libc::sockaddr_in6>() {
-                return Err(io::Error::new(
-                    io::ErrorKind::InvalidData,
-                    "short IPv6 source address from recvmsg",
-                ));
-            }
-            let addr = ptr::read_unaligned(
-                (storage as *const libc::sockaddr_storage).cast::<libc::sockaddr_in6>(),
-            );
-            let ip = Ipv6Addr::from(addr.sin6_addr.s6_addr);
-            let port = u16::from_be(addr.sin6_port);
-            let flowinfo = u32::from_be(addr.sin6_flowinfo);
-            Ok(SocketAddr::V6(SocketAddrV6::new(
-                ip,
-                port,
-                flowinfo,
-                addr.sin6_scope_id,
-            )))
-        }
-        family => Err(io::Error::new(
-            io::ErrorKind::InvalidData,
-            format!("unsupported source address family from recvmsg: {family}"),
-        )),
-    }
-}
-
 fn is_traffic_class_cmsg(cmsg: &libc::cmsghdr) -> bool {
     (cmsg.cmsg_level == libc::IPPROTO_IP && cmsg.cmsg_type == libc::IP_TOS)
         || (cmsg.cmsg_level == libc::IPPROTO_IPV6 && cmsg.cmsg_type == libc::IPV6_TCLASS)