diff --git a/hive-c0re/src/workers/wanted.rs b/hive-c0re/src/workers/wanted.rs index 6fd39393..760a40fb 100644 --- a/hive-c0re/src/workers/wanted.rs +++ b/hive-c0re/src/workers/wanted.rs @@ -171,7 +171,7 @@ fn plan( let Some(intent) = intents.get(agent) else { continue; }; - match decide(decl.state, present.contains(agent), *intent) { + match decide(&decl.state, present.contains(agent), *intent) { Converge::Deploy => plan.deploy.push(agent.clone()), Converge::Start => plan.start.push(agent.clone()), Converge::Stop => plan.stop.push(agent.clone()), @@ -196,7 +196,7 @@ enum Converge { /// The whole decision, pure: no queue, no store, no container. The three /// inputs are exactly what [`converge`] reads per agent, so the table below is /// the behaviour rather than a model of it. -fn decide(state: AgentState, present: bool, intent: Option) -> Converge { +fn decide(state: &AgentState, present: bool, intent: Option) -> Converge { match state { AgentState::Up if !present => Converge::Deploy, AgentState::Up => { @@ -231,7 +231,14 @@ mod tests { HiveWanted { agents: agents .iter() - .map(|(name, state)| ((*name).to_owned(), AgentWanted { state: *state })) + .map(|(name, state)| { + ( + (*name).to_owned(), + AgentWanted { + state: state.clone(), + }, + ) + }) .collect(), } } @@ -285,10 +292,10 @@ mod tests { #[test] fn a_declared_agent_this_hive_does_not_have_is_deployed() { - assert_eq!(decide(AgentState::Up, false, None), Converge::Deploy); + assert_eq!(decide(&AgentState::Up, false, None), Converge::Deploy); // Even holding an intent: a row can outlive its container. assert_eq!( - decide(AgentState::Up, false, Some(Wanted::Up)), + decide(&AgentState::Up, false, Some(Wanted::Up)), Converge::Deploy ); } @@ -296,11 +303,11 @@ mod tests { #[test] fn a_hive_that_already_agrees_queues_nothing() { assert_eq!( - decide(AgentState::Up, true, Some(Wanted::Up)), + decide(&AgentState::Up, true, Some(Wanted::Up)), Converge::Nothing ); assert_eq!( - decide(AgentState::Offline, true, Some(Wanted::Offline)), + decide(&AgentState::Offline, true, Some(Wanted::Offline)), Converge::Nothing ); } @@ -308,11 +315,11 @@ mod tests { #[test] fn a_disagreeing_intent_is_converged_both_ways() { assert_eq!( - decide(AgentState::Up, true, Some(Wanted::Offline)), + decide(&AgentState::Up, true, Some(Wanted::Offline)), Converge::Start ); assert_eq!( - decide(AgentState::Offline, true, Some(Wanted::Up)), + decide(&AgentState::Offline, true, Some(Wanted::Up)), Converge::Stop ); } @@ -322,15 +329,15 @@ mod tests { /// to be doing. #[test] fn an_agent_with_no_intent_row_is_converged() { - assert_eq!(decide(AgentState::Up, true, None), Converge::Start); - assert_eq!(decide(AgentState::Offline, true, None), Converge::Stop); + assert_eq!(decide(&AgentState::Up, true, None), Converge::Start); + assert_eq!(decide(&AgentState::Offline, true, None), Converge::Stop); } /// Declared offline, no container: the one absent case that must not /// deploy. Its control is the `Up` case above, which must. #[test] fn an_absent_agent_declared_offline_is_left_absent() { - assert_eq!(decide(AgentState::Offline, false, None), Converge::Nothing); + assert_eq!(decide(&AgentState::Offline, false, None), Converge::Nothing); } /// Version skew is handled one layer up, at the decode: `AgentState` is @@ -343,8 +350,8 @@ mod tests { fn every_state_this_build_knows_is_covered_above() { for state in [AgentState::Up, AgentState::Offline] { let seen = [true, false].iter().any(|present| { - decide(state, *present, None) != Converge::Nothing - || decide(state, *present, Some(Wanted::Up)) != Converge::Nothing + decide(&state, *present, None) != Converge::Nothing + || decide(&state, *present, Some(Wanted::Up)) != Converge::Nothing }); assert!(seen, "{state:?} produces no action in any combination"); } diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 54aff33c..3cda4121 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -47,7 +47,6 @@ 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 @@ -430,12 +429,6 @@ 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 @@ -688,150 +681,6 @@ 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. /// @@ -1479,7 +1328,6 @@ 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, @@ -1489,19 +1337,6 @@ 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)) @@ -1514,8 +1349,6 @@ fn build_app(state: AppState) -> axum::Router { .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)) @@ -1525,12 +1358,15 @@ fn build_app(state: AppState) -> axum::Router { // 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`. - router + let app = router .route( "/api/openapi.json", get(move || async move { Json(api.clone()) }), ) - .with_state(state) + .with_state(state); + axum::serve(listener, app) + .await + .context("serving swarm-controller") } #[cfg(test)] @@ -1619,9 +1455,6 @@ 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 deleted file mode 100644 index b5d2c7ba..00000000 --- a/swarm-controller/src/wanted.rs +++ /dev/null @@ -1,216 +0,0 @@ -//! 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 f9d6a4ed..60d8e114 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, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "snake_case")] pub enum AgentState { /// Exists on the hive and is running. @@ -72,26 +72,12 @@ 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. 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. +/// 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. #[cfg(feature = "kv")] pub async fn open_or_create( client: &async_nats::Client, @@ -174,22 +160,4 @@ 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()) - ); - } - } }