diff --git a/crates/irtt-app/tests/cli.rs b/crates/irtt-app/tests/cli.rs index a2c3939..206662e 100644 --- a/crates/irtt-app/tests/cli.rs +++ b/crates/irtt-app/tests/cli.rs @@ -256,3 +256,108 @@ fn server_command_line_binds_replace_the_environment_bind() { ); } } + +// Exercise stdin replacement and EOF through the actual application process. +#[cfg(feature = "client")] +#[test] +fn targets_stdin_supersedes_the_active_set_and_stops_on_eof() { + use std::{ + io::{BufRead, BufReader, Read, Write}, + process::{Child, Stdio}, + sync::mpsc, + thread, + time::{Duration, Instant}, + }; + struct ClientProcess(Child); + impl Drop for ClientProcess { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } + } + let server = in_tree_server::InTreeServer::start(ServerConfig::default()); + let mut client = ClientProcess( + Command::new(env!("CARGO_BIN_EXE_irtt-client")) + .env_clear() + .args([ + "--duration", + "0", + "--targets-stdin", + "--interval", + "20ms", + "--format", + "jsonl", + "--columns", + "target,event", + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(), + ); + let stdout = client.0.stdout.take().unwrap(); + let (tx, rx) = mpsc::channel(); + let reader = thread::spawn(move || { + for line in BufReader::new(stdout).lines() { + if tx.send(line.unwrap()).is_err() { + break; + } + } + }); + let mut stdin = client.0.stdin.take().unwrap(); + let mut rows = Vec::new(); + for target in ["first", "second"] { + writeln!(stdin, "{target}={}", server.addr).unwrap(); + stdin.flush().unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let row = rx + .recv_timeout(deadline.saturating_duration_since(Instant::now())) + .expect("stdin target set did not produce a reply"); + let replied = row.contains(&format!("\"target\":\"{target}\"")) + && row.contains("\"event\":\"echo_reply\""); + rows.push(row); + if replied { + break; + } + } + } + // Retirement can finish after the newer target's first reply. + let deadline = Instant::now() + Duration::from_secs(5); + while !rows.iter().any(|row| { + row.contains("\"target\":\"first\"") && row.contains("\"event\":\"session_closed\"") + }) { + rows.push( + rx.recv_timeout(deadline.saturating_duration_since(Instant::now())) + .expect("superseded stdin target did not retire"), + ); + } + drop(stdin); + let deadline = Instant::now() + Duration::from_secs(5); + let status = loop { + if let Some(status) = client.0.try_wait().unwrap() { + break status; + } + assert!( + Instant::now() < deadline, + "client did not terminate on stdin EOF" + ); + thread::sleep(Duration::from_millis(10)); + }; + reader.join().unwrap(); + rows.extend(rx.try_iter()); + let mut stderr = String::new(); + client + .0 + .stderr + .take() + .unwrap() + .read_to_string(&mut stderr) + .unwrap(); + assert!(status.success(), "{stderr}"); + assert!(rows + .iter() + .any(|row| row.contains("\"target\":\"second\"") + && row.contains("\"event\":\"session_closed\""))); +} diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 7ec2e9a..b54f931 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -1139,6 +1139,56 @@ mod tests { use super::*; use irtt_proto::{FLAG_OPEN, FLAG_REPLY}; + #[test] + fn timing_arithmetic_preserves_negative_samples() { + let mono = Instant::now(); + let sent = ClientTimestamp { + mono, + wall: UNIX_EPOCH + Duration::from_nanos(100), + }; + let received = ClientTimestamp { + mono: mono + Duration::from_nanos(10), + wall: UNIX_EPOCH + Duration::from_nanos(110), + }; + // Processing longer than raw RTT must remain negative with either clock. + for timestamps in [ + TimestampFields { + recv_mono: Some(100), + send_mono: Some(120), + ..TimestampFields::default() + }, + TimestampFields { + recv_wall: Some(100), + send_wall: Some(120), + ..TimestampFields::default() + }, + ] { + let rtt = compute_rtt(&sent, &received, ×tamps); + assert_eq!(rtt.raw, Duration::from_nanos(10)); + assert_eq!(rtt.adjusted.unwrap().as_nanos(), -10); + assert_eq!(rtt.effective.as_nanos(), -10); + } + // A server clock behind or ahead can make either direction negative. + for (recv_wall, send_wall, upstream, downstream) in [(90, 95, -10, 15), (115, 120, 15, -10)] + { + let sample = compute_one_way( + &sent, + sent.wall, + &received, + None, + &ReceiveMeta::default(), + &TimestampFields { + recv_wall: Some(recv_wall), + send_wall: Some(send_wall), + ..TimestampFields::default() + }, + ) + .unwrap(); + assert_eq!(sample.client_to_server.unwrap().as_nanos(), upstream); + assert_eq!(sample.server_to_client.unwrap().as_nanos(), downstream); + } + } + fn opened() -> SessionMachine { let mut machine = SessionMachine::new( ClientConfig { diff --git a/crates/irtt-integration-tests/tests/managed_behavior.rs b/crates/irtt-integration-tests/tests/managed_behavior.rs index c709cbe..9972f43 100644 --- a/crates/irtt-integration-tests/tests/managed_behavior.rs +++ b/crates/irtt-integration-tests/tests/managed_behavior.rs @@ -97,6 +97,7 @@ async fn target_updates_preserve_identical_sessions_and_retire_removed_generatio ..ClientConfig::default() }, completion: ManagedCompletionPolicy::ExplicitStop, + outcome_history_limit: 1, ..ManagedClientConfig::default() }; let configured = ManagedTargetConfig::new("one", server.addr.to_string()); @@ -173,6 +174,17 @@ async fn target_updates_preserve_identical_sessions_and_retire_removed_generatio stop.await; assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); assert_eq!(outcome.total_target_outcomes, 2); + assert_eq!(outcome.successful_target_outcomes, 2); + assert_eq!(outcome.failed_target_outcomes, 0); + assert_eq!(outcome.discarded_target_outcomes, 1); + assert_eq!(outcome.recent_target_outcomes.len(), 1); + let status = handle.status(); + assert_eq!(status.total_target_outcomes, 2); + assert_eq!(status.successful_target_outcomes, 2); + assert_eq!(status.failed_target_outcomes, 0); + assert_eq!(status.discarded_target_outcomes, 1); + assert_eq!(status.recent_target_outcomes.len(), 1); + assert_eq!(status.recent_target_outcomes[0].target, second); assert!(outcome.recent_target_outcomes.iter().any(|target| { target.target == second && target.end_reason == ManagedTargetEndReason::Stopped diff --git a/crates/irtt-server/tests/protocol_contracts.rs b/crates/irtt-server/tests/protocol_contracts.rs index 761efb0..319be48 100644 --- a/crates/irtt-server/tests/protocol_contracts.rs +++ b/crates/irtt-server/tests/protocol_contracts.rs @@ -321,39 +321,50 @@ fn negotiated_timestamp_modes_emit_only_the_selected_fields_in_order() { fn duration_close_waits_for_rate_allowance_after_the_deadline() { let refill = Duration::from_secs(4); let maximum = Duration::from_millis(1); - let mut server = ServerCore::new( - ServerConfig::default() - .with_min_send_interval(refill) - .with_burst_allowance(1) - .with_idle_timeout(Duration::MAX) - .with_max_test_duration(maximum), - ); - let opened = open( - &mut server, - &Params { - received_stats: ReceivedStats::Both, - ..Params::with_protocol_defaults() - }, - ); - let first = echo(&mut server, &opened, 0).unwrap(); - assert_eq!(first.flags & FLAG_CLOSE, 0); - thread::sleep(maximum + Duration::from_secs(2)); - assert!( - echo(&mut server, &opened, 63).is_none(), - "exhausted allowance must suppress the deadline-crossing reply" - ); - assert_eq!( - server.session_count(), - 1, - "rate refusal must not release the session" - ); - thread::sleep(refill); - let closed = echo(&mut server, &opened, 1).unwrap(); - assert_ne!(closed.flags & FLAG_CLOSE, 0); - assert_eq!(closed.sequence, 1); - assert_eq!(closed.recv_count, Some(2)); - assert_eq!(closed.recv_window, Some(3)); - assert_eq!(server.session_count(), 0); + for _ in 0..3 { + let mut server = ServerCore::new( + ServerConfig::default() + .with_min_send_interval(refill) + .with_burst_allowance(1) + .with_idle_timeout(Duration::MAX) + .with_max_test_duration(maximum), + ); + let started = Instant::now(); + let opened = open( + &mut server, + &Params { + received_stats: ReceivedStats::Both, + ..Params::with_protocol_defaults() + }, + ); + let first = echo(&mut server, &opened, 0).unwrap(); + assert_eq!(first.flags & FLAG_CLOSE, 0); + thread::sleep(maximum + Duration::from_secs(2)); + let refused = echo(&mut server, &opened, 63); + // Include open/setup and the request itself in the elapsed bound. If the + // scheduler carried us past refill, the reply may legitimately close. + if started.elapsed() >= refill { + continue; + } + assert!( + refused.is_none(), + "exhausted allowance must suppress the deadline-crossing reply" + ); + assert_eq!( + server.session_count(), + 1, + "rate refusal must not release the session" + ); + thread::sleep(refill); + let closed = echo(&mut server, &opened, 1).unwrap(); + assert_ne!(closed.flags & FLAG_CLOSE, 0); + assert_eq!(closed.sequence, 1); + assert_eq!(closed.recv_count, Some(2)); + assert_eq!(closed.recv_window, Some(3)); + assert_eq!(server.session_count(), 0); + return; + } + panic!("scheduler crossed the refill boundary in all three fresh sessions"); } #[test]