738dd1c95b
no behaviour change: a pure move of items and tests into new modules, plus visibility edits (pub(crate)) and path updates in sibling modules that named crate-root items. same routes, same OpenAPI, 561 tests before and after. why: #5470, new features stop landing in one 6.4k-line file by default. files outside swarm-controller/src: docs/agent-lifecycle/agent-roster.md, swarm-ui JobsPage.tsx and AgentsPage.tsx, nix/host-modules/swarm-controller.nix and swarm-grafana.nix hold comments that named main.rs for items that moved.
365 lines
18 KiB
Rust
365 lines
18 KiB
Rust
//! The HTTP state every handler shares, the error shapes they answer with,
|
|
//! and the handlers that live outside a resource module of their own.
|
|
|
|
pub(crate) mod agents;
|
|
pub(crate) mod config_prs;
|
|
pub(crate) mod forge_admin;
|
|
pub(crate) mod hives;
|
|
pub(crate) mod jobq_graph;
|
|
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use crate::jobq::{Late, SwarmNodeKind, SwarmResourceKind};
|
|
use crate::routes::hives::{HiveEntry, ServiceLink};
|
|
use crate::{
|
|
agent_icon, agent_login, agent_status, agents_md, alerts, auth, builder, config_pr, forge,
|
|
inference, linked_accounts, review_classifier, status, wanted,
|
|
};
|
|
|
|
/// 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 <version>\"", body = String)),
|
|
tag = "health"
|
|
)]
|
|
pub(crate) async fn health() -> &'static str {
|
|
concat!("swarm-controller ", env!("CARGO_PKG_VERSION"), "\n")
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub(crate) struct AppState {
|
|
/// Loaded once at startup (`load_hives`); never mutated, so an
|
|
/// `Arc` clone per request is the whole synchronization story.
|
|
pub(crate) hives: Arc<Vec<HiveEntry>>,
|
|
/// Loaded once at startup (`load_links`); same synchronization story
|
|
/// as `hives`.
|
|
pub(crate) links: Arc<Vec<ServiceLink>>,
|
|
/// `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.
|
|
pub(crate) status: Option<Arc<status::StatusReader>>,
|
|
/// Publishes the agent set this swarm declares for each hive.
|
|
///
|
|
/// `None` in exactly the state `status` is: no swarm queue was wired
|
|
/// up, so there is nowhere to publish a declaration to. Shares that
|
|
/// reader's connection rather than opening a second one.
|
|
pub(crate) wanted: Option<Arc<wanted::WantedWriter>>,
|
|
/// Per-agent status, sharing `status`'s connection — same "one queue
|
|
/// connection, several consumers" rationale as `wanted` above. `None`
|
|
/// in exactly the state `status` is.
|
|
pub(crate) agent_status: Option<Arc<agent_status::AgentStatusReader>>,
|
|
/// Per-agent icons, sharing `status`'s connection — same "one queue
|
|
/// connection, several consumers" rationale as `agent_status` above,
|
|
/// and `None` in exactly the state `status` is.
|
|
pub(crate) agent_icons: Option<Arc<agent_icon::AgentIconReader>>,
|
|
/// Each agent's published `AGENTS.md`, on `status`'s connection like
|
|
/// `agent_icons`, and `None` in exactly the state `status` is.
|
|
pub(crate) agents_md: Option<Arc<agents_md::AgentsMdReader>>,
|
|
/// What each agent last reported running on, from its turn-state
|
|
/// headers on `status`'s connection. `None` in exactly the state
|
|
/// `status` is.
|
|
pub(crate) reported_inference: Option<Arc<inference::Reported>>,
|
|
/// Whether each agent last said it needs a login, from its turn-state
|
|
/// headers on `status`'s connection. `None` in exactly the state
|
|
/// `status` is.
|
|
pub(crate) agent_login: Option<Arc<agent_login::Reported>>,
|
|
/// 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.
|
|
pub(crate) jobq: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
|
/// The identity store's only reader, for `GET /api/agents`.
|
|
///
|
|
/// `None` when no bridge is wired up, which is the one state in which
|
|
/// the roster cannot be answered at all — the same shape as `status`
|
|
/// above, and for the same reason: an empty roster and an unreadable
|
|
/// one are different facts.
|
|
///
|
|
/// ⚠️ The two `/api/agents` verbs treat this differently on purpose.
|
|
/// **POST does not need it**: creation only queues a job, and
|
|
/// whether a bridge exists is `run_swarm_node`'s concern (it holds its
|
|
/// own clone from `spawn_jobq_worker`), so a request still queues
|
|
/// cleanly on a bridge-less host and fails loud once claimed. POST reads
|
|
/// it only for a name that breaks a naming rule, to tell a new agent
|
|
/// from an existing one, and warns when it cannot (`name_verdict`). **GET
|
|
/// has nowhere to defer to** — there is no job, only an answer it
|
|
/// either has or does not. `POST /api/forge/users/{name}/admin` reads
|
|
/// it for every name and refuses when it cannot (`admin_unless_agent`).
|
|
pub(crate) auth: Option<Arc<auth::AuthBridge>>,
|
|
/// HMAC secret for swarm-wide forge webhooks, loaded once at startup.
|
|
/// `None` when it could not be read or created — the webhook endpoint
|
|
/// then refuses every delivery with 503 rather than admitting one it
|
|
/// cannot verify. Deliberately not fatal to startup: nothing is
|
|
/// registered against that endpoint yet, and the rest of this daemon's
|
|
/// surface is unaffected. See [`load_or_generate_secret`](crate::webhook::load_or_generate_secret).
|
|
///
|
|
/// `Arc<str>`, not `Arc<String>`: the value is never mutated after
|
|
/// startup, and this way `as_deref()` yields the `&str` the verifier
|
|
/// takes without a second hop through `String`.
|
|
pub(crate) webhook_secret: Option<Arc<str>>,
|
|
/// Last successful `agent-configs/*` scan, kept current by
|
|
/// [`config_pr::spawn`]. Empty while this daemon has no forge access —
|
|
/// same "absent means don't ask" shape as `status` and `forge`, not a
|
|
/// startup failure — and filled when the forge client is.
|
|
pub(crate) config_prs: Late<Arc<config_pr::ConfigPrCache>>,
|
|
/// The swarm's human display name (`services.hyperhive.swarm.name`),
|
|
/// loaded once at startup (`load_swarm_name`). `None` when the
|
|
/// operator never set it — a swarm without a display name is a
|
|
/// supported, if less friendly, state, not a startup failure. Same
|
|
/// `Arc<str>` rationale as `webhook_secret`: never mutated, so a
|
|
/// clone per request is just a refcount bump.
|
|
pub(crate) swarm_name: Option<Arc<str>>,
|
|
/// The forge client itself, for the read-only repo/issue-report
|
|
/// routes (`GET /api/repos`, `GET /api/repos/{org}/{repo}/issue-report`)
|
|
/// and `POST /api/forge/users/{name}/admin` — distinct from `config_prs`, which holds a *cache* built off this
|
|
/// same client rather than the client. Empty while this daemon has no
|
|
/// forge access, and filled the moment `forge::enable` builds the client.
|
|
pub(crate) forge: Late<Arc<forge::Client>>,
|
|
/// Held by `create_agent` from reading where a name is placed until its
|
|
/// graph is queued, and by `set_agent_state` from that read until its
|
|
/// declaration is written, so two placements of one name cannot both
|
|
/// find it unplaced. `tokio`'s mutex, unlike `jobq`'s: the read awaits
|
|
/// the queue.
|
|
pub(crate) create_gate: Arc<tokio::sync::Mutex<()>>,
|
|
/// The store the linked-account routes read and write.
|
|
pub(crate) accounts: linked_accounts::Accounts,
|
|
/// The review classifier try-outs started here, until their results
|
|
/// expire.
|
|
pub(crate) try_outs: review_classifier::TryOuts,
|
|
/// The requested reviews running or waiting.
|
|
pub(crate) reviews: Arc<review_classifier::review::Reviews>,
|
|
/// The review classifier's finished runs.
|
|
pub(crate) runs: review_classifier::runs::Runs,
|
|
/// The swarm builder's finished builds.
|
|
pub(crate) builds: builder::runs::Builds,
|
|
/// Grafana's alerts API, for `GET /api/alerts`. `None` when no Grafana URL
|
|
/// is configured.
|
|
pub(crate) alerts: Option<Arc<alerts::Grafana>>,
|
|
}
|
|
|
|
/// 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.
|
|
///
|
|
/// It travels in `detail` of an RFC 9457 `application/problem+json` body
|
|
/// rather than as a bare string, so the cause is an addressable field
|
|
/// instead of the whole payload — see `error_problem` below.
|
|
pub(crate) struct StatusUnavailable(pub(crate) String);
|
|
|
|
impl axum::response::IntoResponse for StatusUnavailable {
|
|
fn into_response(self) -> axum::response::Response {
|
|
error_problem(axum::http::StatusCode::SERVICE_UNAVAILABLE, &self.0).into_response()
|
|
}
|
|
}
|
|
|
|
/// Every error this daemon returns, in one shape.
|
|
///
|
|
/// RFC 9457 `application/problem+json` is the hive-wide contract for HTTP
|
|
/// error bodies (`docs/process/conventions.md`), and the operator UIs read `detail`
|
|
/// for display. A bare string forces the reader to treat the entire body as
|
|
/// the message, which is the difference between a UI that can offer "copy the
|
|
/// cause" and one that can only dump a response.
|
|
pub(crate) fn error_problem(
|
|
status: axum::http::StatusCode,
|
|
detail: &str,
|
|
) -> problem_details::ProblemDetails {
|
|
problem_details::ProblemDetails::from_status_code(status).with_detail(detail)
|
|
}
|
|
|
|
/// The status a [`wanted::WantedWriter`], [`agent_icon::AgentIconReader`] or
|
|
/// `schedules` queue failure should answer with.
|
|
///
|
|
/// `WantedWriter::view`/`set` call `ensure_connected` internally and
|
|
/// propagate through `anyhow`, so the concrete
|
|
/// [`swarm_queue_client::Error`] survives underneath but is not the type in
|
|
/// hand — `swarm_queue_client`'s own doc says a caller "gets a type it can
|
|
/// match on", which means downcasting back to it rather than string-matching
|
|
/// the rendered message. `NotConnected` is a retryable condition (the queue
|
|
/// is reconnecting, or hasn't yet) and every other queue-backed route here
|
|
/// already answers 503 for it; anything else is a genuine failure to publish
|
|
/// or read, which stays 500.
|
|
pub(crate) fn wanted_error_status(e: &anyhow::Error) -> axum::http::StatusCode {
|
|
match e.downcast_ref::<swarm_queue_client::Error>() {
|
|
Some(swarm_queue_client::Error::NotConnected(_)) => {
|
|
axum::http::StatusCode::SERVICE_UNAVAILABLE
|
|
}
|
|
_ => axum::http::StatusCode::INTERNAL_SERVER_ERROR,
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(crate) mod tests {
|
|
use super::*;
|
|
|
|
/// The 503 this route returns is what the operator UIs' shared error
|
|
/// component renders, so assert the RENDERED response rather than the
|
|
/// `problem_details` crate: the contract a UI depends on is the content
|
|
/// type plus a `detail` it can address, and a handler that built the
|
|
/// value and returned it as a bare string would satisfy any test
|
|
/// written against the type alone.
|
|
#[tokio::test]
|
|
async fn status_unavailable_renders_problem_json_with_the_cause_in_detail() {
|
|
use axum::response::IntoResponse as _;
|
|
|
|
let cause = "listing status bucket keys: timed out";
|
|
let resp = StatusUnavailable(cause.to_owned()).into_response();
|
|
|
|
assert_eq!(resp.status(), axum::http::StatusCode::SERVICE_UNAVAILABLE);
|
|
let ct = resp
|
|
.headers()
|
|
.get(axum::http::header::CONTENT_TYPE)
|
|
.and_then(|v| v.to_str().ok())
|
|
.unwrap_or_default()
|
|
.to_owned();
|
|
assert!(
|
|
ct.starts_with("application/problem+json"),
|
|
"RFC 9457 media type, got {ct:?}"
|
|
);
|
|
|
|
let bytes = axum::body::to_bytes(resp.into_body(), 64 * 1024)
|
|
.await
|
|
.expect("body reads");
|
|
let v: serde_json::Value = serde_json::from_slice(&bytes).expect("problem+json parses");
|
|
assert_eq!(v["status"], 503);
|
|
// The cause is an addressable field, not the entire payload — that
|
|
// distinction is the point of the change, so it is what is asserted.
|
|
assert_eq!(v["detail"], cause);
|
|
}
|
|
|
|
/// Drives `SwarmNodeKind::CreateRepo` through the real
|
|
/// `hive_jobq::scheduler::Scheduler` claim → run → complete path,
|
|
/// rather than only through `create_agent`'s endpoint test (there
|
|
/// isn't one — the endpoint itself is thin, insert-and-return; the
|
|
/// interesting behavior is in `run_swarm_node`'s executor arm, which
|
|
/// this exercises directly).
|
|
///
|
|
/// Deliberately offline: with the forge URL unset,
|
|
/// `forge::Client::from_env` returns `Ok(None)` (see that module's doc
|
|
/// comment), so this exercises the whole claim → run → complete path
|
|
/// through `hive_jobq::scheduler::Scheduler` without a real forge
|
|
/// server — at the cost of only ever observing the
|
|
/// graceful-absence-is-failure branch here. The happy path needs an
|
|
/// actual forge instance and isn't something a unit test in this crate
|
|
/// can reach.
|
|
///
|
|
/// SAFETY: single-threaded mutation of the `forge` env var this
|
|
/// test itself owns, restored before returning — no other test in this
|
|
/// crate reads them.
|
|
/// One roster, one agent name, two hives — the accepted one and a
|
|
/// typo. Built per test so neither can see the other's queue.
|
|
/// The scheduler handle these tests hold onto so they can assert on
|
|
/// what the endpoint queued. Aliased because the full type is three
|
|
/// nested generics deep and reads worse inline than named.
|
|
pub(crate) type SharedSched = std::sync::Arc<
|
|
std::sync::Mutex<
|
|
hive_jobq::scheduler::Scheduler<
|
|
crate::jobq::SwarmNodeKind,
|
|
crate::jobq::SwarmResourceKind,
|
|
>,
|
|
>,
|
|
>;
|
|
|
|
pub(crate) fn state_with_roster() -> (super::AppState, SharedSched) {
|
|
let sched =
|
|
std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new(
|
|
hive_jobq::Graph::new(),
|
|
hive_jobq::resources::ResourceTable::new(),
|
|
)));
|
|
let state = super::AppState {
|
|
hives: std::sync::Arc::new(vec![HiveEntry {
|
|
name: "pr1ma".to_owned(),
|
|
domain: "pr1ma.example".to_owned(),
|
|
dashboard: "pr1ma.hives.example".to_owned(),
|
|
}]),
|
|
links: std::sync::Arc::new(Vec::new()),
|
|
status: None,
|
|
// No queue, for the same reason as `status`: these tests build
|
|
// the creation graph and read its shape, and never run the node
|
|
// that would write a declaration.
|
|
wanted: None,
|
|
agent_status: None,
|
|
agent_icons: None,
|
|
agents_md: None,
|
|
reported_inference: None,
|
|
agent_login: None,
|
|
jobq: std::sync::Arc::clone(&sched),
|
|
webhook_secret: None,
|
|
config_prs: std::sync::Arc::default(),
|
|
swarm_name: None,
|
|
// No bridge: these tests drive agent *creation*, which queues a
|
|
// job and consults one only for a rule-breaking name. The roster
|
|
// read is the verb that needs it, and it has its own test below.
|
|
auth: None,
|
|
forge: std::sync::Arc::default(),
|
|
create_gate: std::sync::Arc::default(),
|
|
accounts: crate::linked_accounts::Accounts::Fake(std::sync::Arc::default()),
|
|
try_outs: crate::review_classifier::TryOuts::default(),
|
|
reviews: std::sync::Arc::default(),
|
|
runs: crate::review_classifier::runs::Runs::Live(None),
|
|
builds: crate::builder::runs::Builds::Live(None),
|
|
alerts: None,
|
|
};
|
|
(state, sched)
|
|
}
|
|
|
|
/// `wanted_error_status` as a pure function first: build the exact
|
|
/// `swarm_queue_client::Error::NotConnected` `ensure_connected` raises,
|
|
/// wrap it the same way `?` inside `WantedWriter::view`/`set` does, and
|
|
/// confirm it downcasts back to the status those routes are meant to
|
|
/// answer.
|
|
#[test]
|
|
fn not_connected_downcasts_to_503() {
|
|
let e: anyhow::Error =
|
|
swarm_queue_client::Error::NotConnected(async_nats::connection::State::Pending).into();
|
|
assert_eq!(
|
|
super::wanted_error_status(&e),
|
|
axum::http::StatusCode::SERVICE_UNAVAILABLE
|
|
);
|
|
}
|
|
|
|
/// Invert-proof for the test above: an `anyhow::Error` that does not
|
|
/// wrap a `swarm_queue_client::Error` at all (the shape a genuine write
|
|
/// or decode failure takes — `apply`'s "declaration is not decodable"
|
|
/// context, for instance) must NOT downcast to `NotConnected`, and must
|
|
/// stay 500. Without this, a `wanted_error_status` that answered 503
|
|
/// unconditionally would still pass the test above.
|
|
#[test]
|
|
fn an_unrelated_error_stays_500() {
|
|
let e = anyhow::anyhow!("the hive's current declaration is not decodable");
|
|
assert_eq!(
|
|
super::wanted_error_status(&e),
|
|
axum::http::StatusCode::INTERNAL_SERVER_ERROR
|
|
);
|
|
}
|
|
|
|
/// Second invert-proof, closer to the actual bug this closes: a
|
|
/// *different* `swarm_queue_client::Error` variant (a real connect
|
|
/// failure, not `NotConnected`) also stays 500 — proving the match
|
|
/// looks at the variant, not just at "was this crate's error type
|
|
/// involved at all".
|
|
#[test]
|
|
fn a_different_queue_client_error_variant_stays_500() {
|
|
let e: anyhow::Error = swarm_queue_client::Error::Connect {
|
|
url: "nats://queue.example:4222".to_owned(),
|
|
source: async_nats::ConnectErrorKind::TimedOut.into(),
|
|
}
|
|
.into();
|
|
assert_eq!(
|
|
super::wanted_error_status(&e),
|
|
axum::http::StatusCode::INTERNAL_SERVER_ERROR
|
|
);
|
|
}
|
|
}
|