From 65a1f8d9022389827d02cc9525e09ded3a1811d0 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 26 Sep 2026 00:15:22 +0200 Subject: [PATCH 1/2] hive-c0re: bound request lines on the agent and manager sockets serve() read each request with read_line into an unbounded String, so size checks such as the 4 KiB Send body limit ran only after the whole line was buffered. Any process in an agent container can write to /run/hive/mcp.sock, and one stream with no newline grew hive-c0re's heap until the host OOM-killed it. Read at most 16 MiB per line. The largest request a client sends is an OperatorMsg from the agent web UI's /send, whose body axum's default limit caps at 2 MiB (the gateway's nginx admits 10 MiB); JSON escaping can roughly double that, and the cap is 4x the result. A longer line gets a Response::Err, the same shape as the parse-error path, and the connection is closed because the rest of the line is still unread. The line is read as bytes and parsed with serde_json::from_slice, so a line with invalid UTF-8 now gets a parse-error response instead of the connection being dropped. Closes #4718 --- hive-c0re/src/socket_server/mod.rs | 124 +++++++++++++++++++++-- hive-c0re/src/socket_server/schedules.rs | 6 +- 2 files changed, 116 insertions(+), 14 deletions(-) diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index 3777cc3e..aff0862f 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; @@ -128,26 +128,77 @@ pub fn start_manager(coord: Arc) -> Result<()> { Ok(()) } +/// 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 +842,55 @@ 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"); + } } 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"), From 386741d38f3096faa2f76af7f4ca668b36a3d777 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 26 Sep 2026 00:17:59 +0200 Subject: [PATCH 2/2] hive-c0re: keep agent and manager listeners alive across accept errors Both accept loops returned on the first accept() error. The AgentSocket stayed in Coordinator.agents, so mcp_sockets::sync_on_start skipped the agent, and only a Start node or a hive-c0re restart bound it again. The unit sets no LimitNOFILE, so a single EMFILE at the 1024-fd soft limit cut the agent off from send/recv/ack with one warn line as the only trace. Both loops now share accept_until_fatal. Connection errors (ECONNABORTED/ECONNRESET/ECONNREFUSED) retry at once, any other error retries after 1 s, following axum's serve loop that the dashboard already runs. Errors that leave the listening fd unusable (EBADF, EFAULT, EINVAL, ENOTSOCK, EOPNOTSUPP) log at error and exit the process: nothing re-binds a listener while hive-c0re runs, the manager listener has no re-bind path at all, and the unit's Restart=on-failure restart runs start_manager and sync_on_start, which re-bind every socket. mcp_sockets.rs now states that invariant instead of asserting that a listener can only disappear on restart. LimitNOFILE is left unset: per-agent fd use is a listener plus a Recv long-poll plus short-lived requests, and a higher limit would raise the memory ceiling of the per-line bound (bound x open connections). Closes #4721 --- hive-c0re/src/socket_server/mod.rs | 138 ++++++++++++++++++++------- hive-c0re/src/workers/mcp_sockets.rs | 8 +- 2 files changed, 110 insertions(+), 36 deletions(-) diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index aff0862f..052efee9 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -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,25 +105,80 @@ 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; } } } - }); - Ok(()) + } +} + +/// 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 @@ -893,4 +946,23 @@ mod tests { ); 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/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;