From 58d11f1dec3b7eb533fc98d1eff3c57955b5bc39 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 28 Jul 2026 02:20:09 +0700 Subject: [PATCH] fix(cli): fail continuous runs on peer closure Keep PeerClosed as a distinct successful library outcome while treating it as an error at the CLI and TUI boundary for uninterrupted continuous runs. Track aggregate peer-close outcomes independently of bounded recent history and cover finite, continuous, grouped, mixed-failure, interruption, and TUI status behavior. --- README.md | 8 +- crates/irtt-cli/src/cmd/client/run.rs | 38 +++- crates/irtt-cli/src/cmd/tui/run.rs | 22 ++- crates/irtt-cli/src/cmd/tui/ui.rs | 23 +++ crates/irtt-cli/src/shared/client/session.rs | 39 ++++ crates/irtt-cli/tests/client_multi_target.rs | 186 ++++++++++++++++++- crates/irtt-client/src/managed/group.rs | 61 ++++++ 7 files changed, 358 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index a453bd6..263425e 100644 --- a/README.md +++ b/README.md @@ -139,6 +139,10 @@ Adjusted RTT can be negative when server processing exceeds the measured raw RTT Default table output prints a final summary after completed finite runs and interrupted continuous runs when the run policy permits it. CSV, TSV, and JSON Lines output do not print this summary. +A peer closing a finite CLI or TUI run is accepted as a terminal outcome. In +continuous mode, peer closure exits nonzero unless the user requested +interruption, allowing a supervisor to restart the client. + ## Terminal UI When built with the `tui` feature, `irtt-tui` provides a live cumulative dashboard: @@ -229,7 +233,9 @@ groups. `ExplicitCancellation` supports long-lived controllers; replacing the desired set with an empty vector removes current targets but leaves the group idle and ready for a later `update_targets` call. `join()` reports aggregate outcome counts and retains only the 256 most recent target outcomes, so -long-running target churn remains bounded. +long-running target churn remains bounded. The aggregate +`peer_closed_target_outcomes` count includes peer closures omitted from that +recent snapshot. Managed event subscriptions remain bounded and nonblocking. When using a dropping overflow policy, check `EventSubscription::dropped_events()` before diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs index b931a7a..9094f36 100644 --- a/crates/irtt-cli/src/cmd/client/run.rs +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -23,7 +23,9 @@ use super::{ #[cfg(feature = "stats")] use crate::shared::client::expected_probe_count; use crate::shared::client::{ - is_shutdown_requested, session::should_print_final_summary, ClientSession, + is_shutdown_requested, + session::{peer_close_run_error, should_print_final_summary}, + ClientSession, }; #[cfg(feature = "stats")] @@ -152,6 +154,12 @@ pub fn run_stream( print_events_with_stats(&mut stream_output, &events, Some(target_label), &mut stats)?; #[cfg(not(feature = "stats"))] print_events_with_stats(&mut stream_output, &events, Some(target_label))?; + interrupted |= is_shutdown_requested(shutdown_requested); + let peer_close_error = peer_close_run_error( + continuous, + interrupted, + if session.is_peer_closed() { 1 } else { 0 }, + ); let print_final_summary = should_print_final_summary(continuous, interrupted); stream_output.print_final_summary = print_final_summary; stream_output.show_running_only_summary_note = continuous && interrupted && print_final_summary; @@ -160,6 +168,9 @@ pub fn run_stream( #[cfg(not(feature = "stats"))] stream_output.print_summary()?; stream_output.out.flush()?; + if let Some(error) = peer_close_error { + return Err(error.into()); + } Ok(()) } @@ -403,9 +414,7 @@ fn run_group_stream( eprintln!("{warning}"); } - if outcome.end_reason == ManagedGroupEndReason::Cancelled && !interrupted { - return Err("managed client group was cancelled".into()); - } + interrupted |= is_shutdown_requested(shutdown_requested); for target in &outcome.targets { if terminal_targets.insert(target.id.as_str().to_owned()) { report_target_failure(target); @@ -413,11 +422,19 @@ fn run_group_stream( } let successful_targets = outcome.successful_target_outcomes; let failed_targets = outcome.failed_target_outcomes; - if !interrupted && successful_targets == 0 && failed_targets > 0 { - return Err( - format!("no managed target completed successfully ({failed_targets} failed)").into(), - ); - } + let terminal_error = if outcome.end_reason == ManagedGroupEndReason::Cancelled && !interrupted { + Some("managed client group was cancelled".to_owned()) + } else if let Some(error) = + peer_close_run_error(continuous, interrupted, outcome.peer_closed_target_outcomes) + { + Some(error) + } else if !interrupted && successful_targets == 0 && failed_targets > 0 { + Some(format!( + "no managed target completed successfully ({failed_targets} failed)" + )) + } else { + None + }; let print_final_summary = should_print_final_summary(continuous, interrupted); stream_output.print_final_summary = print_final_summary; @@ -437,6 +454,9 @@ fn run_group_stream( stream_output.print_summary()?; } stream_output.out.flush()?; + if let Some(error) = terminal_error { + return Err(error.into()); + } Ok(()) } diff --git a/crates/irtt-cli/src/cmd/tui/run.rs b/crates/irtt-cli/src/cmd/tui/run.rs index a5cb14e..4178431 100644 --- a/crates/irtt-cli/src/cmd/tui/run.rs +++ b/crates/irtt-cli/src/cmd/tui/run.rs @@ -14,7 +14,7 @@ use irtt_client::{ use crate::{ cmd::tui::args::{ResolvedTuiTarget, TuiArgs}, - shared::client::{is_shutdown_requested, ClientSession}, + shared::client::{is_shutdown_requested, session::peer_close_run_error, ClientSession}, }; use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal}; @@ -129,6 +129,16 @@ pub fn run_tui( let events = session.close()?; state.process_events(&events); + interrupted |= is_shutdown_requested(shutdown_requested); + if let Some(error) = peer_close_run_error( + continuous, + interrupted, + if session.is_peer_closed() { 1 } else { 0 }, + ) { + state.set_run_error(error.clone()); + render_if_due(&mut terminal, &state, &mut next_render, true)?; + return Err(error.into()); + } state.set_status(TuiStatus::Complete); render_if_due(&mut terminal, &state, &mut next_render, true)?; Ok(()) @@ -249,6 +259,7 @@ fn run_group_tui( } } + interrupted |= is_shutdown_requested(shutdown_requested); if outcome.end_reason == ManagedGroupEndReason::Cancelled && !interrupted { return match exit { GroupLoopExit::IdleGraceElapsed => { @@ -262,6 +273,15 @@ fn run_group_tui( } }; } + if let Some(error) = peer_close_run_error( + args.is_continuous(), + interrupted, + outcome.peer_closed_target_outcomes, + ) { + state.set_run_error(error.clone()); + render_if_due(terminal, state, next_render, true)?; + return Err(error.into()); + } let successful_targets = outcome.successful_target_outcomes; let failed_targets = outcome.failed_target_outcomes; if !interrupted && successful_targets == 0 && failed_targets > 0 { diff --git a/crates/irtt-cli/src/cmd/tui/ui.rs b/crates/irtt-cli/src/cmd/tui/ui.rs index 0104d17..bcaabee 100644 --- a/crates/irtt-cli/src/cmd/tui/ui.rs +++ b/crates/irtt-cli/src/cmd/tui/ui.rs @@ -409,6 +409,12 @@ impl TuiState { self.push_event(format!("error {message}")); } + pub(super) fn set_run_error(&mut self, message: String) { + self.status = TuiStatus::Error; + self.last_warning = Some(message.clone()); + self.push_event(format!("error {message}")); + } + pub(super) fn clear_visible_history(&mut self) { for target in &mut self.targets { target.graph_history.clear(); @@ -2412,6 +2418,23 @@ mod tests { .is_some_and(|warning| warning.contains("pending probe limit"))); } + #[test] + fn run_error_preserves_peer_closed_target_status() { + let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]); + state.process_target_outcome(&target_outcome("a", ManagedTargetEndReason::PeerClosed)); + + state.set_run_error( + "continuous run ended because of peer closure (1 target session)".to_owned(), + ); + + assert_eq!(state.status, TuiStatus::Error); + assert_eq!(state.targets[0].status, TargetStatus::Closed); + assert!(state + .last_warning + .as_deref() + .is_some_and(|warning| warning.contains("peer closure"))); + } + #[test] fn all_failed_terminal_outcomes_set_error_status() { let mut state = diff --git a/crates/irtt-cli/src/shared/client/session.rs b/crates/irtt-cli/src/shared/client/session.rs index 114457f..fb567b0 100644 --- a/crates/irtt-cli/src/shared/client/session.rs +++ b/crates/irtt-cli/src/shared/client/session.rs @@ -106,6 +106,10 @@ impl ClientSession { self.client.close() } + pub fn is_peer_closed(&self) -> bool { + self.client.is_peer_closed() + } + pub fn next_send_deadline(&self) -> Option { self.client.next_send_deadline() } @@ -139,6 +143,25 @@ pub fn should_print_final_summary(continuous: bool, interrupted: bool) -> bool { !continuous || interrupted } +pub fn peer_close_run_error( + continuous: bool, + interrupted: bool, + peer_closed_target_outcomes: u64, +) -> Option { + if !continuous || interrupted || peer_closed_target_outcomes == 0 { + return None; + } + + let sessions = if peer_closed_target_outcomes == 1 { + "target session" + } else { + "target sessions" + }; + Some(format!( + "continuous run ended because of peer closure ({peer_closed_target_outcomes} {sessions})" + )) +} + pub fn final_drain_duration(probe_timeout: Duration) -> Duration { probe_timeout.min(MAX_FINAL_DRAIN) } @@ -201,6 +224,21 @@ mod tests { assert!(!should_drain_final(true, false)); } + #[test] + fn peer_close_run_error_is_run_mode_and_interruption_aware() { + assert_eq!(peer_close_run_error(false, false, 1), None); + assert_eq!(peer_close_run_error(true, false, 0), None); + assert_eq!(peer_close_run_error(true, true, 1), None); + assert_eq!( + peer_close_run_error(true, false, 1).as_deref(), + Some("continuous run ended because of peer closure (1 target session)") + ); + assert_eq!( + peer_close_run_error(true, false, 3).as_deref(), + Some("continuous run ended because of peer closure (3 target sessions)") + ); + } + #[test] fn peer_close_followed_by_session_cleanup_is_successful() { let params = test_params(None, Duration::from_millis(10)); @@ -221,6 +259,7 @@ mod tests { .iter() .any(|event| matches!(event, ClientEvent::SessionClosed { .. }))); assert!(!session.should_continue(&shutdown_requested)); + assert!(session.is_peer_closed()); session.drain_final(|_| {}).unwrap(); assert!(session.poll_timeouts().unwrap().is_empty()); diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs index 95a718d..e5654ca 100644 --- a/crates/irtt-cli/tests/client_multi_target.rs +++ b/crates/irtt-cli/tests/client_multi_target.rs @@ -124,6 +124,76 @@ fn single_target_custom_target_column_renders_positional_label() { ); } +#[test] +fn finite_single_target_peer_close_exits_successfully() { + let server = start_peer_close_server(test_params( + Some(Duration::from_millis(40)), + Duration::from_millis(10), + )); + let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli")); + command.args([ + "--duration", + "40ms", + "--interval", + "10ms", + "--format", + "csv", + "--columns", + "seq", + "--header", + "never", + &server.addr.to_string(), + ]); + + let output = run_with_timeout(command, Duration::from_secs(3)); + server.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.trim().is_empty(), "{stdout}"); + assert!(!stderr.contains("peer closure"), "{stderr}"); +} + +#[test] +fn continuous_single_target_peer_close_exits_nonzero() { + let server = start_peer_close_server(test_params(None, Duration::from_millis(10))); + let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli")); + command.args([ + "--duration", + "0s", + "--interval", + "10ms", + "--format", + "csv", + "--columns", + "seq", + "--header", + "never", + &server.addr.to_string(), + ]); + + let output = run_with_timeout(command, Duration::from_secs(3)); + server.join(); + + assert!( + !output.status.success(), + "status={:?}\nstdout={}", + output.status.code(), + String::from_utf8_lossy(&output.stdout) + ); + let stdout = String::from_utf8(output.stdout).unwrap(); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert!(!stdout.trim().is_empty(), "{stdout}"); + assert!(stderr.contains("continuous run"), "{stderr}"); + assert!(stderr.contains("peer closure"), "{stderr}"); +} + #[test] fn multi_target_csv_emits_rows_for_both_labels() { let params = test_params(Some(Duration::from_millis(40)), Duration::from_millis(10)); @@ -171,9 +241,105 @@ fn multi_target_csv_emits_rows_for_both_labels() { } #[test] -fn continuous_mixed_success_and_open_failure_exits_and_reports_failure() { +fn finite_multi_target_peer_close_is_accepted_as_completion() { + let params = test_params(Some(Duration::from_millis(40)), Duration::from_millis(10)); + let a = start_peer_close_server(params.clone()); + let b = start_peer_close_server(params); + let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli")); + command.args([ + "--duration", + "40ms", + "--interval", + "10ms", + "--format", + "csv", + "--columns", + "target,seq", + "--header", + "never", + "--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={:?}\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("a,")), + "{stdout}" + ); + assert!( + stdout.lines().any(|line| line.starts_with("b,")), + "{stdout}" + ); + assert!(!stderr.contains("peer closure"), "{stderr}"); +} + +#[test] +fn continuous_all_peer_closed_targets_exit_nonzero() { let params = test_params(None, Duration::from_millis(10)); - let success = start_peer_close_server(params.clone()); + let a = start_peer_close_server(params.clone()); + let b = start_peer_close_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!("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 stdout = String::from_utf8(output.stdout).unwrap(); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert!( + stdout.lines().any(|line| line.starts_with("a,")), + "{stdout}" + ); + assert!( + stdout.lines().any(|line| line.starts_with("b,")), + "{stdout}" + ); + assert!(stderr.contains("continuous run"), "{stderr}"); + assert!( + stderr.contains("peer closure (2 target sessions)"), + "{stderr}" + ); +} + +#[test] +fn continuous_mixed_peer_close_and_open_failure_exits_nonzero() { + let params = test_params(None, Duration::from_millis(10)); + let peer_closed = 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([ @@ -188,29 +354,33 @@ fn continuous_mixed_success_and_open_failure_exits_and_reports_failure() { "--header", "never", "--target", - &format!("success={}", success.addr), + &format!("peer={}", peer_closed.addr), "--target", &format!("failure={}", failure.addr), ]); let output = run_with_timeout(command, Duration::from_secs(3)); - success.join(); + peer_closed.join(); failure.join(); assert!( - output.status.success(), - "status={:?}\nstderr={}", + !output.status.success(), + "status={:?}\nstdout={}", output.status.code(), - String::from_utf8_lossy(&output.stderr) + String::from_utf8_lossy(&output.stdout) ); 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.lines().any(|line| line.starts_with("peer,")), "{stdout}" ); assert!(stderr.contains("target failure failed"), "{stderr}"); assert!(stderr.contains("zero token without close flag"), "{stderr}"); + assert!( + stderr.contains("peer closure (1 target session)"), + "{stderr}" + ); } #[test] diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index e6d2a0d..411cba0 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -214,6 +214,12 @@ pub struct ManagedGroupOutcome { pub total_target_outcomes: u64, /// Number of target outcomes classified as successful. pub successful_target_outcomes: u64, + /// Number of target outcomes that ended because the peer closed the session. + /// + /// Peer closure remains a successful library-level outcome and is also + /// included in [`Self::successful_target_outcomes`]. This aggregate covers + /// outcomes omitted from the bounded [`Self::targets`] snapshot. + pub peer_closed_target_outcomes: u64, /// Number of target outcomes carrying structured failure details. pub failed_target_outcomes: u64, /// Number of older outcomes omitted from [`Self::targets`]. @@ -637,6 +643,7 @@ struct TargetOutcomeHistory { recent: VecDeque, total: u64, successful: u64, + peer_closed: u64, failed: u64, discarded: u64, } @@ -647,6 +654,9 @@ impl TargetOutcomeHistory { if outcome.is_success() { self.successful = self.successful.saturating_add(1); } + if matches!(&outcome.end_reason, ManagedTargetEndReason::PeerClosed) { + self.peer_closed = self.peer_closed.saturating_add(1); + } if outcome.end_reason.failure().is_some() { self.failed = self.failed.saturating_add(1); } @@ -663,6 +673,7 @@ impl TargetOutcomeHistory { targets: self.recent.into_iter().collect(), total_target_outcomes: self.total, successful_target_outcomes: self.successful, + peer_closed_target_outcomes: self.peer_closed, failed_target_outcomes: self.failed, discarded_target_outcomes: self.discarded, } @@ -2146,6 +2157,51 @@ mod tests { assert_eq!(target.end_reason, ManagedTargetEndReason::PeerClosed); assert_eq!(target.packets_sent, 1); assert_eq!(target.replies_received, 1); + assert_eq!(outcome.total_target_outcomes, 1); + assert_eq!(outcome.successful_target_outcomes, 1); + assert_eq!(outcome.peer_closed_target_outcomes, 1); + assert_eq!(outcome.failed_target_outcomes, 0); + } + + #[test] + fn peer_close_aggregate_survives_bounded_recent_history() { + let remote: SocketAddr = "127.0.0.1:2112".parse().unwrap(); + let mut history = TargetOutcomeHistory::default(); + history.record(ManagedTargetOutcome { + id: TargetId::from("peer"), + remote, + end_reason: ManagedTargetEndReason::PeerClosed, + packets_sent: 1, + replies_received: 1, + duplicates: 0, + late: 0, + warning_events: 0, + }); + for index in 0..MANAGED_GROUP_OUTCOME_HISTORY_LIMIT { + history.record(ManagedTargetOutcome { + id: TargetId::from(format!("removed-{index}")), + remote, + end_reason: ManagedTargetEndReason::Removed, + packets_sent: 0, + replies_received: 0, + duplicates: 0, + late: 0, + warning_events: 0, + }); + } + + let outcome = history.into_group_outcome(ManagedGroupEndReason::AllTargetsComplete); + let expected_total = u64::try_from(MANAGED_GROUP_OUTCOME_HISTORY_LIMIT + 1).unwrap(); + assert_eq!(outcome.total_target_outcomes, expected_total); + assert_eq!(outcome.successful_target_outcomes, 1); + assert_eq!(outcome.peer_closed_target_outcomes, 1); + assert_eq!(outcome.failed_target_outcomes, 0); + assert_eq!(outcome.targets.len(), MANAGED_GROUP_OUTCOME_HISTORY_LIMIT); + assert_eq!(outcome.discarded_target_outcomes, 1); + assert!(outcome + .targets + .iter() + .all(|target| target.end_reason == ManagedTargetEndReason::Removed)); } #[test] @@ -2196,6 +2252,10 @@ mod tests { ) ))); assert_eq!(outcome.targets.len(), 2); + assert_eq!(outcome.total_target_outcomes, 2); + assert_eq!(outcome.successful_target_outcomes, 1); + assert_eq!(outcome.peer_closed_target_outcomes, 1); + assert_eq!(outcome.failed_target_outcomes, 1); } #[test] @@ -2778,6 +2838,7 @@ mod tests { assert_eq!(published, churn_count + 1); assert_eq!(outcome.total_target_outcomes, expected_total); assert_eq!(outcome.successful_target_outcomes, 0); + assert_eq!(outcome.peer_closed_target_outcomes, 0); assert_eq!(outcome.failed_target_outcomes, 0); assert_eq!(outcome.targets.len(), MANAGED_GROUP_OUTCOME_HISTORY_LIMIT); assert_eq!(