From 00a85c124dd2eef5bf9806773dbdefdf6557758d Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Tue, 22 Sep 2026 18:27:31 -0700 Subject: [PATCH] loopd: read on a socket whose peer closed, instead of failing on macOS macOS refuses every socket option with EINVAL once the peer has closed (XNU sosetoptlock, bsd/kern/uipc_socket.c), even with unread data still buffered. loopd set a read timeout before each read, so on macOS every frame or response that arrived just before the peer closed was lost: BrokerPort reported the broker unavailable with "os error 22", and the llama client failed the same way. Twelve loopd test binaries failed on macOS; Linux never refuses, so the gate on Talos did not see it. Both places now go through socket::set_read_timeout, which on Apple targets takes that one refusal as success: a socket shut in both directions returns its data or the end at once and cannot block. A zero timeout is still an error. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/loopd/src/broker_port.rs | 2 +- crates/loopd/src/http.rs | 4 +-- crates/loopd/src/lib.rs | 1 + crates/loopd/src/socket.rs | 21 +++++++++++++++ crates/loopd/tests/socket.rs | 47 +++++++++++++++++++++++++++++++++ 5 files changed, 71 insertions(+), 4 deletions(-) create mode 100644 crates/loopd/src/socket.rs create mode 100644 crates/loopd/tests/socket.rs diff --git a/crates/loopd/src/broker_port.rs b/crates/loopd/src/broker_port.rs index 15644a7..c16d616 100644 --- a/crates/loopd/src/broker_port.rs +++ b/crates/loopd/src/broker_port.rs @@ -45,7 +45,7 @@ impl std::io::Read for Deadline<'_> { if left.is_zero() { return Err(std::io::Error::from(std::io::ErrorKind::TimedOut)); } - self.stream.set_read_timeout(Some(left))?; + crate::socket::set_read_timeout(self.stream, left)?; let mut stream = self.stream; // `Read` is implemented for `&UnixStream` stream.read(buf) } diff --git a/crates/loopd/src/http.rs b/crates/loopd/src/http.rs index cce3d73..d16b068 100644 --- a/crates/loopd/src/http.rs +++ b/crates/loopd/src/http.rs @@ -151,9 +151,7 @@ impl Connection { } pub fn set_read_timeout(&self, timeout: Duration) -> Result<(), HttpError> { - self.stream - .set_read_timeout(Some(timeout)) - .map_err(HttpError::Io) + crate::socket::set_read_timeout(&self.stream, timeout).map_err(HttpError::Io) } pub fn received_any(&self) -> bool { diff --git a/crates/loopd/src/lib.rs b/crates/loopd/src/lib.rs index e12bf53..084dab9 100644 --- a/crates/loopd/src/lib.rs +++ b/crates/loopd/src/lib.rs @@ -8,6 +8,7 @@ pub mod http; pub mod llama; pub mod selftest; pub mod session; +pub mod socket; pub mod sse; pub mod tools; pub mod turn; diff --git a/crates/loopd/src/socket.rs b/crates/loopd/src/socket.rs new file mode 100644 index 0000000..4d1d201 --- /dev/null +++ b/crates/loopd/src/socket.rs @@ -0,0 +1,21 @@ +//! Unix socket options that behave the same on Linux and macOS. + +use std::io; +use std::os::unix::net::UnixStream; +use std::time::Duration; + +/// `EINVAL` on Darwin (`sys/errno.h`); std does not name it and a crate for one number is too much. +#[cfg(target_vendor = "apple")] +const EINVAL: i32 = 22; + +/// Set the read timeout for the next read. macOS refuses every socket option with `EINVAL` once +/// the peer has closed (XNU `sosetoptlock`: both `SS_CANTRCVMORE` and `SS_CANTSENDMORE` set), even +/// with unread data still buffered. A read on such a socket returns that data or the end at once +/// and cannot block, so there is no timeout left to set and the refusal is not an error. +pub fn set_read_timeout(stream: &UnixStream, timeout: Duration) -> io::Result<()> { + match stream.set_read_timeout(Some(timeout)) { + #[cfg(target_vendor = "apple")] + Err(e) if e.raw_os_error() == Some(EINVAL) => Ok(()), + other => other, + } +} diff --git a/crates/loopd/tests/socket.rs b/crates/loopd/tests/socket.rs new file mode 100644 index 0000000..92b2d91 --- /dev/null +++ b/crates/loopd/tests/socket.rs @@ -0,0 +1,47 @@ +//! Setting a read timeout on a Unix socket whose peer has already gone. + +use std::io::{ErrorKind, Read, Write}; +use std::os::unix::net::UnixStream; +use std::time::Duration; + +use loopd::socket::set_read_timeout; + +#[test] +fn a_peer_that_wrote_and_closed_still_delivers_what_it_wrote() { + let (mut ours, mut theirs) = UnixStream::pair().unwrap(); + theirs.write_all(b"frame").unwrap(); + drop(theirs); + set_read_timeout(&ours, Duration::from_secs(5)).unwrap(); + let mut got = Vec::new(); + ours.read_to_end(&mut got).unwrap(); + assert_eq!(got, b"frame"); +} + +#[test] +fn a_peer_that_closed_without_writing_reads_as_the_end() { + let (mut ours, theirs) = UnixStream::pair().unwrap(); + drop(theirs); + set_read_timeout(&ours, Duration::from_secs(5)).unwrap(); + let mut byte = [0u8; 1]; + assert_eq!(ours.read(&mut byte).unwrap(), 0); +} + +#[test] +fn a_live_peer_gets_the_timeout() { + let (mut ours, _theirs) = UnixStream::pair().unwrap(); + set_read_timeout(&ours, Duration::from_millis(20)).unwrap(); + let mut byte = [0u8; 1]; + let err = ours.read(&mut byte).unwrap_err(); + assert!( + matches!(err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut), + "{err:?}" + ); +} + +#[test] +fn a_zero_timeout_is_still_refused() { + let (ours, theirs) = UnixStream::pair().unwrap(); + drop(theirs); + let err = set_read_timeout(&ours, Duration::ZERO).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::InvalidInput); +}