confium_net/
deadline.rs

1//! Deadline-bounded reads and writes over a `TcpStream`, implemented
2//! with non-blocking polling instead of `SO_RCVTIMEO`.
3//!
4//! Why not [`std::net::TcpStream::set_read_timeout`]: inside the MRI
5//! Ruby process on `x86_64-pc-windows-gnu`, the first `recv` on a
6//! socket that carries a receive timeout fails `WSAENOTSOCK` (os
7//! error 10038) — every other winsock operation (bind, connect,
8//! write, the setsockopt itself, reads WITHOUT the option) succeeds,
9//! and the same code passes in a plain Rust process. Root-caused
10//! empirically across five probe rounds on the gem's Windows CI (see
11//! the audit ledger; gem PRs #97-#101). Polling sidesteps the option
12//! entirely while keeping the deadlines that prevent the
13//! hang-forever failure class.
14
15use std::io;
16use std::io::ErrorKind;
17use std::io::Read;
18use std::io::Write;
19use std::net::TcpStream;
20use std::time::Duration;
21use std::time::Instant;
22
23/// A deadline-bounded view over a connected [`TcpStream`].
24///
25/// Entering ([`Self::new`]) switches the stream to non-blocking;
26/// dropping it restores blocking mode for the established session.
27/// Reads and writes poll with a short sleep while the peer is not
28/// ready and fail with [`ErrorKind::TimedOut`] once the deadline
29/// passes.
30pub struct DeadlineStream<'a> {
31    stream: &'a mut TcpStream,
32    deadline: Instant,
33    poll: Duration,
34}
35
36impl<'a> DeadlineStream<'a> {
37    /// Bound the stream for `timeout`. The stream is switched to
38    /// non-blocking until the adapter is dropped.
39    pub fn new(stream: &'a mut TcpStream, timeout: Duration) -> io::Result<Self> {
40        stream.set_nonblocking(true)?;
41        Ok(Self {
42            stream,
43            deadline: Instant::now() + timeout,
44            poll: Duration::from_millis(2),
45        })
46    }
47
48    fn expired(&self) -> bool {
49        Instant::now() >= self.deadline
50    }
51}
52
53impl Read for DeadlineStream<'_> {
54    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
55        loop {
56            match self.stream.read(buf) {
57                Ok(n) => return Ok(n),
58                Err(ref e)
59                    if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut =>
60                {
61                    if self.expired() {
62                        return Err(io::Error::new(
63                            ErrorKind::TimedOut,
64                            "read deadline exceeded",
65                        ));
66                    }
67                    std::thread::sleep(self.poll);
68                }
69                Err(ref e) if e.kind() == ErrorKind::Interrupted => {}
70                Err(e) => return Err(e),
71            }
72        }
73    }
74}
75
76impl Write for DeadlineStream<'_> {
77    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
78        loop {
79            match self.stream.write(buf) {
80                Ok(n) => return Ok(n),
81                Err(ref e) if e.kind() == ErrorKind::WouldBlock => {
82                    if self.expired() {
83                        return Err(io::Error::new(
84                            ErrorKind::TimedOut,
85                            "write deadline exceeded",
86                        ));
87                    }
88                    std::thread::sleep(self.poll);
89                }
90                Err(ref e) if e.kind() == ErrorKind::Interrupted => {}
91                Err(e) => return Err(e),
92            }
93        }
94    }
95
96    fn flush(&mut self) -> io::Result<()> {
97        self.stream.flush()
98    }
99}
100
101impl Drop for DeadlineStream<'_> {
102    fn drop(&mut self) {
103        // Restore blocking mode for the established session. Drop
104        // cannot surface errors; a failure here leaves the socket
105        // non-blocking, which misbehaves visibly on the next blocking
106        // operation rather than silently.
107        let _ = self.stream.set_nonblocking(false);
108    }
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114
115    // Local-socket tests flake when run in parallel (the known
116    // port/handler race class); serialize them.
117    static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
118
119    #[test]
120    fn round_trip_within_deadline() {
121        let _guard = LOCK.lock().unwrap();
122        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
123        let port = listener.local_addr().unwrap().port();
124        let echo = std::thread::spawn(move || {
125            use std::io::{Read as _, Write as _};
126            let (mut peer, _) = listener.accept().unwrap();
127            let mut buf = [0u8; 16];
128            let n = peer.read(&mut buf).unwrap();
129            peer.write_all(&buf[..n]).unwrap();
130        });
131        let mut stream = TcpStream::connect(("127.0.0.1", port)).unwrap();
132        {
133            let mut bounded = DeadlineStream::new(&mut stream, Duration::from_secs(5)).unwrap();
134            bounded.write_all(b"ping").unwrap();
135            let mut got = [0u8; 4];
136            bounded.read_exact(&mut got).unwrap();
137            assert_eq!(&got, b"ping");
138        }
139        // Blocking mode restored: a plain write on the underlying
140        // stream behaves normally.
141        assert!(stream.set_nonblocking(false).is_ok());
142        echo.join().unwrap();
143    }
144
145    #[test]
146    fn read_deadline_fires_on_silent_peer() {
147        let _guard = LOCK.lock().unwrap();
148        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
149        let port = listener.local_addr().unwrap().port();
150        let silent = std::thread::spawn(move || {
151            let (_peer, _) = listener.accept().unwrap();
152            std::thread::sleep(Duration::from_secs(1));
153        });
154        let mut stream = TcpStream::connect(("127.0.0.1", port)).unwrap();
155        let start = Instant::now();
156        let mut bounded = DeadlineStream::new(&mut stream, Duration::from_millis(150)).unwrap();
157        let mut buf = [0u8; 8];
158        let err = bounded.read(&mut buf).unwrap_err();
159        assert_eq!(err.kind(), ErrorKind::TimedOut);
160        assert!(start.elapsed() >= Duration::from_millis(140));
161        drop(bounded);
162        silent.join().unwrap();
163    }
164}
confium_net/deadline.rs

confium_net/
deadline.rs
