diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index 052efee9..3777cc3e 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::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; use tokio::task::JoinHandle; @@ -63,21 +63,23 @@ pub fn start(agent: &str, socket_path: &Path, coord: Arc) -> Result let path = socket_path.to_path_buf(); let handle = tokio::spawn(async move { - 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); + 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; + } + } + } }); Ok(AgentSocket { path, handle }) } @@ -105,153 +107,47 @@ pub fn start_manager(coord: Arc) -> Result<()> { tracing::info!(socket = %socket.display(), "manager socket listening"); tokio::spawn(async move { - 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); + 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; + } + } + } }); 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 = Vec::new(); + let mut line = String::new(); loop { - 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, - ), + 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 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" - ); - } } } @@ -895,74 +791,4 @@ 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 64d018ad..d41d2dfe 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)] -pub(super) mod tests { +mod tests { use super::*; fn target(name: &str) -> hive_sh4re::schedule::WireScheduleTarget { @@ -377,10 +377,10 @@ pub(super) mod tests { /// anybody's ancestor or descendant. const STRANGER: &str = "requester-with-no-relations"; - /// A real `Coordinator` over a throwaway sqlite dir. The socket-server + /// A real `Coordinator` over a throwaway sqlite dir. The schedule /// handlers take `&Arc`, so there is no lighter way in; /// `open` touches nothing outside the db path it is handed. - pub(in crate::socket_server) fn coordinator() -> (tempfile::TempDir, Arc) { + 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 49d9b860..ee0f3de3 100644 --- a/hive-c0re/src/workers/mcp_sockets.rs +++ b/hive-c0re/src/workers/mcp_sockets.rs @@ -10,11 +10,9 @@ //! - `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. 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. +//! 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. use std::sync::Arc;