diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index f7662c55..13ae10a1 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -113,14 +113,6 @@ pub(crate) fn recv_timeout(wait_seconds: Option) -> std::time::Duration { } } -#[allow( - clippy::too_many_lines, - reason = "flat routing table over the shared request variants — the \ - logic-bearing arms (recv / wake / set_status / get_agent_meta) \ - are extracted to handlers; what's left is one short broker \ - dispatch per variant, kept inline so the routing map reads \ - top-to-bottom" -)] /// Handle the subset of `Request` variants that are identical on both /// the agent socket and the manager socket. Returns `Some(response)` for /// every variant it handles; returns `None` for variants with socket-specific @@ -134,7 +126,6 @@ pub(crate) async fn dispatch_shared( agent: &str, coord: &Arc, ) -> Option { - let broker = &coord.broker; Some(match req { hive_sh4re::Request::Send { to, @@ -144,34 +135,14 @@ pub(crate) async fn dispatch_shared( hive_sh4re::Request::Recv { wait_seconds, max } => { handle_recv(coord, agent, *wait_seconds, *max).await } - hive_sh4re::Request::Status => match broker.count_pending(agent) { - Ok(unread) => hive_sh4re::Response::Status { unread }, - Err(e) => hive_sh4re::Response::Err { - message: format!("{e:#}"), - }, - }, - hive_sh4re::Request::OperatorMsg { body } => match broker.send(&Message { - from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(), - to: agent.to_owned(), - body: body.clone(), - in_reply_to: None, - }) { - Ok(()) => hive_sh4re::Response::Ok, - Err(e) => hive_sh4re::Response::Err { - message: format!("{e:#}"), - }, - }, + hive_sh4re::Request::Status => handle_status(coord, agent), + hive_sh4re::Request::OperatorMsg { body } => handle_operator_msg(coord, agent, body), hive_sh4re::Request::Wake { from, body, transient, } => handle_wake(coord, agent, from, body, *transient), - hive_sh4re::Request::Recent { limit } => match broker.recent_for(agent, *limit) { - Ok(rows) => hive_sh4re::Response::Recent { rows }, - Err(e) => hive_sh4re::Response::Err { - message: format!("{e:#}"), - }, - }, + hive_sh4re::Request::Recent { limit } => handle_recent(coord, agent, *limit), hive_sh4re::Request::Ask { question, options, @@ -212,23 +183,8 @@ pub(crate) async fn dispatch_shared( |()| hive_sh4re::Response::Ok, ) } - hive_sh4re::Request::AckTurn => match broker.ack_turn(agent) { - Ok(_n) => hive_sh4re::Response::Ok, - Err(e) => hive_sh4re::Response::Err { - message: format!("{e:#}"), - }, - }, - hive_sh4re::Request::RequeueInflight => match broker.requeue_inflight(agent) { - Ok(n) => { - if n > 0 { - tracing::info!(%agent, requeued = %n, "requeued in-flight messages"); - } - hive_sh4re::Response::Ok - } - Err(e) => hive_sh4re::Response::Err { - message: format!("{e:#}"), - }, - }, + hive_sh4re::Request::AckTurn => handle_ack_turn(coord, agent), + hive_sh4re::Request::RequeueInflight => handle_requeue_inflight(coord, agent), hive_sh4re::Request::GetHostJournal { unit, container, @@ -355,6 +311,70 @@ async fn handle_get_agent_meta( } } +/// `Status` — count of pending (unread) inbox messages for `agent`. +fn handle_status(coord: &Arc, agent: &str) -> hive_sh4re::Response { + match coord.broker.count_pending(agent) { + Ok(unread) => hive_sh4re::Response::Status { unread }, + Err(e) => hive_sh4re::Response::Err { + message: format!("{e:#}"), + }, + } +} + +/// `OperatorMsg` — deliver an operator-authored message into `agent`'s +/// inbox (from the `operator` recipient). +fn handle_operator_msg(coord: &Arc, agent: &str, body: &str) -> hive_sh4re::Response { + match coord.broker.send(&Message { + from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(), + to: agent.to_owned(), + body: body.to_owned(), + in_reply_to: None, + }) { + Ok(()) => hive_sh4re::Response::Ok, + Err(e) => hive_sh4re::Response::Err { + message: format!("{e:#}"), + }, + } +} + +/// `Recent` — the last `limit` inbox rows for `agent` (read-only, +/// doesn't consume). +fn handle_recent(coord: &Arc, agent: &str, limit: u64) -> hive_sh4re::Response { + match coord.broker.recent_for(agent, limit) { + Ok(rows) => hive_sh4re::Response::Recent { rows }, + Err(e) => hive_sh4re::Response::Err { + message: format!("{e:#}"), + }, + } +} + +/// `AckTurn` — mark `agent`'s in-flight delivered messages acked so +/// they don't redeliver on the next turn. +fn handle_ack_turn(coord: &Arc, agent: &str) -> hive_sh4re::Response { + match coord.broker.ack_turn(agent) { + Ok(_n) => hive_sh4re::Response::Ok, + Err(e) => hive_sh4re::Response::Err { + message: format!("{e:#}"), + }, + } +} + +/// `RequeueInflight` — resurface `agent`'s unacked in-flight messages +/// (crash recovery on harness boot). +fn handle_requeue_inflight(coord: &Arc, agent: &str) -> hive_sh4re::Response { + match coord.broker.requeue_inflight(agent) { + Ok(n) => { + if n > 0 { + tracing::info!(%agent, requeued = %n, "requeued in-flight messages"); + } + hive_sh4re::Response::Ok + } + Err(e) => hive_sh4re::Response::Err { + message: format!("{e:#}"), + }, + } +} + async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> AgentResponse { if let Some(resp) = dispatch_shared(req, agent, coord).await { return resp; diff --git a/hive-priv/src/main.rs b/hive-priv/src/main.rs index ba8e05f9..edf8c6a6 100644 --- a/hive-priv/src/main.rs +++ b/hive-priv/src/main.rs @@ -155,13 +155,6 @@ async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: /// For streaming ops (`CreateContainer`/`UpdateContainer` with `stream: true`) /// output lines are forwarded to `writer` as `PrivEvent::Line` messages and /// the returned strings are empty. -#[allow( - clippy::too_many_lines, - reason = "flat routing table over the privileged request variants — the \ - logic-bearing arms (resource-limits / daemon-reload / \ - restart-matrix / create+update) are extracted to helpers; the \ - rest are one-line validate-then-delegate dispatches kept inline" -)] async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> { match req { PrivRequest::StartContainer { ref name } => { @@ -206,15 +199,7 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, ref container, ref binds, ref isolation, - } => { - validate_container_system_name(container)?; - for bind in binds { - validate_bind_path(&bind.host_path)?; - validate_bind_path(&bind.container_path)?; - } - write_nspawn_flags(container, binds, isolation.as_ref())?; - Ok((String::new(), String::new())) - } + } => handle_write_nspawn_flags(container, binds, isolation.as_ref()), PrivRequest::WriteResourceLimits { ref container, @@ -222,14 +207,7 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, ref cpu_quota, } => write_resource_limits(container, memory_max, cpu_quota), - PrivRequest::RemoveServiceDropin { ref container } => { - validate_container_system_name(container)?; - let dir = format!("/run/systemd/system/container@{container}.service.d"); - if Path::new(&dir).exists() { - std::fs::remove_dir_all(&dir).with_context(|| format!("remove {dir}"))?; - } - Ok((String::new(), String::new())) - } + PrivRequest::RemoveServiceDropin { ref container } => remove_service_dropin(container), PrivRequest::DaemonReload => daemon_reload().await, @@ -239,25 +217,12 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, ref agent_name, uid, gid, - } => { - validate_agent_name(agent_name)?; - let path = socket_dir_path(agent_name); - std::os::unix::fs::chown(&path, Some(uid), Some(gid)) - .with_context(|| format!("chown {} to {uid}:{gid}", path.display()))?; - Ok((String::new(), String::new())) - } + } => chown_socket_dir(agent_name, uid, gid), PrivRequest::ChmodSocketDir { ref agent_name, mode, - } => { - use std::os::unix::fs::PermissionsExt as _; - validate_agent_name(agent_name)?; - let path = socket_dir_path(agent_name); - std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode)) - .with_context(|| format!("chmod {:o} {}", mode, path.display()))?; - Ok((String::new(), String::new())) - } + } => chmod_socket_dir(agent_name, mode), PrivRequest::RunForgeAdmin { ref args } => { for arg in args { @@ -307,6 +272,53 @@ async fn container_flake_action( } } +/// `WriteNspawnFlags` — validate the container + every bind path, then +/// write the container's nspawn flag overrides. +fn handle_write_nspawn_flags( + container: &str, + binds: &[BindMount], + isolation: Option<&NetworkIsolation>, +) -> Result<(String, String)> { + validate_container_system_name(container)?; + for bind in binds { + validate_bind_path(&bind.host_path)?; + validate_bind_path(&bind.container_path)?; + } + write_nspawn_flags(container, binds, isolation)?; + Ok((String::new(), String::new())) +} + +/// `RemoveServiceDropin` — remove the container service's drop-in dir +/// if present (idempotent). +fn remove_service_dropin(container: &str) -> Result<(String, String)> { + validate_container_system_name(container)?; + let dir = format!("/run/systemd/system/container@{container}.service.d"); + if Path::new(&dir).exists() { + std::fs::remove_dir_all(&dir).with_context(|| format!("remove {dir}"))?; + } + Ok((String::new(), String::new())) +} + +/// `ChownSocketDir` — chown the agent's host socket dir to its +/// container uid/gid. +fn chown_socket_dir(agent_name: &str, uid: u32, gid: u32) -> Result<(String, String)> { + validate_agent_name(agent_name)?; + let path = socket_dir_path(agent_name); + std::os::unix::fs::chown(&path, Some(uid), Some(gid)) + .with_context(|| format!("chown {} to {uid}:{gid}", path.display()))?; + Ok((String::new(), String::new())) +} + +/// `ChmodSocketDir` — set the mode on the agent's host socket dir. +fn chmod_socket_dir(agent_name: &str, mode: u32) -> Result<(String, String)> { + use std::os::unix::fs::PermissionsExt as _; + validate_agent_name(agent_name)?; + let path = socket_dir_path(agent_name); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode)) + .with_context(|| format!("chmod {:o} {}", mode, path.display()))?; + Ok((String::new(), String::new())) +} + /// `WriteResourceLimits` — drop a systemd `MemoryMax`/`CPUQuota` /// override into the container service's drop-in dir. fn write_resource_limits(