diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 98139a21..8f673991 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -328,26 +328,11 @@ async fn run_swarm_node( } /// 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". +/// 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, diff --git a/swarm-controller/src/wanted.rs b/swarm-controller/src/wanted.rs index 4f0d4cd7..254ba5cd 100644 --- a/swarm-controller/src/wanted.rs +++ b/swarm-controller/src/wanted.rs @@ -13,19 +13,11 @@ //! declaration to reconcile against, because the current value can be read //! back from the queue whenever it is needed. //! -//! **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. +//! [`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}; @@ -50,17 +42,9 @@ 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. -/// -/// Unconditional otherwise — see the module doc for why there is no -/// terminal-state refusal here any more. +/// `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)> { let mut declaration = match current { Some(raw) => serde_json::from_slice::(raw) @@ -85,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 { - swarm_queue_client::wanted::open_or_create(&self.client, hive).await - } - /// The declaration currently published for `hive`, or `None`. pub async fn view(&self, hive: &str) -> Result> { // An unconnected client does not fail a JetStream request, it hangs // on it — see `swarm_queue_client::ensure_connected`. swarm_queue_client::ensure_connected(&self.client)?; - 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 @@ -122,23 +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. /// - /// 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. + /// 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 { 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