fix(cli): report multi-target failures and return nonzero

Consume structured per-target outcomes in the streaming CLI and TUI so mixed failures remain visible while successful targets still produce output. Return a failure status when every target fails, and keep interruption behavior distinct. Add strict-timeout integration coverage for mixed continuous completion and all-open-failure exit status, plus focused TUI outcome-state tests.
2026-07-28 01:04:42 +07:00
1 parent 97aac829ec
commit 4398d25d4b
@@ -10,7 +10,8 @@ use std::{

 use irtt_client::{
    ClientEvent, EventSubscriptionError, ManagedClientGroup, ManagedClientGroupConfig,
-    ManagedGroupEndReason, ManagedGroupEvent, SubscriberConfig, SubscriberOverflow, TargetEvent,
+    ManagedGroupEndReason, ManagedGroupEvent, ManagedTargetOutcome, SubscriberConfig,
+    SubscriberOverflow, TargetEvent,
 };
 #[cfg(all(test, feature = "stats"))]
 use irtt_client::{ClientTimestamp, PacketMeta, RttSample, SignedDuration};
@@ -347,6 +348,7 @@ fn run_group_stream(
                        }
                    }
                    ManagedGroupEvent::TargetFinished(target) => {
+                        report_target_failure(&target);
                        terminal_targets.insert(target.id.as_str().to_owned());
                    }
                }
@@ -374,18 +376,25 @@ fn run_group_stream(

    let outcome = session.join()?;
    while let Ok(Some(group_event)) = events.try_recv() {
-        if let ManagedGroupEvent::Client(target_event) = group_event {
-            #[cfg(feature = "stats")]
-            {
-                let label = target_event.target.as_str();
-                let stats = stats
-                    .entry(label.to_owned())
-                    .or_insert_with(|| StatsCollector::new(stats_config(continuous)));
-                print_target_event_with_stats(&mut stream_output, &target_event, stats)?;
+        match group_event {
+            ManagedGroupEvent::Client(target_event) => {
+                #[cfg(feature = "stats")]
+                {
+                    let label = target_event.target.as_str();
+                    let stats = stats
+                        .entry(label.to_owned())
+                        .or_insert_with(|| StatsCollector::new(stats_config(continuous)));
+                    print_target_event_with_stats(&mut stream_output, &target_event, stats)?;
+                }
+                #[cfg(not(feature = "stats"))]
+                {
+                    print_target_event_with_stats(&mut stream_output, &target_event)?;
+                }
            }
-            #[cfg(not(feature = "stats"))]
-            {
-                print_target_event_with_stats(&mut stream_output, &target_event)?;
+            ManagedGroupEvent::TargetFinished(target) => {
+                if terminal_targets.insert(target.id.as_str().to_owned()) {
+                    report_target_failure(&target);
+                }
            }
        }
    }
@@ -393,6 +402,26 @@ fn run_group_stream(
    if outcome.end_reason == ManagedGroupEndReason::Cancelled && !interrupted {
        return Err("managed client group was cancelled".into());
    }
+    for target in &outcome.targets {
+        if terminal_targets.insert(target.id.as_str().to_owned()) {
+            report_target_failure(target);
+        }
+    }
+    let successful_targets = outcome
+        .targets
+        .iter()
+        .filter(|target| target.is_success())
+        .count();
+    let failed_targets = outcome
+        .targets
+        .iter()
+        .filter(|target| target.end_reason.failure().is_some())
+        .count();
+    if !interrupted && successful_targets == 0 && failed_targets > 0 {
+        return Err(
+            format!("no managed target completed successfully ({failed_targets} failed)").into(),
+        );
+    }

    let print_final_summary = should_print_final_summary(continuous, interrupted);
    stream_output.print_final_summary = print_final_summary;
@@ -415,6 +444,15 @@ fn run_group_stream(
    Ok(())
 }

+fn report_target_failure(target: &ManagedTargetOutcome) {
+    if let Some(failure) = target.end_reason.failure() {
+        eprintln!(
+            "irtt-rs: target {} failed ({}): {}",
+            target.id, failure.kind, failure.message
+        );
+    }
+}
+
 fn estimated_group_completion_grace(args: &ClientArgs) -> Duration {
    let open_timeout: Duration = args.to_client_config().open_timeouts.iter().sum();
    open_timeout
@@ -193,6 +193,7 @@ fn run_group_tui(
                    }
                    ManagedGroupEvent::TargetFinished(target) => {
                        terminal_targets.insert(target.id.as_str().to_owned());
+                        state.process_target_outcome(&target);
                    }
                }
                render_if_due(terminal, state, next_render, false)?;
@@ -230,8 +231,18 @@ fn run_group_tui(

    let outcome = session.join()?;
    while let Ok(Some(group_event)) = events.try_recv() {
-        if let ManagedGroupEvent::Client(target_event) = group_event {
-            state.process_target_event(&target_event);
+        match group_event {
+            ManagedGroupEvent::Client(target_event) => state.process_target_event(&target_event),
+            ManagedGroupEvent::TargetFinished(target) => {
+                if terminal_targets.insert(target.id.as_str().to_owned()) {
+                    state.process_target_outcome(&target);
+                }
+            }
+        }
+    }
+    for target in &outcome.targets {
+        if terminal_targets.insert(target.id.as_str().to_owned()) {
+            state.process_target_outcome(target);
        }
    }

@@ -248,6 +259,23 @@ fn run_group_tui(
            }
        };
    }
+    let successful_targets = outcome
+        .targets
+        .iter()
+        .filter(|target| target.is_success())
+        .count();
+    let failed_targets = outcome
+        .targets
+        .iter()
+        .filter(|target| target.end_reason.failure().is_some())
+        .count();
+    if !interrupted && successful_targets == 0 && failed_targets > 0 {
+        state.set_status(TuiStatus::Error);
+        render_if_due(terminal, state, next_render, true)?;
+        return Err(
+            format!("no managed target completed successfully ({failed_targets} failed)").into(),
+        );
+    }

    state.set_status(TuiStatus::Complete);
    render_if_due(terminal, state, next_render, true)?;
@@ -9,7 +9,10 @@ use crossterm::{
    execute,
    terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen},
 };
-use irtt_client::{ClientEvent, NegotiatedParams, SignedDuration, TargetEvent};
+use irtt_client::{
+    ClientEvent, ManagedTargetEndReason, ManagedTargetOutcome, NegotiatedParams, SignedDuration,
+    TargetEvent,
+};
 use irtt_stats::{Snapshot, StatsCollector, TimeStats};
 use ratatui::{
    backend::CrosstermBackend,
@@ -163,6 +166,82 @@ impl TuiState {
        self.process_event_for_target(idx, &event.event);
    }

+    pub(super) fn process_target_outcome(&mut self, outcome: &ManagedTargetOutcome) {
+        let label = outcome.id.as_str();
+        let idx = if let Some(idx) = self.target_index.get(label).copied() {
+            idx
+        } else {
+            let idx = self.targets.len();
+            self.targets.push(TuiTargetState::new(
+                label.to_owned(),
+                stats_config(self.config.duration.is_none()),
+            ));
+            self.target_index.insert(label.to_owned(), idx);
+            idx
+        };
+
+        let (status, recent, warning) = match &outcome.end_reason {
+            ManagedTargetEndReason::TestComplete => {
+                (TargetStatus::Closed, "test completed".to_owned(), None)
+            }
+            ManagedTargetEndReason::PeerClosed => (
+                TargetStatus::Closed,
+                "session closed by peer".to_owned(),
+                None,
+            ),
+            ManagedTargetEndReason::NoTestComplete => (
+                TargetStatus::NoTest,
+                "no-test negotiation completed".to_owned(),
+                None,
+            ),
+            ManagedTargetEndReason::Cancelled => {
+                (TargetStatus::Cancelled, "target cancelled".to_owned(), None)
+            }
+            ManagedTargetEndReason::Removed => {
+                (TargetStatus::Removed, "target removed".to_owned(), None)
+            }
+            ManagedTargetEndReason::OpenFailed(failure)
+            | ManagedTargetEndReason::Failed(failure) => {
+                let message = format!("{}: {}", failure.kind, failure.message);
+                (
+                    TargetStatus::Failed,
+                    format!("target failed: {message}"),
+                    Some(message),
+                )
+            }
+        };
+
+        if let Some(target) = self.targets.get_mut(idx) {
+            target.remote = Some(outcome.remote.to_string());
+            target.status = status;
+            if let Some(warning) = &warning {
+                target.last_warning = Some(warning.clone());
+            }
+        }
+        if let Some(warning) = warning {
+            self.last_warning = Some(warning);
+        }
+        self.push_event(format!("{label}: {recent}"));
+
+        if self
+            .targets
+            .iter()
+            .all(|target| target.status.is_terminal())
+        {
+            self.status = if self.targets.iter().any(|target| target.status.is_success()) {
+                TuiStatus::Complete
+            } else if self
+                .targets
+                .iter()
+                .any(|target| target.status == TargetStatus::Failed)
+            {
+                TuiStatus::Error
+            } else {
+                TuiStatus::Complete
+            };
+        }
+    }
+
    fn process_event_for_target(&mut self, target_idx: usize, event: &ClientEvent) {
        let recent;
        let mut global_status = None;
@@ -456,6 +535,8 @@ pub(super) enum TargetStatus {
    Closed,
    Failed,
    NoTest,
+    Cancelled,
+    Removed,
    Unknown,
 }

@@ -467,12 +548,21 @@ impl TargetStatus {
            Self::Closed => "closed",
            Self::Failed => "failed",
            Self::NoTest => "no-test",
+            Self::Cancelled => "cancelled",
+            Self::Removed => "removed",
            Self::Unknown => "unknown",
        }
    }

    fn is_terminal(self) -> bool {
-        matches!(self, Self::Closed | Self::Failed | Self::NoTest)
+        matches!(
+            self,
+            Self::Closed | Self::Failed | Self::NoTest | Self::Cancelled | Self::Removed
+        )
+    }
+
+    fn is_success(self) -> bool {
+        matches!(self, Self::Closed | Self::NoTest)
    }
 }

@@ -2019,6 +2109,19 @@ mod tests {
        }
    }

+    fn target_outcome(label: &str, end_reason: ManagedTargetEndReason) -> ManagedTargetOutcome {
+        ManagedTargetOutcome {
+            id: TargetId::from(label),
+            remote: remote(),
+            end_reason,
+            packets_sent: 0,
+            replies_received: 0,
+            duplicates: 0,
+            late: 0,
+            warning_events: 0,
+        }
+    }
+
    #[test]
    fn formats_signed_durations_and_missing_values() {
        assert_eq!(format_ns_i128(Some(-1_500_000)), "-1.5ms");
@@ -2246,6 +2349,71 @@ mod tests {
        assert_eq!(state.status, TuiStatus::Complete);
    }

+    #[test]
+    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(),
+        };
+
+        state.process_target_outcome(&target_outcome(
+            "b",
+            ManagedTargetEndReason::OpenFailed(failure),
+        ));
+
+        assert_eq!(state.targets[0].status, TargetStatus::Opening);
+        assert_eq!(state.targets[1].status, TargetStatus::Failed);
+        assert!(state.targets[1]
+            .last_warning
+            .as_deref()
+            .is_some_and(|warning| warning.contains("opening timeout")));
+    }
+
+    #[test]
+    fn mixed_terminal_outcomes_preserve_success_and_show_failure() {
+        let mut state =
+            TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
+        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(),
+            }),
+        ));
+
+        assert_eq!(state.targets[0].status, TargetStatus::Closed);
+        assert_eq!(state.targets[1].status, TargetStatus::Failed);
+        assert_eq!(state.status, TuiStatus::Complete);
+        assert!(state
+            .last_warning
+            .as_deref()
+            .is_some_and(|warning| warning.contains("pending probe limit")));
+    }
+
+    #[test]
+    fn all_failed_terminal_outcomes_set_error_status() {
+        let mut state =
+            TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
+        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(),
+                }),
+            ));
+        }
+
+        assert_eq!(state.status, TuiStatus::Error);
+        assert!(state
+            .targets
+            .iter()
+            .all(|target| target.status == TargetStatus::Failed));
+    }
+
    #[test]
    fn reset_clears_graph_history_for_all_targets() {
        let mut state =
@@ -1,11 +1,12 @@
 #![cfg(feature = "client")]

 use std::{
+    io::Read,
    net::{SocketAddr, UdpSocket},
-    process::Command,
+    process::{Command, Output, Stdio},
    thread,
    thread::JoinHandle,
-    time::Duration,
+    time::{Duration, Instant},
 };

 use irtt_proto::{
@@ -169,6 +170,117 @@ fn multi_target_csv_emits_rows_for_both_labels() {
    );
 }

+#[test]
+fn continuous_mixed_success_and_open_failure_exits_and_reports_failure() {
+    let params = test_params(None, Duration::from_millis(10));
+    let success = start_peer_close_server(params.clone());
+    let failure = start_open_failure_server(params);
+    let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli"));
+    command.args([
+        "--duration",
+        "0s",
+        "--interval",
+        "10ms",
+        "--format",
+        "csv",
+        "--columns",
+        "target,seq",
+        "--header",
+        "never",
+        "--target",
+        &format!("success={}", success.addr),
+        "--target",
+        &format!("failure={}", failure.addr),
+    ]);
+
+    let output = run_with_timeout(command, Duration::from_secs(3));
+    success.join();
+    failure.join();
+
+    assert!(
+        output.status.success(),
+        "status={:?}\nstderr={}",
+        output.status.code(),
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let stdout = String::from_utf8(output.stdout).unwrap();
+    let stderr = String::from_utf8(output.stderr).unwrap();
+    assert!(
+        stdout.lines().any(|line| line.starts_with("success,")),
+        "{stdout}"
+    );
+    assert!(stderr.contains("target failure failed"), "{stderr}");
+    assert!(stderr.contains("zero token without close flag"), "{stderr}");
+}
+
+#[test]
+fn all_open_failures_exit_nonzero_with_diagnostics() {
+    let params = test_params(None, Duration::from_millis(10));
+    let a = start_open_failure_server(params.clone());
+    let b = start_open_failure_server(params);
+    let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli"));
+    command.args([
+        "--duration",
+        "0s",
+        "--interval",
+        "10ms",
+        "--target",
+        &format!("a={}", a.addr),
+        "--target",
+        &format!("b={}", b.addr),
+    ]);
+
+    let output = run_with_timeout(command, Duration::from_secs(3));
+    a.join();
+    b.join();
+
+    assert!(
+        !output.status.success(),
+        "status={:?}\nstdout={}",
+        output.status.code(),
+        String::from_utf8_lossy(&output.stdout)
+    );
+    let stderr = String::from_utf8(output.stderr).unwrap();
+    assert!(stderr.contains("target a failed"), "{stderr}");
+    assert!(stderr.contains("target b failed"), "{stderr}");
+    assert!(
+        stderr.contains("no managed target completed successfully"),
+        "{stderr}"
+    );
+}
+
+fn run_with_timeout(mut command: Command, timeout: Duration) -> Output {
+    let mut child = command
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .unwrap();
+    let deadline = Instant::now() + timeout;
+    loop {
+        if child.try_wait().unwrap().is_some() {
+            return child.wait_with_output().unwrap();
+        }
+        if Instant::now() >= deadline {
+            let _ = child.kill();
+            let mut stdout = Vec::new();
+            let mut stderr = Vec::new();
+            if let Some(mut pipe) = child.stdout.take() {
+                let _ = pipe.read_to_end(&mut stdout);
+            }
+            if let Some(mut pipe) = child.stderr.take() {
+                let _ = pipe.read_to_end(&mut stderr);
+            }
+            let _ = child.wait();
+            panic!(
+                "CLI timed out after {timeout:?}\nstdout={}\nstderr={}",
+                String::from_utf8_lossy(&stdout),
+                String::from_utf8_lossy(&stderr)
+            );
+        }
+        thread::sleep(Duration::from_millis(5));
+    }
+}
+
 fn start_echo_server(params: Params) -> FakeServer {
    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
    socket
@@ -205,6 +317,50 @@ fn start_echo_server(params: Params) -> FakeServer {
    FakeServer { addr, done }
 }

+fn start_peer_close_server(params: Params) -> FakeServer {
+    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+    socket
+        .set_read_timeout(Some(Duration::from_secs(2)))
+        .unwrap();
+    let addr = socket.local_addr().unwrap();
+    let done = thread::spawn(move || {
+        let (_, peer) = recv_request(&socket);
+        socket
+            .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
+            .unwrap();
+        let (packet, peer) = recv_request(&socket);
+        let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap());
+        socket
+            .send_to(
+                &echo_reply_packet(
+                    TOKEN,
+                    seq,
+                    &params,
+                    &TimestampFields::default(),
+                    FLAG_REPLY | FLAG_CLOSE,
+                ),
+                peer,
+            )
+            .unwrap();
+    });
+    FakeServer { addr, done }
+}
+
+fn start_open_failure_server(params: Params) -> FakeServer {
+    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+    socket
+        .set_read_timeout(Some(Duration::from_secs(2)))
+        .unwrap();
+    let addr = socket.local_addr().unwrap();
+    let done = thread::spawn(move || {
+        let (_, peer) = recv_request(&socket);
+        socket
+            .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, 0, &params), peer)
+            .unwrap();
+    });
+    FakeServer { addr, done }
+}
+
 fn recv_request(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
    let mut buf = [0_u8; 2048];
    let (size, peer) = socket.recv_from(&mut buf).unwrap();
@@ -211,6 +211,13 @@ pub struct ManagedTargetOutcome {
    pub warning_events: u64,
 }

+impl ManagedTargetOutcome {
+    /// Return whether this target completed successfully.
+    pub fn is_success(&self) -> bool {
+        self.end_reason.is_success()
+    }
+}
+
 /// Reason an individual target stopped.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum ManagedTargetEndReason {
@@ -290,6 +297,19 @@ pub enum ManagedTargetFailureKind {
    InternalWorker,
 }

+impl fmt::Display for ManagedTargetFailureKind {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(match self {
+            Self::OpeningTimeout => "opening timeout",
+            Self::OpeningProtocol => "opening protocol/authentication/negotiation",
+            Self::Socket => "socket",
+            Self::RuntimeProtocol => "runtime protocol",
+            Self::InvalidConfiguration => "invalid configuration",
+            Self::InternalWorker => "internal worker",
+        })
+    }
+}
+
 fn classify_target_failure(error: &ClientError, opening: bool) -> ManagedTargetFailureKind {
    match error {
        ClientError::OpenTimeout => ManagedTargetFailureKind::OpeningTimeout,