vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

swarm-controller: drop the Intent split, one unconditional wanted-state write

Browse source 
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).
This commit is contained in:
iris 2026-09-18 22:34:15 +02:00 • committed by mara
commit c908a71f3e
2 changed files with 87 additions and 270 deletions
Download patch file Download diff file Expand all files Collapse all files

49
swarm-controller/src/main.rs
View file

@ -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<AgentDeclaration>),
(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::<wanted::TerminalStateError>() {
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,

306
swarm-controller/src/wanted.rs
View file

@ -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<u8>)> {
/// 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<u8>)> {
let mut declaration = match current {
Some(raw) => serde_json::from_slice::<HiveWanted>(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<HiveWanted> {
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<HiveWanted> {
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<HiveWanted> {
pub async fn set(&self, hive: &str, agent: &str, state: AgentState) -> Result<HiveWanted> {
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<u8>)> {
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::<super::TerminalStateError>().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}"
);
}
}