//! Swarm-level controller daemon. Runs as the unprivileged //! `swarm-controller` user on whichever host the operator flips //! `services.hyperhive.swarm.controller.enable` on, and serves HTTP over a //! unix socket that the hive-gateway's nginx proxies to. //! //! Holds one piece of read-only state: the swarm's hive directory, loaded //! once at startup from an env var the NixOS module sets //! (`services.hyperhive.swarm.controller`) — see `load_hives`. Still no //! persistence and no writes; a config change means a redeploy, same as //! every other option this process reads. //! //! Distinct from `hive-c0re`, which is per-hive: c0re owns the agents on //! one host, this owns what is true across hives. //! //! `OpenAPI` spec generation mirrors `hive-c0re/src/dashboard/mod.rs` //! exactly: `#[utoipa::path(...)]` per handler, an `ApiDoc` root, and a //! raw JSON route at `/api/openapi.json`. Swagger UI itself is //! nginx-hosted from the nix store, same shape as the per-hive //! dashboard's (`nix/host-modules/hive-gateway/vhosts.nix`'s //! `swarmUiVhost` — a swagger-ui-theme dist under `/api/docs/`, no //! fallback to this daemon). Only annotated routes appear in the spec; //! an unannotated one just doesn't show up, nothing breaks. use std::os::unix::fs::PermissionsExt as _; use std::path::PathBuf; use std::sync::{Arc, Mutex}; use anyhow::{Context, Result}; use axum::{Json, extract::State, routing::get}; use hive_jobq_wire::GraphWire as _; use serde::{Deserialize, Serialize}; use utoipa::{OpenApi, ToSchema}; use utoipa_axum::{router::OpenApiRouter, routes}; mod status; /// Placeholder node payload for the swarm-level job graph — uninhabited on /// purpose, and named `Swarm*` rather than the bare `NodeKind`/`Resource` /// `hive-c0re::job_queue::model` already uses, so a grep for either doesn't /// land on both crates. The *scheduler loop* below is real and running /// (`spawn_jobq_worker`, mirroring `hive-c0re/src/job_queue/scheduler.rs`'s /// `run_worker`) — what's still missing is a real job to give it: no /// variant exists yet, so nothing is ever inserted into the graph and /// `claim_next` always returns `None`. Giving this real variants (starting /// with `CreateRepo`) is the next slice, landing together with /// `swarm-controller::forge`, the client those nodes will call. `WireNode` /// is trivially satisfiable on an empty enum (`match *self {}`), so the /// wire machinery below is real and typechecked today, with nothing yet to /// put in it. #[derive(Clone, Debug)] enum SwarmNodeKind {} impl hive_jobq_wire::WireNode for SwarmNodeKind { fn label(&self) -> String { match *self {} } fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value { match *self {} } } /// Placeholder resource name — same rationale and same "no variants until a /// real node needs one" shape as [`SwarmNodeKind`]. #[derive(Clone, Debug, PartialEq, Eq, Hash)] enum SwarmResourceKind {} impl hive_jobq_wire::WireResource for SwarmResourceKind { fn name(&self) -> String { match *self {} } } /// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/ /// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind` /// variant turns into a real effect. Trivially exhaustive today /// (`match kind {}`) because the enum has no variants yet; the first /// real arm (`CreateRepo`, calling `swarm-controller::forge`) lands /// alongside that variant, not before. async fn run_swarm_node( _id: hive_jobq::NodeId, kind: SwarmNodeKind, builder: hive_jobq::builder::JobBuilder, ) -> ( hive_jobq::builder::JobBuilder, hive_jobq::scheduler::Outcome, ) { let _ = builder; match kind {} } /// Spawn the swarm-level job-graph scheduler loop. Mirrors `hive-c0re/src/ /// job_queue/scheduler.rs::run_worker`'s shape: claim one runnable node, /// spawn the future that runs + completes it, loop again immediately if /// something started (more may now be runnable), otherwise back off /// briefly before re-polling. /// /// No shutdown signal to wire in — unlike `hive-c0re`'s `coord.shutdown_rx()`, /// this daemon has no graceful-shutdown machinery at all yet (`main`'s /// `axum::serve` runs unconditionally to process exit), so this loop /// matches that: it rides the runtime down with the process, same as /// every in-flight HTTP request does. /// /// Cheap to run with an empty graph: `claim_next` on a graph nothing was /// ever inserted into just returns `None` every poll, so this is a /// harmless idle loop until the first real node kind exists. fn spawn_jobq_worker( sched: Arc>>, ) { tokio::spawn(async move { loop { let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, run_swarm_node); match runner { Some(runner) => { tokio::spawn(async move { let (id, grew) = runner.await; if let Err(e) = grew { tracing::warn!( node = id.get(), error = %e, "swarm jobq: grown job rejected" ); } }); // Something just started — more may be runnable right // now, so loop again immediately rather than sleeping. } None => { // Nothing runnable. Bounded poll rather than an event // wake (unlike hive-c0re's `notify.notify_one()`, // there is no completion-signal channel here yet) — // fine at this daemon's scale (one graph, no // submitters yet); revisit if/when that stops holding. tokio::time::sleep(std::time::Duration::from_millis(200)).await; } } } }); } /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`. /// /// A compiled-in default is legitimate here and is *not* the mistake that /// a hardcoded remote address would be: this is a path this process /// **creates**, not an address it hopes to find something at. systemd's /// `RuntimeDirectory=swarm-controller` makes the parent exist before /// `ExecStart`, so the default names a directory the unit just produced. /// /// The directory is its own — deliberately not shared with hive-c0re's /// `/run/hyperhive`. The socket is `0666`, so its directory is the only /// access control it has; co-locating it with c0re's admin socket would /// put both within reach of whatever can reach either. nginx runs on the /// host, so nothing narrows its reach for you. const DEFAULT_SOCKET: &str = "/run/swarm-controller/controller.sock"; fn socket_path() -> PathBuf { std::env::var_os("SWARM_CONTROLLER_SOCKET") .map_or_else(|| PathBuf::from(DEFAULT_SOCKET), PathBuf::from) } /// Root of the auto-generated `OpenAPI` spec, served raw at /// `/api/openapi.json` — see the module doc comment above. Tag list /// grows alongside the swarm-level surfaces this daemon picks up, same /// as `hive-c0re::dashboard::ApiDoc`'s tag list did. #[derive(OpenApi)] #[openapi( info( title = "hyperhive swarm-controller API", description = "swarm-controller's HTTP surface, served over its unix \ socket behind the gateway's swarm-UI vhost." ), tags( (name = "health", description = "liveness probe"), (name = "hives", description = "the swarm's hive directory"), (name = "links", description = "swarm service quick links"), (name = "jobq", description = "the swarm-level job graph"), ) )] struct ApiDoc; /// Liveness probe. Returns the build's version so an operator can tell /// *which* controller answered without shelling onto the host. #[utoipa::path( get, path = "/health", responses((status = 200, description = "process is up, body is \"swarm-controller \"", body = String)), tag = "health" )] async fn health() -> &'static str { concat!("swarm-controller ", env!("CARGO_PKG_VERSION"), "\n") } /// One hive in the swarm's directory — the same `name`/`domain` pair /// `services.hyperhive.swarm.hives` (nix/host-modules/swarm.nix) declares, /// carried across unchanged rather than reshaped, so this stays a direct /// mirror of the nix source of truth instead of a second vocabulary for /// the same two fields. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, ToSchema)] struct HiveEntry { name: String, domain: String, } #[derive(Clone)] struct AppState { /// Loaded once at startup (`load_hives`); never mutated, so an /// `Arc` clone per request is the whole synchronization story. hives: Arc>, /// Loaded once at startup (`load_links`); same synchronization story /// as `hives`. links: Arc>, /// `None` when this deployment wired up no swarm queue — the only /// state in which `/api/hives/status` cannot answer at all. A queue /// that is merely *unreachable* still yields a reader, because /// `async-nats` reconnects underneath it. status: Option>, /// The swarm-level job graph, wrapped in its /// [`hive_jobq::scheduler::Scheduler`] now that something drives it /// (`spawn_jobq_worker`) — the graph alone was enough for the /// read-only endpoints, the scheduler is what a `claim_next` loop /// needs. `std::sync::Mutex`, not `tokio`'s — every lock scope below /// is synchronous (no `.await` while held). Always present, never /// gated on the swarm queue: this is process state, not something /// read over the network. jobq: Arc>>, } /// Env var the controller's NixOS module sets from /// `services.hyperhive.swarm.hives`, JSON-encoded — the full directory /// (this daemon has no "self" to exclude) rather than peers-minus-self /// (`services.hyperhive.swarm.peerHives`, which other consumers use). const HIVES_ENV: &str = "SWARM_CONTROLLER_HIVES"; /// Parses [`HIVES_ENV`] into the swarm's hive directory. Unset or /// unparseable both fall back to an empty list with a warning rather than /// failing startup — a swarm-controller that can't yet see its own config /// (dev run, a module not wired up yet) should still serve `/health`. fn load_hives() -> Vec { let Some(raw) = std::env::var_os(HIVES_ENV) else { return Vec::new(); }; match serde_json::from_str(&raw.to_string_lossy()) { Ok(hives) => hives, Err(e) => { tracing::warn!(error = %e, env = HIVES_ENV, "failed to parse hive directory, serving an empty list"); Vec::new() } } } /// The swarm's hive directory — every hive, including whichever one this /// controller instance happens to run on (there is no "self" to exclude /// at the swarm level, unlike `hive-c0re`'s peer list). #[utoipa::path( get, path = "/api/hives", responses((status = 200, description = "every hive in the swarm", body = Vec)), tag = "hives" )] async fn get_hives(State(state): State) -> Json> { Json((*state.hives).clone()) } /// One quick link to a swarm-wide service (authelia, matrix, forge, this /// daemon's own swagger UI, …). Deliberately generic rather than named /// fields per service: each service's own nix module contributes its own /// entry to `services.hyperhive.swarm.controller.links` (same list-merge /// idiom `services.hyperhive.gateway.localNames` already uses), so adding /// a new one is a nix-only change — no new field here, no swarm-ui change. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, ToSchema)] struct ServiceLink { label: String, /// Emoji or short glyph. Empty string, not `Option`, when a /// contributing module has none — one fewer null-vs-absent case for /// the frontend to handle, and every real contributor sets one today. icon: String, url: String, } /// Env var the controller's NixOS module sets from the merged /// `services.hyperhive.swarm.controller.links` list, JSON-encoded — same /// shape/rationale as [`HIVES_ENV`]. Consumed by `GET /api/links`. const LINKS_ENV: &str = "SWARM_CONTROLLER_LINKS"; /// Parses [`LINKS_ENV`] into the swarm's service-link list. Same /// fall-back-to-empty rationale as `load_hives`: a daemon that can't yet /// see this config should still serve `/health` rather than fail startup. fn load_links() -> Vec { let Some(raw) = std::env::var_os(LINKS_ENV) else { return Vec::new(); }; match serde_json::from_str(&raw.to_string_lossy()) { Ok(links) => links, Err(e) => { tracing::warn!(error = %e, env = LINKS_ENV, "failed to parse service links, serving an empty list"); Vec::new() } } } /// Quick links to swarm-wide services (authelia, matrix, forge, this /// daemon's own swagger UI, …), as contributed by each service's own nix /// module. Empty when nothing was configured to contribute — a caller /// renders 0 links the same way it renders any other count, no special /// "not configured" case. #[utoipa::path( get, path = "/api/links", responses((status = 200, description = "swarm service quick links", body = Vec)), tag = "links" )] async fn get_links(State(state): State) -> Json> { Json((*state.links).clone()) } /// Why the status route answers 503 rather than an empty list. /// /// "I cannot reach the store" and "every hive is silent" are different /// answers, and rendering the second when the first is true is exactly /// the smoothing this endpoint exists to avoid — a caller would draw a /// swarm-wide outage out of a local one. The cause is carried in the /// body because a bare 503 on an operator-facing diagnostic is how a /// misconfiguration costs an afternoon; it is a queue/JetStream error /// string, and this surface is already behind the swarm's SSO. struct StatusUnavailable(String); impl axum::response::IntoResponse for StatusUnavailable { fn into_response(self) -> axum::response::Response { (axum::http::StatusCode::SERVICE_UNAVAILABLE, self.0).into_response() } } /// What each hive last said about itself, read from the swarm queue at /// request time. /// /// Every hive in the roster gets a row whether or not it has ever /// reported — see the `status` module for why absence, not presence, is /// the case this is built around. #[utoipa::path( get, path = "/api/hives/status", responses( (status = 200, description = "a row per hive, freshness derived now", body = Vec), (status = 503, description = "no swarm queue is configured here, or its store could not be read", body = String), ), tag = "hives" )] async fn get_hives_status( State(state): State, ) -> Result>, StatusUnavailable> { let Some(reader) = state.status.as_ref() else { return Err(StatusUnavailable( "no swarm queue is configured on this host".to_owned(), )); }; match reader .view(&state.hives, std::time::SystemTime::now()) .await { Ok(rows) => Ok(Json(rows)), Err(e) => { let detail = format!("{e:#}"); tracing::warn!(error = %detail, "reading the swarm status bucket failed"); Err(StatusUnavailable(detail)) } } } /// Query params for `GET /api/jobq/graph` — `?states=` narrows to root /// groups in the named states, same shape `hive_jobq_wire::parse_states` /// parses. #[derive(Deserialize, utoipa::IntoParams)] struct JobqGraphQuery { states: Option, } /// Every node of every root group in the swarm-level job graph. Nothing /// filters "done and old" here — with zero nodes ever submitted there is /// nothing to bound yet, so `graph.roots()` (everything) is the whole /// roster passed to [`hive_jobq_wire::GraphWire::wire_snapshot`]. #[utoipa::path( get, path = "/api/jobq/graph", params(JobqGraphQuery), responses((status = 200, description = "every node of every root group, as generic \ `hive_jobq` graph nodes. `?states=` narrows to root groups in the named states.", body = Vec)), tag = "jobq" )] async fn get_jobq_graph( State(state): State, axum::extract::Query(q): axum::extract::Query, ) -> Json> { let states = hive_jobq_wire::parse_states(q.states.as_deref()); let sched = state .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let graph = sched.graph(); let roots: Vec = graph.roots().map(|n| n.id).collect(); let nodes = graph.wire_snapshot(roots); Json(hive_jobq_wire::filter_nodes_by_state( nodes, states.as_deref(), )) } /// Counts by lifecycle state over the same groups `/api/jobq/graph` serves. #[utoipa::path( get, path = "/api/jobq/rollup", responses((status = 200, description = "counts by lifecycle state", body = Vec)), tag = "jobq" )] async fn get_jobq_rollup(State(state): State) -> Json> { let sched = state .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let graph = sched.graph(); let roots: Vec = graph.roots().map(|n| n.id).collect(); Json(hive_jobq_wire::state_rollup(graph, roots)) } #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), ) .init(); let path = socket_path(); // `RuntimeDirectoryPreserve=yes` keeps the directory across a restart, // so a socket file from the previous run can outlive the process that // owned it and `bind` would fail with EADDRINUSE. Unlinking a stale // socket is safe precisely because the directory is ours alone: nothing // else can have put a file at this path. if let Err(e) = std::fs::remove_file(&path) && e.kind() != std::io::ErrorKind::NotFound { return Err(e).with_context(|| format!("clearing stale socket at {}", path.display())); } let listener = tokio::net::UnixListener::bind(&path) .with_context(|| format!("binding {}", path.display()))?; // `bind` leaves the socket 0755, and connecting needs write — the // gateway's nginx is a different user, so it would be locked out. // 0666 matches how hive-c0re publishes the per-agent sockets // (`socket_server::start`), and rests on the same argument: **the // containing directory is the access control, not the socket mode.** // This directory holds one socket and is bind-mounted into exactly // one container. That is also why it must not be shared with // hive-c0re's `/run/hyperhive` — with a 0666 socket, a directory // that carries more than it should is the whole vulnerability. std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o666)) .with_context(|| format!("chmod {}", path.display()))?; tracing::info!(socket = %path.display(), "swarm-controller listening"); // Connect to the swarm queue when this deployment wired one up. // // Deliberately NOT fatal on failure: the controller's HTTP surface is // useful without the queue, and a hive that cannot be read from renders // as `unknown` rather than as an outage of this daemon. What IS fatal is // a half-set environment — `QueueConfig::from_env` refuses that, because // silently behaving like an unconfigured host is how every hive ends up // reading `never_reported` with nothing to point at. let status = match swarm_queue_client::QueueConfig::from_env("SWARM_CONTROLLER")? { None => { tracing::info!("no swarm queue configured; status aggregation is off"); None } Some(cfg) => match swarm_queue_client::connect(cfg).await { Ok(client) => { // NOT "connected": `retry_on_initial_connect` returns a client // before any connection has been established, so claiming a // connection here would put "connected to the swarm queue" in // the journal moments before every request 503s with "not // connected" — and a reader would rightly distrust the second // line rather than the first. The connection's real state is // reported by the status endpoint, which checks it per request. tracing::info!("swarm queue configured; connecting in the background"); Some(Arc::new(status::StatusReader::new( client, status::StatusReader::stale_after_from_env(), ))) } Err(e) => { // `chain`, not `{:#}`: this is the queue client's own // error type, and thiserror's Display ignores the // alternate flag — the source would be dropped silently. tracing::warn!( error = swarm_queue_client::chain(&e), "swarm queue unreachable" ); None } }, }; let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ))); spawn_jobq_worker(Arc::clone(&jobq)); let state = AppState { hives: Arc::new(load_hives()), links: Arc::new(load_links()), status, jobq, }; let (router, api) = OpenApiRouter::::with_openapi(ApiDoc::openapi()) .routes(routes!(health)) .routes(routes!(get_hives)) .routes(routes!(get_hives_status)) .routes(routes!(get_links)) .routes(routes!(get_jobq_graph)) .routes(routes!(get_jobq_rollup)) .split_for_parts(); // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from // the nix store (see the module doc comment above). `api` is // `Clone`; each request gets its own owned copy for `Json` to // serialize, same as `hive-c0re::dashboard::serve`. let app = router .route( "/api/openapi.json", get(move || async move { Json(api.clone()) }), ) .with_state(state); axum::serve(listener, app) .await .context("serving swarm-controller") } #[cfg(test)] mod tests { use super::{ DEFAULT_SOCKET, HIVES_ENV, HiveEntry, LINKS_ENV, ServiceLink, load_hives, load_links, }; use std::path::Path; /// The socket must not share a directory with anything else, because /// the socket is `0666` and the directory is therefore the only access /// control it has. `/run/hyperhive` in particular holds hive-c0re's /// **admin** socket, and nginx — a host service — is bounded by /// nothing but the directory itself. /// /// A test rather than a comment: the failure this guards against is a /// one-word edit that looks tidier and reads fine in review. #[test] fn socket_lives_in_its_own_runtime_dir() { let parent = Path::new(DEFAULT_SOCKET) .parent() .expect("socket path has a parent directory"); assert_eq!( parent, Path::new("/run/swarm-controller"), "the socket's directory is its access control — moving it under a shared \ directory (notably /run/hyperhive, which holds the host admin socket) \ exposes everything else in that directory to the gateway's nginx" ); } /// One test, not three, deliberately: `HIVES_ENV` is a real process /// env var, and `cargo test`'s default parallel runner would race /// separate missing/malformed/valid tests against each other. Driving /// all three states sequentially inside one test needs no mutex and /// still proves each branch of `load_hives`. /// /// SAFETY: single-threaded mutation of a process env var no other /// test in this crate reads; restored (removed) before returning. #[test] fn load_hives_covers_missing_malformed_and_valid() { unsafe { std::env::remove_var(HIVES_ENV); } assert_eq!( load_hives(), Vec::::new(), "unset env var is an empty directory, not a startup failure" ); unsafe { std::env::set_var(HIVES_ENV, "not json"); } assert_eq!( load_hives(), Vec::::new(), "unparseable env var falls back to empty rather than panicking — \ /health must still answer even if this daemon's own config is wrong" ); unsafe { std::env::set_var( HIVES_ENV, r#"[{"name":"pr1ma","domain":"pr1ma.example.com"},{"name":"umbra","domain":"umbra.example.com"}]"#, ); } assert_eq!( load_hives(), vec![ HiveEntry { name: "pr1ma".to_string(), domain: "pr1ma.example.com".to_string(), }, HiveEntry { name: "umbra".to_string(), domain: "umbra.example.com".to_string(), }, ] ); unsafe { std::env::remove_var(HIVES_ENV); } } /// Same three-state coverage as `load_hives_covers_missing_malformed_and_valid`, /// same reason for one test rather than three (a shared process env var). /// /// SAFETY: single-threaded mutation of a process env var no other test /// in this crate reads; restored (removed) before returning. #[test] fn load_links_covers_missing_malformed_and_valid() { unsafe { std::env::remove_var(LINKS_ENV); } assert_eq!( load_links(), Vec::::new(), "unset env var is an empty list, not a startup failure" ); unsafe { std::env::set_var(LINKS_ENV, "not json"); } assert_eq!( load_links(), Vec::::new(), "unparseable env var falls back to empty rather than panicking" ); unsafe { std::env::set_var( LINKS_ENV, r#"[{"label":"Authelia","icon":"🔑","url":"https://auth.example.com/"}]"#, ); } assert_eq!( load_links(), vec![ServiceLink { label: "Authelia".to_string(), icon: "🔑".to_string(), url: "https://auth.example.com/".to_string(), }] ); unsafe { std::env::remove_var(LINKS_ENV); } } }