test: address review gaps in timing and dynamic targets

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\"")));
+}

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, &timestamps);
+            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 {

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

crates/irtt-server/tests/protocol_contracts.rs

@@ -321,6 +321,7 @@ 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);
+    for _ in 0..3 {
         let mut server = ServerCore::new(
             ServerConfig::default()
                 .with_min_send_interval(refill)
@@ -328,6 +329,7 @@ fn duration_close_waits_for_rate_allowance_after_the_deadline() {
                 .with_idle_timeout(Duration::MAX)
                 .with_max_test_duration(maximum),
         );
+        let started = Instant::now();
         let opened = open(
             &mut server,
             &Params {
@@ -338,8 +340,14 @@ fn duration_close_waits_for_rate_allowance_after_the_deadline() {
         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!(
-        echo(&mut server, &opened, 63).is_none(),
+            refused.is_none(),
             "exhausted allowance must suppress the deadline-crossing reply"
         );
         assert_eq!(
@@ -354,6 +362,9 @@ fn duration_close_waits_for_rate_allowance_after_the_deadline() {
         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]