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) <noreply@anthropic.com>
This commit is contained in:
@@ -45,7 +45,7 @@ impl std::io::Read for Deadline<'_> {
|
|||||||
if left.is_zero() {
|
if left.is_zero() {
|
||||||
return Err(std::io::Error::from(std::io::ErrorKind::TimedOut));
|
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`
|
let mut stream = self.stream; // `Read` is implemented for `&UnixStream`
|
||||||
stream.read(buf)
|
stream.read(buf)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -151,9 +151,7 @@ impl Connection {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn set_read_timeout(&self, timeout: Duration) -> Result<(), HttpError> {
|
pub fn set_read_timeout(&self, timeout: Duration) -> Result<(), HttpError> {
|
||||||
self.stream
|
crate::socket::set_read_timeout(&self.stream, timeout).map_err(HttpError::Io)
|
||||||
.set_read_timeout(Some(timeout))
|
|
||||||
.map_err(HttpError::Io)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn received_any(&self) -> bool {
|
pub fn received_any(&self) -> bool {
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ pub mod http;
|
|||||||
pub mod llama;
|
pub mod llama;
|
||||||
pub mod selftest;
|
pub mod selftest;
|
||||||
pub mod session;
|
pub mod session;
|
||||||
|
pub mod socket;
|
||||||
pub mod sse;
|
pub mod sse;
|
||||||
pub mod tools;
|
pub mod tools;
|
||||||
pub mod turn;
|
pub mod turn;
|
||||||
|
|||||||
@@ -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,
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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);
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user