diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 3cda4121..54aff33c 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -47,6 +47,7 @@ mod issue_report; mod otel_http_client; mod status; mod vcs_metrics; +mod wanted; mod webhook; /// Node payload for the swarm-level job graph. Named `Swarm*` rather than @@ -429,6 +430,12 @@ struct AppState { /// that is merely *unreachable* still yields a reader, because /// `async-nats` reconnects underneath it. status: Option>, + /// 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. + wanted: 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 @@ -681,6 +688,150 @@ fn error_problem(status: axum::http::StatusCode, detail: &str) -> problem_detail problem_details::ProblemDetails::from_status_code(status).with_detail(detail) } +/// The state to declare for one agent. +#[derive(Debug, Deserialize, ToSchema)] +struct SetAgentStateRequest { + /// Typed as a string in the schema only — the parse is the real enum, so + /// a value this build does not know is a 400 rather than a field that + /// silently does nothing. + #[schema(value_type = String, example = "up")] + state: swarm_queue_client::wanted::AgentState, +} + +/// One agent's line in a hive's declaration. +#[derive(Clone, Debug, Serialize, ToSchema)] +struct AgentDeclaration { + agent: String, + state: String, +} + +fn render(declaration: &swarm_queue_client::wanted::HiveWanted) -> Vec { + declaration + .agents + .iter() + .map(|(agent, wanted)| AgentDeclaration { + agent: agent.clone(), + state: wanted.state.as_str().to_owned(), + }) + .collect() +} + +/// The declaration writer, sharing the status reader's connection. +/// +/// The controller holds exactly one queue connection by design — see +/// `StatusReader::queue_client`. A second connect would double the +/// auth-callout traffic and give the two paths independent reconnect state, +/// so one could be serving while the other was still down. +fn wanted_writer(status: Option<&Arc>) -> Option> { + status.map(|s| Arc::new(wanted::WantedWriter::new(s.queue_client()))) +} + +/// Both handlers below take the same hive name and reject it the same way. +/// +/// Reports the status and the detail rather than a rendered +/// `ProblemDetails`: that type is 232 bytes, which makes every `Result` in +/// this path pay for the error case it usually does not take. The handlers +/// render at the boundary, where the body is actually needed. +fn declaration_target( + state: &AppState, + hive: &str, +) -> Result<(Arc, String), (axum::http::StatusCode, String)> { + let writer = state.wanted.clone().ok_or_else(|| { + ( + axum::http::StatusCode::SERVICE_UNAVAILABLE, + "this deployment wired up no swarm queue, so there is nowhere to publish a declaration" + .to_owned(), + ) + })?; + let hive = hive_types::Ident::parse(hive) + .map_err(|reason| (axum::http::StatusCode::BAD_REQUEST, reason.to_owned()))? + .into_string(); + // Shaped like a hive name, and actually one. Same two checks + // `create_agent` makes, for the same reason: a typo otherwise publishes a + // declaration under a key no hive will ever read. + if !state.hives.iter().any(|h| h.name == hive) { + return Err(( + axum::http::StatusCode::BAD_REQUEST, + format!("{hive:?} is not a hive in this swarm"), + )); + } + Ok((writer, hive)) +} + +/// Declare what this swarm wants of one agent on one hive. +/// +/// The whole declaration is returned as published, because the value is the +/// hive's entire agent map and a caller that changed one agent still wants to +/// render the rest. +#[utoipa::path( + put, + path = "/api/hives/{hive}/agents/{agent}/state", + params( + ("hive" = String, Path, description = "hive whose declaration this is"), + ("agent" = String, Path, description = "agent to declare"), + ), + request_body = SetAgentStateRequest, + responses( + (status = 200, description = "the declaration as now published", body = Vec), + (status = 400, description = "a name is not an identifier, the hive is not in this swarm, or the state is unknown (problem+json)", body = String), + (status = 503, description = "no swarm queue is wired up (problem+json)", body = String), + (status = 500, description = "the declaration could not be published (problem+json)", body = String), + ), + tag = "agents" +)] +async fn set_agent_state( + State(state): State, + axum::extract::Path((hive, agent)): axum::extract::Path<(String, String)>, + Json(req): Json, +) -> Result>, problem_details::ProblemDetails> { + let (writer, hive) = + declaration_target(&state, &hive).map_err(|(s, d)| error_problem(s, &d))?; + let agent = hive_types::Ident::parse(&agent) + .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))? + .into_string(); + + let declaration = writer.set(&hive, &agent, req.state).await.map_err(|e| { + tracing::warn!(hive = %hive, agent = %agent, error = %format!("{e:#}"), "declaring agent state failed"); + error_problem( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + &format!("{e:#}"), + ) + })?; + Ok(Json(render(&declaration))) +} + +/// What this swarm currently declares for a hive. +/// +/// Read back from the bucket rather than from a second copy kept here: the +/// published value is the record. +#[utoipa::path( + get, + path = "/api/hives/{hive}/wanted", + params(("hive" = String, Path, description = "hive whose declaration to read")), + responses( + (status = 200, description = "the declaration, empty when nothing is published yet", body = Vec), + (status = 400, description = "not an identifier, or not a hive in this swarm (problem+json)", body = String), + (status = 503, description = "no swarm queue is wired up (problem+json)", body = String), + (status = 500, description = "the declaration could not be read (problem+json)", body = String), + ), + tag = "agents" +)] +async fn get_hive_wanted( + State(state): State, + axum::extract::Path(hive): axum::extract::Path, +) -> Result>, problem_details::ProblemDetails> { + let (writer, hive) = + declaration_target(&state, &hive).map_err(|(s, d)| error_problem(s, &d))?; + let declaration = writer.view(&hive).await.map_err(|e| { + tracing::warn!(hive = %hive, error = %format!("{e:#}"), "reading the declaration failed"); + error_problem( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + &format!("{e:#}"), + ) + })?; + Ok(Json(declaration.as_ref().map(render).unwrap_or_default())) +} + /// What each hive last said about itself, read from the swarm queue at /// request time. /// @@ -1328,6 +1479,7 @@ async fn main() -> Result<()> { let state = AppState { hives: Arc::new(load_hives()), links: Arc::new(load_links()), + wanted: wanted_writer(status.as_ref()), status, jobq, webhook_secret, @@ -1337,6 +1489,19 @@ async fn main() -> Result<()> { forge: state_forge, }; + let app = build_app(state); + axum::serve(listener, app) + .await + .context("serving swarm-controller") +} + +/// Every route this daemon serves, wired to its state. +/// +/// Split out of `main` because the route list is the part that grows, and +/// `main` sat exactly on the `too_many_lines` limit — adding an endpoint +/// tripped a lint about the startup sequence, which is not where the change +/// was. A new route now costs one line here and none there. +fn build_app(state: AppState) -> axum::Router { let (router, api) = OpenApiRouter::::with_openapi(ApiDoc::openapi()) .routes(routes!(health)) .routes(routes!(get_hives)) @@ -1349,6 +1514,8 @@ async fn main() -> Result<()> { .routes(routes!(get_config_prs)) .routes(routes!(create_agent)) .routes(routes!(get_agents)) + .routes(routes!(set_agent_state)) + .routes(routes!(get_hive_wanted)) .routes(routes!(issue_report::get_repos)) .routes(routes!(issue_report::get_issue_report_all)) .routes(routes!(issue_report::get_issue_report)) @@ -1358,15 +1525,12 @@ async fn main() -> Result<()> { // 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 + router .route( "/api/openapi.json", get(move || async move { Json(api.clone()) }), ) - .with_state(state); - axum::serve(listener, app) - .await - .context("serving swarm-controller") + .with_state(state) } #[cfg(test)] @@ -1455,6 +1619,9 @@ mod tests { }]), links: std::sync::Arc::new(Vec::new()), status: None, + // No queue, for the same reason as `status`: these tests drive + // agent creation, which publishes no declaration. + wanted: None, jobq: std::sync::Arc::clone(&sched), webhook_secret: None, config_prs: None, diff --git a/swarm-controller/src/wanted.rs b/swarm-controller/src/wanted.rs new file mode 100644 index 00000000..b5d2c7ba --- /dev/null +++ b/swarm-controller/src/wanted.rs @@ -0,0 +1,216 @@ +//! Writes the agent set this swarm declares for each hive. +//! +//! The mirror of [`crate::status`]: that module reads what hives report, +//! this one writes what they are told, and both address the same queue. +//! The lifecycle is deliberately identical — a NATS client rather than a +//! bucket handle, resolved on first use and cached, so a controller that +//! starts before the bucket exists picks it up without a restart. +//! +//! The bucket is the record. Nothing here keeps a second copy of the +//! declaration to reconcile against, because the current value can be read +//! back from the queue whenever it is needed. + +use anyhow::{Context, Result}; +use async_nats::jetstream::kv::{CreateErrorKind, UpdateErrorKind}; +use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted}; + +/// The three outcomes of one write attempt, which the two KV verbs report +/// through separate error types. +enum Wrote { + Ok, + /// Another writer won the race; re-read and re-apply. + LostRace, + Failed(anyhow::Error), +} + +/// How many times a losing writer re-reads and re-applies before giving up. +/// +/// A conflict means another writer changed a *different* agent between this +/// one's read and its write, so a retry re-reads and re-applies onto the +/// winner. Bounded because an unbounded loop against a hot key is a spin, +/// and a caller that gets an error can ask again with fresh intent. +const MAX_ATTEMPTS: usize = 5; + +/// Apply one agent's declared state to a hive's current declaration. +/// +/// Split out from the write loop because it holds the invariant that matters: +/// the value under a hive's key is the map of **every** agent on that hive, +/// so declaring one agent must preserve the rest. +/// +/// `current` is `None` when the hive has no declaration yet. An undecodable +/// value is an **error**, never treated as absent: overwriting a document +/// nobody can read discards the declarations of every other agent on that +/// hive, which is exactly what a fresh-start fallback would do quietly. +fn apply(current: Option<&[u8]>, agent: &str, state: AgentState) -> Result<(HiveWanted, Vec)> { + let mut declaration = match current { + Some(raw) => serde_json::from_slice::(raw) + .context("the hive's current declaration is not decodable")?, + None => HiveWanted::default(), + }; + declaration + .agents + .insert(agent.to_owned(), AgentWanted { state }); + let encoded = serde_json::to_vec(&declaration).context("encoding the new declaration")?; + Ok((declaration, encoded)) +} + +/// Writes the wanted-state bucket, and reads it back. +pub struct WantedWriter { + client: async_nats::Client, + store: tokio::sync::OnceCell, +} + +impl WantedWriter { + #[must_use] + pub fn new(client: async_nats::Client) -> Self { + Self { + client, + store: tokio::sync::OnceCell::new(), + } + } + + /// The bucket handle, created on first use if nothing has made it yet. + /// + /// Creation lives in [`swarm_queue_client::wanted`] because a bucket is + /// described identically by everyone who may create it. Only the + /// controller creates this one; a hive opens it read-only. + async fn store( + &self, + ) -> std::result::Result<&async_nats::jetstream::kv::Store, swarm_queue_client::Error> { + self.store + .get_or_try_init(|| swarm_queue_client::wanted::open_or_create(&self.client)) + .await + } + + /// The declaration currently published for `hive`, or `None`. + pub async fn view(&self, hive: &str) -> Result> { + // An unconnected client does not fail a JetStream request, it hangs + // on it — see `swarm_queue_client::ensure_connected`. + swarm_queue_client::ensure_connected(&self.client)?; + let store = self.store().await?; + let Some(entry) = store + .entry(hive) + .await + .with_context(|| format!("reading the declaration for {hive}"))? + else { + return Ok(None); + }; + serde_json::from_slice(&entry.value) + .map(Some) + .with_context(|| format!("the declaration for {hive} is not decodable")) + } + + /// Declare `agent` on `hive` to be in `state`, and return the whole + /// declaration as published. + /// + /// Read-modify-write against the entry's revision rather than a plain + /// `put`: the value is the hive's whole agent map, so a blind write + /// would drop a concurrent change to a different agent. The bucket keeps + /// one revision of history, so a lost write is not recoverable after the + /// fact — the conflict has to be caught here. + pub async fn set(&self, hive: &str, agent: &str, state: AgentState) -> Result { + swarm_queue_client::ensure_connected(&self.client)?; + let store = self.store().await?; + + for _ in 0..MAX_ATTEMPTS { + let entry = store + .entry(hive) + .await + .with_context(|| format!("reading the declaration for {hive}"))?; + let revision = entry.as_ref().map(|e| e.revision); + let (declaration, encoded) = + apply(entry.as_ref().map(|e| e.value.as_ref()), agent, state)?; + + // `update` and `create` have separate error types, and only one + // variant of each means "someone else got there first". Every + // other failure returns immediately: retrying a disconnect or a + // permission error would spin the loop and then report a + // conflict, blaming a concurrent writer that never existed. + let written = match revision { + Some(revision) => match store.update(hive, encoded.into(), revision).await { + Ok(_) => Wrote::Ok, + Err(e) if matches!(e.kind(), UpdateErrorKind::WrongLastRevision) => { + Wrote::LostRace + } + Err(e) => Wrote::Failed(anyhow::Error::new(e)), + }, + None => match store.create(hive, encoded.into()).await { + Ok(_) => Wrote::Ok, + Err(e) if matches!(e.kind(), CreateErrorKind::AlreadyExists) => Wrote::LostRace, + Err(e) => Wrote::Failed(anyhow::Error::new(e)), + }, + }; + match written { + Wrote::Ok => { + tracing::info!(hive, agent, ?state, "declared agent state"); + return Ok(declaration); + } + // Re-read and re-apply onto the winner's value, not over it. + Wrote::LostRace => { + tracing::debug!(hive, agent, "declaration write lost a race, retrying"); + } + Wrote::Failed(e) => { + return Err(e).with_context(|| format!("declaring {agent} on {hive}")); + } + } + } + anyhow::bail!("gave up declaring {agent} on {hive} after {MAX_ATTEMPTS} conflicting writes") + } +} + +#[cfg(test)] +mod tests { + use super::apply; + use swarm_queue_client::wanted::AgentState; + + #[test] + fn declaring_one_agent_preserves_every_other() { + let current = br#"{"agents":{"iris":{"state":"up"},"argus":{"state":"offline"}}}"#; + let (declaration, _) = apply(Some(current), "atlas", AgentState::Up).unwrap(); + assert_eq!(declaration.agents.len(), 3); + assert_eq!(declaration.agents["iris"].state, AgentState::Up); + assert_eq!(declaration.agents["argus"].state, AgentState::Offline); + assert_eq!(declaration.agents["atlas"].state, AgentState::Up); + } + + #[test] + fn redeclaring_an_agent_replaces_only_its_own_state() { + let current = br#"{"agents":{"iris":{"state":"up"},"atlas":{"state":"up"}}}"#; + let (declaration, _) = apply(Some(current), "atlas", AgentState::Offline).unwrap(); + assert_eq!(declaration.agents.len(), 2); + assert_eq!(declaration.agents["iris"].state, AgentState::Up); + assert_eq!(declaration.agents["atlas"].state, AgentState::Offline); + } + + #[test] + fn a_hive_with_no_declaration_yet_gets_a_one_agent_one() { + let (declaration, _) = apply(None, "atlas", AgentState::Up).unwrap(); + assert_eq!(declaration.agents.len(), 1); + assert_eq!(declaration.agents["atlas"].state, AgentState::Up); + } + + // The failure this function exists to prevent: a fresh-start fallback + // here would publish a one-agent document over a hive's whole set. + #[test] + fn an_undecodable_declaration_is_an_error_not_a_fresh_start() { + let err = apply(Some(b"{not json"), "atlas", AgentState::Up).unwrap_err(); + assert!( + err.to_string().contains("not decodable"), + "unexpected error: {err}" + ); + } + + #[test] + fn an_unknown_state_in_the_current_value_is_also_an_error() { + let current = br#"{"agents":{"iris":{"state":"sideways"}}}"#; + assert!(apply(Some(current), "atlas", AgentState::Up).is_err()); + } + + #[test] + fn the_encoded_form_round_trips() { + let (_, encoded) = apply(None, "atlas", AgentState::Offline).unwrap(); + let (again, _) = apply(Some(&encoded), "iris", AgentState::Up).unwrap(); + assert_eq!(again.agents["atlas"].state, AgentState::Offline); + assert_eq!(again.agents["iris"].state, AgentState::Up); + } +} diff --git a/swarm-queue-client/src/wanted.rs b/swarm-queue-client/src/wanted.rs index 60d8e114..f9d6a4ed 100644 --- a/swarm-queue-client/src/wanted.rs +++ b/swarm-queue-client/src/wanted.rs @@ -63,7 +63,7 @@ pub struct AgentWanted { /// than part of a declaration it only half understands. Adding a state means /// adding a variant here and shipping it to both ends — which is the intended /// workflow, not an obstacle to route around with a catch-all variant. -#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "snake_case")] pub enum AgentState { /// Exists on the hive and is running. @@ -72,12 +72,26 @@ pub enum AgentState { Offline, } +impl AgentState { + /// The wire spelling, for a reader that renders rather than decodes. + /// + /// Kept beside the enum so it cannot drift from the `rename_all` above; + /// a test pins the two together. + #[must_use] + pub fn as_str(self) -> &'static str { + match self { + AgentState::Up => "up", + AgentState::Offline => "offline", + } + } +} + /// Open the wanted-state bucket for writing, creating it if nothing has yet. /// /// **Controller-side only.** `history: 1` because a hive converges to the -/// current declaration and never asks what the previous one was — that -/// question is answered by the controller's own records, not by replaying a -/// bucket. +/// current declaration and never asks what the previous one was. The +/// controller keeps no second copy either: the value published here is the +/// record, and it reads it back from this bucket when it needs it. #[cfg(feature = "kv")] pub async fn open_or_create( client: &async_nats::Client, @@ -160,4 +174,22 @@ mod tests { fn a_missing_state_is_an_error() { assert!(serde_json::from_str::(r#"{"agents":{"a":{}}}"#).is_err()); } + + /// `as_str` and the `rename_all` are two spellings of one fact, and a + /// renderer that disagrees with the wire is worse than one that does not + /// exist. The `match` is what makes this exhaustive: a new variant fails + /// to compile here rather than quietly going untested. + #[test] + fn as_str_matches_the_serde_spelling() { + let every = [AgentState::Up, AgentState::Offline]; + for state in every { + match state { + AgentState::Up | AgentState::Offline => {} + } + assert_eq!( + serde_json::to_string(&state).expect("serialises"), + format!("\"{}\"", state.as_str()) + ); + } + } }