test: speed up integration tests with channel notification and parallel execution

AGENTS.md

@@ -29,9 +29,15 @@ Full CI-equivalent before PRs, shared transport/request/response changes, or unc
 cargo fmt --check
 cargo clippy --locked --all-targets --all-features -- -D warnings
 cargo test --locked --all-features --lib --bins
-cargo test --locked --all-features --test cli --test formatting --test grpc --test http --test network --test terminal --test update --test websocket -- --test-threads=1
+cargo test --locked --all-features --test cli --test formatting --test grpc --test http --test network --test terminal --test update --test websocket
 ```
 
+The integration test suite uses `TcpListener::bind("127.0.0.1:0")` and
+`UdpSocket::bind("127.0.0.1:0")` for all test servers, so tests are safe to run in
+parallel. If transient "Connection refused" errors occur from port reuse races, add
+`-- --test-threads=1` to force sequential execution. The `TestServer` and
+`H3TestServer` use `mpsc` channels instead of polling for request notification.
+
 Docs-only changes: skip Cargo unless examples/generated CLI output changed; format changed docs only:
 
 ```bash
@@ -58,7 +64,7 @@ cargo build --release --locked
 | gRPC/protobuf | `src/grpc`, `src/proto` | Framing/status/reflection, local schema/discovery/conversion/JSON streams. Reuse standard gRPC headers/status/framed-body helpers. |
 | WebSocket | `src/websocket` | Interactive and non-interactive message loops; custom dialer for DNS/proxy/TLS. |
 | Auth/session/update | `src/auth`, `src/session.rs`, `src/update`, `install.sh` | Auth helpers; locked cookie sessions; HTTPS self-update with checksum/archive validation and no origin-specific TLS overrides. |
-| Tests | `tests/`, `tests/support/` | Integration tests run the compiled binary; support code is split by domain. `run_fetch` isolates HTTP/3 cache by default. |
+| Tests | `tests/`, `tests/support/` | Integration tests run the compiled binary; support code is split by domain. `run_fetch` isolates HTTP/3 cache by default. `TestServer`/`H3TestServer` use `mpsc` channel notification (not polling). `wait_for_requests` blocks via `recv_timeout` on the notification channel. |
 
 Request flow: CLI parse → config merge → request build (gRPC may load/reflect schema and frame protobuf) → transport execute → response format/output/pager/clipboard.
 

src/output/clipboard.rs

@@ -156,23 +156,42 @@ fn write_to_command(command: &ClipboardCommand, bytes: &[u8]) -> CopyOutcome {
     write_to_command_with_timeout(command, bytes, CLIPBOARD_COMMAND_TIMEOUT)
 }
 
+fn spawn_with_retry(command: &ClipboardCommand) -> Result<Child, String> {
+    let mut last_error = String::new();
+    for i in 0..5 {
+        match Command::new(command.program)
+            .args(command.args)
+            .stdin(Stdio::piped())
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .spawn()
+        {
+            Ok(child) => return Ok(child),
+            Err(err) => {
+                last_error = err.to_string();
+                if err.raw_os_error() == Some(26) {
+                    // ETXTBUSY: transient; retry with backoff.
+                    thread::sleep(Duration::from_millis(10 << i));
+                    continue;
+                }
+                return Err(last_error);
+            }
+        }
+    }
+    Err(last_error)
+}
+
 fn write_to_command_with_timeout(
     command: &ClipboardCommand,
     bytes: &[u8],
     timeout: Duration,
 ) -> CopyOutcome {
-    let mut child = match Command::new(command.program)
-        .args(command.args)
-        .stdin(Stdio::piped())
-        .stdout(Stdio::null())
-        .stderr(Stdio::null())
-        .spawn()
-    {
+    let mut child = match spawn_with_retry(command) {
         Ok(child) => child,
-        Err(err) => {
+        Err(message) => {
             return CopyOutcome::Failed {
                 command: command.label(),
-                message: err.to_string(),
+                message,
             };
         }
     };

tests/http.rs

@@ -100,7 +100,7 @@ fn schemeless_https_connect_error_suggests_plaintext_url() {
     let join = thread::spawn(move || {
         for _ in 0..2 {
             let Ok(mut stream) =
-                accept_tcp_connection(&listener, Duration::from_secs(5), "plaintext TLS hint")
+                accept_tcp_connection(&listener, Duration::from_secs(3), "plaintext TLS hint")
             else {
                 return;
             };

tests/network.rs

@@ -66,7 +66,7 @@ fn custom_dns_connects_after_fast_a_without_waiting_for_slow_aaaa() {
     let dns_addr = start_udp_dns_server_with_delayed_aaaa(
         "fetch-happy-a-first.test.",
         Ipv4Addr::new(127, 0, 0, 1),
-        Duration::from_secs(4),
+        Duration::from_secs(1),
     );
 
     let start = std::time::Instant::now();
@@ -322,7 +322,7 @@ fn http3_go_harness_cases() {
             return H3Response::ok(body)
                 .header("Content-Type", "text/plain")
                 .header("Content-Encoding", "zstd")
-                .delay_body(Duration::from_millis(100));
+                .delay_body(Duration::from_millis(50));
         }
         if req.path == "/h3-grpc-trailer-error" {
             return H3Response::ok("")
@@ -729,7 +729,7 @@ fn default_https_races_cached_alt_svc_without_waiting_for_slow_https_record_look
             }
             TestResponse::status(404, "Not Found", "")
         },
-        Duration::from_millis(250),
+        Duration::from_millis(100),
     );
     let h2_url = format!("{}/cached-alt-svc-race", h2.url);
     let dns_addr = start_udp_dns_server_dropping_https("localhost.", Ipv4Addr::new(127, 0, 0, 1));

tests/support/common.rs

@@ -116,7 +116,7 @@ pub(crate) fn run_fetch_once(opts: FetchOpts, args: &[&str]) -> FetchOutput {
         if child.try_wait().expect("poll fetch").is_some() {
             break;
         }
-        if start.elapsed() > Duration::from_secs(30) {
+        if start.elapsed() > Duration::from_secs(15) {
             let _ = child.kill();
             let out = child.wait_with_output().expect("wait killed fetch");
             return FetchOutput {

tests/support/http.rs

@@ -85,8 +85,9 @@ impl TestResponse {
 pub(crate) struct TestServer {
     pub(crate) url: String,
     pub(crate) requests: Arc<Mutex<Vec<TestRequest>>>,
-    pub(crate) shutdown: Option<mpsc::Sender<()>>,
-    pub(crate) join: Option<thread::JoinHandle<()>>,
+    request_notify: mpsc::Receiver<()>,
+    shutdown: Option<mpsc::Sender<()>>,
+    join: Option<thread::JoinHandle<()>>,
 }
 
 pub(crate) struct PartialBodyReplayServer {
@@ -106,24 +107,27 @@ impl TestServer {
         let url = format!("http://{}", listener.local_addr().expect("local addr"));
         let requests = Arc::new(Mutex::new(Vec::new()));
         let handler = Arc::new(handler);
-        let (tx, rx) = mpsc::channel();
+        let (shutdown_tx, shutdown_rx) = mpsc::channel();
+        let (notify_tx, notify_rx) = mpsc::channel();
         let request_log = Arc::clone(&requests);
         let join = thread::spawn(move || {
             loop {
-                if rx.try_recv().is_ok() {
+                if shutdown_rx.try_recv().is_ok() {
                     break;
                 }
                 match listener.accept() {
                     Ok((stream, _)) => {
                         let _ = stream.set_nonblocking(false);
                         let handler = Arc::clone(&handler);
                         let request_log = Arc::clone(&request_log);
+                        let notify = notify_tx.clone();
                         thread::spawn(move || {
                             let mut writer = stream.try_clone().expect("clone response stream");
                             let mut reader = BufReader::new(stream);
                             while let Some(req) = read_request(&mut reader) {
                                 let close = req.header("connection").eq_ignore_ascii_case("close");
                                 request_log.lock().unwrap().push(req.clone());
+                                let _ = notify.send(());
                                 let resp = handler(req);
                                 write_response(&mut writer, resp);
                                 if close {
@@ -142,7 +146,8 @@ impl TestServer {
         Self {
             url,
             requests,
-            shutdown: Some(tx),
+            request_notify: notify_rx,
+            shutdown: Some(shutdown_tx),
             join: Some(join),
         }
     }
@@ -178,7 +183,7 @@ impl PartialBodyReplayServer {
         let requests = Arc::new(Mutex::new(Vec::new()));
         let request_log = Arc::clone(&requests);
         let join = thread::spawn(move || {
-            let deadline = Instant::now() + Duration::from_secs(10);
+            let deadline = Instant::now() + Duration::from_secs(5);
             let mut first = true;
             while Instant::now() < deadline {
                 match listener.accept() {
@@ -206,7 +211,7 @@ impl PartialBodyReplayServer {
                             let _ = stream.write_all(&body);
                             let _ = stream.flush();
                             thread::spawn(move || {
-                                let deadline = Instant::now() + Duration::from_secs(5);
+                                let deadline = Instant::now() + Duration::from_secs(3);
                                 let _ = stream.set_read_timeout(Some(Duration::from_millis(100)));
                                 let mut buf = [0_u8; 1024];
                                 while Instant::now() < deadline {
@@ -355,16 +360,16 @@ pub(crate) fn write_response(stream: &mut impl Write, resp: TestResponse) {
 
 pub(crate) fn wait_for_requests(server: &TestServer, count: usize) -> Vec<TestRequest> {
     let start = Instant::now();
-    loop {
-        let requests = server.requests();
-        if requests.len() >= count {
-            return requests;
+    while server.requests().len() < count {
+        let remaining = Duration::from_secs(2).saturating_sub(start.elapsed());
+        if remaining.is_zero() {
+            let requests = server.requests();
+            panic!(
+                "timed out waiting for {count} requests; got {}",
+                requests.len()
+            );
         }
-        assert!(
-            start.elapsed() < Duration::from_secs(2),
-            "timed out waiting for {count} requests; got {}",
-            requests.len()
-        );
-        thread::sleep(Duration::from_millis(10));
+        let _ = server.request_notify.recv_timeout(remaining);
     }
+    server.requests()
 }

tests/support/http3.rs

@@ -90,23 +90,24 @@ pub(crate) struct H3TestServer {
     pub(crate) url: String,
     pub(crate) ca_cert_path: PathBuf,
     pub(crate) requests: Arc<Mutex<Vec<H3Request>>>,
+    request_notify: mpsc::Receiver<()>,
     pub(crate) connections: Arc<AtomicUsize>,
 }
 
 pub(crate) fn wait_for_h3_requests(server: &H3TestServer, count: usize) -> Vec<H3Request> {
     let start = Instant::now();
-    loop {
-        let requests = server.requests();
-        if requests.len() >= count {
-            return requests;
+    while server.requests().len() < count {
+        let remaining = Duration::from_secs(3).saturating_sub(start.elapsed());
+        if remaining.is_zero() {
+            let requests = server.requests();
+            panic!(
+                "timed out waiting for {count} HTTP/3 requests; got {}",
+                requests.len()
+            );
         }
-        assert!(
-            start.elapsed() < Duration::from_secs(3),
-            "timed out waiting for {count} HTTP/3 requests; got {}",
-            requests.len()
-        );
-        thread::sleep(Duration::from_millis(10));
+        let _ = server.request_notify.recv_timeout(remaining);
     }
+    server.requests()
 }
 
 pub(crate) fn start_http3_server(
@@ -138,6 +139,7 @@ pub(crate) fn start_http3_server(
     let requests_for_thread = Arc::clone(&requests);
     let connections_for_thread = Arc::clone(&connections);
     let (addr_tx, addr_rx) = mpsc::channel();
+    let (notify_tx, notify_rx) = mpsc::channel();
     thread::spawn(move || {
         let runtime = tokio::runtime::Builder::new_current_thread()
             .enable_all()
@@ -154,6 +156,7 @@ pub(crate) fn start_http3_server(
                 let handler = Arc::clone(&handler);
                 let requests = Arc::clone(&requests_for_thread);
                 let connections = Arc::clone(&connections_for_thread);
+                let notify = notify_tx.clone();
                 tokio::spawn(async move {
                     let Ok(connection) = incoming.await else {
                         return;
@@ -170,6 +173,7 @@ pub(crate) fn start_http3_server(
                         };
                         let handler = Arc::clone(&handler);
                         let requests = Arc::clone(&requests);
+                        let notify = notify.clone();
                         tokio::spawn(async move {
                             let Ok((req, mut stream)) = resolver.resolve_request().await else {
                                 return;
@@ -207,6 +211,7 @@ pub(crate) fn start_http3_server(
                                 body,
                             };
                             requests.lock().unwrap().push(h3_req.clone());
+                            let _ = notify.send(());
                             let response = handler(h3_req);
                             let mut builder = http::Response::builder().status(response.status);
                             for (name, value) in response.headers {
@@ -259,6 +264,7 @@ pub(crate) fn start_http3_server(
         url,
         ca_cert_path,
         requests,
+        request_notify: notify_rx,
         connections,
     }
 }

tests/support/proxy.rs

@@ -266,7 +266,7 @@ pub(crate) fn start_stalling_proxy(scheme: &str) -> String {
                 let _ = conn.set_read_timeout(Some(Duration::from_millis(200)));
                 let mut buf = [0_u8; 1024];
                 let _ = conn.read(&mut buf);
-                thread::sleep(Duration::from_secs(5));
+                thread::sleep(Duration::from_secs(3));
             });
         }
     });

tests/support/terminal.rs

@@ -74,7 +74,7 @@ pub(crate) fn run_image_render_pty(env: Vec<(String, String)>) -> String {
     let mut child = cmd.spawn().expect("spawn fetch under PTY");
     drop(pty.slave);
     let capture = start_pty_capture(&pty.master);
-    let status = wait_child(&mut child, Duration::from_secs(5))
+    let status = wait_child(&mut child, Duration::from_secs(3))
         .unwrap_or_else(|| {
             let _ = child.kill();
             panic!(
@@ -158,7 +158,7 @@ pub(crate) fn run_fetch_pty_with_fake_less_env(
     let mut child = cmd.spawn().expect("spawn fetch under PTY");
     drop(pty.slave);
     let capture = start_pty_capture(&pty.master);
-    let status = wait_child(&mut child, Duration::from_secs(5))
+    let status = wait_child(&mut child, Duration::from_secs(3))
         .unwrap_or_else(|| {
             let _ = child.kill();
             panic!(
@@ -221,7 +221,7 @@ pub(crate) fn run_binary_pty_with_fake_less(
     let mut child = cmd.spawn().expect("spawn fetch under PTY");
     drop(pty.slave);
     let capture = start_pty_capture(&pty.master);
-    let status = wait_child(&mut child, Duration::from_secs(5))
+    let status = wait_child(&mut child, Duration::from_secs(3))
         .unwrap_or_else(|| {
             let _ = child.kill();
             panic!(
@@ -286,7 +286,7 @@ pub(crate) fn run_image_pty_with_fake_less(
     let mut child = cmd.spawn().expect("spawn fetch under PTY");
     drop(pty.slave);
     let capture = start_pty_capture(&pty.master);
-    let status = wait_child(&mut child, Duration::from_secs(5))
+    let status = wait_child(&mut child, Duration::from_secs(3))
         .unwrap_or_else(|| {
             let _ = child.kill();
             panic!(

tests/terminal.rs

@@ -202,11 +202,11 @@ fn request_ctrl_c_exits_interrupted() {
     let mut child = cmd.spawn().expect("spawn fetch for signal test");
 
     started_rx
-        .recv_timeout(Duration::from_secs(5))
+        .recv_timeout(Duration::from_secs(3))
         .expect("request did not reach server");
     let rc = unsafe { libc::kill(child.id() as libc::pid_t, libc::SIGINT) };
     assert_eq!(rc, 0, "failed to send SIGINT");
-    let status = wait_child(&mut child, Duration::from_secs(5))
+    let status = wait_child(&mut child, Duration::from_secs(3))
         .unwrap_or_else(|| {
             let _ = child.kill();
             panic!("fetch did not exit after SIGINT")