Watch
0
0
Fork
You've already forked hyperhive
0

Compare commits

...
Author SHA1 Message Date
iris
8aafe4eaee swarm-controller: trim oversized comments, drop a trivial wrapper fn
mara: "pls dont make comments longer than functions or fns that just
call a single other fn". Trimmed wanted.rs's module doc, apply()'s
doc, and declare_new_agent's doc down to what's non-obvious; deleted
WantedWriter::store (a one-line call to open_or_create with a
10-line doc comment above it) and inlined its body into its two
callers.
2026-09-18 22:59:29 +02:00
iris
c908a71f3e 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).
2026-09-18 22:59:29 +02:00
iris
75af27abe0 swarm-controller: make agent-creation's wanted-state write idempotent
Intent::Create now no-ops when the agent already has a non-Destroyed
declaration, instead of unconditionally overwriting it to the
create path's state. Without this, re-running create against a name
that already has a wanted-state entry (an operator migrating a
pre-existing agent into this bookkeeping, or a retried request)
would silently pause an agent already running under some other
state. write() skips the network round-trip entirely when apply
returns the declaration unchanged.

mara: this does not match the expectation that agent creation is
idempotent so pre existing agents can be migrated
2026-09-18 22:59:29 +02:00
iris
48aa7a1e79 swarm-controller: let creating an agent reuse a destroyed name
Review caught a path the new declaration node breaks: an operator may
already declare an agent `Destroyed` over the per-agent state endpoint,
and `apply` then refuses any transition off that state. Before the
wanted-state node existed the refusal was inert at creation time, but
now creating an agent under a previously-destroyed name builds its
identity, repo and config, fails the declaration, and silently cancels
the deploy — while the caller sees a 200 and a job id.

`AgentState::Destroyed`'s own doc comment already sanctions this case
("no state that brings a destroyed agent back short of a fresh deploy");
nothing implemented it. Give `apply` an `Intent`, keep the refusal for
redeclares, and add `WantedWriter::create` for the one caller that is a
fresh deploy. A separate method rather than a parameter on `set`, so no
other caller can reach the override by passing an argument wrong.
2026-09-18 22:59:29 +02:00
iris
bfd019fe8d swarm-controller: declare a new agent paused at creation
Creating an agent queued its identity, forge repo and deploy, but never
wrote a wanted-state declaration for it — so the agent showed up in
swarm-ui as "no declaration", and the hive brought it up with nothing
saying whether it should be driving turns.

Add a `SetAgentWanted` job node that declares the agent `Paused` in its
hive's wanted-state bucket, using the same `WantedWriter::set` primitive
the per-agent state HTTP handler already uses. A fresh agent therefore
sits paused until the operator explicitly flips it to `Up`.

The node is a root — it needs only the hive and agent names known at
request time — but the deploy trigger now waits on it, so the pause is
in the store before the hive brings the container up rather than landing
some time after a freshly deployed agent has already started taking
turns.
2026-09-18 22:59:29 +02:00
2 changed files with 263 additions and 114 deletions

View file

@ -105,6 +105,12 @@ enum SwarmNodeKind {
/// its hive's shared queue credential, so the document cannot be rendered /// its hive's shared queue credential, so the document cannot be rendered
/// without knowing which hive the agent belongs to. /// without knowing which hive the agent belongs to.
MintAgentIdentity { hive: String, agent: String }, 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 /// Tell `hive` to rebuild `agent`, by publishing on the swarm's deploy
/// subject. The one node kind whose effect leaves this host. /// subject. The one node kind whose effect leaves this host.
/// ///
@ -124,6 +130,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(), SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(),
SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(), SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(),
SwarmNodeKind::MintAgentIdentity { .. } => "mint_agent_identity".to_owned(), SwarmNodeKind::MintAgentIdentity { .. } => "mint_agent_identity".to_owned(),
SwarmNodeKind::SetAgentWanted { .. } => "set_agent_wanted".to_owned(),
SwarmNodeKind::TriggerDeploy { .. } => "trigger_deploy".to_owned(), SwarmNodeKind::TriggerDeploy { .. } => "trigger_deploy".to_owned(),
} }
} }
@ -144,7 +151,8 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
serde_json::json!({ "agent": agent }) serde_json::json!({ "agent": agent })
} }
SwarmNodeKind::TriggerDeploy { hive, agent } SwarmNodeKind::TriggerDeploy { hive, agent }
| SwarmNodeKind::MintAgentIdentity { hive, agent } => { | SwarmNodeKind::MintAgentIdentity { hive, agent }
| SwarmNodeKind::SetAgentWanted { hive, agent } => {
serde_json::json!({ "agent": agent, "hive": hive }) serde_json::json!({ "agent": agent, "hive": hive })
} }
} }
@ -186,8 +194,22 @@ struct WorkerDeps {
/// a per-node chance to read one. `None` on a host the operator has not /// a per-node chance to read one. `None` on a host the operator has not
/// given an authority — see `agent_identity::Authority::from_env`. /// given an authority — see `agent_identity::Authority::from_env`.
agent_ca: Option<std::sync::Arc<agent_identity::Authority>>, agent_ca: Option<std::sync::Arc<agent_identity::Authority>>,
/// 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<std::sync::Arc<wanted::WantedWriter>>,
} }
/// 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/ /// 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` /// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind`
/// variant turns into a real effect. /// variant turns into a real effect.
@ -286,6 +308,14 @@ 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 { SwarmNodeKind::TriggerDeploy { hive, agent } => match deps.queue {
None => Outcome::Failed( None => Outcome::Failed(
"no swarm queue is configured on this host, so no hive can be told to deploy" "no swarm queue is configured on this host, so no hive can be told to deploy"
@ -297,6 +327,41 @@ async fn run_swarm_node(
(builder, outcome) (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. /// Publish one deploy request and report whether it left this process.
/// ///
/// The flush is not belt-and-braces: `publish` hands the message to the /// The flush is not belt-and-braces: `publish` hands the message to the
@ -840,7 +905,6 @@ fn declaration_target(
responses( responses(
(status = 200, description = "the declaration as now published", body = Vec<AgentDeclaration>), (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 = 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 = 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), (status = 500, description = "the declaration could not be published (problem+json)", body = String),
), ),
@ -858,14 +922,6 @@ async fn set_agent_state(
.into_string(); .into_string();
let declaration = writer.set(&hive, &agent, req.state).await.map_err(|e| { 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"); tracing::warn!(hive = %hive, agent = %agent, error = %format!("{e:#}"), "declaring agent state failed");
error_problem( error_problem(
axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::http::StatusCode::INTERNAL_SERVER_ERROR,
@ -1316,6 +1372,15 @@ fn declare_agent_job(
agent: agent.to_owned(), agent: agent.to_owned(),
}) })
.after_ok(create_identity); .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 // Last, and specifically after the config repo is seeded: the hive
// deploys by reading that repo, so a deploy asked for any earlier would // 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 // find nothing to build. This is the edge that makes creating an agent at
@ -1327,13 +1392,21 @@ fn declare_agent_job(
// overtake the mint — but a host with no authority configured must still // 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 // create agents exactly as it does today. `after_ok` there would turn an
// unconfigured option into an agent nobody runs. // 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 let _trigger_deploy = b
.node(SwarmNodeKind::TriggerDeploy { .node(SwarmNodeKind::TriggerDeploy {
hive: hive.to_owned(), hive: hive.to_owned(),
agent: agent.to_owned(), agent: agent.to_owned(),
}) })
.after_ok(init_config) .after_ok(init_config)
.after_any(mint_identity); .after_any(mint_identity)
.after_ok(set_wanted);
vec![create_identity.guid()] vec![create_identity.guid()]
} }
@ -1664,6 +1737,7 @@ async fn main() -> Result<()> {
forge: forge_client.clone(), forge: forge_client.clone(),
queue: status.as_ref().map(|s| s.queue_client()), queue: status.as_ref().map(|s| s.queue_client()),
agent_ca: load_agent_authority(), agent_ca: load_agent_authority(),
wanted: wanted_writer(status.as_ref()),
}; };
let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new( let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new(
@ -1882,8 +1956,9 @@ mod tests {
}]), }]),
links: std::sync::Arc::new(Vec::new()), links: std::sync::Arc::new(Vec::new()),
status: None, status: None,
// No queue, for the same reason as `status`: these tests drive // No queue, for the same reason as `status`: these tests build
// agent creation, which publishes no declaration. // the creation graph and read its shape, and never run the node
// that would write a declaration.
wanted: None, wanted: None,
agent_status: None, agent_status: None,
jobq: std::sync::Arc::clone(&sched), jobq: std::sync::Arc::clone(&sched),
@ -2031,6 +2106,7 @@ mod tests {
forge: None, forge: None,
queue: None, queue: None,
agent_ca: None, agent_ca: None,
wanted: None,
}; };
let runner = let runner =
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
@ -2084,6 +2160,7 @@ mod tests {
forge: None, forge: None,
queue: None, queue: None,
agent_ca: None, agent_ca: None,
wanted: None,
}; };
let runner = let runner =
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
@ -2137,6 +2214,7 @@ mod tests {
forge: None, forge: None,
queue: None, queue: None,
agent_ca: None, agent_ca: None,
wanted: None,
}; };
let runner = let runner =
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
@ -2159,6 +2237,137 @@ 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` /// 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 /// 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 /// the assertion that the new variant joined the two-field arm rather

View file

@ -12,42 +12,17 @@
//! The bucket is the record. Nothing here keeps a second copy of the //! The bucket is the record. Nothing here keeps a second copy of the
//! declaration to reconcile against, because the current value can be read //! declaration to reconcile against, because the current value can be read
//! back from the queue whenever it is needed. //! back from the queue whenever it is needed.
//!
use std::fmt; //! [`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 anyhow::{Context, Result}; use anyhow::{Context, Result};
use async_nats::jetstream::kv::{CreateErrorKind, UpdateErrorKind}; use async_nats::jetstream::kv::{CreateErrorKind, UpdateErrorKind};
use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted}; 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 /// The three outcomes of one write attempt, which the two KV verbs report
/// through separate error types. /// through separate error types.
enum Wrote { enum Wrote {
@ -67,33 +42,15 @@ const MAX_ATTEMPTS: usize = 5;
/// Apply one agent's declared state to a hive's current declaration. /// 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: /// `current` is the hive's whole agent map — an undecodable value is an
/// the value under a hive's key is the map of **every** agent on that hive, /// **error**, not treated as absent, so a bad read never discards every
/// so declaring one agent must preserve the rest. /// other agent's declaration under a fresh-start fallback.
///
/// `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<u8>)> { fn apply(current: Option<&[u8]>, agent: &str, state: AgentState) -> Result<(HiveWanted, Vec<u8>)> {
let mut declaration = match current { let mut declaration = match current {
Some(raw) => serde_json::from_slice::<HiveWanted>(raw) Some(raw) => serde_json::from_slice::<HiveWanted>(raw)
.context("the hive's current declaration is not decodable")?, .context("the hive's current declaration is not decodable")?,
None => HiveWanted::default(), 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 declaration
.agents .agents
.insert(agent.to_owned(), AgentWanted { state }); .insert(agent.to_owned(), AgentWanted { state });
@ -112,31 +69,15 @@ impl WantedWriter {
Self { client } 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<async_nats::jetstream::kv::Store, swarm_queue_client::Error> {
swarm_queue_client::wanted::open_or_create(&self.client, hive).await
}
/// The declaration currently published for `hive`, or `None`. /// The declaration currently published for `hive`, or `None`.
pub async fn view(&self, hive: &str) -> Result<Option<HiveWanted>> { pub async fn view(&self, hive: &str) -> Result<Option<HiveWanted>> {
// An unconnected client does not fail a JetStream request, it hangs // An unconnected client does not fail a JetStream request, it hangs
// on it — see `swarm_queue_client::ensure_connected`. // on it — see `swarm_queue_client::ensure_connected`.
swarm_queue_client::ensure_connected(&self.client)?; swarm_queue_client::ensure_connected(&self.client)?;
let store = self.store(hive).await?; // 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 Some(entry) = store let Some(entry) = store
.entry(hive) .entry(hive)
.await .await
@ -149,17 +90,15 @@ impl WantedWriter {
.with_context(|| format!("the declaration for {hive} is not decodable")) .with_context(|| format!("the declaration for {hive} is not decodable"))
} }
/// Declare `agent` on `hive` to be in `state`, and return the whole /// Declare `agent` on `hive` to be in `state`, unconditionally (see the
/// declaration as published. /// module doc), and return the whole declaration as published.
/// ///
/// Read-modify-write against the entry's revision rather than a plain /// Read-modify-write against the entry's revision, not a plain `put`:
/// `put`: the value is the hive's whole agent map, so a blind write /// the value is the hive's whole agent map, so a blind write would drop
/// would drop a concurrent change to a different agent. The bucket keeps /// a concurrent change to a different agent.
/// 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<HiveWanted> { pub async fn set(&self, hive: &str, agent: &str, state: AgentState) -> Result<HiveWanted> {
swarm_queue_client::ensure_connected(&self.client)?; swarm_queue_client::ensure_connected(&self.client)?;
let store = self.store(hive).await?; let store = swarm_queue_client::wanted::open_or_create(&self.client, hive).await?;
for _ in 0..MAX_ATTEMPTS { for _ in 0..MAX_ATTEMPTS {
let entry = store let entry = store
@ -170,6 +109,18 @@ impl WantedWriter {
let (declaration, encoded) = let (declaration, encoded) =
apply(entry.as_ref().map(|e| e.value.as_ref()), agent, state)?; 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 // `update` and `create` have separate error types, and only one
// variant of each means "someone else got there first". Every // variant of each means "someone else got there first". Every
// other failure returns immediately: retrying a disconnect or a // other failure returns immediately: retrying a disconnect or a
@ -263,33 +214,22 @@ mod tests {
assert_eq!(again.agents["iris"].state, AgentState::Up); assert_eq!(again.agents["iris"].state, AgentState::Up);
} }
/// The bug this module exists to fix: nothing used to stop a caller from /// No terminal-state refusal any more (mara: "do not refuse to recreate
/// declaring a destroyed agent back to `Up`/`Offline`, silently /// an agent") — `apply` writes whatever it is asked to, including moving
/// un-terminaling a state whose own doc comment says that is impossible. /// 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] #[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"}}}"#; let current = br#"{"agents":{"atlas":{"state":"destroyed"}}}"#;
for state in [AgentState::Up, AgentState::Offline, AgentState::Paused] { for state in [AgentState::Up, AgentState::Offline, AgentState::Paused] {
let err = apply(Some(current), "atlas", state).unwrap_err(); let (declaration, _) = apply(Some(current), "atlas", state).unwrap();
assert!( assert_eq!(declaration.agents["atlas"].state, state);
err.downcast_ref::<super::TerminalStateError>().is_some(),
"expected a TerminalStateError for {state:?}, got: {err}"
);
} }
} }
/// Redeclaring `Destroyed` while already `Destroyed` is a no-op, not a /// The write is per-agent — a destroyed agent on the same hive as a live
/// transition — a client that already sees the terminal state and asks /// one must not block declaring the live one, and declaring one must not
/// again should get success back, not an error for stating a fact. /// disturb the other.
#[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] #[test]
fn a_destroyed_agent_does_not_block_declaring_a_different_one() { fn a_destroyed_agent_does_not_block_declaring_a_different_one() {
let current = br#"{"agents":{"atlas":{"state":"destroyed"},"iris":{"state":"up"}}}"#; let current = br#"{"agents":{"atlas":{"state":"destroyed"},"iris":{"state":"up"}}}"#;