From df22db85fabb3130fe6c026356d6b4a9f9ff7062 Mon Sep 17 00:00:00 2001 From: LaZzyMan Date: Fri, 26 Jun 2026 11:47:11 +0800 Subject: [PATCH] fix(cua-driver): retry daemon socket writes on EAGAIN instead of failing fatally MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit send_request's write side had a 5s SO_SNDTIMEO but no retry, so when the daemon was momentarily too busy to read a request (backpressure under concurrent slow system tools — list_apps / get_accessibility_tree / launch_app), the write timed out and surfaced as a fatal "daemon transport error forwarding '': Resource temporarily unavailable (os error 35)" — even for a tiny request. The read side already tolerates the mirror case (#1997 for #1864): a daemon still working is not a transport failure. Add write_all_with_retry (cua-driver-core::socket_io): write with offset bookkeeping, treat WouldBlock/TimedOut as "keep waiting" until an overall deadline, mirroring the read loop. send_request uses it with the same 120s budget. Extracted to core so the retry logic is unit-tested without linking the platform crates (and their Swift/Metal interop) the cua-driver binary needs. cargo test -p cua-driver-core green (3 socket_io tests: transient-EAGAIN retry, deadline timeout, partial-write accumulation); cargo check -p cua-driver green. --- .../rust/crates/cua-driver-core/src/lib.rs | 1 + .../crates/cua-driver-core/src/socket_io.rs | 135 ++++++++++++++++++ .../rust/crates/cua-driver/src/serve.rs | 12 +- 3 files changed, 145 insertions(+), 3 deletions(-) create mode 100644 libs/cua-driver/rust/crates/cua-driver-core/src/socket_io.rs diff --git a/libs/cua-driver/rust/crates/cua-driver-core/src/lib.rs b/libs/cua-driver/rust/crates/cua-driver-core/src/lib.rs index 6e4886270c..c2b6056621 100644 --- a/libs/cua-driver/rust/crates/cua-driver-core/src/lib.rs +++ b/libs/cua-driver/rust/crates/cua-driver-core/src/lib.rs @@ -30,6 +30,7 @@ pub mod recording_zoom; pub mod server; pub mod session; pub mod session_tools; +pub mod socket_io; pub mod text_sanitize; pub mod tool; pub mod tool_args; diff --git a/libs/cua-driver/rust/crates/cua-driver-core/src/socket_io.rs b/libs/cua-driver/rust/crates/cua-driver-core/src/socket_io.rs new file mode 100644 index 0000000000..41ca0dd409 --- /dev/null +++ b/libs/cua-driver/rust/crates/cua-driver-core/src/socket_io.rs @@ -0,0 +1,135 @@ +//! Reliable framed write to the cua-driver daemon socket. +//! +//! Split out of `serve.rs::send_request` so its EAGAIN-retry logic is unit +//! testable without linking the platform crates (and their Swift/Metal interop) +//! that the `cua-driver` binary pulls in. + +use std::io::Write; +use std::time::Instant; + +/// Write `bytes` in full to a daemon socket that has `SO_SNDTIMEO` set, +/// treating a write timeout (`WouldBlock`/`TimedOut`, i.e. EAGAIN) as "the +/// daemon is still draining, keep waiting" rather than a fatal transport error. +/// +/// This is the write-side mirror of `send_request`'s read loop (#1997 for +/// #1864): a daemon momentarily too busy to read our request is not a transport +/// failure, just as a daemon still computing a slow response is not. Without it, +/// a single 5s `SO_SNDTIMEO` write timeout surfaced as a fatal `daemon transport +/// error forwarding '': Resource temporarily unavailable (os error 35)` +/// even for a tiny request. Bounded by `deadline` so a genuinely stuck daemon +/// still surfaces an error instead of blocking forever. +pub fn write_all_with_retry( + w: &mut W, + bytes: &[u8], + deadline: Instant, +) -> std::io::Result<()> { + let mut written = 0; + while written < bytes.len() { + match w.write(&bytes[written..]) { + Ok(0) => { + return Err(std::io::Error::new( + std::io::ErrorKind::WriteZero, + "daemon closed the connection mid-request", + )); + } + Ok(n) => written += n, + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, + // Write timeout (SO_SNDTIMEO) / non-blocking EAGAIN: the daemon is + // momentarily not reading. Keep waiting until the deadline rather + // than surfacing a fatal transport error. + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + ) => + { + if Instant::now() >= deadline { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + format!( + "daemon did not drain the socket in time \ + (wrote {written}/{} bytes)", + bytes.len() + ), + )); + } + continue; + } + Err(e) => return Err(e), + } + } + w.flush() +} + +#[cfg(test)] +mod tests { + use super::write_all_with_retry; + use std::io::{Error, ErrorKind, Write}; + use std::time::{Duration, Instant}; + + /// Returns `WouldBlock` (EAGAIN) for its first `eagain_left` write attempts, + /// then accepts data — models a daemon briefly too busy to drain the socket + /// (the backpressure that triggers the bug). + struct FlakyWriter { + eagain_left: usize, + written: Vec, + } + impl Write for FlakyWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + if self.eagain_left > 0 { + self.eagain_left -= 1; + return Err(Error::new(ErrorKind::WouldBlock, "eagain")); + } + self.written.extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + #[test] + fn retries_through_transient_eagain() { + let mut w = FlakyWriter { eagain_left: 3, written: vec![] }; + let deadline = Instant::now() + Duration::from_secs(5); + write_all_with_retry(&mut w, b"hello\n", deadline) + .expect("transient EAGAIN should be retried, not fatal"); + assert_eq!(w.written, b"hello\n"); + } + + #[test] + fn times_out_when_daemon_never_drains() { + let mut w = FlakyWriter { eagain_left: usize::MAX, written: vec![] }; + let deadline = Instant::now() + Duration::from_millis(30); + let err = write_all_with_retry(&mut w, b"hello\n", deadline) + .expect_err("a never-draining daemon must eventually surface an error"); + assert_eq!( + err.kind(), + ErrorKind::TimedOut, + "deadline breach should report TimedOut, not a bare EAGAIN" + ); + } + + #[test] + fn accumulates_partial_writes() { + // Accepts only 2 bytes per call — exercises the offset bookkeeping so a + // short write keeps going until every byte lands. + struct ChunkWriter { + written: Vec, + } + impl Write for ChunkWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let n = buf.len().min(2); + self.written.extend_from_slice(&buf[..n]); + Ok(n) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let mut w = ChunkWriter { written: vec![] }; + let deadline = Instant::now() + Duration::from_secs(5); + write_all_with_retry(&mut w, b"abcdefg\n", deadline).unwrap(); + assert_eq!(w.written, b"abcdefg\n"); + } +} diff --git a/libs/cua-driver/rust/crates/cua-driver/src/serve.rs b/libs/cua-driver/rust/crates/cua-driver/src/serve.rs index 7efa76ec38..4301433fea 100644 --- a/libs/cua-driver/rust/crates/cua-driver/src/serve.rs +++ b/libs/cua-driver/rust/crates/cua-driver/src/serve.rs @@ -358,7 +358,7 @@ pub fn read_pid_file(pid_file_path: &str) -> Option { /// Uses a 3-second connect timeout (by polling) and a 10-second read timeout. #[cfg(unix)] pub fn send_request(socket_path: &str, req: &DaemonRequest) -> anyhow::Result { - use std::io::{Read, Write}; + use std::io::Read; use std::os::unix::net::UnixStream; use std::time::{Duration, Instant}; @@ -372,8 +372,14 @@ pub fn send_request(socket_path: &str, req: &DaemonRequest) -> anyhow::Result