diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs index 9094f36..33c62d8 100644 --- a/crates/irtt-cli/src/cmd/client/run.rs +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -24,7 +24,10 @@ use super::{ use crate::shared::client::expected_probe_count; use crate::shared::client::{ is_shutdown_requested, - session::{peer_close_run_error, should_print_final_summary}, + session::{ + peer_close_run_error, request_group_stop_once, should_print_final_summary, + should_stop_group_on_peer_close, + }, ClientSession, }; @@ -326,6 +329,8 @@ fn run_group_stream( }; let mut interrupted = false; + let mut stop_requested = false; + let mut peer_close_requested_stop = false; let mut terminal_targets = HashSet::new(); let mut last_event_at = Instant::now(); let mut saw_target_event = false; @@ -333,7 +338,9 @@ fn run_group_stream( loop { if is_shutdown_requested(shutdown_requested) { interrupted = true; - session.stop(); + if request_group_stop_once(&mut stop_requested) { + session.stop(); + } } match events.try_recv() { @@ -362,6 +369,16 @@ fn run_group_stream( ManagedGroupEvent::TargetFinished(target) => { report_target_failure(&target); terminal_targets.insert(target.id.as_str().to_owned()); + if should_stop_group_on_peer_close( + continuous, + interrupted, + &target.end_reason, + ) { + peer_close_requested_stop = true; + if request_group_stop_once(&mut stop_requested) { + session.stop(); + } + } } } } @@ -422,12 +439,18 @@ fn run_group_stream( } let successful_targets = outcome.successful_target_outcomes; let failed_targets = outcome.failed_target_outcomes; - 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) + let peer_closed_target_outcomes = outcome + .peer_closed_target_outcomes + .max(u64::from(peer_close_requested_stop)); + let terminal_error = if let Some(error) = + peer_close_run_error(continuous, interrupted, peer_closed_target_outcomes) { Some(error) + } else if outcome.end_reason == ManagedGroupEndReason::Cancelled + && !interrupted + && !peer_close_requested_stop + { + Some("managed client group was cancelled".to_owned()) } else if !interrupted && successful_targets == 0 && failed_targets > 0 { Some(format!( "no managed target completed successfully ({failed_targets} failed)" diff --git a/crates/irtt-cli/src/cmd/tui/run.rs b/crates/irtt-cli/src/cmd/tui/run.rs index 4178431..1a633d9 100644 --- a/crates/irtt-cli/src/cmd/tui/run.rs +++ b/crates/irtt-cli/src/cmd/tui/run.rs @@ -14,7 +14,11 @@ use irtt_client::{ use crate::{ cmd::tui::args::{ResolvedTuiTarget, TuiArgs}, - shared::client::{is_shutdown_requested, session::peer_close_run_error, ClientSession}, + shared::client::{ + is_shutdown_requested, + session::{peer_close_run_error, request_group_stop_once, should_stop_group_on_peer_close}, + ClientSession, + }, }; use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal}; @@ -177,6 +181,8 @@ fn run_group_tui( )?; let mut interrupted = false; + let mut stop_requested = false; + let mut peer_close_requested_stop = false; let mut terminal_targets = std::collections::HashSet::new(); let mut saw_target_event = false; let mut last_event_at = Instant::now(); @@ -184,7 +190,9 @@ fn run_group_tui( let exit = loop { if is_shutdown_requested(shutdown_requested) { interrupted = true; - session.stop(); + if request_group_stop_once(&mut stop_requested) { + session.stop(); + } } if handle_input(state, shutdown_requested)? { @@ -192,7 +200,9 @@ fn run_group_tui( } if state.quit_requested { interrupted = true; - session.stop(); + if request_group_stop_once(&mut stop_requested) { + session.stop(); + } } match events.try_recv() { @@ -206,6 +216,16 @@ fn run_group_tui( ManagedGroupEvent::TargetFinished(target) => { terminal_targets.insert(target.id.as_str().to_owned()); state.process_target_outcome(&target); + if should_stop_group_on_peer_close( + args.is_continuous(), + interrupted, + &target.end_reason, + ) { + peer_close_requested_stop = true; + if request_group_stop_once(&mut stop_requested) { + session.stop(); + } + } } } render_if_due(terminal, state, next_render, false)?; @@ -229,7 +249,7 @@ fn run_group_tui( } }; - if exit.should_stop_before_join() { + if exit.should_stop_before_join() && request_group_stop_once(&mut stop_requested) { session.stop(); } @@ -260,7 +280,22 @@ fn run_group_tui( } interrupted |= is_shutdown_requested(shutdown_requested); - if outcome.end_reason == ManagedGroupEndReason::Cancelled && !interrupted { + let peer_closed_target_outcomes = outcome + .peer_closed_target_outcomes + .max(u64::from(peer_close_requested_stop)); + if let Some(error) = peer_close_run_error( + args.is_continuous(), + interrupted, + peer_closed_target_outcomes, + ) { + state.set_run_error(error.clone()); + render_if_due(terminal, state, next_render, true)?; + return Err(error.into()); + } + if outcome.end_reason == ManagedGroupEndReason::Cancelled + && !interrupted + && !peer_close_requested_stop + { return match exit { GroupLoopExit::IdleGraceElapsed => { Err("managed client group stayed idle before all targets completed".into()) @@ -273,15 +308,6 @@ 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 { @@ -507,4 +533,23 @@ mod tests { assert!(GroupLoopExit::SubscriptionDisconnected.should_stop_before_join()); assert!(!GroupLoopExit::AllTargetsTerminal.should_stop_before_join()); } + + #[test] + fn continuous_group_peer_close_requests_stop_unless_interrupted() { + assert!(should_stop_group_on_peer_close( + true, + false, + &irtt_client::ManagedTargetEndReason::PeerClosed + )); + assert!(!should_stop_group_on_peer_close( + false, + false, + &irtt_client::ManagedTargetEndReason::PeerClosed + )); + assert!(!should_stop_group_on_peer_close( + true, + true, + &irtt_client::ManagedTargetEndReason::PeerClosed + )); + } } diff --git a/crates/irtt-cli/src/shared/client/session.rs b/crates/irtt-cli/src/shared/client/session.rs index fb567b0..444173f 100644 --- a/crates/irtt-cli/src/shared/client/session.rs +++ b/crates/irtt-cli/src/shared/client/session.rs @@ -4,7 +4,9 @@ use std::{ time::{Duration, Instant}, }; -use irtt_client::{Client, ClientConfig, ClientEvent, OpenOutcome, RecvBudget}; +use irtt_client::{ + Client, ClientConfig, ClientEvent, ManagedTargetEndReason, OpenOutcome, RecvBudget, +}; pub const RECV_BUDGET: RecvBudget = RecvBudget { max_packets: 16 }; const MAX_FINAL_DRAIN: Duration = Duration::from_secs(30); @@ -162,6 +164,22 @@ pub fn peer_close_run_error( )) } +pub fn should_stop_group_on_peer_close( + continuous: bool, + interrupted: bool, + end_reason: &ManagedTargetEndReason, +) -> bool { + continuous && !interrupted && matches!(end_reason, ManagedTargetEndReason::PeerClosed) +} + +pub fn request_group_stop_once(stop_requested: &mut bool) -> bool { + if *stop_requested { + return false; + } + *stop_requested = true; + true +} + pub fn final_drain_duration(probe_timeout: Duration) -> Duration { probe_timeout.min(MAX_FINAL_DRAIN) } @@ -239,6 +257,34 @@ mod tests { ); } + #[test] + fn grouped_peer_close_stop_policy_is_run_mode_and_interruption_aware() { + assert!(should_stop_group_on_peer_close( + true, + false, + &ManagedTargetEndReason::PeerClosed + )); + assert!(!should_stop_group_on_peer_close( + false, + false, + &ManagedTargetEndReason::PeerClosed + )); + assert!(!should_stop_group_on_peer_close( + true, + true, + &ManagedTargetEndReason::PeerClosed + )); + assert!(!should_stop_group_on_peer_close( + true, + false, + &ManagedTargetEndReason::TestComplete + )); + + let mut stop_requested = false; + assert!(request_group_stop_once(&mut stop_requested)); + assert!(!request_group_stop_once(&mut stop_requested)); + } + #[test] fn peer_close_followed_by_session_cleanup_is_successful() { let params = test_params(None, Duration::from_millis(10)); diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs index e5654ca..62701cc 100644 --- a/crates/irtt-cli/tests/client_multi_target.rs +++ b/crates/irtt-cli/tests/client_multi_target.rs @@ -4,6 +4,7 @@ use std::{ io::Read, net::{SocketAddr, UdpSocket}, process::{Command, Output, Stdio}, + sync::mpsc, thread, thread::JoinHandle, time::{Duration, Instant}, @@ -29,6 +30,24 @@ impl FakeServer { } } +struct InterruptibleFakeServer { + addr: SocketAddr, + opened: mpsc::Receiver<()>, + done: JoinHandle<()>, +} + +impl InterruptibleFakeServer { + fn wait_until_opened(&self) { + self.opened + .recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting for CLI to open target"); + } + + fn join(self) { + self.done.join().unwrap(); + } +} + #[test] fn list_columns_succeeds_without_target() { let output = Command::new(env!("CARGO_BIN_EXE_irtt-cli")) @@ -337,10 +356,11 @@ fn continuous_all_peer_closed_targets_exit_nonzero() { } #[test] -fn continuous_mixed_peer_close_and_open_failure_exits_nonzero() { +fn continuous_partial_peer_close_stops_promptly_and_preserves_queued_rows() { 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 (healthy_reply_tx, healthy_reply_rx) = mpsc::channel(); + let healthy = start_echo_server_with_first_reply(params.clone(), healthy_reply_tx); + let peer_closed = start_gated_peer_close_server(params, vec![healthy_reply_rx]); let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli")); command.args([ "--duration", @@ -354,14 +374,79 @@ fn continuous_mixed_peer_close_and_open_failure_exits_nonzero() { "--header", "never", "--target", + &format!("healthy={}", healthy.addr), + "--target", &format!("peer={}", peer_closed.addr), + ]); + + let started_at = Instant::now(); + let output = run_with_timeout(command, Duration::from_secs(2)); + let elapsed = started_at.elapsed(); + healthy.join(); + peer_closed.join(); + + assert!( + !output.status.success(), + "status={:?}\nstdout={}", + output.status.code(), + String::from_utf8_lossy(&output.stdout) + ); + assert!( + elapsed < Duration::from_secs(1), + "partial peer-close shutdown took {elapsed:?}" + ); + let stdout = String::from_utf8(output.stdout).unwrap(); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert!( + stdout.lines().any(|line| line.starts_with("healthy,")), + "{stdout}" + ); + assert!( + stdout.lines().any(|line| line.starts_with("peer,")), + "{stdout}" + ); + assert!( + stderr.contains("peer closure (1 target session)"), + "{stderr}" + ); + assert!( + !stderr.contains("managed client group was cancelled"), + "{stderr}" + ); +} + +#[test] +fn continuous_mixed_peer_close_and_open_failure_exits_nonzero() { + let params = test_params(None, Duration::from_millis(10)); + let (failure_tx, failure_rx) = mpsc::channel(); + let (healthy_reply_tx, healthy_reply_rx) = mpsc::channel(); + let failure = start_open_failure_server_with_signal(params.clone(), failure_tx); + let healthy = start_echo_server_with_first_reply(params.clone(), healthy_reply_tx); + let peer_closed = start_gated_peer_close_server(params, vec![failure_rx, healthy_reply_rx]); + 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!("failure={}", failure.addr), + "--target", + &format!("healthy={}", healthy.addr), + "--target", + &format!("peer={}", peer_closed.addr), ]); let output = run_with_timeout(command, Duration::from_secs(3)); - peer_closed.join(); failure.join(); + healthy.join(); + peer_closed.join(); assert!( !output.status.success(), @@ -375,6 +460,10 @@ fn continuous_mixed_peer_close_and_open_failure_exits_nonzero() { stdout.lines().any(|line| line.starts_with("peer,")), "{stdout}" ); + assert!( + stdout.lines().any(|line| line.starts_with("healthy,")), + "{stdout}" + ); assert!(stderr.contains("target failure failed"), "{stderr}"); assert!(stderr.contains("zero token without close flag"), "{stderr}"); assert!( @@ -383,6 +472,67 @@ fn continuous_mixed_peer_close_and_open_failure_exits_nonzero() { ); } +#[cfg(unix)] +#[test] +fn explicit_group_interruption_succeeds_without_peer_close_error() { + let params = test_params(None, Duration::from_millis(10)); + let a = start_interruptible_echo_server(params.clone()); + let b = start_interruptible_echo_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 child = command + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + a.wait_until_opened(); + b.wait_until_opened(); + thread::sleep(Duration::from_millis(50)); + let signal_status = Command::new("kill") + .args(["-INT", &child.id().to_string()]) + .status() + .unwrap(); + assert!(signal_status.success()); + let output = wait_with_timeout(child, Duration::from_secs(2)); + 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("interrupted, closing group"), "{stderr}"); + assert!(!stderr.contains("peer closure"), "{stderr}"); +} + #[test] fn all_open_failures_exit_nonzero_with_diagnostics() { let params = test_params(None, Duration::from_millis(10)); @@ -420,11 +570,15 @@ fn all_open_failures_exit_nonzero_with_diagnostics() { } fn run_with_timeout(mut command: Command, timeout: Duration) -> Output { - let mut child = command + let child = command .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn() .unwrap(); + wait_with_timeout(child, timeout) +} + +fn wait_with_timeout(mut child: std::process::Child, timeout: Duration) -> Output { let deadline = Instant::now() + timeout; loop { if child.try_wait().unwrap().is_some() { @@ -452,6 +606,20 @@ fn run_with_timeout(mut command: Command, timeout: Duration) -> Output { } fn start_echo_server(params: Params) -> FakeServer { + start_echo_server_inner(params, None) +} + +fn start_echo_server_with_first_reply( + params: Params, + first_reply_tx: mpsc::Sender<()>, +) -> FakeServer { + start_echo_server_inner(params, Some(first_reply_tx)) +} + +fn start_echo_server_inner( + params: Params, + mut first_reply_tx: Option>, +) -> FakeServer { let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); socket .set_read_timeout(Some(Duration::from_secs(2))) @@ -482,12 +650,62 @@ fn start_echo_server(params: Params) -> FakeServer { peer, ) .unwrap(); + if let Some(tx) = first_reply_tx.take() { + tx.send(()).unwrap(); + } } }); FakeServer { addr, done } } +fn start_interruptible_echo_server(params: Params) -> InterruptibleFakeServer { + 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 (opened_tx, opened) = mpsc::channel(); + let done = thread::spawn(move || { + let (_, peer) = recv_request(&socket); + opened_tx.send(()).unwrap(); + socket + .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), peer) + .unwrap(); + socket + .set_read_timeout(Some(Duration::from_millis(500))) + .unwrap(); + + while let Some((packet, peer)) = recv_request_timeout(&socket) { + let flags = packet[3]; + if flags & FLAG_CLOSE != 0 { + break; + } + let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap()); + socket + .send_to( + &echo_reply_packet( + TOKEN, + seq, + ¶ms, + &TimestampFields::default(), + FLAG_REPLY, + ), + peer, + ) + .unwrap(); + } + }); + InterruptibleFakeServer { addr, opened, done } +} + fn start_peer_close_server(params: Params) -> FakeServer { + start_gated_peer_close_server(params, Vec::new()) +} + +fn start_gated_peer_close_server( + params: Params, + release_gates: Vec>, +) -> FakeServer { let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); socket .set_read_timeout(Some(Duration::from_secs(2))) @@ -499,6 +717,10 @@ fn start_peer_close_server(params: Params) -> FakeServer { .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), peer) .unwrap(); let (packet, peer) = recv_request(&socket); + for gate in release_gates { + gate.recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting to release peer-close reply"); + } let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap()); socket .send_to( @@ -517,6 +739,20 @@ fn start_peer_close_server(params: Params) -> FakeServer { } fn start_open_failure_server(params: Params) -> FakeServer { + start_open_failure_server_inner(params, None) +} + +fn start_open_failure_server_with_signal( + params: Params, + failure_tx: mpsc::Sender<()>, +) -> FakeServer { + start_open_failure_server_inner(params, Some(failure_tx)) +} + +fn start_open_failure_server_inner( + params: Params, + failure_tx: Option>, +) -> FakeServer { let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); socket .set_read_timeout(Some(Duration::from_secs(2))) @@ -527,6 +763,9 @@ fn start_open_failure_server(params: Params) -> FakeServer { socket .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, 0, ¶ms), peer) .unwrap(); + if let Some(tx) = failure_tx { + tx.send(()).unwrap(); + } }); FakeServer { addr, done } } @@ -537,6 +776,14 @@ fn recv_request(socket: &UdpSocket) -> (Vec, SocketAddr) { (buf[..size].to_vec(), peer) } +fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec, SocketAddr)> { + let mut buf = [0_u8; 2048]; + socket + .recv_from(&mut buf) + .ok() + .map(|(size, peer)| (buf[..size].to_vec(), peer)) +} + fn open_reply(flags: u8, token: u64, params: &Params) -> Vec { let mut packet = Vec::new(); packet.extend_from_slice(&MAGIC);