Merge pull request #111 from Lochnair/fix/linux-receive-eintr

fix(io): retry interrupted receives

crates/irtt-cli/tests/client_multi_target.rs

@@ -4,7 +4,7 @@
 mod in_tree_server;
 
 use std::{
-    io::Read,
+    io::{self, ErrorKind, Read},
     net::{SocketAddr, UdpSocket},
     process::{Command, Output, Stdio},
     sync::mpsc,
@@ -1177,16 +1177,66 @@ fn start_open_failure_server_inner(
     FakeServer { addr, done }
 }
 
-fn recv_request(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
+/// A `recv_from` that retries an interrupted syscall
-    let mut buf = [0_u8; 2048];
+/// (`EINTR`/`ErrorKind::Interrupted`) instead of failing outright: real
-    let (size, peer) = socket.recv_from(&mut buf).unwrap();
+/// signal delivery (e.g. process-group signals during CI) can interrupt a
-    (buf[..size].to_vec(), peer)
+/// 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];
+    match recv_from_retrying_interrupted(socket, &mut buf) {
+        Ok((size, peer)) => (buf[..size].to_vec(), peer),
+        Err(err) => panic!("fake server recv_from failed: {err}"),
+    }
+}
+
+/// Like [`recv_request`], but treats any error, including a configured read
+/// timeout, as "nothing more to receive" rather than a failure, since
+/// callers use this to detect the peer going quiet.
 fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec<u8>, SocketAddr)> {
     let mut buf = [0_u8; 2048];
-    socket
+    recv_from_retrying_interrupted(socket, &mut buf)
-        .recv_from(&mut buf)
         .ok()
         .map(|(size, peer)| (buf[..size].to_vec(), peer))
 }

crates/irtt-client/src/async_client.rs

@@ -41,6 +41,19 @@ use crate::{
 /// so one managed target cannot retain the current-thread runtime.
 const OPEN_POLL_WORK_BUDGET: usize = 64;
 
+/// Maximum immediately available receive work performed by one
+/// [`AsyncClient::poll_recv`] poll.
+///
+/// A `WouldBlock` readiness false positive is naturally bounded: `try_io`
+/// clears the reactor's readiness on `WouldBlock`, so the next
+/// `poll_recv_ready` genuinely awaits a fresh edge. An `Interrupted`
+/// (`EINTR`) nonblocking `recvmsg` does not clear that readiness — the
+/// syscall was interrupted, not told there was nothing to read — so without
+/// an explicit bound a sustained signal delivery rate could otherwise retry
+/// in a tight loop on this executor thread. Bounding both cases the same way
+/// this crate already bounds open-attempt work keeps that impossible.
+const RECV_POLL_WORK_BUDGET: usize = 64;
+
 #[cfg(test)]
 use crate::client::ProbeSendTimestamps;
 
@@ -174,6 +187,11 @@ struct AsyncClientTestHooks {
     sends: RefCell<VecDeque<InjectedSend>>,
     send_attempts: Cell<usize>,
     receive_would_block: Cell<usize>,
+    /// Number of subsequent receive attempts (`poll_open` or `poll_recv`)
+    /// that should report `io::ErrorKind::Interrupted` instead of performing
+    /// a real nonblocking receive, deterministically exercising interrupted-
+    /// syscall retry behavior.
+    inject_recv_interrupted: Cell<usize>,
     fail_open_dscp: Cell<bool>,
     fail_cleanup_send: Cell<bool>,
     fail_peer_close_dscp: Cell<bool>,
@@ -205,6 +223,18 @@ impl AsyncClientTestHooks {
     fn take_probe_timestamps(&self) -> Option<ProbeSendTimestamps> {
         self.probe_timestamps.borrow_mut().pop_front()
     }
+
+    /// Consumes one pending injected interruption, if any, reporting whether
+    /// the caller should report `io::ErrorKind::Interrupted` instead of
+    /// attempting a real receive.
+    fn take_injected_interrupted(&self) -> bool {
+        let remaining = self.inject_recv_interrupted.get();
+        if remaining == 0 {
+            return false;
+        }
+        self.inject_recv_interrupted.set(remaining - 1);
+        true
+    }
 }
 
 /// Low-level Tokio IRTT client for one connected UDP target.
@@ -408,9 +438,26 @@ impl AsyncClient {
                 Poll::Ready(Ok(true)) => {}
             }
 
+            #[cfg(test)]
+            if self.test_hooks.take_injected_interrupted() {
+                continue;
+            }
             let datagram = match try_recv_tokio_datagram(&self.socket, &mut state.buffer) {
                 Ok(datagram) => datagram,
-                Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue,
+                // WouldBlock is a readiness false positive; Interrupted
+                // (EINTR) is a transient nonblocking-recvmsg interruption.
+                // Both retry through the surrounding `OPEN_POLL_WORK_BUDGET`-
+                // bounded loop above, which re-checks readiness before the
+                // next attempt and yields back to the executor once that
+                // budget is exhausted.
+                Err(error)
+                    if matches!(
+                        error.kind(),
+                        io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted
+                    ) =>
+                {
+                    continue
+                }
                 Err(error) => return Poll::Ready(Err(ClientError::Socket(error))),
             };
             if datagram.received_at.mono > state.deadline() {
@@ -613,7 +660,7 @@ impl AsyncClient {
             return Poll::Ready(Err(error));
         }
 
-        loop {
+        for _ in 0..RECV_POLL_WORK_BUDGET {
             match self.socket.poll_recv_ready(cx) {
                 Poll::Pending => return Poll::Pending,
                 Poll::Ready(Ok(())) => {}
@@ -628,9 +675,23 @@ impl AsyncClient {
                     .set(self.test_hooks.receive_would_block.get() - 1);
                 continue;
             }
+            #[cfg(test)]
+            if self.test_hooks.take_injected_interrupted() {
+                continue;
+            }
             let datagram = match try_recv_tokio_datagram(&self.socket, &mut self.recv_buffer) {
                 Ok(datagram) => datagram,
-                Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue,
+                // See `RECV_POLL_WORK_BUDGET`: both a readiness false
+                // positive and an interrupted nonblocking recvmsg retry
+                // within this bounded loop rather than failing the receive.
+                Err(error)
+                    if matches!(
+                        error.kind(),
+                        io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted
+                    ) =>
+                {
+                    continue
+                }
                 Err(error) => return Poll::Ready(Err(ClientError::Socket(error))),
             };
 
@@ -659,6 +720,12 @@ impl AsyncClient {
             }
             return Poll::Ready(Ok(events));
         }
+
+        // Truncated known immediate receive work, exactly as `poll_open`
+        // does: arrange for another poll even if no new socket-readiness
+        // edge occurs, rather than spinning this executor thread further.
+        cx.waker().wake_by_ref();
+        Poll::Pending
     }
 
     pub(crate) fn poll_close(

crates/irtt-client/src/async_client/tests.rs

@@ -883,6 +883,53 @@ fn receive_retries_false_readiness_and_shares_packet_classification() {
     assert_eq!(server.finish().len(), 2);
 }
 
+#[test]
+fn receive_retries_interrupted_syscall_and_shares_packet_classification() {
+    let server = start_server(move |socket, tx| {
+        let (params, peer) = open_session(&socket, &tx, None);
+        let (probe_packet, _) = recv_packet(&socket, &tx);
+        let sequence = echo_request_sequence(&probe_packet, None);
+        socket
+            .send_to(
+                &echo_reply(&params, sequence, TOKEN, flags::FLAG_REPLY, None),
+                peer,
+            )
+            .unwrap();
+    });
+
+    runtime().block_on(async {
+        let mut client = opened_client(server.addr, 0).await;
+        client.send_probe().await.unwrap();
+        client.test_hooks.inject_recv_interrupted.set(2);
+
+        assert!(matches!(
+            client.recv().await.unwrap().as_slice(),
+            [ClientEvent::EchoReply { seq: 0, .. }]
+        ));
+    });
+    assert_eq!(server.finish().len(), 2);
+}
+
+#[test]
+fn open_retries_interrupted_syscall_within_the_same_attempt() {
+    let server = start_open_server(None, 0);
+    runtime().block_on(async {
+        let mut client = AsyncClient::connect(config(server.addr, None, 0))
+            .await
+            .unwrap();
+        client.test_hooks.inject_recv_interrupted.set(3);
+
+        assert!(matches!(
+            client.open().await,
+            Ok(OpenOutcome::Started { .. })
+        ));
+        // Only one open request reached the server: interruptions were
+        // absorbed inside the single open attempt, with no retransmit.
+        assert_eq!(client.test_hooks.send_attempts.get(), 1);
+    });
+    server.finish();
+}
+
 #[test]
 fn peer_close_dscp_failure_preserves_events_and_logical_ownership() {
     let server = start_peer_close_server();

crates/irtt-client/src/client.rs

@@ -14,7 +14,7 @@ use crate::{
     config::{ClientConfig, RecvBudget},
     error::ClientError,
     event::{ClientEvent, OpenOutcome},
-    receive::{drain_tx_timestamps, recv_datagram, try_enable_tx_timestamping},
+    receive::{drain_tx_timestamps, recv_datagram, try_enable_tx_timestamping, ReceivedDatagram},
     session::machine::{
         recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, ProbeSent,
         SessionMachine, MAX_OPEN_PACKET_SIZE,
@@ -69,6 +69,10 @@ struct ClientTestHooks {
     fail_dscp_restore: Cell<bool>,
     last_restored_read_timeout: Cell<Option<Duration>>,
     prepared_active_session_before_dscp: Cell<bool>,
+    /// Number of subsequent receive attempts that should report
+    /// `io::ErrorKind::Interrupted` before a real receive is attempted.
+    /// Decremented on every injected receive; `0` means inject nothing.
+    inject_recv_interrupted: Cell<usize>,
 }
 
 #[cfg(test)]
@@ -264,18 +268,40 @@ impl Client {
         self.recv_once_inner()
     }
 
-    fn recv_once_inner(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
+    /// Receive one datagram into `self.recv_buffer`. In test builds, a
-        let datagram = match recv_datagram(&self.socket, &mut self.recv_buffer) {
+    /// pending `inject_recv_interrupted` count reports
-            Ok(datagram) => datagram,
+    /// `io::ErrorKind::Interrupted` instead of performing a real receive,
-            Err(err)
+    /// deterministically exercising interrupted-syscall retry behavior.
-                if matches!(
+    fn recv_datagram_once(&mut self) -> Result<ReceivedDatagram, io::Error> {
-                    err.kind(),
+        #[cfg(test)]
-                    io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
+        {
-                ) =>
+            let remaining = self.test_hooks.inject_recv_interrupted.get();
-            {
+            if remaining > 0 {
-                return Ok(vec![]);
+                self.test_hooks.inject_recv_interrupted.set(remaining - 1);
+                return Err(io::Error::from(io::ErrorKind::Interrupted));
             }
-            Err(err) => return Err(ClientError::Socket(err)),
+        }
+        recv_datagram(&self.socket, &mut self.recv_buffer)
+    }
+
+    /// Same as [`Self::recv_datagram_once`], receiving into a caller-supplied
+    /// buffer instead of `self.recv_buffer` (used before the negotiated
+    /// receive buffer size is known, i.e. during Open).
+    fn recv_datagram_once_into(&self, buf: &mut [u8]) -> Result<ReceivedDatagram, io::Error> {
+        #[cfg(test)]
+        {
+            let remaining = self.test_hooks.inject_recv_interrupted.get();
+            if remaining > 0 {
+                self.test_hooks.inject_recv_interrupted.set(remaining - 1);
+                return Err(io::Error::from(io::ErrorKind::Interrupted));
+            }
+        }
+        recv_datagram(&self.socket, buf)
+    }
+
+    fn recv_once_inner(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
+        let Some(datagram) = self.recv_datagram_retrying_interrupted()? else {
+            return Ok(vec![]);
         };
 
         // One final bounded drain before processing the reply, so a TX
@@ -297,6 +323,71 @@ impl Client {
         Ok(events)
     }
 
+    /// Receive one datagram, transparently retrying an interrupted syscall
+    /// (`io::ErrorKind::Interrupted`, e.g. `EINTR`) without treating it as a
+    /// genuine socket error, timeout, or empty receive.
+    ///
+    /// One logical receive keeps one timeout budget: when a receive timeout
+    /// is configured, an absolute deadline is computed once here and each
+    /// retry after an interruption restores the socket's remaining time
+    /// against that same deadline, so repeated interruptions cannot extend
+    /// this receive past its configured timeout. An unconfigured timeout
+    /// (`None`) keeps blocking indefinitely after an interruption, matching
+    /// its no-timeout contract. The socket's read timeout is restored to the
+    /// caller-configured value before returning, so a later call in the same
+    /// `recv_available` budget is unaffected by an earlier interruption here.
+    fn recv_datagram_retrying_interrupted(
+        &mut self,
+    ) -> Result<Option<ReceivedDatagram>, ClientError> {
+        let configured_timeout = self.runtime.config().socket_config.recv_timeout;
+        let deadline = match configured_timeout {
+            Some(timeout) => Some(
+                Instant::now()
+                    .checked_add(timeout)
+                    .ok_or(ClientError::DurationOverflow)?,
+            ),
+            None => None,
+        };
+        let mut timeout_narrowed = false;
+
+        let result = loop {
+            match self.recv_datagram_once() {
+                Ok(datagram) => break Ok(Some(datagram)),
+                Err(err)
+                    if matches!(
+                        err.kind(),
+                        io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
+                    ) =>
+                {
+                    break Ok(None);
+                }
+                Err(err) if err.kind() == io::ErrorKind::Interrupted => {
+                    if let Some(deadline) = deadline {
+                        let Some(remaining) = deadline.checked_duration_since(Instant::now())
+                        else {
+                            break Ok(None);
+                        };
+                        if remaining.is_zero() {
+                            break Ok(None);
+                        }
+                        if let Err(err) = self.socket.set_read_timeout(Some(remaining)) {
+                            break Err(ClientError::Socket(err));
+                        }
+                        timeout_narrowed = true;
+                    }
+                }
+                Err(err) => break Err(ClientError::Socket(err)),
+            }
+        };
+
+        if timeout_narrowed {
+            if let Err(err) = self.socket.set_read_timeout(configured_timeout) {
+                return Err(ClientError::Socket(err));
+            }
+        }
+        result
+    }
+
     /// Receive and classify datagrams until a receive produces no events or the
     /// receive budget is exhausted.
     pub fn recv_available(&mut self, budget: RecvBudget) -> Result<Vec<ClientEvent>, ClientError> {
@@ -496,12 +587,22 @@ impl Client {
                 }
                 self.socket.set_read_timeout(Some(remaining))?;
 
-                let datagram = match recv_datagram(&self.socket, &mut buf) {
+                let datagram = match self.recv_datagram_once_into(&mut buf) {
                     Ok(datagram) => datagram,
+                    // WouldBlock/TimedOut mean the attempt's own deadline
+                    // elapsed; Interrupted (EINTR) means the syscall was
+                    // merely interrupted and carries no timing information of
+                    // its own. Both retry through the same loop, which
+                    // recomputes `remaining` and re-applies it as the read
+                    // timeout on every iteration — so an interruption resumes
+                    // within this same absolute deadline rather than
+                    // resetting or extending it.
                     Err(err)
                         if matches!(
                             err.kind(),
-                            io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
+                            io::ErrorKind::WouldBlock
+                                | io::ErrorKind::TimedOut
+                                | io::ErrorKind::Interrupted
                         ) =>
                     {
                         if Instant::now() >= deadline {

crates/irtt-client/src/client/tests/open.rs

@@ -64,6 +64,63 @@ fn open_retries_after_first_timeout() {
     server.join();
 }
 
+#[test]
+fn open_survives_interrupted_receive_within_same_attempt() {
+    let server = start_fake_server(|socket, tx| {
+        let (first, peer) = recv_request(&socket, &tx);
+        let params = Params::decode(&first[4..]).unwrap();
+        let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
+        socket.send_to(&reply, peer).unwrap();
+    });
+    let config = ClientConfig {
+        open_timeouts: vec![Duration::from_millis(500)],
+        ..default_test_config(server.addr)
+    };
+    let mut client = Client::connect(config).unwrap();
+    // Several interrupted receive attempts before the real reply arrives
+    // must not be treated as a timeout, must not retransmit the open
+    // request, and must not advance to a next configured attempt.
+    client.test_hooks.inject_recv_interrupted.set(5);
+
+    let outcome = client.open().unwrap();
+    assert_open_started(outcome);
+    // Exactly one request reached the server: interruptions were absorbed
+    // inside the single configured attempt.
+    assert_eq!(server.rx.iter().take(1).count(), 1);
+    server.join();
+}
+
+#[test]
+fn open_interrupted_receive_does_not_extend_the_attempt_deadline() {
+    // A fake server that never replies: with a short attempt timeout, the
+    // real question is whether interruptions can push the effective wait
+    // past the configured deadline. They must not: the deadline stays
+    // absolute across any number of interruptions.
+    let server = start_fake_server(|socket, tx| {
+        let _ = recv_request(&socket, &tx);
+    });
+    let config = ClientConfig {
+        open_timeouts: vec![Duration::from_millis(200)],
+        ..default_test_config(server.addr)
+    };
+    let mut client = Client::connect(config).unwrap();
+    // Each interruption carries no timing information of its own, so the
+    // loop must keep re-checking the real deadline rather than treating an
+    // interruption as "more time available" or a reason to fail outright.
+    client.test_hooks.inject_recv_interrupted.set(20);
+
+    let started = Instant::now();
+    let result = client.open();
+    let elapsed = started.elapsed();
+
+    assert!(matches!(result, Err(ClientError::OpenTimeout)));
+    assert!(
+        elapsed < Duration::from_millis(800),
+        "interrupted receives must not extend the attempt deadline, took {elapsed:?}"
+    );
+    server.join();
+}
+
 #[test]
 fn open_timeout_after_all_timeouts() {
     let server = timeout_server(Duration::from_millis(700));

crates/irtt-client/src/client/tests/probes_replies/interrupted_receive.rs

@@ -0,0 +1,80 @@
+use super::*;
+
+/// An interrupted receive (`io::ErrorKind::Interrupted`, e.g. `EINTR`) must be
+/// invisible to the caller: retried transparently, with the eventual reply
+/// classified exactly as if no interruption had occurred.
+#[test]
+fn interrupted_receive_is_retried_and_the_reply_is_still_classified() {
+    let params = default_params();
+    let server = start_fake_server(move |socket, tx| {
+        let (_, peer) = recv_request(&socket, &tx);
+        let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
+        socket.send_to(&reply, peer).unwrap();
+
+        let mut buf = [0_u8; 2048];
+        let (size, _) = socket.recv_from(&mut buf).unwrap();
+        let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap());
+        let _ = size;
+        let ts = TimestampFields::default();
+        let reply_packet = echo_reply_packet(TOKEN, seq, &params, &ts, None);
+        socket.send_to(&reply_packet, peer).unwrap();
+    });
+    let config = ClientConfig {
+        socket_config: crate::SocketConfig {
+            recv_timeout: Some(Duration::from_secs(2)),
+            ..Default::default()
+        },
+        ..default_test_config(server.addr)
+    };
+    let mut client = Client::connect(config).unwrap();
+    assert_open_started(client.open().unwrap());
+
+    client.send_probe().unwrap();
+    thread::sleep(Duration::from_millis(50));
+
+    client.test_hooks.inject_recv_interrupted.set(10);
+    let events = client.recv_once().unwrap();
+
+    assert_eq!(events.len(), 1);
+    assert!(
+        matches!(&events[0], ClientEvent::EchoReply { seq: 0, .. }),
+        "expected a normal EchoReply despite interrupted receive attempts, got {:?}",
+        events[0]
+    );
+
+    client.close().unwrap();
+    server.join();
+}
+
+/// One logical receive keeps one timeout budget: repeated interruptions must
+/// not extend a `recv_once` call past its configured receive timeout.
+#[test]
+fn interrupted_receive_does_not_extend_the_configured_timeout() {
+    let params = default_params();
+    let server = silent_open_server(params);
+    let config = ClientConfig {
+        socket_config: crate::SocketConfig {
+            recv_timeout: Some(Duration::from_millis(100)),
+            ..Default::default()
+        },
+        ..default_test_config(server.addr)
+    };
+    let mut client = Client::connect(config).unwrap();
+    assert_open_started(client.open().unwrap());
+
+    // No reply is ever coming; only interrupted attempts occur before the
+    // real socket takes over and legitimately times out.
+    client.test_hooks.inject_recv_interrupted.set(20);
+    let started = Instant::now();
+    let events = client.recv_once().unwrap();
+    let elapsed = started.elapsed();
+
+    assert!(events.is_empty());
+    assert!(
+        elapsed < Duration::from_millis(500),
+        "interrupted receives must not extend the configured receive timeout, took {elapsed:?}"
+    );
+
+    client.close().unwrap();
+    server.join();
+}

crates/irtt-client/src/client/tests/probes_replies/mod.rs

@@ -20,6 +20,7 @@ fn kernel_rx_timestamp_unavailable_skip(test_name: &str) {
 }
 
 mod burst_recv;
+mod interrupted_receive;
 mod packet_length;
 mod probe_schedule;
 mod reply_classification;

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

@@ -153,7 +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`).
+/// Returns `Ok(None)` when the queue is empty (`EAGAIN`/`EWOULDBLOCK`) — the
+/// caller in [`super::drain_tx_timestamps`] stops its drain there, exactly
+/// as it always has. An interrupted read (`EINTR`) is different: it says
+/// nothing about whether the queue is empty, so treating it the same as
+/// `EWOULDBLOCK` could end a drain early and strand a still-queued record
+/// past the point its matching probe is looked up and removed (see the
+/// crate's `AGENTS.md`). TX timestamp capture is optional best-effort
+/// 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.
@@ -175,6 +186,7 @@ pub(crate) fn try_recv_error_queue_record(fd: RawFd) -> io::Result<Option<ErrorQ
             Ok(Some(classify(cmsgs)))
         }
         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,11 +4,12 @@ mod in_tree_server;
 mod real_irtt;
 
 use std::{
+    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};
@@ -447,18 +448,64 @@ where
     FakeServer { addr, rx, done }
 }
 
-fn recv_request(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
+/// A `recv_from` that retries an interrupted syscall
-    let mut buf = [0_u8; 8192];
+/// (`EINTR`/`ErrorKind::Interrupted`) instead of failing outright, keeping
-    let (size, peer) = socket.recv_from(&mut buf).unwrap();
+/// one absolute deadline across any number of interruptions exactly like the
-    (buf[..size].to_vec(), peer)
+/// 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];
+    match recv_from_retrying_interrupted(socket, &mut buf) {
+        Ok((size, peer)) => (buf[..size].to_vec(), peer),
+        Err(err) => panic!("fake server recv_from failed: {err}"),
+    }
+}
+
+/// Like [`recv_request`], but any error, including a configured read
+/// timeout, means "nothing more to receive" rather than a failure.
 fn recv_request_timeout(socket: &UdpSocket) -> Option<Vec<u8>> {
     let mut buf = [0_u8; 8192];
-    match socket.recv_from(&mut buf) {
+    recv_from_retrying_interrupted(socket, &mut buf)
-        Ok((size, _)) => Some(buf[..size].to_vec()),
+        .ok()
-        Err(_) => None,
+        .map(|(size, _)| buf[..size].to_vec())
-    }
 }
 
 fn assert_started(outcome: OpenOutcome) -> NegotiatedParams {

crates/irtt-server/src/socket_io/ancillary.rs

@@ -152,6 +152,18 @@ pub(super) async fn receive_bound(
     receive_ancillary(socket, buffer, false).await
 }
 
+/// Consecutive interrupted (`EINTR`) nonblocking receives tolerated before
+/// this receive loop explicitly yields to the runtime.
+///
+/// `try_io` only clears the reactor's readiness state on `WouldBlock`; an
+/// `Interrupted` result leaves it untouched, so `socket.readable().await`
+/// can resolve immediately again without a genuine new readiness edge. Under
+/// a sustained signal delivery rate that could otherwise retry in a tight
+/// loop without ever suspending this task. Yielding periodically bounds that
+/// without treating the interruption as any kind of failure or terminating
+/// the listener.
+const INTERRUPTED_YIELD_THRESHOLD: u32 = 64;
+
 /// `require_destination` is the listener's own wildcard-ness. It decides both
 /// what this receive must recover and how severe a truncated control buffer is.
 async fn receive_ancillary(
@@ -159,12 +171,24 @@ async fn receive_ancillary(
     buffer: &mut [u8],
     require_destination: bool,
 ) -> io::Result<Option<ReceivedDatagram>> {
+    let mut consecutive_interrupted = 0_u32;
     loop {
         socket.readable().await?;
         match socket.try_io(Interest::READABLE, || {
             receive_once(socket, buffer, require_destination)
         }) {
-            Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue,
+            Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+                consecutive_interrupted = 0;
+                continue;
+            }
+            Err(error) if error.kind() == io::ErrorKind::Interrupted => {
+                consecutive_interrupted += 1;
+                if consecutive_interrupted >= INTERRUPTED_YIELD_THRESHOLD {
+                    consecutive_interrupted = 0;
+                    tokio::task::yield_now().await;
+                }
+                continue;
+            }
             result => return result,
         }
     }