starred/herdr
1
0
Fork 0
mirror of https://github.com/herdrdev/herdr.git synced 2026-09-22 00:01:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: keep windows wait connections open

Browse Source 
refs #963
This commit is contained in:
Ogulcan Celik
2026-07-03 20:38:05 +03:00
parent ef67a97047
commit 045f506ec8
7 changed files with 493 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1
View File
@@ -53,6 +53,7 @@ windows-sys = { version = "0.61.2", features = [
"Win32_System_Diagnostics_ToolHelp",
"Win32_System_Console",
"Win32_System_Kernel",
"Win32_System_Pipes",
"Win32_System_Threading",
"Win32_UI_Shell",
"Win32_UI_WindowsAndMessaging",
src/api/server.rs
+149 -37
View File
@@ -14,11 +14,11 @@ use crate::api::schema::{
ErrorBody, ErrorResponse, Method, Request, ResponseResult, ServerCapabilities, SuccessResponse,
};
use crate::api::subscriptions::ActiveSubscription;
use crate::api::wait::wait_for_output;
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, remove_socket_file_if_owned, socket_file_identity, LocalStream,
SocketFileIdentity,
bind_local_listener, is_connection_closed_error, local_stream_peer_closed,
remove_socket_file_if_owned, socket_file_identity, LocalStream, SocketFileIdentity,
};
const SOCKET_PERMISSION_MODE: u32 = 0o600;
@@ -197,6 +197,38 @@ fn handle_connection(
}
result
}
Method::EventsWait(params) => {
let Some(response) = wait_for_event(
request_id.clone(),
params,
&mut stream,
api_tx,
event_hub,
running,
)?
else {
crate::logging::api_request_completed(
&request_id,
method,
"client_disconnected",
changes_ui,
);
return Ok(());
};
let result = write_text_line_allow_disconnect(&mut stream, &response);
match &result {
Ok(()) => crate::logging::api_request_completed(
&request_id,
method,
api_response_outcome(&response),
changes_ui,
),
Err(err) => {
crate::logging::api_request_failed(&request_id, method, &err.to_string())
}
}
result
}
Method::PaneWaitForOutput(params) => {
let Some(response) =
wait_for_output(request_id.clone(), params, &mut stream, api_tx, running)?
@@ -515,40 +547,7 @@ pub(super) fn should_stop_connection(
return Ok(true);
}
probe_stream_closed(stream)
}
fn probe_stream_closed(stream: &mut LocalStream) -> std::io::Result<bool> {
stream.set_nonblocking(true)?;
let mut probe = [0u8; 1];
let status = match stream.read(&mut probe) {
Ok(0) => Ok(true),
Ok(_) => Ok(true),
Err(err)
if matches!(
err.kind(),
std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted
) =>
{
Ok(false)
}
Err(err) if is_connection_closed_error(&err) => Ok(true),
Err(err) => Err(err),
};
stream.set_nonblocking(false)?;
status
}
fn is_connection_closed_error(err: &std::io::Error) -> bool {
matches!(
err.kind(),
std::io::ErrorKind::BrokenPipe
| std::io::ErrorKind::ConnectionAborted
| std::io::ErrorKind::ConnectionReset
| std::io::ErrorKind::NotConnected
| std::io::ErrorKind::UnexpectedEof
| std::io::ErrorKind::WriteZero
)
local_stream_peer_closed(stream)
}
fn dispatch_to_app(request: Request, api_tx: &ApiRequestSender) -> String {
@@ -620,6 +619,7 @@ fn error_response_json(id: String, code: &str, message: String) -> String {
mod tests {
use super::*;
use interprocess::local_socket::traits::Listener as _;
use std::collections::HashMap;
use std::io::{BufRead, BufReader};
use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::UnixListener;
@@ -654,6 +654,64 @@ mod tests {
(client, server, path)
}
fn pane_info(
pane_id: &str,
agent_status: crate::api::schema::AgentStatus,
) -> crate::api::schema::PaneInfo {
crate::api::schema::PaneInfo {
pane_id: pane_id.into(),
terminal_id: "term_1".into(),
workspace_id: "ws_1".into(),
tab_id: "tab_1".into(),
focused: true,
cwd: None,
foreground_cwd: None,
label: None,
agent: Some("pi".into()),
title: None,
display_agent: None,
agent_status,
custom_status: None,
state_labels: HashMap::new(),
agent_session: None,
revision: 0,
}
}
fn spawn_pane_get_responder(
agent_status: crate::api::schema::AgentStatus,
) -> (ApiRequestSender, std::thread::JoinHandle<()>) {
let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
let responder = std::thread::spawn(move || {
while let Some(msg) = api_rx.blocking_recv() {
match msg.request.method {
Method::PaneGet(_) => msg
.respond_to
.send(
serde_json::to_string(&SuccessResponse {
id: msg.request.id,
result: ResponseResult::PaneInfo {
pane: pane_info("pane_1", agent_status),
},
})
.unwrap(),
)
.unwrap(),
Method::EventsWait(_) => msg
.respond_to
.send(error_response_json(
msg.request.id,
"unexpected_dispatch",
"events.wait should be handled by the api server".into(),
))
.unwrap(),
other => panic!("unexpected request: {other:?}"),
}
}
});
(api_tx, responder)
}
#[test]
fn socket_path_prefers_explicit_env_override() {
let _guard = env_lock().lock().unwrap();
@@ -783,6 +841,60 @@ mod tests {
assert_eq!(parsed.id, "req_2");
}
#[test]
fn events_wait_agent_status_returns_initial_match() {
let (api_tx, responder) =
spawn_pane_get_responder(crate::api::schema::AgentStatus::Blocked);
let (mut client, server, _path) = local_stream_pair("api-events-wait-initial");
client
.write_all(br#"{"id":"wait_1","method":"events.wait","params":{"match_event":{"event":"pane_agent_status_changed","pane_id":"pane_1","agent_status":"blocked"},"timeout_ms":1000}}"#)
.unwrap();
client.write_all(b"\n").unwrap();
client.flush().unwrap();
let running = Arc::new(AtomicBool::new(true));
let event_hub = EventHub::default();
handle_connection(server, &api_tx, &event_hub, &running, None).unwrap();
let response: serde_json::Value = serde_json::from_str(&read_line(&mut client)).unwrap();
assert_eq!(response["id"], "wait_1");
assert_eq!(response["result"]["type"], "wait_matched");
assert_eq!(
response["result"]["event"]["data"]["agent_status"],
"blocked"
);
drop(api_tx);
responder.join().unwrap();
}
#[test]
fn events_wait_agent_status_times_out_server_side() {
let (api_tx, responder) =
spawn_pane_get_responder(crate::api::schema::AgentStatus::Unknown);
let (mut client, server, _path) = local_stream_pair("api-events-wait-timeout");
client
.write_all(br#"{"id":"wait_2","method":"events.wait","params":{"match_event":{"event":"pane_agent_status_changed","pane_id":"pane_1","agent_status":"blocked"},"timeout_ms":30}}"#)
.unwrap();
client.write_all(b"\n").unwrap();
client.flush().unwrap();
let running = Arc::new(AtomicBool::new(true));
let event_hub = EventHub::default();
handle_connection(server, &api_tx, &event_hub, &running, None).unwrap();
let response: serde_json::Value = serde_json::from_str(&read_line(&mut client)).unwrap();
assert_eq!(response["id"], "wait_2");
assert_eq!(response["error"]["code"], "timeout");
assert_eq!(
response["error"]["message"],
"timed out waiting for event match"
);
drop(api_tx);
responder.join().unwrap();
}
#[test]
fn wait_for_output_stops_when_client_disconnects() {
let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
src/api/wait.rs
+119 -2
View File
@@ -4,14 +4,17 @@ use std::sync::Arc;
use regex::Regex;
use crate::api::schema::{
ErrorBody, ErrorResponse, Method, Request, ResponseResult, SuccessResponse,
ErrorBody, ErrorResponse, EventData, EventEnvelope, EventKind, EventMatch, EventsWaitParams,
Method, Request, ResponseResult, Subscription, SubscriptionEventData,
SubscriptionEventEnvelope, SuccessResponse,
};
use crate::api::server::{
dispatch_to_app_with_timeout, should_stop_connection, APP_RESPONSE_TIMEOUT,
CONNECTION_POLL_INTERVAL,
};
use crate::api::subscriptions::ActiveSubscription;
use crate::api::subscriptions::{match_output, output_match_read_source};
use crate::api::ApiRequestSender;
use crate::api::{ApiRequestSender, EventHub};
use crate::ipc::LocalStream;
pub(super) fn wait_for_output(
@@ -122,3 +125,117 @@ pub(super) fn wait_for_output(
std::thread::sleep(CONNECTION_POLL_INTERVAL);
}
}
pub(super) fn wait_for_event(
request_id: String,
params: EventsWaitParams,
stream: &mut LocalStream,
api_tx: &ApiRequestSender,
event_hub: &EventHub,
running: &Arc<AtomicBool>,
) -> std::io::Result<Option<String>> {
let deadline = params
.timeout_ms
.map(|ms| std::time::Instant::now() + std::time::Duration::from_millis(ms));
let subscription = match event_match_subscription(&request_id, params.match_event) {
Ok(subscription) => subscription,
Err(response) => return Ok(Some(serde_json::to_string(&response).unwrap())),
};
let mut active = match ActiveSubscription::new(subscription, &request_id, 0, api_tx, event_hub)
{
Ok(active) => active,
Err(response) => return Ok(Some(serde_json::to_string(&response).unwrap())),
};
loop {
if should_stop_connection(stream, running)? {
return Ok(None);
}
if let Some(event) = active.poll(api_tx, event_hub) {
return Ok(Some(wait_matched_response(&request_id, event)));
}
if deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) {
return Ok(Some(
serde_json::to_string(&ErrorResponse {
id: request_id,
error: ErrorBody {
code: "timeout".into(),
message: "timed out waiting for event match".into(),
},
})
.unwrap(),
));
}
std::thread::sleep(CONNECTION_POLL_INTERVAL);
}
}
fn event_match_subscription(
request_id: &str,
match_event: EventMatch,
) -> Result<Subscription, ErrorResponse> {
match match_event {
EventMatch::PaneAgentStatusChanged {
pane_id,
agent_status,
} => Ok(Subscription::PaneAgentStatusChanged {
pane_id,
agent_status: Some(agent_status),
}),
_ => Err(ErrorResponse {
id: request_id.into(),
error: ErrorBody {
code: "unsupported_event_wait_match".into(),
message: "events.wait currently supports pane agent status matches".into(),
},
}),
}
}
fn wait_matched_response(request_id: &str, event: serde_json::Value) -> String {
let Ok(event) = serde_json::from_value::<SubscriptionEventEnvelope>(event) else {
return serde_json::to_string(&ErrorResponse {
id: request_id.into(),
error: ErrorBody {
code: "internal_error".into(),
message: "failed to decode matched event".into(),
},
})
.unwrap();
};
let SubscriptionEventData::PaneAgentStatusChanged(data) = event.data else {
return serde_json::to_string(&ErrorResponse {
id: request_id.into(),
error: ErrorBody {
code: "unsupported_event_wait_match".into(),
message: "events.wait currently supports pane agent status matches".into(),
},
})
.unwrap();
};
serde_json::to_string(&SuccessResponse {
id: request_id.into(),
result: ResponseResult::WaitMatched {
event: EventEnvelope {
event: EventKind::PaneAgentStatusChanged,
data: EventData::PaneAgentStatusChanged {
pane_id: data.pane_id,
workspace_id: data.workspace_id,
agent_status: data.agent_status,
agent: data.agent,
title: data.title,
display_agent: data.display_agent,
custom_status: data.custom_status,
state_labels: data.state_labels,
},
},
},
})
.unwrap()
}
src/cli.rs
+75 -15
View File
@@ -4,8 +4,10 @@ use serde::Serialize;
use crate::api::client::{ApiClient, ApiClientError};
use crate::api::schema::{
AgentStatus, ClientWindowTitleSetParams, EmptyParams, Method, OutputMatch, PaneAgentState,
PaneWaitForOutputParams, ReadFormat, ReadSource, Request, SplitDirection, Subscription,
AgentStatus, ClientWindowTitleSetParams, EmptyParams, EventData, EventMatch, EventsWaitParams,
Method, OutputMatch, PaneAgentState, PaneWaitForOutputParams, ReadFormat, ReadSource, Request,
ResponseResult, SplitDirection, SubscriptionEventData, SubscriptionEventEnvelope,
SubscriptionEventKind,
};
mod agent;
@@ -820,19 +822,77 @@ fn wait_agent_status(args: &[String]) -> std::io::Result<i32> {
return Ok(2);
};
wait_for_agent_change(
Request {
id: "cli:wait:agent-status".into(),
method: Method::EventsSubscribe(crate::api::schema::EventsSubscribeParams {
subscriptions: vec![Subscription::PaneAgentStatusChanged {
pane_id,
agent_status: Some(agent_status),
}],
}),
},
timeout_ms,
"timed out waiting for agent status change",
)
wait_for_agent_status_change(pane_id, agent_status, timeout_ms)
}
fn wait_for_agent_status_change(
pane_id: String,
agent_status: AgentStatus,
timeout_ms: Option<u64>,
) -> std::io::Result<i32> {
let request = Request {
id: "cli:wait:agent-status".into(),
method: Method::EventsWait(EventsWaitParams {
match_event: EventMatch::PaneAgentStatusChanged {
pane_id,
agent_status,
},
timeout_ms,
}),
};
let response = send_request(&request)?;
match crate::api::client::parse_response_value(response) {
Ok(success) => {
let ResponseResult::WaitMatched { event } = success.result else {
return Err(std::io::Error::other("unexpected wait response result"));
};
let EventData::PaneAgentStatusChanged {
pane_id,
workspace_id,
agent_status,
agent,
title,
display_agent,
custom_status,
state_labels,
} = event.data
else {
return Err(std::io::Error::other("unexpected wait event data"));
};
let event = SubscriptionEventEnvelope {
event: SubscriptionEventKind::PaneAgentStatusChanged,
data: SubscriptionEventData::PaneAgentStatusChanged(
crate::api::schema::PaneAgentStatusChangedEvent {
pane_id,
workspace_id,
agent_status,
agent,
custom_status,
title,
display_agent,
state_labels,
},
),
};
println!(
"{}",
serde_json::to_string(&event).map_err(std::io::Error::other)?
);
Ok(0)
}
Err(ApiClientError::ErrorResponse(response)) => {
if response.error.code == "timeout" {
eprintln!("timed out waiting for agent status change");
} else {
eprintln!(
"{}",
serde_json::to_string(&response).map_err(std::io::Error::other)?
);
}
Ok(1)
}
Err(err) => Err(api_client_error_to_io(err)),
}
}
pub(super) fn wait_for_agent_change(
src/ipc.rs
+104
View File
@@ -1,9 +1,14 @@
use std::fs;
use std::io;
#[cfg(unix)]
use std::io::Read;
#[cfg(unix)]
use std::os::unix::fs::{MetadataExt, PermissionsExt};
use std::path::Path;
#[cfg(unix)]
use interprocess::local_socket::traits::Stream as _;
pub(crate) type LocalListener = interprocess::local_socket::Listener;
pub(crate) type LocalStream = interprocess::local_socket::Stream;
@@ -99,6 +104,75 @@ fn stale_socket_connect_error(kind: io::ErrorKind) -> bool {
) || (cfg!(windows) && kind == io::ErrorKind::WouldBlock)
}
pub(crate) fn local_stream_peer_closed(stream: &mut LocalStream) -> io::Result<bool> {
probe_stream_closed(stream)
}
#[cfg(unix)]
fn probe_stream_closed(stream: &mut LocalStream) -> io::Result<bool> {
stream.set_nonblocking(true)?;
let mut probe = [0u8; 1];
let status = match stream.read(&mut probe) {
Ok(0) => Ok(true),
Ok(_) => Ok(true),
Err(err)
if matches!(
err.kind(),
io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted
) =>
{
Ok(false)
}
Err(err) if is_connection_closed_error(&err) => Ok(true),
Err(err) => Err(err),
};
stream.set_nonblocking(false)?;
status
}
#[cfg(windows)]
fn probe_stream_closed(stream: &mut LocalStream) -> io::Result<bool> {
use std::os::windows::io::{AsHandle, AsRawHandle};
let LocalStream::NamedPipe(pipe) = stream;
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(),
std::ptr::null_mut(),
)
};
if ok != 0 {
return Ok(false);
}
let err = io::Error::last_os_error();
if is_connection_closed_error(&err) || windows_named_pipe_closed_error(&err) {
return Ok(true);
}
Err(err)
}
pub(crate) fn is_connection_closed_error(err: &io::Error) -> bool {
matches!(
err.kind(),
io::ErrorKind::BrokenPipe
| io::ErrorKind::ConnectionAborted
| io::ErrorKind::ConnectionReset
| io::ErrorKind::NotConnected
| io::ErrorKind::UnexpectedEof
| io::ErrorKind::WriteZero
)
}
#[cfg(windows)]
fn windows_named_pipe_closed_error(err: &io::Error) -> bool {
matches!(err.raw_os_error(), Some(6 | 109 | 232 | 233))
}
pub(crate) fn socket_file_identity(path: &Path) -> io::Result<SocketFileIdentity> {
#[cfg(windows)]
{
@@ -163,6 +237,8 @@ pub(crate) fn restrict_socket_permissions(_path: &Path, _mode: u32) -> io::Resul
mod tests {
use super::*;
#[cfg(windows)]
use interprocess::local_socket::traits::Listener as _;
#[cfg(windows)]
use std::path::PathBuf;
#[test]
@@ -193,6 +269,34 @@ mod tests {
let _ = fs::remove_file(&path);
}
#[cfg(windows)]
#[test]
fn idle_named_pipe_peer_is_not_treated_as_closed() {
let path = temp_socket_marker_path("idle-pipe");
let listener = bind_local_listener(&path).unwrap();
let _client = connect_local_stream(&path).unwrap();
let mut server = listener.accept().unwrap();
assert!(!local_stream_peer_closed(&mut server).unwrap());
let _ = fs::remove_file(path);
}
#[cfg(windows)]
#[test]
fn disconnected_named_pipe_peer_is_treated_as_closed() {
let path = temp_socket_marker_path("disconnected-pipe");
let listener = bind_local_listener(&path).unwrap();
let client = connect_local_stream(&path).unwrap();
let mut server = listener.accept().unwrap();
drop(client);
assert!(local_stream_peer_closed(&mut server).unwrap());
let _ = fs::remove_file(path);
}
#[cfg(windows)]
fn temp_socket_marker_path(name: &str) -> PathBuf {
std::env::temp_dir().join(format!("herdr-{name}-{}.sock", std::process::id()))
src/server/headless.rs
+3 -5
View File
@@ -193,7 +193,8 @@ pub struct HeadlessServer {
app: app::App,
#[cfg(unix)]
api_tx: Option<api::ApiRequestSender>,
#[cfg(unix)]
// Kept on every platform so dropping HeadlessServer owns API server shutdown.
#[cfg_attr(windows, allow(dead_code))]
api_server: Option<api::ServerHandle>,
#[cfg(unix)]
client_listener: LocalListener,
@@ -386,13 +387,11 @@ impl HeadlessServer {
let (server_config_diagnostic, server_config_diagnostic_without_keybindings) =
server_config_diagnostic_summaries(config_diagnostics);
#[cfg(not(unix))]
let _ = (&api_tx, &api_server);
let _ = api_tx;
Ok(Self {
app,
#[cfg(unix)]
api_tx,
#[cfg(unix)]
api_server,
#[cfg(unix)]
client_listener: listener,
@@ -4173,7 +4172,6 @@ mod tests {
app,
#[cfg(unix)]
api_tx: None,
#[cfg(unix)]
api_server: None,
#[cfg(unix)]
client_listener: listener,
tests/cli_wrapper.rs
+42
View File
@@ -4217,6 +4217,48 @@ fn wait_agent_status_exits_immediately_when_status_already_matches() {
cleanup_spawned_herdr(herdr, base);
}
#[test]
fn wait_agent_status_times_out_when_status_does_not_match() {
let base = unique_test_dir();
let config_home = base.join("config");
let runtime_dir = base.join("runtime");
let socket_path = runtime_dir.join("herdr.sock");
let herdr = spawn_herdr(&config_home, &runtime_dir, &socket_path);
wait_for_socket(&socket_path, Duration::from_secs(5));
let created = send_request(
&socket_path,
&format!(
r#"{{"id":"req_cli_timeout_1","method":"workspace.create","params":{{"cwd":"{}","focus":true}}}}"#,
base.display()
),
);
assert_eq!(created["result"]["type"], "workspace_created");
let waited = run_cli(
&socket_path,
&[
"wait",
"agent-status",
"1-1",
"--status",
"blocked",
"--timeout",
"100",
],
);
assert!(!waited.status.success());
assert!(
String::from_utf8_lossy(&waited.stderr)
.contains("timed out waiting for agent status change"),
"stderr: {}",
String::from_utf8_lossy(&waited.stderr)
);
cleanup_spawned_herdr(herdr, base);
}
#[test]
fn wait_agent_status_exits_when_done_status_matches() {
let base = unique_test_dir();