diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index 3777cc3e..052efee9 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -15,7 +15,7 @@ use anyhow::{Context, Result}; use hive_core_agent_sock::{Request, Response}; use hive_sh4re::inbox::Message; use hive_sh4re::manager::MANAGER_AGENT; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; use tokio::task::JoinHandle; @@ -63,23 +63,21 @@ pub fn start(agent: &str, socket_path: &Path, coord: Arc) -> Result let path = socket_path.to_path_buf(); let handle = tokio::spawn(async move { - loop { - match listener.accept().await { - Ok((stream, _)) => { - let agent = agent.clone(); - let coord = coord.clone(); - tokio::spawn(async move { - if let Err(e) = serve(stream, agent, coord).await { - tracing::warn!(error = ?e, "agent connection failed"); - } - }); - } - Err(e) => { - tracing::warn!(error = ?e, "agent listener accept failed; exiting"); - return; - } - } - } + let err = accept_until_fatal( + &agent, + || async { listener.accept().await.map(|(stream, _)| stream) }, + |stream| { + let agent = agent.clone(); + let coord = coord.clone(); + tokio::spawn(async move { + if let Err(e) = serve(stream, agent, coord).await { + tracing::warn!(error = ?e, "agent connection failed"); + } + }); + }, + ) + .await; + exit_on_dead_listener(&agent, &err); }); Ok(AgentSocket { path, handle }) } @@ -107,47 +105,153 @@ pub fn start_manager(coord: Arc) -> Result<()> { tracing::info!(socket = %socket.display(), "manager socket listening"); tokio::spawn(async move { - loop { - match listener.accept().await { - Ok((stream, _)) => { - let coord = coord.clone(); - tokio::spawn(async move { - // Pure transport: serve as `ruth`, no privilege grant. - if let Err(e) = serve(stream, MANAGER_AGENT.to_owned(), coord).await { - tracing::warn!(error = ?e, "manager connection failed"); - } - }); - } - Err(e) => { - tracing::warn!(error = ?e, "manager listener accept failed"); - return; - } - } - } + let err = accept_until_fatal( + MANAGER_AGENT, + || async { listener.accept().await.map(|(stream, _)| stream) }, + |stream| { + let coord = coord.clone(); + tokio::spawn(async move { + // Pure transport: serve as `ruth`, no privilege grant. + if let Err(e) = serve(stream, MANAGER_AGENT.to_owned(), coord).await { + tracing::warn!(error = ?e, "manager connection failed"); + } + }); + }, + ) + .await; + exit_on_dead_listener(MANAGER_AGENT, &err); }); Ok(()) } +/// Pause after an accept error that is not about a single connection. EMFILE +/// and ENFILE leave the listener readable, so retrying at once would spin until +/// some fd is closed. Same policy as axum's `serve`, which the dashboard runs. +const ACCEPT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1); + +/// Accept errors that mean the listening fd itself is unusable: retrying it +/// returns the same error forever. +fn accept_error_is_fatal(e: &std::io::Error) -> bool { + matches!( + e.raw_os_error(), + Some(libc::EBADF | libc::EFAULT | libc::EINVAL | libc::ENOTSOCK | libc::EOPNOTSUPP) + ) +} + +/// Hand each accepted connection to `on_conn`, retrying every accept error +/// except a fatal one, which is returned. `agent` labels the log lines. +async fn accept_until_fatal( + agent: &str, + mut accept: A, + mut on_conn: impl FnMut(UnixStream), +) -> std::io::Error +where + A: FnMut() -> F, + F: std::future::Future>, +{ + loop { + match accept().await { + Ok(stream) => on_conn(stream), + Err(e) if accept_error_is_fatal(&e) => return e, + Err(e) => { + tracing::warn!(%agent, error = ?e, "socket accept failed; retrying"); + if !matches!( + e.kind(), + std::io::ErrorKind::ConnectionAborted + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionRefused + ) { + tokio::time::sleep(ACCEPT_RETRY_DELAY).await; + } + } + } + } +} + +/// Exit on a listener that can no longer accept. A running hive-c0re re-binds +/// an agent socket only on that agent's next start, and the manager socket +/// never; on restart (`Restart = "on-failure"` on the unit) `start_manager` and +/// `mcp_sockets::sync_on_start` re-bind every one. +fn exit_on_dead_listener(agent: &str, e: &std::io::Error) -> ! { + tracing::error!( + %agent, + error = ?e, + "socket listener cannot accept; exiting so hive-c0re restarts and re-binds all sockets" + ); + std::process::exit(1) +} + +/// Longest request line accepted on the agent and manager sockets, newline +/// included. The largest request a client sends is an `OperatorMsg` from the +/// agent web UI's `/send` (`hive-agent/src/web_ui/actions.rs:21`), whose body +/// axum's default limit caps at 2 MiB; JSON escaping can roughly double that to +/// 4 MiB, and this is 4× that. +const REQUEST_LINE_MAX_BYTES: usize = 16 * 1024 * 1024; + +enum RequestLine { + Eof, + Complete, + TooLong, +} + +/// Read one request line into `buf`, consuming at most +/// [`REQUEST_LINE_MAX_BYTES`] from `reader`. +async fn read_request_line( + reader: &mut R, + buf: &mut Vec, +) -> std::io::Result { + buf.clear(); + let n = (&mut *reader) + .take(REQUEST_LINE_MAX_BYTES as u64) + .read_until(b'\n', buf) + .await?; + Ok(if n == 0 { + RequestLine::Eof + } else if n == REQUEST_LINE_MAX_BYTES && !buf.ends_with(b"\n") { + RequestLine::TooLong + } else { + RequestLine::Complete + }) +} + +/// Answer each request line on `stream` until EOF. An over-long line gets a +/// `Response::Err` and then the connection is closed, because the rest of the +/// line is still unread. async fn serve(stream: UnixStream, agent: String, coord: Arc) -> Result<()> { let (read, mut write) = stream.into_split(); let mut reader = BufReader::new(read); - let mut line = String::new(); + let mut line = Vec::new(); loop { - line.clear(); - let n = reader.read_line(&mut line).await?; - if n == 0 { - return Ok(()); - } - let resp = match serde_json::from_str::(line.trim()) { - Ok(req) => dispatch(&req, &agent, &coord).await, - Err(e) => Response::Err { - message: format!("parse error: {e}"), + let (resp, too_long) = match read_request_line(&mut reader, &mut line).await? { + RequestLine::Eof => return Ok(()), + RequestLine::TooLong => ( + Response::Err { + message: format!( + "request too long: no newline within {REQUEST_LINE_MAX_BYTES} bytes; \ + connection closed" + ), + }, + true, + ), + RequestLine::Complete => match serde_json::from_slice::(&line) { + Ok(req) => (dispatch(&req, &agent, &coord).await, false), + Err(e) => ( + Response::Err { + message: format!("parse error: {e}"), + }, + false, + ), }, }; let mut payload = serde_json::to_string(&resp)?; payload.push('\n'); write.write_all(payload.as_bytes()).await?; write.flush().await?; + if too_long { + anyhow::bail!( + "`{agent}` sent a request line over {REQUEST_LINE_MAX_BYTES} bytes; closed" + ); + } } } @@ -791,4 +895,74 @@ mod tests { let err = check_can_cancel_approval("nobody-with-no-groups").unwrap_err(); assert!(err.contains("approvals` tool group"), "{err}"); } + + #[tokio::test] + async fn over_long_line_is_rejected_after_reading_only_the_bound() { + let total = 2 * REQUEST_LINE_MAX_BYTES as u64; + let mut reader = BufReader::new(tokio::io::repeat(b'x').take(total)); + let mut buf = Vec::new(); + let outcome = read_request_line(&mut reader, &mut buf) + .await + .expect("read"); + // At most one `BufReader` fill past the bound left the source. + let unread = reader.into_inner().limit(); + let consumed = total - unread; + assert!( + consumed <= (REQUEST_LINE_MAX_BYTES + 8 * 1024) as u64, + "consumed {consumed} of {total} bytes" + ); + assert_eq!(buf.len(), REQUEST_LINE_MAX_BYTES); + assert!(matches!(outcome, RequestLine::TooLong)); + } + + #[tokio::test] + async fn serve_answers_an_over_long_line_with_an_error_and_closes() { + let (_dir, coord) = schedules::tests::coordinator(); + let (client, server) = UnixStream::pair().expect("socketpair"); + let handler = tokio::spawn(serve(server, "iris".to_owned(), coord)); + let (read, mut write) = client.into_split(); + let writer = tokio::spawn(async move { + let chunk = vec![b'x'; 64 * 1024]; + for _ in 0..2 * REQUEST_LINE_MAX_BYTES / chunk.len() { + if write.write_all(&chunk).await.is_err() { + return; + } + } + let _ = write.shutdown().await; + }); + let mut reply = String::new(); + BufReader::new(read) + .read_line(&mut reply) + .await + .expect("reply"); + let resp: Response = serde_json::from_str(&reply).expect("response json"); + assert!( + matches!(&resp, Response::Err { message } if message.starts_with("request too long")), + "{reply}" + ); + assert!( + handler.await.expect("join").is_err(), + "connection not closed" + ); + writer.await.expect("join"); + } + + #[tokio::test] + async fn accept_loop_survives_a_transient_error() { + let (client, _peer) = UnixStream::pair().expect("socketpair"); + let mut script = std::collections::VecDeque::from([ + Err(std::io::Error::from_raw_os_error(libc::EMFILE)), + Ok(client), + Err(std::io::Error::from_raw_os_error(libc::EBADF)), + ]); + let mut served = 0; + let err = accept_until_fatal( + "iris", + || std::future::ready(script.pop_front().expect("accept after a fatal error")), + |_stream| served += 1, + ) + .await; + assert_eq!(served, 1, "connection after EMFILE not served"); + assert_eq!(err.raw_os_error(), Some(libc::EBADF)); + } } diff --git a/hive-c0re/src/socket_server/schedules.rs b/hive-c0re/src/socket_server/schedules.rs index d41d2dfe..64d018ad 100644 --- a/hive-c0re/src/socket_server/schedules.rs +++ b/hive-c0re/src/socket_server/schedules.rs @@ -302,7 +302,7 @@ fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::schedu } #[cfg(test)] -mod tests { +pub(super) mod tests { use super::*; fn target(name: &str) -> hive_sh4re::schedule::WireScheduleTarget { @@ -377,10 +377,10 @@ mod tests { /// anybody's ancestor or descendant. const STRANGER: &str = "requester-with-no-relations"; - /// A real `Coordinator` over a throwaway sqlite dir. The schedule + /// A real `Coordinator` over a throwaway sqlite dir. The socket-server /// handlers take `&Arc`, so there is no lighter way in; /// `open` touches nothing outside the db path it is handed. - fn coordinator() -> (tempfile::TempDir, Arc) { + pub(in crate::socket_server) fn coordinator() -> (tempfile::TempDir, Arc) { let dir = tempfile::tempdir().expect("tempdir"); let coord = Coordinator::open( &dir.path().join("broker.sqlite"), diff --git a/hive-c0re/src/workers/mcp_sockets.rs b/hive-c0re/src/workers/mcp_sockets.rs index ee0f3de3..49d9b860 100644 --- a/hive-c0re/src/workers/mcp_sockets.rs +++ b/hive-c0re/src/workers/mcp_sockets.rs @@ -10,9 +10,11 @@ //! - `run_reconcile` calls `register_agent` immediately after `start_with_fallback`. //! - `kill`/`destroy` paths call `unregister_agent`. //! -//! No recurring poll is needed because c0re owns the listener lifecycle — -//! a listener can only disappear when c0re itself restarts, which is exactly -//! the case `sync_on_start` covers. +//! No recurring poll is needed because c0re owns the listener lifecycle. An +//! accept loop retries every error except one that leaves its fd unusable, +//! and on that one hive-c0re exits (`socket_server::exit_on_dead_listener`), +//! so a listener only disappears across a restart — exactly the case +//! `sync_on_start` covers. use std::sync::Arc;