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

swarm-controller: trim oversized comments, drop a trivial wrapper fn

Browse source 
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.
This commit is contained in:
iris 2026-09-18 22:42:05 +02:00 • committed by mara
commit 8aafe4eaee
2 changed files with 23 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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,

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

@ -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<u8>)> {
let mut declaration = match current {
Some(raw) => serde_json::from_slice::<HiveWanted>(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<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
@ -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<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