d4f3d1ef8e
The forge and matrix are swarm services; per mara's ruling on #5475 (shape a) a hive no longer starts, stops or restarts them. - hive-priv-sock: InfraContainer keeps only Gateway; InfraTarget goes, service_unit maps the gateway to nginx.service directly. A ControlInfraContainer naming Forge/Matrix fails to decode. - hive-host-sock: LifecycleScope drops forge/matrix and refuses unknown fields, so an older hivectl's --forge is an error rather than an all-false scope meaning everything. - hive-c0re: scoped_infra selects only the gateway. - hivectl: --forge/--matrix removed from stop/start/restart. - docs: services.md says how the forge and matrix are managed; security.md, coordinator.md and hivectl-cli.md follow. - hive-host-sock, hivectl/power.rs: Stop and RestartScoped doc comments name the gateway (host nginx.service), not infra containers.
950 lines
39 KiB
Rust
950 lines
39 KiB
Rust
use std::path::Path;
|
|
use std::sync::Arc;
|
|
|
|
use anyhow::{Context, Result};
|
|
use hive_host_sock::{HostRequest, HostResponse, LifecycleScope};
|
|
use hive_priv_sock::{InfraAction, InfraContainer};
|
|
use hive_sock_client::{LineDeadlines, REQUEST_LINE_DEADLINE, RequestLine, read_request_line};
|
|
use tokio::io::{AsyncWriteExt, BufReader};
|
|
use tokio::net::{UnixListener, UnixStream};
|
|
|
|
use crate::actions;
|
|
use crate::coordinator::Coordinator;
|
|
use crate::lifecycle;
|
|
|
|
pub async fn serve(socket: &Path, coord: Arc<Coordinator>) -> Result<()> {
|
|
// Prefer a socket passed by systemd socket-activation (LISTEN_FDS).
|
|
// When running under a `.socket` unit, systemd has already created,
|
|
// bound, and chmod-ed the socket for us — we just accept on it.
|
|
// Fall back to the traditional bind path when not socket-activated
|
|
// (direct invocation, dev, tests).
|
|
let listener = {
|
|
let mut listenfd = listenfd::ListenFd::from_env();
|
|
if let Some(std_listener) = listenfd
|
|
.take_unix_listener(0)
|
|
.context("take socket-activated unix listener")?
|
|
{
|
|
std_listener.set_nonblocking(true)?;
|
|
UnixListener::from_std(std_listener)
|
|
.context("convert socket-activated listener to tokio")?
|
|
} else {
|
|
// Standalone: create parent dir, remove any stale socket, bind.
|
|
if let Some(parent) = socket.parent() {
|
|
std::fs::create_dir_all(parent)
|
|
.with_context(|| format!("create socket parent {}", parent.display()))?;
|
|
}
|
|
if socket.exists() {
|
|
std::fs::remove_file(socket).context("remove stale socket")?;
|
|
}
|
|
UnixListener::bind(socket)
|
|
.with_context(|| format!("bind admin socket {}", socket.display()))?
|
|
}
|
|
};
|
|
tracing::info!(socket = %socket.display(), hyperhive_flake = %coord.hyperhive_flake, "hive-c0re admin listening");
|
|
|
|
loop {
|
|
let (stream, _) = listener.accept().await.context("accept connection")?;
|
|
let coord = coord.clone();
|
|
tokio::spawn(async move {
|
|
if let Err(e) = handle(stream, coord).await {
|
|
tracing::warn!(error = ?e, "connection failed");
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
async fn handle(stream: UnixStream, coord: Arc<Coordinator>) -> Result<()> {
|
|
handle_with(stream, coord, ADMIN_LINE_DEADLINES).await
|
|
}
|
|
|
|
/// A client may idle between requests on one connection, so only the time to
|
|
/// finish a started line is bounded.
|
|
const ADMIN_LINE_DEADLINES: LineDeadlines = LineDeadlines {
|
|
idle: None,
|
|
line: Some(REQUEST_LINE_DEADLINE),
|
|
};
|
|
|
|
async fn handle_with(
|
|
stream: UnixStream,
|
|
coord: Arc<Coordinator>,
|
|
deadlines: LineDeadlines,
|
|
) -> Result<()> {
|
|
let (read, mut write) = stream.into_split();
|
|
let mut reader = BufReader::new(read);
|
|
let mut line = Vec::new();
|
|
|
|
loop {
|
|
let outcome = read_request_line(&mut reader, &mut line, deadlines).await?;
|
|
if outcome == RequestLine::Eof {
|
|
return Ok(());
|
|
}
|
|
if let Some(message) = outcome.refusal() {
|
|
write_response(&mut write, &HostResponse::error(message)).await?;
|
|
return Ok(());
|
|
}
|
|
let req = match serde_json::from_slice::<HostRequest>(line.trim_ascii()) {
|
|
Ok(req) => req,
|
|
Err(e) => {
|
|
write_response(
|
|
&mut write,
|
|
&HostResponse::error(format!("parse error: {e}")),
|
|
)
|
|
.await?;
|
|
continue;
|
|
}
|
|
};
|
|
// The one verb that answers with more than one response — see its
|
|
// doc comment. Hands `write` over and stops reading further
|
|
// requests on this connection; a client wanting anything else
|
|
// opens a fresh one.
|
|
if matches!(req, HostRequest::SubscribeAgentStatus) {
|
|
return stream_agent_status(write, coord).await;
|
|
}
|
|
let resp = dispatch(&req, coord.clone()).await;
|
|
write_response(&mut write, &resp).await?;
|
|
}
|
|
}
|
|
|
|
/// Serialize one `HostResponse` as a JSON line and flush it. Shared by the
|
|
/// ordinary one-response-per-request path and `stream_agent_status`'s
|
|
/// multi-response one, so the two can't quietly drift on framing.
|
|
async fn write_response(
|
|
write: &mut tokio::net::unix::OwnedWriteHalf,
|
|
resp: &HostResponse,
|
|
) -> Result<()> {
|
|
let mut payload = serde_json::to_string(resp)?;
|
|
payload.push('\n');
|
|
write.write_all(payload.as_bytes()).await?;
|
|
write.flush().await?;
|
|
Ok(())
|
|
}
|
|
|
|
#[allow(
|
|
clippy::too_many_lines,
|
|
reason = "flat one-arm-per-HostRequest-variant router; each arm just \
|
|
delegates to a handler. Splitting the match would scatter the \
|
|
wire-command routing without shrinking it."
|
|
)]
|
|
async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
|
|
let result: anyhow::Result<HostResponse> = async {
|
|
Ok(match req {
|
|
HostRequest::Kill { name } => submit_single(&coord, name.as_str(), Verb::Kill).await,
|
|
HostRequest::Restart { name } => {
|
|
submit_single(&coord, name.as_str(), Verb::Restart).await
|
|
}
|
|
HostRequest::SetPaused { name, paused } => {
|
|
handle_set_paused(&coord, name, *paused).await
|
|
}
|
|
HostRequest::RestartScoped { scope, graceful } => {
|
|
handle_restart_scoped(&coord, scope, *graceful).await?
|
|
}
|
|
HostRequest::Stop { scope, graceful } => {
|
|
// Resolve the scope to explicit container names at the entry
|
|
// point, then operate on names — never pass the bare "all
|
|
// agents" flag deeper (it'd force every consumer, incl. the
|
|
// graceful-stop queue, to re-expand it).
|
|
let agents = scoped_agents(scope).await?;
|
|
let infra = scoped_infra(scope);
|
|
// On a broad stop, remember which agents were actually
|
|
// running so a later broad `start` restores only those
|
|
// (not every configured container). A targeted `--agent`
|
|
// stop must not redefine the restore set.
|
|
if is_broad_scope(scope) {
|
|
let mut running = Vec::new();
|
|
for a in &agents {
|
|
if lifecycle::is_running(a).await {
|
|
running.push(a.clone());
|
|
}
|
|
}
|
|
coord.set_last_stopped_running(running);
|
|
}
|
|
handle_stop(&coord, &agents, &infra, *graceful).await?
|
|
}
|
|
HostRequest::Start { scope } => {
|
|
let mut agents = scoped_agents(scope).await?;
|
|
// A broad start restores only the set recorded at the
|
|
// last broad stop, if any. No record (cold "bring the
|
|
// hive up", or a daemon restart since the stop) → start
|
|
// all. Targeted `--agent` start is never filtered.
|
|
if is_broad_scope(scope)
|
|
&& let Some(prev) = coord.take_last_stopped_running()
|
|
{
|
|
agents.retain(|a| prev.contains(a));
|
|
}
|
|
let infra = scoped_infra(scope);
|
|
handle_start(&coord, &agents, &infra).await?
|
|
}
|
|
HostRequest::Destroy { name, purge } => {
|
|
actions::destroy(&coord, name.as_str(), *purge);
|
|
HostResponse::success()
|
|
}
|
|
HostRequest::Rebuild { name } => {
|
|
submit_single(&coord, name.as_str(), Verb::Rebuild).await
|
|
}
|
|
HostRequest::QueueNodes { ids } => {
|
|
HostResponse::nodes(coord.job_queue.node_subtrees(ids))
|
|
}
|
|
HostRequest::List => HostResponse::list(lifecycle::list().await?),
|
|
// The agents root is ours and not world-traversable, so this
|
|
// question is only answerable on this side of the socket —
|
|
// see the request's doc comment for why the client asks.
|
|
HostRequest::AgentExists { name } => HostResponse::agent_exists(agent_exists(name)?),
|
|
HostRequest::AgentStatus => handle_agent_status().await,
|
|
// Intercepted in `handle()` before a request ever reaches
|
|
// `dispatch` — see `stream_agent_status`'s doc comment. This
|
|
// arm exists only so the match stays exhaustive; reaching it
|
|
// would mean a future caller invoked `dispatch` directly.
|
|
HostRequest::SubscribeAgentStatus => {
|
|
HostResponse::error("SubscribeAgentStatus is handled by the connection loop")
|
|
}
|
|
// The hive domain + per-surface public URLs are injected into
|
|
// c0re's service env by the hyperhive module; surface them so the
|
|
// operator CLI can fill in this hive's own identity (the
|
|
// federation peer-config block) and open the web surfaces
|
|
// (`hivectl open`).
|
|
HostRequest::Urls => HostResponse::urls(hive_urls()),
|
|
HostRequest::MatrixInvite { user, room } => {
|
|
handle_matrix_invite(user, room.as_deref()).await?
|
|
}
|
|
HostRequest::ReconcileConfigStatus { agent, verbose } => {
|
|
crate::forge::reconcile_config_status(agent.as_str(), *verbose).await?
|
|
}
|
|
HostRequest::ReconcileConfigApply { agent, direction } => {
|
|
crate::forge::reconcile_config_apply(agent.as_str(), *direction).await?
|
|
}
|
|
HostRequest::QuotaEnable => handle_quota_enable().await?,
|
|
HostRequest::QuotaLimit { name, limit } => {
|
|
handle_quota_limit(name.as_str(), *limit).await?
|
|
}
|
|
HostRequest::QuotaShow { name } => {
|
|
handle_quota_show(name.as_ref().map(hive_types::Ident::as_str)).await?
|
|
}
|
|
HostRequest::UpgradeSubvolume { name } => {
|
|
handle_upgrade_subvolume(name.as_str()).await?
|
|
}
|
|
HostRequest::SnapshotSubvolume { name, label } => {
|
|
handle_snapshot_subvolume(name.as_str(), label).await?
|
|
}
|
|
HostRequest::DeleteSnapshot { name, label } => {
|
|
handle_delete_snapshot(name.as_str(), label).await?
|
|
}
|
|
HostRequest::SendSnapshot {
|
|
name,
|
|
label,
|
|
parent,
|
|
dest,
|
|
} => handle_send_snapshot(name.as_str(), label, parent.as_deref(), dest).await?,
|
|
HostRequest::PushSnapshot {
|
|
name,
|
|
label,
|
|
parent,
|
|
} => handle_push_snapshot(name.as_str(), label, parent.as_deref()).await?,
|
|
})
|
|
}
|
|
.await;
|
|
match result {
|
|
Ok(r) => r,
|
|
Err(e) => HostResponse::error(format!("{e:#}")),
|
|
}
|
|
}
|
|
|
|
/// `hivectl pause|resume` / the dashboard toggle: write or remove the
|
|
/// agent's pause marker.
|
|
///
|
|
/// **Resume** stays synchronous, direct marker removal — there's
|
|
/// nothing to acknowledge (a resumed agent just starts driving turns
|
|
/// again on its own next poll, no handshake needed), and nobody has
|
|
/// asked resume to wait.
|
|
///
|
|
/// **Pause** rides the job queue (`PauseSignal → PauseDrain`, see
|
|
/// `job_queue::power::pause_many`) instead of writing the marker
|
|
/// synchronously here: pausing wants the same "confirmed, not just
|
|
/// requested" signal a graceful stop gets from its `Signal → Drain`
|
|
/// pair, and the DAG's agent lease is what stops a pause from racing
|
|
/// an in-flight rebuild/stop of the same agent — a synchronous write
|
|
/// had no such protection. Still works on a stopped agent (the queued
|
|
/// `PauseSignal` writes the same sticky marker either way, container
|
|
/// running or not).
|
|
async fn handle_set_paused(
|
|
coord: &std::sync::Arc<Coordinator>,
|
|
name: &hive_types::Ident,
|
|
paused: bool,
|
|
) -> HostResponse {
|
|
if !paused {
|
|
if let Err(e) = Coordinator::set_paused(name, false).await {
|
|
return HostResponse::error(format!("set paused=false for {name}: {e}"));
|
|
}
|
|
tracing::info!(%name, "agent pause marker cleared");
|
|
// Refresh the dashboard's view so the paused badge flips without
|
|
// waiting for the next periodic rescan.
|
|
coord.rescan_containers_and_emit().await;
|
|
return HostResponse::success();
|
|
}
|
|
match crate::job_queue::power::pause_many(coord, std::slice::from_ref(&name.to_string())) {
|
|
Ok(ids) => {
|
|
tracing::info!(%name, "agent pause queued");
|
|
HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect())
|
|
}
|
|
Err(e) => HostResponse::error(format!("queue pause for {name}: {e}")),
|
|
}
|
|
}
|
|
|
|
/// Collect per-agent status rows for `hivectl status` and the dashboard.
|
|
async fn handle_agent_status() -> HostResponse {
|
|
match crate::container_view::build_all().await {
|
|
Ok(views) => HostResponse::agent_statuses(
|
|
views
|
|
.into_iter()
|
|
.map(hive_sh4re::container::AgentStatusRow::from)
|
|
.collect(),
|
|
),
|
|
Err(e) => HostResponse::error(format!("listing containers: {e:#}")),
|
|
}
|
|
}
|
|
|
|
/// `SubscribeAgentStatus` — ack once, then push one single-row
|
|
/// [`HostResponse::agent_statuses`] per agent-status change for as long as
|
|
/// the client stays connected. Consumes `write` (rather than borrowing it,
|
|
/// like every other handler here) because this is the one verb that keeps
|
|
/// writing after its own response, so `handle()` hands over ownership and
|
|
/// stops reading further requests on this connection once it calls this.
|
|
///
|
|
/// Rides the same `Coordinator.dashboard_events` broadcast channel the
|
|
/// dashboard's own SSE route reads (`dashboard/state_snapshot.rs`) —
|
|
/// `SetStatus` already triggers a `rescan_containers_and_emit` on every
|
|
/// status change (`socket_server/mod.rs::handle_set_status`), so that
|
|
/// channel already carries every event this needs; no separate plumbing.
|
|
async fn stream_agent_status(
|
|
mut write: tokio::net::unix::OwnedWriteHalf,
|
|
coord: Arc<Coordinator>,
|
|
) -> Result<()> {
|
|
let mut events = coord.dashboard_subscribe();
|
|
write_response(&mut write, &HostResponse::success()).await?;
|
|
loop {
|
|
match events.recv().await {
|
|
Ok(crate::dashboard_events::DashboardEvent::ContainerStateChanged {
|
|
container,
|
|
..
|
|
}) => {
|
|
let row = hive_sh4re::container::AgentStatusRow::from(container);
|
|
write_response(&mut write, &HostResponse::agent_statuses(vec![row])).await?;
|
|
}
|
|
// Every other event kind on this channel is dashboard-only
|
|
// (broker traffic, the queue, …) — nothing this
|
|
// subscriber asked for.
|
|
Ok(_) => {}
|
|
// Best-effort, same contract the dashboard's own live channel
|
|
// already has (see `HostRequest::SubscribeAgentStatus`'s doc
|
|
// comment) — a slow reader drops updates rather than stalling
|
|
// the broadcaster for everyone else on the channel.
|
|
Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
|
|
tracing::warn!(skipped, "agent-status subscribe: receiver lagged");
|
|
}
|
|
Err(tokio::sync::broadcast::error::RecvError::Closed) => return Ok(()),
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Matrix provisioning handlers
|
|
//
|
|
// The `hivectl matrix` subcommands used to run these in-process, which forced
|
|
// the standalone CLI to link the whole daemon crate (matrix-sdk, reqwest, …).
|
|
// They now run daemon-side over the host socket, where the daemon already holds
|
|
// the sender token. Each op returns the operator-facing lines hivectl used to
|
|
// `println!` in `HostResponse::messages` for the client to print verbatim.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Shared reqwest client for the matrix admin HTTP calls (30s timeout,
|
|
/// mirroring the old in-CLI client).
|
|
fn matrix_http_client() -> Result<reqwest::Client> {
|
|
reqwest::Client::builder()
|
|
.timeout(std::time::Duration::from_secs(30))
|
|
.build()
|
|
.context("build reqwest client")
|
|
}
|
|
|
|
/// True when `name` has a state dir under the agents root, i.e. it's a
|
|
/// managed agent rather than a bare (operator/human) account.
|
|
///
|
|
/// Used by the provisioning handlers below to tell an agent account from
|
|
/// a human one, and exposed over the socket as
|
|
/// [`HostRequest::AgentExists`] for clients that can't read the agents
|
|
/// root themselves (it's `0700` and owned by the daemon's user).
|
|
fn agent_exists(name: &hive_types::Ident) -> Result<bool> {
|
|
crate::paths::agent_state_dir(name)
|
|
.try_exists()
|
|
.with_context(|| format!("check agent state dir for {name}"))
|
|
}
|
|
|
|
/// Guard: matrix provisioning needs a homeserver to provision against.
|
|
///
|
|
/// Answers "is one configured", not "is one running here": the address is
|
|
/// the swarm's homeserver, wherever it runs.
|
|
fn require_matrix_present() -> Result<()> {
|
|
if crate::matrix::is_present() {
|
|
return Ok(());
|
|
}
|
|
anyhow::bail!(
|
|
"no matrix homeserver configured — set services.hyperhive.swarm.matrix.apiUrl (it defaults \
|
|
to the swarm.matrix.gatewayHost vhost, and is null when that is) before provisioning \
|
|
matrix users"
|
|
)
|
|
}
|
|
|
|
async fn handle_quota_enable() -> Result<HostResponse> {
|
|
crate::priv_client::ensure_btrfs_quota()
|
|
.await
|
|
.context("enable btrfs qgroup accounting")?;
|
|
Ok(HostResponse::messages(vec![
|
|
"btrfs qgroup accounting enabled on the agent-state filesystem.".to_owned(),
|
|
"(usage may read 0 until btrfs finishes its background rescan)".to_owned(),
|
|
]))
|
|
}
|
|
|
|
async fn handle_quota_limit(name: &str, limit: Option<u64>) -> Result<HostResponse> {
|
|
crate::priv_client::set_subvolume_quota(name, limit)
|
|
.await
|
|
.with_context(|| format!("set quota for {name}"))?;
|
|
// Bare success: the client prints the human-readable confirmation from
|
|
// the value it sent (it holds the `human_bytes` formatter).
|
|
Ok(HostResponse::success())
|
|
}
|
|
|
|
async fn handle_quota_show(name: Option<&str>) -> Result<HostResponse> {
|
|
let agents: Vec<String> = match name {
|
|
Some(n) => vec![n.to_owned()],
|
|
None => Coordinator::kept_state_names()
|
|
.into_iter()
|
|
.map(hive_types::Ident::into_string)
|
|
.collect(),
|
|
};
|
|
let mut rows = Vec::with_capacity(agents.len());
|
|
for agent in &agents {
|
|
match crate::priv_client::read_subvolume_usage(agent).await {
|
|
Ok((referenced, exclusive)) => rows.push(hive_host_sock::QuotaRow {
|
|
agent: agent.clone(),
|
|
referenced: Some(referenced),
|
|
exclusive: Some(exclusive),
|
|
note: None,
|
|
}),
|
|
Err(e) => {
|
|
let msg = format!("{e:#}");
|
|
// btrfs-progs prints "ERROR: ... quota not enabled" to stderr
|
|
// when qgroups are off; short-circuit the whole sweep with the
|
|
// enable hint (case-insensitive fragment match — the wording
|
|
// varies across btrfs-progs versions).
|
|
if msg.to_ascii_lowercase().contains("quota not enabled") {
|
|
return Ok(HostResponse::error(
|
|
"btrfs quota not enabled — run `hivectl quota-enable` first",
|
|
));
|
|
}
|
|
// A plain-dir agent (no subvolume) has no qgroup; note it
|
|
// inline and keep going rather than aborting the whole sweep.
|
|
rows.push(hive_host_sock::QuotaRow {
|
|
agent: agent.clone(),
|
|
referenced: None,
|
|
exclusive: None,
|
|
note: Some(format!("no qgroup data — plain dir or: {msg}")),
|
|
});
|
|
}
|
|
}
|
|
}
|
|
Ok(HostResponse::quota(rows))
|
|
}
|
|
|
|
async fn handle_upgrade_subvolume(name: &str) -> Result<HostResponse> {
|
|
crate::priv_client::upgrade_agent_subvolume(name)
|
|
.await
|
|
.with_context(|| format!("upgrade {name} state subvolume"))?;
|
|
Ok(HostResponse::success())
|
|
}
|
|
|
|
async fn handle_snapshot_subvolume(name: &str, label: &str) -> Result<HostResponse> {
|
|
let path = crate::priv_client::snapshot_agent_subvolume(name, label)
|
|
.await
|
|
.with_context(|| format!("snapshot {name} state subvolume (label {label:?})"))?;
|
|
Ok(HostResponse::messages(vec![path]))
|
|
}
|
|
|
|
async fn handle_delete_snapshot(name: &str, label: &str) -> Result<HostResponse> {
|
|
crate::priv_client::delete_agent_snapshot(name, label)
|
|
.await
|
|
.with_context(|| format!("delete {name} snapshot (label {label:?})"))?;
|
|
Ok(HostResponse::success())
|
|
}
|
|
|
|
async fn handle_send_snapshot(
|
|
name: &str,
|
|
label: &str,
|
|
parent: Option<&str>,
|
|
dest: &str,
|
|
) -> Result<HostResponse> {
|
|
let path = crate::priv_client::send_agent_snapshot_to_file(name, label, parent, dest)
|
|
.await
|
|
.with_context(|| format!("send {name} snapshot (label {label:?}) to file {dest:?}"))?;
|
|
Ok(HostResponse::messages(vec![path]))
|
|
}
|
|
|
|
/// Push a snapshot to the swarm's store. The network sibling of
|
|
/// [`handle_send_snapshot`]: nothing lands on this host, so success is
|
|
/// bare rather than a path.
|
|
async fn handle_push_snapshot(
|
|
name: &str,
|
|
label: &str,
|
|
parent: Option<&str>,
|
|
) -> Result<HostResponse> {
|
|
crate::snapshot_push::push_agent_snapshot(name, label, parent)
|
|
.await
|
|
.with_context(|| format!("push {name} snapshot (label {label:?}) to the swarm store"))?;
|
|
Ok(HostResponse::success())
|
|
}
|
|
|
|
async fn handle_matrix_invite(user: &str, room: Option<&str>) -> Result<HostResponse> {
|
|
require_matrix_present()?;
|
|
let sender_token = crate::matrix::read_sender_token()?;
|
|
let client = matrix_http_client()?;
|
|
let server_name = crate::matrix::discover_server_name(&client)
|
|
.await
|
|
.context("discover matrix server_name")?;
|
|
let room_id = crate::matrix::invite_user(&client, &sender_token, user, room, &server_name)
|
|
.await
|
|
.with_context(|| format!("matrix invite {user}"))?;
|
|
let target = if user.starts_with('@') {
|
|
user.to_owned()
|
|
} else {
|
|
format!("@{user}:{server_name}")
|
|
};
|
|
Ok(HostResponse::messages(vec![format!(
|
|
"matrix: invited {target} to {room_id}"
|
|
)]))
|
|
}
|
|
|
|
/// Single-agent queue verbs the admin socket exposes. Each submits the
|
|
/// matching DAG (persisting the `wanted` intent, serializing on the
|
|
/// agent's lease, with the transient/crash-watch suppression the old
|
|
/// direct lifecycle calls lacked) and returns the DAG id for the
|
|
/// client's wait loop.
|
|
#[derive(Clone, Copy)]
|
|
enum Verb {
|
|
/// Stop DAG (`wanted = Offline`; Reconcile kills + unregisters +
|
|
/// fires `Killed`).
|
|
Kill,
|
|
/// Restart DAG (`wanted = Up`; mechanical stop + reconcile-start).
|
|
Restart,
|
|
/// Rebuild DAG — the Swap tail owns the manager `Rebuilt` events +
|
|
/// kick, so the CLI path can't drift from the dashboard's.
|
|
Rebuild,
|
|
}
|
|
|
|
async fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostResponse {
|
|
use crate::job_queue::power;
|
|
// A single-target op is the N-target one with N = 1 — the shapes are
|
|
// identical, so there is no separate builder to keep in sync.
|
|
let targets = [name.to_owned()];
|
|
let ids = match verb {
|
|
Verb::Kill => {
|
|
tracing::info!(%name, "kill");
|
|
power::stop_many(coord, &targets, false).await
|
|
}
|
|
Verb::Restart => {
|
|
tracing::info!(%name, "restart");
|
|
power::restart_many(coord, &targets, false).await
|
|
}
|
|
Verb::Rebuild => {
|
|
tracing::info!(%name, "rebuild");
|
|
// Not a power op: a rebuild's shape doesn't depend on live state,
|
|
// so it is a plain template insert rather than a `*_many` gather.
|
|
let inserted = coord
|
|
.job_queue
|
|
.insert_job(|b| crate::job_queue::templates::rebuild(b, name, true));
|
|
if inserted.is_ok() {
|
|
coord.emit_rebuild_queue_snapshot();
|
|
}
|
|
inserted
|
|
}
|
|
};
|
|
match ids {
|
|
Ok(ids) => HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect()),
|
|
Err(e) => HostResponse::error(format!("queue insert failed: {e}")),
|
|
}
|
|
}
|
|
|
|
/// Stop the given `agents` (resolved logical names) then `infra` services
|
|
/// (`hivectl stop`). Agents go down before infra so they're not mid-request
|
|
/// through a gateway that's already gone. Per-target failures are
|
|
/// aggregated rather than aborting on the first error, mirroring
|
|
/// `finish_lifecycle` below. Callers resolve the [`LifecycleScope`] to these
|
|
/// explicit name lists up front — this never sees the "all" flag.
|
|
///
|
|
/// Every agent rides the job queue: a `graceful` stop submits the
|
|
/// quiesce DAG (signal → drain → reconcile-stop; all drains overlap),
|
|
/// a hard stop a plain stop DAG — both persist `wanted = Offline` and
|
|
/// serialize on the agent's lease so nothing races an in-flight
|
|
/// rebuild. The response carries the DAG ids so `hivectl` can wait
|
|
/// with per-node progress. Infra services have no harness / lease
|
|
/// and stay direct + synchronous.
|
|
async fn handle_stop(
|
|
coord: &Arc<Coordinator>,
|
|
agents: &[String],
|
|
infra: &[InfraContainer],
|
|
graceful: bool,
|
|
) -> Result<HostResponse> {
|
|
tracing::info!(?agents, ?infra, graceful, "stop");
|
|
let mut ok_items: Vec<String> = Vec::new();
|
|
let mut errors: Vec<String> = Vec::new();
|
|
let mut queued: Vec<u64> = Vec::new();
|
|
|
|
// One insert for all targeted agents — a per-agent stop subgraph each
|
|
// (`SetWanted(Offline) → [Signal → Drain →] Reconcile`), independent
|
|
// roots that run concurrently on their own leases.
|
|
if !agents.is_empty() {
|
|
match crate::job_queue::power::stop_many(coord, agents, graceful).await {
|
|
Ok(ids) => {
|
|
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
|
|
ok_items.extend(agents.iter().cloned());
|
|
}
|
|
Err(e) => errors.push(format!("queue stop: {e}")),
|
|
}
|
|
}
|
|
|
|
// Agents go down before infra so they're not mid-request through a
|
|
// gateway that's already gone. Hard stops are quick kills —
|
|
// await their DAGs (bounded) before touching infra. Graceful stops
|
|
// keep the immediate return (drains take minutes and the
|
|
// agents-then-infra race pre-existed there).
|
|
if !graceful && !infra.is_empty() {
|
|
await_dags(coord, &queued, std::time::Duration::from_mins(2)).await;
|
|
}
|
|
for &container in infra {
|
|
let name = container.name();
|
|
match crate::priv_client::control_infra_container(container, InfraAction::Stop).await {
|
|
Ok(()) => ok_items.push(name.to_owned()),
|
|
Err(e) => {
|
|
tracing::warn!(%name, error = ?e, "stop: infra stop failed");
|
|
errors.push(format!("{name}: {e:#}"));
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut resp = finish_lifecycle(ok_items, &errors);
|
|
resp.queued_dags = Some(queued);
|
|
Ok(resp)
|
|
}
|
|
|
|
/// Best-effort server-side wait for a set of DAGs to settle terminal,
|
|
/// bounded by `timeout` — used to preserve ordering invariants inside
|
|
/// one request (agent stops before infra stops) without trusting the
|
|
/// client to wait.
|
|
async fn await_dags(coord: &Arc<Coordinator>, ids: &[u64], timeout: std::time::Duration) {
|
|
let deadline = std::time::Instant::now() + timeout;
|
|
loop {
|
|
// Every node under the named roots, terminal ones included — unlike the
|
|
// old typed snapshot, a settled group does not drop out of this view.
|
|
// So "pending" is simply "some node hasn't finished", with no second
|
|
// rule for the disappeared case.
|
|
let pending = coord
|
|
.job_queue
|
|
.node_subtrees(ids)
|
|
.iter()
|
|
.any(|n| !n.state.is_terminal());
|
|
if !pending {
|
|
return;
|
|
}
|
|
if std::time::Instant::now() >= deadline {
|
|
tracing::warn!(?ids, "await_dags: timed out; proceeding");
|
|
return;
|
|
}
|
|
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
|
|
}
|
|
}
|
|
|
|
/// Start the given `infra` services then `agents` (`hivectl start`) — the
|
|
/// inverse of [`handle_stop`]. Infra comes up before agents so the agents
|
|
/// find the gateway ready. Per-target failures aggregated. Callers
|
|
/// resolve the [`LifecycleScope`] to these explicit name lists up front.
|
|
async fn handle_start(
|
|
coord: &Arc<Coordinator>,
|
|
agents: &[String],
|
|
infra: &[InfraContainer],
|
|
) -> Result<HostResponse> {
|
|
tracing::info!(?agents, ?infra, "start");
|
|
let mut ok_items: Vec<String> = Vec::new();
|
|
let mut errors: Vec<String> = Vec::new();
|
|
|
|
for &container in infra {
|
|
let name = container.name();
|
|
match crate::priv_client::control_infra_container(container, InfraAction::Start).await {
|
|
Ok(()) => ok_items.push(name.to_owned()),
|
|
Err(e) => {
|
|
tracing::warn!(%name, error = ?e, "start: infra start failed");
|
|
errors.push(format!("{name}: {e:#}"));
|
|
}
|
|
}
|
|
}
|
|
|
|
// One DAG for all targeted agents — a per-agent start subgraph each
|
|
// (`SetWanted(Up) → Reconcile`, or a rebuild-then-start for a stale
|
|
// rev), independent roots that run concurrently on their own leases. A
|
|
// hive-wide `hivectl start` is now a single DAG, not N. Through the
|
|
// queue: persists `wanted = Up`, per-agent stale-rev upgrade to a full
|
|
// rebuild, serializes on each agent's lease. The id rides back for
|
|
// hivectl's wait loop.
|
|
let mut queued: Vec<u64> = Vec::new();
|
|
if !agents.is_empty() {
|
|
match crate::job_queue::power::start_many(coord, agents).await {
|
|
Ok(ids) => {
|
|
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
|
|
ok_items.extend(agents.iter().cloned());
|
|
}
|
|
Err(e) => errors.push(format!("queue start: {e}")),
|
|
}
|
|
}
|
|
|
|
let mut resp = finish_lifecycle(ok_items, &errors);
|
|
resp.queued_dags = Some(queued);
|
|
Ok(resp)
|
|
}
|
|
|
|
/// Restart containers hive-wide (`hivectl restart`) — the DAG-based
|
|
/// sibling of [`handle_stop`]/[`handle_start`]. All targeted agents ride
|
|
/// **one** DAG (a per-agent restart subgraph each: `SetWanted → [Signal →
|
|
/// Drain →] StopForUpdate → Reconcile`, independent roots that run
|
|
/// concurrently on their own leases), not N separate DAGs — a hive-wide
|
|
/// restart is one job. `graceful` prepends signal→drain per agent. No
|
|
/// client-side stop-then-start composition, so a dropped `hivectl`
|
|
/// connection never strands an agent. Infra services have no lease/DAG
|
|
/// and restart synchronously (stop then start).
|
|
async fn handle_restart_scoped(
|
|
coord: &Arc<Coordinator>,
|
|
scope: &LifecycleScope,
|
|
graceful: bool,
|
|
) -> Result<HostResponse> {
|
|
tracing::info!(?scope, graceful, "restart");
|
|
let agents = scoped_agents(scope).await?;
|
|
let infra = scoped_infra(scope);
|
|
let mut ok_items: Vec<String> = Vec::new();
|
|
let mut errors: Vec<String> = Vec::new();
|
|
let mut queued: Vec<u64> = Vec::new();
|
|
|
|
// One insert for all targeted agents — a per-agent restart subgraph each
|
|
// (`[Signal → Drain →] StopForUpdate → Reconcile`; no `SetWanted`, since a
|
|
// restart converges to the agent's *existing* intent rather than rewriting
|
|
// it), independent roots running concurrently on their own leases. No
|
|
// client-side stop-then-start composition — the whole restart survives a
|
|
// dropped connection because the graph owns it.
|
|
if !agents.is_empty() {
|
|
match crate::job_queue::power::restart_many(coord, &agents, graceful).await {
|
|
Ok(ids) => {
|
|
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
|
|
ok_items.extend(agents.iter().cloned());
|
|
}
|
|
Err(e) => errors.push(format!("queue restart: {e}")),
|
|
}
|
|
}
|
|
|
|
for &container in &infra {
|
|
let name = container.name();
|
|
let res = async {
|
|
crate::priv_client::control_infra_container(container, InfraAction::Stop).await?;
|
|
crate::priv_client::control_infra_container(container, InfraAction::Start).await
|
|
}
|
|
.await;
|
|
match res {
|
|
Ok(()) => ok_items.push(name.to_owned()),
|
|
Err(e) => {
|
|
tracing::warn!(%name, error = ?e, "restart: infra restart failed");
|
|
errors.push(format!("{name}: {e:#}"));
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut resp = finish_lifecycle(ok_items, &errors);
|
|
resp.queued_dags = Some(queued);
|
|
Ok(resp)
|
|
}
|
|
|
|
/// Resolve which sub-agent logical names a scope targets: every live
|
|
/// container (from `lifecycle::list`) when `agents` is set or the scope is
|
|
/// "everything", plus any explicit `agent_names`. Returns de-duplicated
|
|
/// logical names with the `h-` container prefix stripped.
|
|
/// A scope that targets *every* agent rather than an explicit
|
|
/// `--agent <name>` list: either the `agents` flag or a bare
|
|
/// "everything" scope. Broad scopes are the ones whose stop/start pair
|
|
/// drives the previously-running restore set (see `handle` Stop/Start
|
|
/// arms); a targeted `--agent` stop/start must not redefine it.
|
|
fn is_broad_scope(scope: &LifecycleScope) -> bool {
|
|
scope.agents || scope.is_everything()
|
|
}
|
|
|
|
/// Assemble this hive's domain + browser-facing web URLs from c0re's
|
|
/// service env (injected by the hyperhive NixOS module). Each field is `None` when its
|
|
/// surface isn't browser-reachable (domain unset, forge `publicUrl`
|
|
/// null, matrix `gatewayHost` null), so the CLI can hint precisely instead of
|
|
/// opening a dead link. Scheme matches the existing `HIVE_FORGE_PUBLIC_URL`
|
|
/// convention (gateway terminates TLS, so https).
|
|
fn hive_urls() -> hive_host_sock::HiveUrls {
|
|
// Treat an empty env value as unset everywhere — an empty domain would
|
|
// otherwise render `swarm.hives."" = …` (invalid nix) and `https:///`.
|
|
let env = |k: &str| std::env::var(k).ok().filter(|v| !v.is_empty());
|
|
let domain = env("HYPERHIVE_HIVE_DOMAIN");
|
|
hive_host_sock::HiveUrls {
|
|
home: env("HYPERHIVE_DASHBOARD_DOMAIN").map(|d| format!("https://{d}/")),
|
|
forge: env("HIVE_FORGE_PUBLIC_URL"),
|
|
matrix: env("HIVE_MATRIX_PUBLIC_URL"),
|
|
domain,
|
|
}
|
|
}
|
|
|
|
async fn scoped_agents(scope: &LifecycleScope) -> Result<Vec<String>> {
|
|
use std::collections::BTreeSet;
|
|
let mut set: BTreeSet<String> = BTreeSet::new();
|
|
if is_broad_scope(scope) {
|
|
for c in lifecycle::list().await? {
|
|
let logical = c
|
|
.strip_prefix(lifecycle::AGENT_PREFIX)
|
|
.unwrap_or(&c)
|
|
.to_owned();
|
|
set.insert(logical);
|
|
}
|
|
}
|
|
for n in &scope.agent_names {
|
|
set.insert(n.clone());
|
|
}
|
|
Ok(set.into_iter().collect())
|
|
}
|
|
|
|
/// Resolve which infra services a scope targets: the gateway, when the
|
|
/// scope is "everything" (no flags set) or names it. The swarm's forge and
|
|
/// matrix are never a target.
|
|
fn scoped_infra(scope: &LifecycleScope) -> Vec<InfraContainer> {
|
|
if scope.is_everything() || scope.gateway {
|
|
vec![InfraContainer::Gateway]
|
|
} else {
|
|
Vec::new()
|
|
}
|
|
}
|
|
|
|
/// Build the aggregated lifecycle response: `ok` with the touched names when
|
|
/// every target succeeded, otherwise `ok: false` with the joined errors and
|
|
/// the partial success list.
|
|
fn finish_lifecycle(ok_items: Vec<String>, errors: &[String]) -> HostResponse {
|
|
if errors.is_empty() {
|
|
HostResponse::list(ok_items)
|
|
} else {
|
|
HostResponse {
|
|
ok: false,
|
|
error: Some(errors.join("; ")),
|
|
agents: Some(ok_items),
|
|
..HostResponse::default()
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use std::time::Duration;
|
|
|
|
use hive_sock_client::REQUEST_LINE_MAX_BYTES;
|
|
use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
|
|
|
|
use super::*;
|
|
|
|
/// A real `Coordinator` over a throwaway sqlite dir; `handle_with` takes
|
|
/// an `Arc<Coordinator>` and `open` touches nothing outside the db path.
|
|
fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
|
|
let dir = tempfile::tempdir().expect("tempdir");
|
|
let coord = Coordinator::open(
|
|
&dir.path().join("broker.sqlite"),
|
|
crate::coordinator::HiveEnv::default(),
|
|
crate::hive_stats::PriceTable::new(),
|
|
1,
|
|
)
|
|
.expect("open coordinator");
|
|
(dir, Arc::new(coord))
|
|
}
|
|
|
|
async fn reply_error(client: &mut tokio::net::unix::OwnedReadHalf) -> String {
|
|
let mut reply = String::new();
|
|
BufReader::new(client)
|
|
.read_line(&mut reply)
|
|
.await
|
|
.expect("reply");
|
|
let resp: HostResponse = serde_json::from_str(&reply).expect("response json");
|
|
resp.error.expect("error reply")
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn over_long_request_gets_an_error_and_the_connection_closes() {
|
|
let (_dir, coord) = coordinator();
|
|
let (client, server) = UnixStream::pair().expect("socketpair");
|
|
let handler = tokio::spawn(handle_with(server, coord, ADMIN_LINE_DEADLINES));
|
|
let (mut 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 message = reply_error(&mut read).await;
|
|
assert!(message.starts_with("request too long"), "{message}");
|
|
handler.await.expect("join").expect("clean close");
|
|
writer.await.expect("join");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn stalled_request_gets_an_error_and_the_connection_closes() {
|
|
let (_dir, coord) = coordinator();
|
|
let (client, server) = UnixStream::pair().expect("socketpair");
|
|
let deadlines = LineDeadlines {
|
|
idle: None,
|
|
line: Some(Duration::from_millis(50)),
|
|
};
|
|
let handler = tokio::spawn(handle_with(server, coord, deadlines));
|
|
let (mut read, mut write) = client.into_split();
|
|
write.write_all(b"{\"partial\"").await.expect("write");
|
|
let message = reply_error(&mut read).await;
|
|
assert!(message.starts_with("request timed out"), "{message}");
|
|
handler.await.expect("join").expect("clean close");
|
|
}
|
|
|
|
/// A bare `hivectl stop` / `start` / `restart` touches the gateway and no
|
|
/// swarm service. Control: an agents-only scope touches no infra.
|
|
#[test]
|
|
fn everything_scope_selects_only_the_gateway() {
|
|
assert_eq!(
|
|
scoped_infra(&LifecycleScope::default()),
|
|
[InfraContainer::Gateway]
|
|
);
|
|
let gateway = LifecycleScope {
|
|
gateway: true,
|
|
..LifecycleScope::default()
|
|
};
|
|
assert_eq!(scoped_infra(&gateway), [InfraContainer::Gateway]);
|
|
let agents = LifecycleScope {
|
|
agents: true,
|
|
..LifecycleScope::default()
|
|
};
|
|
assert!(scoped_infra(&agents).is_empty());
|
|
}
|
|
|
|
/// A stop from an older `hivectl stop --forge` is refused rather than
|
|
/// read as an all-false scope, which would stop everything. Control: the
|
|
/// same request with `gateway` decodes.
|
|
#[test]
|
|
fn stop_naming_forge_or_matrix_is_refused() {
|
|
let req = |field: &str| {
|
|
format!(r#"{{"cmd":"stop","scope":{{"{field}":true}},"graceful":false}}"#)
|
|
};
|
|
for field in ["forge", "matrix"] {
|
|
let err = serde_json::from_str::<HostRequest>(&req(field)).unwrap_err();
|
|
assert!(err.to_string().contains("unknown field"), "{field}: {err}");
|
|
}
|
|
assert!(matches!(
|
|
serde_json::from_str::<HostRequest>(&req("gateway")),
|
|
Ok(HostRequest::Stop { scope, .. }) if scope.gateway && !scope.is_everything()
|
|
));
|
|
}
|
|
}
|