fix(io): address review feedback on interrupted-receive retries

- Error-queue drain: an interrupted MSG_ERRQUEUE read no longer ends
  the bounded drain early as if the queue were empty (which could
  strand a still-queued TX timestamp past the point its probe is
  looked up and removed). It now reports ErrorQueueRecord::Ignored,
  consuming one attempt of the drain's existing per-call work budget
  and retrying the same slot instead.
- Fake-server recv_from test helpers (client_multi_target.rs and
  tests/support/mod.rs) now keep one absolute deadline across
  interruptions, re-applying only the remaining time on each retry
  and restoring the configured timeout afterward, mirroring the
  production blocking-receive fix these tests exist to guard rather
  than restarting the full timeout on every interruption.

crates/irtt-cli/tests/client_multi_target.rs

@@ -4,7 +4,7 @@
 mod in_tree_server;
 
 use std::{
-    io::{ErrorKind, Read},
+    io::{self, ErrorKind, Read},
     net::{SocketAddr, UdpSocket},
     process::{Command, Output, Stdio},
     sync::mpsc,
@@ -1177,35 +1177,68 @@ fn start_open_failure_server_inner(
     FakeServer { addr, done }
 }
 
-/// A fake server's `recv_from`, retrying an interrupted syscall
+/// A `recv_from` that retries an interrupted syscall
-/// (`EINTR`/`ErrorKind::Interrupted`) instead of panicking on it: real
+/// (`EINTR`/`ErrorKind::Interrupted`) instead of failing outright: real
 /// signal delivery (e.g. process-group signals during CI) can interrupt a
 /// blocking receive here without there being anything actually wrong.
+///
+/// Keeps one absolute deadline across any number of interruptions, exactly
+/// like the production blocking receive fix this test file exists to guard:
+/// each retry re-applies only the *remaining* time against the socket's
+/// currently configured read timeout, so a signal storm cannot turn this
+/// into an unbounded wait. The configured timeout is restored before
+/// returning so a later call is unaffected by an earlier interruption here.
+fn recv_from_retrying_interrupted(
+    socket: &UdpSocket,
+    buf: &mut [u8],
+) -> io::Result<(usize, SocketAddr)> {
+    let configured_timeout = socket.read_timeout()?;
+    let deadline = configured_timeout.and_then(|timeout| Instant::now().checked_add(timeout));
+    let mut timeout_narrowed = false;
+
+    let result = loop {
+        match socket.recv_from(buf) {
+            Ok(received) => break Ok(received),
+            Err(err) if err.kind() == ErrorKind::Interrupted => {
+                if let Some(deadline) = deadline {
+                    let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
+                        break Err(io::Error::from(ErrorKind::TimedOut));
+                    };
+                    if remaining.is_zero() {
+                        break Err(io::Error::from(ErrorKind::TimedOut));
+                    }
+                    if let Err(err) = socket.set_read_timeout(Some(remaining)) {
+                        break Err(err);
+                    }
+                    timeout_narrowed = true;
+                }
+            }
+            Err(err) => break Err(err),
+        }
+    };
+
+    if timeout_narrowed {
+        socket.set_read_timeout(configured_timeout)?;
+    }
+    result
+}
+
 fn recv_request(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
     let mut buf = [0_u8; 2048];
-    loop {
+    match recv_from_retrying_interrupted(socket, &mut buf) {
-        match socket.recv_from(&mut buf) {
+        Ok((size, peer)) => (buf[..size].to_vec(), peer),
-            Ok((size, peer)) => return (buf[..size].to_vec(), peer),
+        Err(err) => panic!("fake server recv_from failed: {err}"),
-            Err(err) if err.kind() == ErrorKind::Interrupted => continue,
-            Err(err) => panic!("fake server recv_from failed: {err}"),
-        }
     }
 }
 
-/// Like [`recv_request`], but treats any non-interrupting error (including a
+/// Like [`recv_request`], but treats any error, including a configured read
-/// configured read timeout) as "nothing more to receive" rather than a
+/// timeout, as "nothing more to receive" rather than a failure, since
-/// failure, since callers use this to detect the peer going quiet. An
+/// callers use this to detect the peer going quiet.
-/// interrupted syscall carries no such information and is retried instead of
-/// being misclassified as quiet.
 fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec<u8>, SocketAddr)> {
     let mut buf = [0_u8; 2048];
-    loop {
+    recv_from_retrying_interrupted(socket, &mut buf)
-        match socket.recv_from(&mut buf) {
+        .ok()
-            Ok((size, peer)) => return Some((buf[..size].to_vec(), peer)),
+        .map(|(size, peer)| (buf[..size].to_vec(), peer))
-            Err(err) if err.kind() == ErrorKind::Interrupted => continue,
-            Err(_) => return None,
-        }
-    }
 }
 
 fn open_reply(flags: u8, token: u64, params: &Params) -> Vec<u8> {

crates/irtt-client/src/receive/linux/error_queue.rs

@@ -153,14 +153,18 @@ impl ErrorQueueControlBuffer {
 
 /// Nonblocking drain of a single record from `fd`'s `MSG_ERRQUEUE`.
 ///
-/// Returns `Ok(None)` when the queue is empty (`EAGAIN`/`EWOULDBLOCK`) or
+/// Returns `Ok(None)` when the queue is empty (`EAGAIN`/`EWOULDBLOCK`) — the
-/// when this one read was interrupted (`EINTR`). TX timestamp capture is
+/// caller in [`super::drain_tx_timestamps`] stops its drain there, exactly
-/// optional best-effort metadata, not a network operation: an interrupted
+/// as it always has. An interrupted read (`EINTR`) is different: it says
-/// read here is not a socket/network error and must not become one, so it is
+/// nothing about whether the queue is empty, so treating it the same as
-/// treated as no record obtained on this drain rather than retried in a
+/// `EWOULDBLOCK` could end a drain early and strand a still-queued record
-/// loop. The bounded caller in [`super::drain_tx_timestamps`] naturally
+/// past the point its matching probe is looked up and removed (see the
-/// revisits the queue on its next call, which is how every other record is
+/// crate's `AGENTS.md`). TX timestamp capture is optional best-effort
-/// picked up too.
+/// metadata, not a network operation, so an interruption here must also
+/// never become a socket/network error; instead it is reported as
+/// [`ErrorQueueRecord::Ignored`], which consumes one attempt of the bounded
+/// caller's existing per-drain work budget and lets the loop immediately
+/// retry the same slot rather than stopping or spinning unboundedly.
 /// `SOF_TIMESTAMPING_OPT_TSONLY` notifications carry no meaningful payload,
 /// so a zero-length iovec is enough: correlation is by `id`, never by
 /// payload content.
@@ -181,7 +185,8 @@ pub(crate) fn try_recv_error_queue_record(fd: RawFd) -> io::Result<Option<ErrorQ
                 .map_err(|_| io::Error::from(io::ErrorKind::InvalidData))?;
             Ok(Some(classify(cmsgs)))
         }
-        Err(Errno::EWOULDBLOCK | Errno::EINTR) => Ok(None),
+        Err(Errno::EWOULDBLOCK) => Ok(None),
+        Err(Errno::EINTR) => Ok(Some(ErrorQueueRecord::Ignored)),
         Err(errno) => Err(io::Error::from(errno)),
     }
 }

crates/irtt-client/tests/support/mod.rs

@@ -4,12 +4,12 @@ mod in_tree_server;
 mod real_irtt;
 
 use std::{
-    io::ErrorKind,
+    io::{self, ErrorKind},
     net::{SocketAddr, UdpSocket},
     process::Command,
     sync::mpsc,
     thread::{self, JoinHandle},
-    time::{Duration, SystemTime},
+    time::{Duration, Instant, SystemTime},
 };
 
 use irtt_client::{Client, ClientConfig, ClientEvent, NegotiatedParams, OpenOutcome, SocketConfig};
@@ -448,32 +448,64 @@ where
     FakeServer { addr, rx, done }
 }
 
-/// A fake server's `recv_from`, retrying an interrupted syscall
+/// A `recv_from` that retries an interrupted syscall
-/// (`EINTR`/`ErrorKind::Interrupted`) instead of panicking on it.
+/// (`EINTR`/`ErrorKind::Interrupted`) instead of failing outright, keeping
+/// one absolute deadline across any number of interruptions exactly like the
+/// production blocking receive fix this test file exists to guard: each
+/// retry re-applies only the *remaining* time against the socket's
+/// currently configured read timeout, so a signal storm cannot turn this
+/// into an unbounded wait. The configured timeout is restored before
+/// returning so a later call is unaffected by an earlier interruption here.
+fn recv_from_retrying_interrupted(
+    socket: &UdpSocket,
+    buf: &mut [u8],
+) -> io::Result<(usize, SocketAddr)> {
+    let configured_timeout = socket.read_timeout()?;
+    let deadline = configured_timeout.and_then(|timeout| Instant::now().checked_add(timeout));
+    let mut timeout_narrowed = false;
+
+    let result = loop {
+        match socket.recv_from(buf) {
+            Ok(received) => break Ok(received),
+            Err(err) if err.kind() == ErrorKind::Interrupted => {
+                if let Some(deadline) = deadline {
+                    let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
+                        break Err(io::Error::from(ErrorKind::TimedOut));
+                    };
+                    if remaining.is_zero() {
+                        break Err(io::Error::from(ErrorKind::TimedOut));
+                    }
+                    if let Err(err) = socket.set_read_timeout(Some(remaining)) {
+                        break Err(err);
+                    }
+                    timeout_narrowed = true;
+                }
+            }
+            Err(err) => break Err(err),
+        }
+    };
+
+    if timeout_narrowed {
+        socket.set_read_timeout(configured_timeout)?;
+    }
+    result
+}
+
 fn recv_request(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
     let mut buf = [0_u8; 8192];
-    loop {
+    match recv_from_retrying_interrupted(socket, &mut buf) {
-        match socket.recv_from(&mut buf) {
+        Ok((size, peer)) => (buf[..size].to_vec(), peer),
-            Ok((size, peer)) => return (buf[..size].to_vec(), peer),
+        Err(err) => panic!("fake server recv_from failed: {err}"),
-            Err(err) if err.kind() == ErrorKind::Interrupted => continue,
-            Err(err) => panic!("fake server recv_from failed: {err}"),
-        }
     }
 }
 
-/// Like [`recv_request`], but any non-interrupting error (including a
+/// Like [`recv_request`], but any error, including a configured read
-/// configured read timeout) means "nothing more to receive" rather than a
+/// timeout, means "nothing more to receive" rather than a failure.
-/// failure. An interrupted syscall carries no such information and is
-/// retried instead of being misclassified as quiet.
 fn recv_request_timeout(socket: &UdpSocket) -> Option<Vec<u8>> {
     let mut buf = [0_u8; 8192];
-    loop {
+    recv_from_retrying_interrupted(socket, &mut buf)
-        match socket.recv_from(&mut buf) {
+        .ok()
-            Ok((size, _)) => return Some(buf[..size].to_vec()),
+        .map(|(size, _)| buf[..size].to_vec())
-            Err(err) if err.kind() == ErrorKind::Interrupted => continue,
-            Err(_) => return None,
-        }
-    }
 }
 
 fn assert_started(outcome: OpenOutcome) -> NegotiatedParams {