vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(#1474): extract remaining dispatch_shared + hive-priv arms, drop their too_many_lines allows

Browse source
This commit is contained in:
damocles 2026-06-09 12:22:25 +02:00
commit 5804e986ce
2 changed files with 120 additions and 88 deletions
Download patch file Download diff file Expand all files Collapse all files

118
hive-c0re/src/agent_server.rs
View file

@ -113,14 +113,6 @@ pub(crate) fn recv_timeout(wait_seconds: Option<u64>) -> 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<Coordinator>,
) -> Option<hive_sh4re::Response> {
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<Coordinator>, 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<Coordinator>, 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<Coordinator>, 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<Coordinator>, 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<Coordinator>, 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<Coordinator>) -> AgentResponse {
if let Some(resp) = dispatch_shared(req, agent, coord).await {
return resp;

90
hive-priv/src/main.rs
View file

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