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.
This commit is contained in:
Nils Andreas Svee committed 2026-07-28 10:23:47 +07:00
1 parent 8ea4c2f8a4
commit 58d11f1dec
7 files changed
+358 -19

No files matched your search

+29 -9
View File
@@ -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(())
}
+21 -1
View File
@@ -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 {
+23
View File
@@ -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 =
@@ -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<Instant> {
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<String> {
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());
+178 -8
View File
@@ -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]