From c908a71f3ecd4cce20f4d6384244c300764c5be8 Mon Sep 17 00:00:00 2001 From: iris Date: Fri, 18 Sep 2026 22:34:15 +0200 Subject: [PATCH] swarm-controller: drop the Intent split, one unconditional wanted-state write mara: "i dont want any logic differene between the two cases" and "do not refuse to recreate an agent". wanted.rs goes back to a single write path (set), with no terminal-state refusal at all, used identically by the pause/resume endpoint and the agent-creation job node. The agent-creation node's own idempotency requirement (re-running create against a name that already has a declaration must not silently pause it) now lives entirely in declare_new_agent: it reads the current declaration first and only writes Paused when the agent has no entry, or its entry is Destroyed (recreating a previously- destroyed name is the fresh deploy that state's own doc comment names as the way back). --- swarm-controller/src/main.rs | 49 ++++-- swarm-controller/src/wanted.rs | 306 ++++++--------------------------- 2 files changed, 86 insertions(+), 269 deletions(-) diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index f394f011..98139a21 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -327,25 +327,51 @@ async fn run_swarm_node( (builder, outcome) } -/// Declare a brand-new agent at [`NEW_AGENT_WANTED_STATE`]. +/// Declare a brand-new agent at [`NEW_AGENT_WANTED_STATE`] — unless it turns +/// out not to be new. /// /// A named function beside [`publish_deploy`] rather than the two lines /// inline, for the same reason that one is: `run_swarm_node`'s match is a /// per-variant index, and every arm that grows a body past a call pushes the /// next reader further from the variant they came to read. +/// +/// Reads the current declaration first and only calls +/// [`WantedWriter::set`](wanted::WantedWriter::set) when this agent has no +/// entry yet — `wanted.rs` itself no longer distinguishes creation from any +/// other write (mara: "i dont want any logic difference between the two +/// cases"), so the idempotency this node needs — calling create against a +/// name that already has a declaration (an operator migrating a pre-existing +/// agent into this bookkeeping, or a retried request) must not silently +/// pause an agent already running under some other state — lives entirely +/// here, the one caller it applies to. An existing `Destroyed` entry counts +/// as "no entry" for this check and gets written over: recreating a +/// previously-destroyed name is exactly the fresh deploy that state's own +/// doc comment names as the way back, and mara separately ruled "do not +/// refuse to recreate an agent". async fn declare_new_agent( writer: &wanted::WantedWriter, hive: &str, agent: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; + use swarm_queue_client::wanted::AgentState; - // `create`, not `set`: a name that was destroyed earlier still carries a - // terminal entry, and recreating the agent is the fresh deploy that - // entry's own doc comment names as the way back. `set` would refuse it, - // cancelling the deploy of an agent whose identity, repo and config the - // swarm has just built. - match writer.create(hive, agent, NEW_AGENT_WANTED_STATE).await { + let existing = match writer.view(hive).await { + Ok(declaration) => declaration.and_then(|d| d.agents.get(agent).map(|a| a.state)), + Err(e) => return Outcome::Failed(format!("{e:#}")), + }; + if let Some(state) = existing + && state != AgentState::Destroyed + { + tracing::debug!( + hive, + agent, + ?state, + "agent already declared, leaving its wanted state alone" + ); + return Outcome::Done; + } + match writer.set(hive, agent, NEW_AGENT_WANTED_STATE).await { Ok(_) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } @@ -894,7 +920,6 @@ fn declaration_target( 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 = 409, description = "the agent is already declared destroyed, a terminal state the request tries to move it off of (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), ), @@ -912,14 +937,6 @@ async fn set_agent_state( .into_string(); let declaration = writer.set(&hive, &agent, req.state).await.map_err(|e| { - // A `TerminalStateError` is the caller's mistake, but not a malformed - // request — the request is well-formed, it just conflicts with the - // target resource's current (terminal) state, which per RFC 9110 is - // what 409 exists for, not 400. Everything else here is the - // pre-existing catch-all. - if let Some(terminal) = e.downcast_ref::() { - return error_problem(axum::http::StatusCode::CONFLICT, &terminal.to_string()); - } tracing::warn!(hive = %hive, agent = %agent, error = %format!("{e:#}"), "declaring agent state failed"); error_problem( axum::http::StatusCode::INTERNAL_SERVER_ERROR, diff --git a/swarm-controller/src/wanted.rs b/swarm-controller/src/wanted.rs index a80eaf47..4f0d4cd7 100644 --- a/swarm-controller/src/wanted.rs +++ b/swarm-controller/src/wanted.rs @@ -12,42 +12,25 @@ //! 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 std::fmt; +//! +//! **One write path, no per-caller behaviour.** An earlier revision of this +//! module carried an `Intent` enum that let the agent-creation job node +//! write over a `Destroyed` entry while the ordinary `PUT .../state` +//! endpoint could not — mara's ruling on that: "i dont want any logic +//! difference between the two cases" (plus "do not refuse to recreate an +//! agent", which the old refusal made impossible for *either* caller). So +//! [`WantedWriter::set`] just writes whatever it is told, unconditionally, +//! for every caller alike. A caller that wants "don't touch an agent that +//! already has a declaration" (the agent-creation node's own idempotency +//! requirement — recreating an already-`Up` agent must not silently pause +//! it) makes that decision itself, by reading the current declaration via +//! [`WantedWriter::view`] first — that policy lives with the one caller it +//! applies to, not inside this module's shared write. use anyhow::{Context, Result}; use async_nats::jetstream::kv::{CreateErrorKind, UpdateErrorKind}; use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted}; -/// The one failure out of [`apply`] that is the caller's mistake, not a -/// server fault — worth a distinct type so `set_agent_state` can tell it -/// apart from every other error this module raises (an undecodable -/// declaration, a lost write) and answer 400 instead of 500. -/// -/// Carried through as a plain `anyhow::Error` like every other error here -/// (this crate has no typed-error convention to join), and recovered at the -/// HTTP boundary with `downcast_ref` — the shape every other "kind of error -/// that came from underneath" check in this module already uses -/// ([`Wrote`]'s `UpdateErrorKind`/`CreateErrorKind` matching), just inspecting -/// an `anyhow::Error` instead of a client library's own error enum. -#[derive(Debug)] -pub struct TerminalStateError { - agent: String, -} - -impl fmt::Display for TerminalStateError { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!( - f, - "{} is already declared destroyed — terminal, per AgentState::Destroyed's own doc \ - comment: no transition brings it back short of a fresh deploy", - self.agent - ) - } -} - -impl std::error::Error for TerminalStateError {} - /// The three outcomes of one write attempt, which the two KV verbs report /// through separate error types. enum Wrote { @@ -57,28 +40,6 @@ enum Wrote { Failed(anyhow::Error), } -/// Why a caller is declaring a state, which is what decides whether a -/// `Destroyed` entry stands in its way. -/// -/// The distinction is the one [`AgentState::Destroyed`]'s own doc comment -/// already promises — "no state that brings a destroyed agent back **short of -/// a fresh deploy**". Redeclaring is not a fresh deploy and must keep being -/// refused; creating the agent again *is* one, and the whole of its identity, -/// repo and config has just been built anew alongside this declaration. -/// -/// An enum rather than a bool because the two call sites read as opposites -/// only if the argument says which one it is — `apply(.., true)` at the create -/// path would be a flag nobody can check at a glance. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum Intent { - /// An operator (or anything else) changing what an existing agent should - /// be doing. Cannot move off `Destroyed`. - Redeclare, - /// The agent is being created from nothing — a fresh deploy. Replaces a - /// `Destroyed` entry left behind by an agent that had this name before. - Create, -} - /// 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 @@ -98,54 +59,14 @@ const MAX_ATTEMPTS: usize = 5; /// nobody can read discards the declarations of every other agent on that /// hive, which is exactly what a fresh-start fallback would do quietly. /// -/// `intent` decides whether an existing `Destroyed` entry for this agent is a -/// wall or something to write over — see [`Intent`]. -fn apply( - current: Option<&[u8]>, - agent: &str, - state: AgentState, - intent: Intent, -) -> Result<(HiveWanted, Vec)> { +/// Unconditional otherwise — see the module doc for why there is no +/// terminal-state refusal here any more. +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(), }; - // Redeclaring the same terminal state is a no-op, not a transition — a - // client that already sees the agent as destroyed and asks again should - // not be punished for it. Anything else moving off `Destroyed` is the - // one transition this module exists to refuse — unless the caller is - // creating the agent from nothing, which is the "fresh deploy" carve-out - // `Intent` documents. - if intent == Intent::Redeclare - && let Some(existing) = declaration.agents.get(agent) - && existing.state == AgentState::Destroyed - && state != AgentState::Destroyed - { - return Err(TerminalStateError { - agent: agent.to_owned(), - } - .into()); - } - // `Create` against a name that already has a *non-terminal* declaration - // is not a creation — it's the create-agent endpoint being asked to - // declare a name that turns out to already exist (a retried request, or - // an operator migrating a pre-existing agent into this bookkeeping for - // the first time). Creation is expected to be idempotent, so this is a - // no-op that leaves the existing declaration exactly as it stood rather - // than stomping it back to whatever state the create path passes — - // otherwise calling create against a name that is already `Up` would - // silently pause a running agent. `Destroyed` is excluded on purpose: - // that's the one case a fresh create *does* need to write over (see - // `Intent::Create`'s own doc comment). - if intent == Intent::Create - && let Some(existing) = declaration.agents.get(agent) - && existing.state != AgentState::Destroyed - { - let encoded = - serde_json::to_vec(&declaration).context("encoding the current declaration")?; - return Ok((declaration, encoded)); - } declaration .agents .insert(agent.to_owned(), AgentWanted { state }); @@ -201,53 +122,21 @@ impl WantedWriter { .with_context(|| format!("the declaration for {hive} is not decodable")) } - /// Redeclare `agent` on `hive` to be in `state`, and return the whole + /// Declare `agent` on `hive` to be in `state`, and return the whole /// declaration as published. /// - /// Refuses to move an agent off `Destroyed` — see - /// [`create`](Self::create) for the one caller that may. - pub async fn set(&self, hive: &str, agent: &str, state: AgentState) -> Result { - self.write(hive, agent, state, Intent::Redeclare).await - } - - /// Declare a *newly created* `agent` on `hive`, and return the whole - /// declaration as published. - /// - /// Same write as [`set`](Self::set) but with [`Intent::Create`], so a - /// `Destroyed` entry left by an earlier agent of this name does not - /// refuse it. Recreating a destroyed name is precisely the "fresh deploy" - /// [`AgentState::Destroyed`]'s doc comment names as the way back, and by - /// the time this runs the swarm has already built that agent's identity, - /// repo and config from nothing. - /// - /// Idempotent against any other existing declaration: an agent already - /// declared in a non-terminal state (an operator migrating a pre-existing - /// agent into this bookkeeping, or a retried request) is left exactly as - /// it stood — creation must not silently pause an agent already running - /// under some other wanted state. - /// - /// A separate method rather than an `Intent` parameter on `set`: the - /// caller that may overwrite a terminal state is the agent-creation job - /// node and nothing else, and an argument every other caller has to pass - /// correctly is an argument one of them eventually passes wrong. - pub async fn create(&self, hive: &str, agent: &str, state: AgentState) -> Result { - self.write(hive, agent, state, Intent::Create).await - } - - /// The write both of the above are. + /// Unconditional — see the module doc. Every caller (the pause/resume + /// endpoint, the agent-creation job node, anything else) gets the exact + /// same write; a caller that needs to decide *whether* to call this at + /// all (idempotency, terminal-state handling, …) makes that call itself + /// by reading [`view`](Self::view) first. /// /// 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. - async fn write( - &self, - hive: &str, - agent: &str, - state: AgentState, - intent: Intent, - ) -> Result { + pub async fn set(&self, hive: &str, agent: &str, state: AgentState) -> Result { swarm_queue_client::ensure_connected(&self.client)?; let store = self.store(hive).await?; @@ -257,17 +146,16 @@ impl WantedWriter { .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, - intent, - )?; + let (declaration, encoded) = + apply(entry.as_ref().map(|e| e.value.as_ref()), agent, state)?; - // `apply`'s `Intent::Create` no-op (an already-declared, - // non-terminal agent) re-encodes the declaration unchanged — - // skip the network round-trip for it entirely rather than - // writing identical bytes back. + // A redeclare of the identical state re-encodes the declaration + // unchanged (a caller that already sees this state and asks + // again is not doing anything) — skip the network round-trip for + // it entirely rather than writing identical bytes back. + // `HiveWanted::agents` is a `BTreeMap`, so this byte comparison + // is deterministic, not vulnerable to a map re-ordering itself + // between reads. if entry.as_ref().map(|e| e.value.as_ref()) == Some(encoded.as_slice()) { tracing::debug!(hive, agent, "declaration already current, no write needed"); return Ok(declaration); @@ -312,24 +200,13 @@ impl WantedWriter { #[cfg(test)] mod tests { - use super::{Intent, apply}; - use swarm_queue_client::wanted::{AgentState, HiveWanted}; - - /// Every case below that predates the create carve-out is a redeclare; - /// spelling the intent out in each of them would say nothing the test - /// name doesn't. The carve-out's own tests call `apply` directly. - fn redeclare( - current: Option<&[u8]>, - agent: &str, - state: AgentState, - ) -> anyhow::Result<(HiveWanted, Vec)> { - apply(current, agent, state, Intent::Redeclare) - } + 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, _) = redeclare(Some(current), "atlas", AgentState::Up).unwrap(); + 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); @@ -339,7 +216,7 @@ mod tests { #[test] fn redeclaring_an_agent_replaces_only_its_own_state() { let current = br#"{"agents":{"iris":{"state":"up"},"atlas":{"state":"up"}}}"#; - let (declaration, _) = redeclare(Some(current), "atlas", AgentState::Offline).unwrap(); + 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); @@ -347,7 +224,7 @@ mod tests { #[test] fn a_hive_with_no_declaration_yet_gets_a_one_agent_one() { - let (declaration, _) = redeclare(None, "atlas", AgentState::Up).unwrap(); + let (declaration, _) = apply(None, "atlas", AgentState::Up).unwrap(); assert_eq!(declaration.agents.len(), 1); assert_eq!(declaration.agents["atlas"].state, AgentState::Up); } @@ -356,7 +233,7 @@ mod tests { // 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 = redeclare(Some(b"{not json"), "atlas", AgentState::Up).unwrap_err(); + let err = apply(Some(b"{not json"), "atlas", AgentState::Up).unwrap_err(); assert!( err.to_string().contains("not decodable"), "unexpected error: {err}" @@ -366,115 +243,38 @@ mod tests { #[test] fn an_unknown_state_in_the_current_value_is_also_an_error() { let current = br#"{"agents":{"iris":{"state":"sideways"}}}"#; - assert!(redeclare(Some(current), "atlas", AgentState::Up).is_err()); + assert!(apply(Some(current), "atlas", AgentState::Up).is_err()); } #[test] fn the_encoded_form_round_trips() { - let (_, encoded) = redeclare(None, "atlas", AgentState::Offline).unwrap(); - let (again, _) = redeclare(Some(&encoded), "iris", AgentState::Up).unwrap(); + 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); } - /// The bug this module exists to fix: nothing used to stop a caller from - /// declaring a destroyed agent back to `Up`/`Offline`, silently - /// un-terminaling a state whose own doc comment says that is impossible. + /// No terminal-state refusal any more (mara: "do not refuse to recreate + /// an agent") — `apply` writes whatever it is asked to, including moving + /// a `Destroyed` agent to any other state. Whether that write should + /// happen at all is now the caller's own decision (see the module doc). #[test] - fn moving_a_destroyed_agent_to_any_other_state_is_refused() { + fn a_destroyed_agent_can_be_declared_into_any_other_state() { let current = br#"{"agents":{"atlas":{"state":"destroyed"}}}"#; for state in [AgentState::Up, AgentState::Offline, AgentState::Paused] { - let err = redeclare(Some(current), "atlas", state).unwrap_err(); - assert!( - err.downcast_ref::().is_some(), - "expected a TerminalStateError for {state:?}, got: {err}" - ); + let (declaration, _) = apply(Some(current), "atlas", state).unwrap(); + assert_eq!(declaration.agents["atlas"].state, state); } } - /// Redeclaring `Destroyed` while already `Destroyed` is a no-op, not a - /// transition — a client that already sees the terminal state and asks - /// again should get success back, not an error for stating a fact. - #[test] - fn redeclaring_destroyed_as_destroyed_is_not_a_transition() { - let current = br#"{"agents":{"atlas":{"state":"destroyed"}}}"#; - let (declaration, _) = redeclare(Some(current), "atlas", AgentState::Destroyed).unwrap(); - assert_eq!(declaration.agents["atlas"].state, AgentState::Destroyed); - } - - /// The refusal is per-agent — a destroyed agent on the same hive as a - /// live one must not block declaring the live one. + /// The write is per-agent — a destroyed agent on the same hive as a live + /// one must not block declaring the live one, and declaring one must not + /// disturb the other. #[test] fn a_destroyed_agent_does_not_block_declaring_a_different_one() { let current = br#"{"agents":{"atlas":{"state":"destroyed"},"iris":{"state":"up"}}}"#; - let (declaration, _) = redeclare(Some(current), "iris", AgentState::Offline).unwrap(); + let (declaration, _) = apply(Some(current), "iris", AgentState::Offline).unwrap(); assert_eq!(declaration.agents["atlas"].state, AgentState::Destroyed); assert_eq!(declaration.agents["iris"].state, AgentState::Offline); } - - /// The carve-out `AgentState::Destroyed`'s doc comment already promised - /// and nothing implemented: a *fresh deploy* does bring a destroyed name - /// back. Without this, creating an agent under a name that was destroyed - /// earlier builds its identity, repo and config and then fails the - /// declaration — cancelling the deploy while the operator sees a 200. - #[test] - fn creating_an_agent_may_reuse_a_destroyed_name() { - let current = br#"{"agents":{"atlas":{"state":"destroyed"}}}"#; - let (declaration, _) = - apply(Some(current), "atlas", AgentState::Paused, Intent::Create).unwrap(); - assert_eq!(declaration.agents["atlas"].state, AgentState::Paused); - } - - /// The carve-out is scoped to the agent being created — recreating one - /// name must not disturb another destroyed agent on the same hive. - #[test] - fn creating_an_agent_leaves_other_destroyed_agents_destroyed() { - let current = br#"{"agents":{"atlas":{"state":"destroyed"},"iris":{"state":"destroyed"}}}"#; - let (declaration, _) = - apply(Some(current), "atlas", AgentState::Paused, Intent::Create).unwrap(); - assert_eq!(declaration.agents["atlas"].state, AgentState::Paused); - assert_eq!(declaration.agents["iris"].state, AgentState::Destroyed); - } - - /// mara: "this does not match the expectation that agent creation is - /// idempotent so pre existing agents can be migrated" — an agent already - /// declared in some other, non-terminal state (say an operator migrating - /// a pre-existing agent into this bookkeeping) must not be stomped back - /// to whatever state the create path passes. - #[test] - fn creating_an_agent_does_not_disturb_an_existing_non_terminal_declaration() { - let current = br#"{"agents":{"atlas":{"state":"up"}}}"#; - let (declaration, _) = - apply(Some(current), "atlas", AgentState::Paused, Intent::Create).unwrap(); - assert_eq!(declaration.agents["atlas"].state, AgentState::Up); - } - - /// The idempotent no-op only inspects the agent being created — it must - /// not touch any other agent's declaration on the same hive. - #[test] - fn creating_an_agent_that_already_exists_leaves_every_other_agent_alone() { - let current = br#"{"agents":{"atlas":{"state":"up"},"iris":{"state":"offline"}}}"#; - let (declaration, _) = - apply(Some(current), "atlas", AgentState::Paused, Intent::Create).unwrap(); - assert_eq!(declaration.agents["atlas"].state, AgentState::Up); - assert_eq!(declaration.agents["iris"].state, AgentState::Offline); - } - - /// `Intent::Create` relaxes exactly one rule. An undecodable current - /// value is still an error, for the same reason it is on a redeclare: - /// writing over it discards every other agent on the hive. - #[test] - fn creating_an_agent_still_refuses_an_undecodable_declaration() { - let err = apply( - Some(b"{not json"), - "atlas", - AgentState::Paused, - Intent::Create, - ) - .unwrap_err(); - assert!( - err.to_string().contains("not decodable"), - "unexpected error: {err}" - ); - } }