diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 8f673991..5594654e 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -105,12 +105,6 @@ enum SwarmNodeKind { /// its hive's shared queue credential, so the document cannot be rendered /// without knowing which hive the agent belongs to. MintAgentIdentity { hive: String, agent: String }, - /// Declare `agent` on `hive` as `Paused` in the swarm's wanted-state - /// store, so a freshly created agent does not start driving turns the - /// moment it's deployed — the operator has to explicitly flip it to `Up`. - /// Carries the hive for the same reason `TriggerDeploy`/`MintAgentIdentity` - /// do: the wanted-state bucket is keyed per hive. - SetAgentWanted { hive: String, agent: String }, /// Tell `hive` to rebuild `agent`, by publishing on the swarm's deploy /// subject. The one node kind whose effect leaves this host. /// @@ -130,7 +124,6 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(), SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(), SwarmNodeKind::MintAgentIdentity { .. } => "mint_agent_identity".to_owned(), - SwarmNodeKind::SetAgentWanted { .. } => "set_agent_wanted".to_owned(), SwarmNodeKind::TriggerDeploy { .. } => "trigger_deploy".to_owned(), } } @@ -151,8 +144,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { serde_json::json!({ "agent": agent }) } SwarmNodeKind::TriggerDeploy { hive, agent } - | SwarmNodeKind::MintAgentIdentity { hive, agent } - | SwarmNodeKind::SetAgentWanted { hive, agent } => { + | SwarmNodeKind::MintAgentIdentity { hive, agent } => { serde_json::json!({ "agent": agent, "hive": hive }) } } @@ -194,22 +186,8 @@ struct WorkerDeps { /// a per-node chance to read one. `None` on a host the operator has not /// given an authority — see `agent_identity::Authority::from_env`. agent_ca: Option>, - /// The wanted-state writer, for nodes that declare what a hive should - /// converge an agent to. Shares the status reader's queue connection — - /// see `wanted_writer`. `None` exactly when no swarm queue is configured - /// on this host, same as `queue`. - wanted: Option>, } -/// What [`SwarmNodeKind::SetAgentWanted`] declares a brand-new agent to be. -/// -/// A named constant rather than the literal inline, because this *is* the -/// decision the node exists to carry — writing a declaration at all is the -/// mechanism, `Paused` is the policy — and a test can pin a name where it -/// cannot pin a value written into a KV bucket no unit test has a server for. -const NEW_AGENT_WANTED_STATE: swarm_queue_client::wanted::AgentState = - swarm_queue_client::wanted::AgentState::Paused; - /// 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. @@ -308,14 +286,6 @@ async fn run_swarm_node( } } }, - SwarmNodeKind::SetAgentWanted { hive, agent } => match deps.wanted { - None => Outcome::Failed( - "no swarm queue is configured on this host, so no wanted-state \ - declaration can be written" - .to_owned(), - ), - Some(writer) => declare_new_agent(&writer, &hive, &agent).await, - }, SwarmNodeKind::TriggerDeploy { hive, agent } => match deps.queue { None => Outcome::Failed( "no swarm queue is configured on this host, so no hive can be told to deploy" @@ -327,41 +297,6 @@ async fn run_swarm_node( (builder, outcome) } -/// Declare a brand-new agent at [`NEW_AGENT_WANTED_STATE`] — unless it turns -/// out not to be new: an agent that already has a declaration (other than -/// `Destroyed`, which this treats as reusable) is left alone, so a retried -/// or migrating create is idempotent rather than pausing an already-running -/// agent. `wanted.rs` itself has no such carve-out, so this node makes the -/// call. -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; - - 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:#}")), - } -} - /// Publish one deploy request and report whether it left this process. /// /// The flush is not belt-and-braces: `publish` hands the message to the @@ -905,6 +840,7 @@ 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), ), @@ -922,6 +858,14 @@ 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, @@ -1372,15 +1316,6 @@ fn declare_agent_job( agent: agent.to_owned(), }) .after_ok(create_identity); - // Declared before the deploy trigger so the pause is visible in the - // wanted-state store before the hive brings the container up — see the - // node's own doc comment for why "before", not just "eventually". Needs - // no parent of its own: the declaration is written from the hive/agent - // names known at request time, same as `create_forge_user`. - let set_wanted = b.node(SwarmNodeKind::SetAgentWanted { - hive: hive.to_owned(), - agent: agent.to_owned(), - }); // Last, and specifically after the config repo is seeded: the hive // deploys by reading that repo, so a deploy asked for any earlier would // find nothing to build. This is the edge that makes creating an agent at @@ -1392,21 +1327,13 @@ fn declare_agent_job( // overtake the mint — but a host with no authority configured must still // create agents exactly as it does today. `after_ok` there would turn an // unconfigured option into an agent nobody runs. - // - // `after_ok` on the declaration, though: an agent deployed without its - // pause landing first is the race this node exists to close, so a - // declaration that did not land must not be deployed past. A host with no - // queue at all fails this node — but `TriggerDeploy` has nothing to - // publish to on such a host either, so the strict edge cancels a node - // that could not have succeeded anyway. let _trigger_deploy = b .node(SwarmNodeKind::TriggerDeploy { hive: hive.to_owned(), agent: agent.to_owned(), }) .after_ok(init_config) - .after_any(mint_identity) - .after_ok(set_wanted); + .after_any(mint_identity); vec![create_identity.guid()] } @@ -1737,7 +1664,6 @@ async fn main() -> Result<()> { forge: forge_client.clone(), queue: status.as_ref().map(|s| s.queue_client()), agent_ca: load_agent_authority(), - wanted: wanted_writer(status.as_ref()), }; let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new( @@ -1956,9 +1882,8 @@ mod tests { }]), 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. + // No queue, for the same reason as `status`: these tests drive + // agent creation, which publishes no declaration. wanted: None, agent_status: None, jobq: std::sync::Arc::clone(&sched), @@ -2106,7 +2031,6 @@ mod tests { forge: None, queue: None, agent_ca: None, - wanted: None, }; let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { @@ -2160,7 +2084,6 @@ mod tests { forge: None, queue: None, agent_ca: None, - wanted: None, }; let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { @@ -2214,7 +2137,6 @@ mod tests { forge: None, queue: None, agent_ca: None, - wanted: None, }; let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { @@ -2237,137 +2159,6 @@ mod tests { ); } - /// Fourth sibling, same shape and the same caveat once more: with no - /// queue configured this reaches only the graceful-absence-is-failure - /// branch. Writing the declaration for real needs a `JetStream` server, - /// which no test in this crate has — `wanted::apply`'s own unit tests - /// cover what gets written, and [`a_new_agent_is_declared_paused`] below - /// pins the state this node asks for. - #[tokio::test] - async fn set_agent_wanted_node_runs_end_to_end_and_fails_without_a_queue() { - let mut sched = hive_jobq::scheduler::Scheduler::new( - hive_jobq::Graph::new(), - hive_jobq::resources::ResourceTable::new(), - ); - let id = sched - .append( - SwarmNodeKind::SetAgentWanted { - hive: "pr1ma".to_owned(), - agent: "atlas".to_owned(), - }, - Vec::new(), - None, - ) - .expect("insert"); - let sched = std::sync::Arc::new(std::sync::Mutex::new(sched)); - - let deps = WorkerDeps { - auth: None, - forge: None, - queue: None, - agent_ca: None, - wanted: None, - }; - let runner = - hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { - run_swarm_node(id, kind, builder, deps) - }) - .expect("the node just inserted is runnable"); - runner - .await - .1 - .expect("no growth declared, nothing to reject"); - - let guard = sched.lock().unwrap(); - let node = guard.graph().node(id).expect("node still present"); - assert_eq!(node.state, hive_jobq::State::Failed); - assert!( - node.error.as_deref().unwrap_or_default().contains("queue"), - "expected a no-queue-configured error, got {:?}", - node.error - ); - } - - /// The whole point of the node: a freshly created agent is declared - /// `Paused`, so it does not start driving turns the moment its hive - /// brings it up. Any other state here is the bug this fixes. - #[test] - fn a_new_agent_is_declared_paused() { - assert_eq!( - crate::NEW_AGENT_WANTED_STATE, - swarm_queue_client::wanted::AgentState::Paused - ); - } - - /// Same reasoning as `a_mint_node_renders_both_the_agent_and_the_hive`: - /// this variant carries a hive too, and has to have joined `data`'s - /// two-field or-pattern rather than the agent-only one. - #[test] - fn a_set_wanted_node_renders_both_the_agent_and_the_hive() { - use hive_jobq_wire::WireNode as _; - - let kind = SwarmNodeKind::SetAgentWanted { - hive: "pr1ma".to_owned(), - agent: "atlas".to_owned(), - }; - assert_eq!(kind.label(), "set_agent_wanted"); - let data = kind.data(1); - assert_eq!(data["agent"], "atlas"); - assert_eq!(data["hive"], "pr1ma"); - } - - /// The ordering the fix depends on: the pause has to be in the - /// wanted-state store *before* the hive is told to deploy, or a freshly - /// created agent runs undeclared for the window between the two. - /// - /// Asserted on the graph `create_agent` builds, same as the mint edge's - /// test above and for the same reason — the edge is one line in a - /// builder closure whose absence changes nothing observable until a real - /// agent boots and starts taking turns nobody asked for. - #[tokio::test] - async fn the_deploy_waits_for_the_pause_to_be_declared() { - use hive_jobq_wire::WireNode as _; - - let (state, sched) = state_with_roster(); - let _queued = super::create_agent( - axum::extract::State(state), - axum::Json(super::CreateAgentRequest { - name: "atlas".to_owned(), - hive: "pr1ma".to_owned(), - }), - ) - .await - .expect("a hive in the roster must be accepted"); - - let guard = sched - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let graph = guard.graph(); - let id_of = |label: &str| { - graph - .nodes() - .find(|n| n.payload.label() == label) - .unwrap_or_else(|| panic!("the graph holds a {label} node")) - .id - }; - let set_wanted = id_of("set_agent_wanted"); - let deploy = graph.node(id_of("trigger_deploy")).expect("just found"); - - let when = deploy - .deps - .iter() - .find_map(|d| match d { - hive_jobq::Dep::Node { id, when } if *id == set_wanted => Some(*when), - _ => None, - }) - .expect("the deploy waits for the declaration"); - assert!( - !when.accepts(hive_jobq::TerminalState::Failed), - "a declaration that never landed must not be deployed past; this \ - edge has to be `after_ok`, not `after_any`" - ); - } - /// The node carries the hive, and the viewer has to see it. The `data` /// match is an or-pattern on purpose (see its own comment), and this is /// the assertion that the new variant joined the two-field arm rather diff --git a/swarm-controller/src/wanted.rs b/swarm-controller/src/wanted.rs index 254ba5cd..6a4d0863 100644 --- a/swarm-controller/src/wanted.rs +++ b/swarm-controller/src/wanted.rs @@ -12,17 +12,42 @@ //! 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. -//! -//! [`WantedWriter::set`] writes unconditionally, for every caller alike — no -//! per-caller `Intent`, no terminal-state refusal. A caller that needs to -//! decide *whether* to write at all (the agent-creation node's own -//! idempotency check) reads [`WantedWriter::view`] first and makes that call -//! itself. + +use std::fmt; 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 { @@ -42,15 +67,33 @@ const MAX_ATTEMPTS: usize = 5; /// Apply one agent's declared state to a hive's current declaration. /// -/// `current` is the hive's whole agent map — an undecodable value is an -/// **error**, not treated as absent, so a bad read never discards every -/// other agent's declaration under a fresh-start fallback. +/// 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(), }; + // 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. + if let Some(existing) = declaration.agents.get(agent) + && existing.state == AgentState::Destroyed + && state != AgentState::Destroyed + { + return Err(TerminalStateError { + agent: agent.to_owned(), + } + .into()); + } declaration .agents .insert(agent.to_owned(), AgentWanted { state }); @@ -69,15 +112,31 @@ impl WantedWriter { Self { client } } + /// One hive's 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 these; a hive opens its own read-only. + /// + /// Resolved per call rather than cached: there is one bucket per hive, so + /// a single cached handle cannot serve them, and the declarations this + /// writes change on operator action rather than on a loop — the extra + /// lookup is per *declaration*, not per tick. A cache here would be a map + /// whose invalidation nobody needs yet. + async fn store( + &self, + hive: &str, + ) -> std::result::Result { + swarm_queue_client::wanted::open_or_create(&self.client, hive).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)?; - // One bucket per hive, resolved per call, created on first use — see - // `swarm_queue_client::wanted` for why (only the controller creates - // these; a hive opens its own read-only). - let store = swarm_queue_client::wanted::open_or_create(&self.client, hive).await?; + let store = self.store(hive).await?; let Some(entry) = store .entry(hive) .await @@ -90,15 +149,17 @@ impl WantedWriter { .with_context(|| format!("the declaration for {hive} is not decodable")) } - /// Declare `agent` on `hive` to be in `state`, unconditionally (see the - /// module doc), and return the whole declaration as published. + /// Declare `agent` on `hive` to be in `state`, and return the whole + /// declaration as published. /// - /// Read-modify-write against the entry's revision, not 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. + /// 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 = swarm_queue_client::wanted::open_or_create(&self.client, hive).await?; + let store = self.store(hive).await?; for _ in 0..MAX_ATTEMPTS { let entry = store @@ -109,18 +170,6 @@ impl WantedWriter { let (declaration, encoded) = apply(entry.as_ref().map(|e| e.value.as_ref()), agent, state)?; - // 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); - } - // `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 @@ -214,22 +263,33 @@ mod tests { assert_eq!(again.agents["iris"].state, AgentState::Up); } - /// 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). + /// 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. #[test] - fn a_destroyed_agent_can_be_declared_into_any_other_state() { + fn moving_a_destroyed_agent_to_any_other_state_is_refused() { let current = br#"{"agents":{"atlas":{"state":"destroyed"}}}"#; for state in [AgentState::Up, AgentState::Offline, AgentState::Paused] { - let (declaration, _) = apply(Some(current), "atlas", state).unwrap(); - assert_eq!(declaration.agents["atlas"].state, state); + let err = apply(Some(current), "atlas", state).unwrap_err(); + assert!( + err.downcast_ref::().is_some(), + "expected a TerminalStateError for {state:?}, got: {err}" + ); } } - /// 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. + /// 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, _) = apply(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. #[test] fn a_destroyed_agent_does_not_block_declaring_a_different_one() { let current = br#"{"agents":{"atlas":{"state":"destroyed"},"iris":{"state":"up"}}}"#;