Nils Andreas Svee
ca2389174f
Real signal delivery on Linux (observed via x86_64-musl workspace testing) can interrupt blocking and nonblocking recv/recvmsg calls with EINTR. Several production and test receive paths treated that as a fatal error rather than a transient interruption: - Client::open's blocking receive loop and Client::recv_once/ recv_available's blocking receive returned ClientError::Socket on Interrupted instead of retrying within their existing timeout budget. - AsyncClient::poll_open and poll_recv's nonblocking recvmsg path (Linux ancillary) failed the same way instead of retrying through the existing readiness/work-budget machinery. - The server's ancillary receive loop could terminate the listener on an interrupted nonblocking recvmsg. - The client's MSG_ERRQUEUE TX-timestamp drain turned an interrupted read of optional metadata into a fatal ClientError::Socket. - Fake-server test helpers panicked or misclassified EINTR as a timeout instead of retrying it. Interrupted is now retried transparently everywhere a receive can observe it, while preserving each path's existing timing contract: - Blocking Open reuses its already-absolute per-attempt deadline, so interruptions resume within the same attempt without retransmitting or extending the deadline. - recv_once/recv_available compute one absolute deadline per logical receive (when a timeout is configured) so repeated interruptions cannot extend it past the configured timeout, and restore the socket's configured read timeout afterward so a later call in the same recv_available budget is unaffected. - AsyncClient's poll_open already bounds its retry loop with OPEN_POLL_WORK_BUDGET; poll_recv gains an equivalent RECV_POLL_WORK_BUDGET, since unlike WouldBlock, Interrupted does not clear Tokio's readiness state, so an unbounded retry loop was otherwise possible under sustained signal delivery. - The server's receive loop retries Interrupted the same way it already retries WouldBlock, yielding to the runtime after a bounded number of consecutive interruptions for the same reason. - The error-queue drain treats an interrupted read as no record obtained this drain (deferring to the next natural drain point) rather than a socket error, matching its existing EWOULDBLOCK handling. Deterministic regression tests use a small test-only injection hook (mirroring existing send-injection test hooks) rather than real signal timing, covering: Open completing despite interruptions with no retransmit, Open's deadline not extending under repeated interruptions, recv_once classifying a reply normally despite interruptions, recv_once's timeout not extending under repeated interruptions, and the AsyncClient equivalents for open and recv. No protocol, session, or timeout-classification behavior changes; no new dependencies; send-side EINTR handling is unchanged and out of scope.
81 lines
2.9 KiB
Rust
81 lines
2.9 KiB
Rust
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, ¶ms, 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, ¶ms, &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();
|
|
}
|