test(cli): synchronize multi-target peer close

crates/irtt-cli/tests/client_multi_target.rs

@@ -4,7 +4,7 @@ use std::{
     io::Read,
     net::{SocketAddr, UdpSocket},
     process::{Command, Output, Stdio},
-    sync::mpsc,
+    sync::{mpsc, Arc, Barrier},
     thread,
     thread::JoinHandle,
     time::{Duration, Instant},
@@ -425,8 +425,9 @@ fn finite_multi_target_peer_close_is_accepted_as_completion() {
 #[test]
 fn continuous_all_peer_closed_targets_exit_nonzero() {
     let params = test_params(None, Duration::from_millis(10));
-    let a = start_peer_close_server(params.clone());
+    let first_probe_barrier = Arc::new(Barrier::new(2));
-    let b = start_peer_close_server(params);
+    let a = start_synchronized_peer_close_server(params.clone(), Arc::clone(&first_probe_barrier));
+    let b = start_synchronized_peer_close_server(params, first_probe_barrier);
     let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli"));
     command.args([
         "--duration",
@@ -860,6 +861,49 @@ fn start_peer_close_server(params: Params) -> FakeServer {
     start_gated_peer_close_server(params, Vec::new())
 }
 
+fn start_synchronized_peer_close_server(
+    params: Params,
+    first_probe_barrier: Arc<Barrier>,
+) -> FakeServer {
+    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 done = thread::spawn(move || {
+        let (_, peer) = recv_request(&socket);
+        socket
+            .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
+            .unwrap();
+        let (packet, peer) = recv_request(&socket);
+        let flags = packet[3];
+        assert_eq!(
+            flags, 0,
+            "target closed before synchronized first peer-close probe"
+        );
+        assert_eq!(
+            packet.len(),
+            echo_packet_len(false, &params).unwrap(),
+            "expected a complete echo probe before synchronized peer-close reply"
+        );
+        let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap());
+        first_probe_barrier.wait();
+        socket
+            .send_to(
+                &echo_reply_packet(
+                    TOKEN,
+                    seq,
+                    &params,
+                    &TimestampFields::default(),
+                    FLAG_REPLY | FLAG_CLOSE,
+                ),
+                peer,
+            )
+            .unwrap();
+    });
+    FakeServer { addr, done }
+}
+
 fn start_gated_peer_close_server(
     params: Params,
     release_gates: Vec<mpsc::Receiver<()>>,
@@ -875,10 +919,19 @@ fn start_gated_peer_close_server(
             .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
             .unwrap();
         let (packet, peer) = recv_request(&socket);
+        let flags = packet[3];
+        if flags & FLAG_CLOSE != 0 {
+            return;
+        }
         for gate in release_gates {
             gate.recv_timeout(Duration::from_secs(2))
                 .expect("timed out waiting to release peer-close reply");
         }
+        assert_eq!(
+            packet.len(),
+            echo_packet_len(false, &params).unwrap(),
+            "expected a complete echo probe before peer-close reply"
+        );
         let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap());
         socket
             .send_to(