Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8aafe4eaee | ||
|
|
c908a71f3e | ||
|
|
75af27abe0 | ||
|
|
48aa7a1e79 | ||
|
|
bfd019fe8d |
2 changed files with 263 additions and 114 deletions
|
|
@ -105,6 +105,12 @@ 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.
|
||||
///
|
||||
|
|
@ -124,6 +130,7 @@ 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(),
|
||||
}
|
||||
}
|
||||
|
|
@ -144,7 +151,8 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
|
|||
serde_json::json!({ "agent": agent })
|
||||
}
|
||||
SwarmNodeKind::TriggerDeploy { hive, agent }
|
||||
| SwarmNodeKind::MintAgentIdentity { hive, agent } => {
|
||||
| SwarmNodeKind::MintAgentIdentity { hive, agent }
|
||||
| SwarmNodeKind::SetAgentWanted { hive, agent } => {
|
||||
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
|
||||
/// given an authority — see `agent_identity::Authority::from_env`.
|
||||
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/
|
||||
/// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind`
|
||||
/// 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 {
|
||||
None => Outcome::Failed(
|
||||
"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)
|
||||
}
|
||||
|
||||
/// 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
|
||||
|
|
@ -840,7 +905,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),
|
||||
),
|
||||
|
|
@ -858,14 +922,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,
|
||||
|
|
@ -1316,6 +1372,15 @@ 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
|
||||
|
|
@ -1327,13 +1392,21 @@ 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_any(mint_identity)
|
||||
.after_ok(set_wanted);
|
||||
vec![create_identity.guid()]
|
||||
}
|
||||
|
||||
|
|
@ -1664,6 +1737,7 @@ 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(
|
||||
|
|
@ -1882,8 +1956,9 @@ mod tests {
|
|||
}]),
|
||||
links: std::sync::Arc::new(Vec::new()),
|
||||
status: None,
|
||||
// No queue, for the same reason as `status`: these tests drive
|
||||
// agent creation, which publishes no declaration.
|
||||
// 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.
|
||||
wanted: None,
|
||||
agent_status: None,
|
||||
jobq: std::sync::Arc::clone(&sched),
|
||||
|
|
@ -2031,6 +2106,7 @@ mod tests {
|
|||
forge: None,
|
||||
queue: None,
|
||||
agent_ca: None,
|
||||
wanted: None,
|
||||
};
|
||||
let runner =
|
||||
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
|
||||
|
|
@ -2084,6 +2160,7 @@ mod tests {
|
|||
forge: None,
|
||||
queue: None,
|
||||
agent_ca: None,
|
||||
wanted: None,
|
||||
};
|
||||
let runner =
|
||||
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
|
||||
|
|
@ -2137,6 +2214,7 @@ mod tests {
|
|||
forge: None,
|
||||
queue: None,
|
||||
agent_ca: None,
|
||||
wanted: None,
|
||||
};
|
||||
let runner =
|
||||
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`
|
||||
/// 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
|
||||
|
|
|
|||
|
|
@ -12,42 +12,17 @@
|
|||
//! 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;
|
||||
//!
|
||||
//! [`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 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 {
|
||||
|
|
@ -67,33 +42,15 @@ const MAX_ATTEMPTS: usize = 5;
|
|||
|
||||
/// 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:
|
||||
/// 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.
|
||||
/// `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.
|
||||
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.
|
||||
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 });
|
||||
|
|
@ -112,31 +69,15 @@ 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<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`.
|
||||
pub async fn view(&self, hive: &str) -> Result<Option<HiveWanted>> {
|
||||
// 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)?;
|
||||
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
|
||||
.entry(hive)
|
||||
.await
|
||||
|
|
@ -149,17 +90,15 @@ impl WantedWriter {
|
|||
.with_context(|| format!("the declaration for {hive} is not decodable"))
|
||||
}
|
||||
|
||||
/// Declare `agent` on `hive` to be in `state`, and return the whole
|
||||
/// declaration as published.
|
||||
/// Declare `agent` on `hive` to be in `state`, unconditionally (see the
|
||||
/// module doc), and return the whole declaration as published.
|
||||
///
|
||||
/// 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.
|
||||
/// 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.
|
||||
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?;
|
||||
let store = swarm_queue_client::wanted::open_or_create(&self.client, hive).await?;
|
||||
|
||||
for _ in 0..MAX_ATTEMPTS {
|
||||
let entry = store
|
||||
|
|
@ -170,6 +109,18 @@ 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
|
||||
|
|
@ -263,33 +214,22 @@ mod tests {
|
|||
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 = apply(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, _) = 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.
|
||||
/// 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"}}}"#;
|
||||
|
|
|
|||
Loading…
Reference in a new issue