vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

Compare commits

base: vinzenz:386741d38f3096faa2f76af7f4ca668b36a3d777
Branches Tags
vinzenz:main
..
compare: vinzenz:7b1fe5f9d37cf66d1c53e331d7996c7db1c32d7f
Branches Tags
vinzenz:main
3 changed files with 50 additions and 226 deletions
Download patch file Download diff file Expand all files Collapse all files

264
hive-c0re/src/socket_server/mod.rs
View file

@ -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<Coordinator>) -> 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<Coordinator>) -> 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<A, F>(
agent: &str,
mut accept: A,
mut on_conn: impl FnMut(UnixStream),
) -> std::io::Error
where
A: FnMut() -> F,
F: std::future::Future<Output = std::io::Result<UnixStream>>,
{
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<R: AsyncBufRead + Unpin>(
reader: &mut R,
buf: &mut Vec<u8>,
) -> std::io::Result<RequestLine> {
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<Coordinator>) -> 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::<Request>(&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::<Request>(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));
}
}

6
hive-c0re/src/socket_server/schedules.rs
View file

@ -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<Coordinator>`, 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<Coordinator>) {
fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
let dir = tempfile::tempdir().expect("tempdir");
let coord = Coordinator::open(
&dir.path().join("broker.sqlite"),

8
hive-c0re/src/workers/mcp_sockets.rs
View file

@ -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;