diff --git a/docs/next/CHANGELOG.md b/docs/next/CHANGELOG.md index 22e80931..7db02764 100644 --- a/docs/next/CHANGELOG.md +++ b/docs/next/CHANGELOG.md @@ -8,6 +8,7 @@ ### Fixed - Native Windows servers now detach from the terminal console that launched them, so closing WezTerm, Windows Terminal, or another host terminal no longer stops persistent pane processes. (#1329) +- Windows API clients now remain connected while waiting for initial named-pipe request bytes, so `status server`, `api snapshot`, and other socket commands no longer intermittently fail with BrokenPipe. (#1279) - `herdr --remote` now installs remote helper binaries without routing the binary stream through a multiline `/bin/sh -c` command, fixing installs for non-POSIX login shells such as xonsh. (#1203, thanks @nhumrich) ## [0.7.3] - 2026-07-08 diff --git a/src/api/server.rs b/src/api/server.rs index 67093c72..692e9388 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -1,4 +1,4 @@ -use std::io::{self, Read, Write}; +use std::io::{self, Write}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; @@ -18,7 +18,8 @@ use crate::api::wait::{wait_for_event, wait_for_output}; use crate::api::{request_changes_ui, socket_path, ApiRequestMessage, ApiRequestSender, EventHub}; use crate::ipc::{ bind_local_listener, is_connection_closed_error, local_stream_peer_closed, - remove_socket_file_if_owned, socket_file_identity, LocalStream, SocketFileIdentity, + poll_local_stream_read, remove_socket_file_if_owned, set_local_stream_polling, + socket_file_identity, LocalStream, LocalStreamRead, SocketFileIdentity, }; const SOCKET_PERMISSION_MODE: u32 = 0o600; @@ -405,48 +406,189 @@ fn api_response_outcome(response: &str) -> &'static str { } fn read_initial_request_line(stream: &mut LocalStream) -> std::io::Result> { - stream.set_nonblocking(true)?; - let deadline = Instant::now() + INITIAL_REQUEST_TIMEOUT; + read_initial_request_line_with_timeout(stream, INITIAL_REQUEST_TIMEOUT) +} + +fn read_initial_request_line_with_timeout( + stream: &mut LocalStream, + timeout: Duration, +) -> std::io::Result> { + read_initial_request_line_with_limits(stream, timeout, MAX_INITIAL_REQUEST_BYTES) +} + +fn read_initial_request_line_with_limits( + stream: &mut LocalStream, + timeout: Duration, + max_bytes: usize, +) -> std::io::Result> { + set_local_stream_polling(stream, true)?; + let deadline = Instant::now() + timeout; let mut bytes = Vec::new(); let mut byte = [0u8; 1]; - loop { - match stream.read(&mut byte) { - Ok(0) => { - stream.set_nonblocking(false)?; - return Ok(None); - } - Ok(_) => { + let result = loop { + let read = match poll_local_stream_read(stream, &mut byte) { + Ok(read) => read, + Err(err) => break Err(err), + }; + match read { + LocalStreamRead::Closed => break Ok(None), + LocalStreamRead::Data => { bytes.push(byte[0]); if byte[0] == b'\n' { - stream.set_nonblocking(false)?; - return String::from_utf8(bytes) + break String::from_utf8(bytes) .map(Some) .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err)); } - if bytes.len() > MAX_INITIAL_REQUEST_BYTES { - stream.set_nonblocking(false)?; - return Err(io::Error::new( + if bytes.len() > max_bytes { + break Err(io::Error::new( io::ErrorKind::InvalidData, "api request line is too large", )); } } - Err(err) if err.kind() == io::ErrorKind::WouldBlock => { + LocalStreamRead::Pending => { if Instant::now() >= deadline { - stream.set_nonblocking(false)?; - return Err(io::Error::new( + break Err(io::Error::new( io::ErrorKind::TimedOut, "timed out reading api request", )); } std::thread::sleep(CONNECTION_POLL_INTERVAL); } - Err(err) => { - stream.set_nonblocking(false)?; - return Err(err); - } } + }; + set_local_stream_polling(stream, false)?; + result +} + +#[cfg(all(test, windows))] +mod windows_tests { + use super::*; + use interprocess::local_socket::traits::Listener as _; + use std::io::{BufRead, BufReader}; + use std::sync::mpsc::{self, Receiver}; + + fn local_stream_pair(name: &str) -> (LocalStream, LocalStream, PathBuf) { + let path = std::env::temp_dir().join(format!( + "herdr-api-{name}-{}-{}.sock", + std::process::id(), + Instant::now().elapsed().as_nanos() + )); + let listener = crate::ipc::bind_local_listener(&path).unwrap(); + let client = crate::ipc::connect_local_stream(&path).unwrap(); + let server = listener.accept().unwrap(); + (client, server, path) + } + + fn spawn_connection( + server: LocalStream, + ) -> (Receiver>, std::thread::JoinHandle<()>) { + let (done_tx, done_rx) = mpsc::channel(); + let thread = std::thread::spawn(move || { + let (api_tx, _api_rx) = tokio::sync::mpsc::unbounded_channel(); + let result = handle_connection( + server, + &api_tx, + &EventHub::default(), + &Arc::new(AtomicBool::new(true)), + None, + ); + done_tx.send(result).unwrap(); + }); + (done_rx, thread) + } + + #[test] + fn windows_delayed_partial_initial_request_returns_pong() { + let (mut client, server, path) = local_stream_pair("delayed-request"); + let (done_rx, server_thread) = spawn_connection(server); + + std::thread::sleep(Duration::from_millis(300)); + assert!( + done_rx.try_recv().is_err(), + "idle connected client must not be treated as closed" + ); + + client + .write_all(br#"{"id":"delayed","method":"ping","params":{}}"#) + .unwrap(); + client.flush().unwrap(); + std::thread::sleep(Duration::from_millis(150)); + assert!( + done_rx.try_recv().is_err(), + "partial request must wait for its newline" + ); + client.write_all(b"\n").unwrap(); + client.flush().unwrap(); + + let mut response = String::new(); + BufReader::new(&mut client) + .read_line(&mut response) + .unwrap(); + let response: serde_json::Value = serde_json::from_str(&response).unwrap(); + assert_eq!(response["id"], "delayed"); + assert_eq!(response["result"]["type"], "pong"); + + done_rx + .recv_timeout(Duration::from_secs(2)) + .unwrap() + .unwrap(); + server_thread.join().unwrap(); + let _ = std::fs::remove_file(path); + } + + #[test] + fn windows_disconnected_initial_request_returns_promptly() { + let (client, server, path) = local_stream_pair("disconnected-request"); + let (done_rx, server_thread) = spawn_connection(server); + + drop(client); + + done_rx + .recv_timeout(Duration::from_secs(2)) + .expect("disconnected connection handler must finish promptly") + .unwrap(); + server_thread.join().unwrap(); + let _ = std::fs::remove_file(path); + } + + #[test] + fn windows_idle_initial_request_honors_timeout() { + let (_client, mut server, path) = local_stream_pair("request-timeout"); + + let err = read_initial_request_line_with_timeout(&mut server, Duration::from_millis(50)) + .unwrap_err(); + + assert_eq!(err.kind(), io::ErrorKind::TimedOut); + let _ = std::fs::remove_file(path); + } + + #[test] + fn windows_initial_request_enforces_size_limit() { + let (mut client, mut server, path) = local_stream_pair("request-size-limit"); + client.write_all(b"12345").unwrap(); + client.flush().unwrap(); + + let err = read_initial_request_line_with_limits(&mut server, Duration::from_secs(1), 4) + .unwrap_err(); + + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + assert_eq!(err.to_string(), "api request line is too large"); + let _ = std::fs::remove_file(path); + } + + #[test] + fn windows_initial_request_rejects_invalid_utf8() { + let (mut client, mut server, path) = local_stream_pair("request-invalid-utf8"); + client.write_all(&[0xff, b'\n']).unwrap(); + client.flush().unwrap(); + + let err = read_initial_request_line_with_timeout(&mut server, Duration::from_secs(1)) + .unwrap_err(); + + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + let _ = std::fs::remove_file(path); } } diff --git a/src/ipc.rs b/src/ipc.rs index 8b72dd4c..c293a278 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -1,7 +1,5 @@ use std::fs; -use std::io; -#[cfg(unix)] -use std::io::Read; +use std::io::{self, Read}; #[cfg(unix)] use std::os::unix::fs::{MetadataExt, PermissionsExt}; use std::path::Path; @@ -12,6 +10,12 @@ use interprocess::local_socket::traits::Stream as _; pub(crate) type LocalListener = interprocess::local_socket::Listener; pub(crate) type LocalStream = interprocess::local_socket::Stream; +pub(crate) enum LocalStreamRead { + Data, + Pending, + Closed, +} + #[derive(Clone, Debug, Eq, PartialEq)] pub(crate) struct SocketFileIdentity { #[cfg(unix)] @@ -108,6 +112,48 @@ pub(crate) fn local_stream_peer_closed(stream: &mut LocalStream) -> io::Result io::Result<()> { + #[cfg(unix)] + { + stream.set_nonblocking(enabled) + } + + #[cfg(windows)] + { + let _ = (stream, enabled); + Ok(()) + } +} + +pub(crate) fn poll_local_stream_read( + stream: &mut LocalStream, + buf: &mut [u8], +) -> io::Result { + #[cfg(unix)] + { + match stream.read(buf) { + Ok(0) => Ok(LocalStreamRead::Closed), + Ok(_) => Ok(LocalStreamRead::Data), + Err(err) if err.kind() == io::ErrorKind::WouldBlock => Ok(LocalStreamRead::Pending), + Err(err) => Err(err), + } + } + + #[cfg(windows)] + { + match windows_named_pipe_available(stream)? { + None => Ok(LocalStreamRead::Closed), + Some(0) => Ok(LocalStreamRead::Pending), + Some(_) => match stream.read(buf) { + Ok(0) => Ok(LocalStreamRead::Closed), + Ok(_) => Ok(LocalStreamRead::Data), + Err(err) if is_connection_closed_error(&err) => Ok(LocalStreamRead::Closed), + Err(err) => Err(err), + }, + } + } +} + #[cfg(unix)] fn probe_stream_closed(stream: &mut LocalStream) -> io::Result { stream.set_nonblocking(true)?; @@ -132,26 +178,32 @@ fn probe_stream_closed(stream: &mut LocalStream) -> io::Result { #[cfg(windows)] fn probe_stream_closed(stream: &mut LocalStream) -> io::Result { + Ok(windows_named_pipe_available(stream)?.is_none()) +} + +#[cfg(windows)] +fn windows_named_pipe_available(stream: &mut LocalStream) -> io::Result> { use std::os::windows::io::{AsHandle, AsRawHandle}; let LocalStream::NamedPipe(pipe) = stream; + let mut available = 0; let ok = unsafe { windows_sys::Win32::System::Pipes::PeekNamedPipe( pipe.as_handle().as_raw_handle(), std::ptr::null_mut(), 0, std::ptr::null_mut(), - std::ptr::null_mut(), + &mut available, std::ptr::null_mut(), ) }; if ok != 0 { - return Ok(false); + return Ok(Some(available)); } let err = io::Error::last_os_error(); if is_connection_closed_error(&err) || windows_named_pipe_closed_error(&err) { - return Ok(true); + return Ok(None); } Err(err) }