vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

Compare commits

base: vinzenz:8aafe4eaee2347e68054b79aed63984f5888c4dc
Branches Tags
vinzenz:main
..
compare: vinzenz:4a1b5f5d519881fd583f213891f6d3abc3d80116
Branches Tags
vinzenz:main
2 changed files with 114 additions and 263 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -105,12 +105,6 @@ 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.
/// ///
@ -130,7 +124,6 @@ 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(),
} }
} }
@ -151,8 +144,7 @@ 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 })
} }
} }
@ -194,22 +186,8 @@ 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.
@ -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 { 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"
@ -327,41 +297,6 @@ 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
@ -905,6 +840,7 @@ 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),
), ),
@ -922,6 +858,14 @@ 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,
@ -1372,15 +1316,6 @@ 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
@ -1392,21 +1327,13 @@ 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()]
} }
@ -1737,7 +1664,6 @@ 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(
@ -1956,9 +1882,8 @@ 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 build // No queue, for the same reason as `status`: these tests drive
// the creation graph and read its shape, and never run the node // agent creation, which publishes no declaration.
// 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),
@ -2106,7 +2031,6 @@ 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| {
@ -2160,7 +2084,6 @@ 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| {
@ -2214,7 +2137,6 @@ 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| {
@ -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` /// 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

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

@ -12,17 +12,42 @@
//! 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.
//!
//! [`WantedWriter::set`] writes unconditionally, for every caller alike — no use std::fmt;
//! 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 {
@ -42,15 +67,33 @@ 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.
/// ///
/// `current` is the hive's whole agent map — an undecodable value is an /// Split out from the write loop because it holds the invariant that matters:
/// **error**, not treated as absent, so a bad read never discards every /// the value under a hive's key is the map of **every** agent on that hive,
/// other agent's declaration under a fresh-start fallback. /// 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<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 });
@ -69,15 +112,31 @@ 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)?;
// One bucket per hive, resolved per call, created on first use — see let store = self.store(hive).await?;
// `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
@ -90,15 +149,17 @@ 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`, unconditionally (see the /// Declare `agent` on `hive` to be in `state`, and return the whole
/// module doc), and return the whole declaration as published. /// declaration as published.
/// ///
/// Read-modify-write against the entry's revision, not a plain `put`: /// Read-modify-write against the entry's revision rather than a plain
/// the value is the hive's whole agent map, so a blind write would drop /// `put`: the value is the hive's whole agent map, so a blind write
/// a concurrent change to a different agent. /// 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<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 = swarm_queue_client::wanted::open_or_create(&self.client, hive).await?; let store = self.store(hive).await?;
for _ in 0..MAX_ATTEMPTS { for _ in 0..MAX_ATTEMPTS {
let entry = store let entry = store
@ -109,18 +170,6 @@ 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
@ -214,22 +263,33 @@ mod tests {
assert_eq!(again.agents["iris"].state, AgentState::Up); assert_eq!(again.agents["iris"].state, AgentState::Up);
} }
/// No terminal-state refusal any more (mara: "do not refuse to recreate /// The bug this module exists to fix: nothing used to stop a caller from
/// an agent") — `apply` writes whatever it is asked to, including moving /// declaring a destroyed agent back to `Up`/`Offline`, silently
/// a `Destroyed` agent to any other state. Whether that write should /// un-terminaling a state whose own doc comment says that is impossible.
/// happen at all is now the caller's own decision (see the module doc).
#[test] #[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"}}}"#; 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 (declaration, _) = apply(Some(current), "atlas", state).unwrap(); let err = apply(Some(current), "atlas", state).unwrap_err();
assert_eq!(declaration.agents["atlas"].state, state); assert!(
err.downcast_ref::<super::TerminalStateError>().is_some(),
"expected a TerminalStateError for {state:?}, got: {err}"
);
} }
} }
/// The write is per-agent — a destroyed agent on the same hive as a live /// Redeclaring `Destroyed` while already `Destroyed` is a no-op, not a
/// one must not block declaring the live one, and declaring one must not /// transition — a client that already sees the terminal state and asks
/// disturb the other. /// 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] #[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"}}}"#;