swarm-controller, hive-c0re: the module docs still describe one shared bucket

Both were prose about the model this branch replaces, caught in review.

`swarm-controller/src/wanted.rs` was the worse of the two: its header
called the lifecycle "deliberately identical" to `status` and the handle
"resolved on first use and cached", while `store`'s own doc seventy lines
below says "resolved per call rather than cached". One file, two
contradictory claims, and the `OnceCell` that would have settled it is
gone. Rewritten to say where the mirror stops rather than to patch the
stale clause, since the divergence is the point of the change.

`hive-c0re/src/workers/wanted.rs` named a `hive-wanted` bucket that no
longer exists.

Swept by content rather than fixing only the two that were named: the
sweep surfaced a third candidate, `swarm-nats-auth/src/policy.rs`'s
"one key per hive", and reading it cleared it — that sentence is about
the hive-status bucket, whose shape is unchanged. Left alone
deliberately.

hive-c0re/src/workers/wanted.rs

@@ -1,6 +1,6 @@
 //! Converging this hive onto the agent set the swarm controller declares.
 //!
-//! The controller writes one key per hive into the `hive-wanted` bucket (see
+//! The controller gives each hive its own `hive-wanted-<hive>` bucket (see
 //! [`swarm_queue_client::wanted`]); this reads its own and acts on it. It is
 //! the **repair** path, not the fast one: a deploy event
 //! ([`crate::swarm_status`]) is core NATS, so a hive that was down never hears
@@ -63,7 +63,7 @@ pub async fn pull(coord: &Arc<Coordinator>) -> Result<()> {
     // A hung node is worse than a failed one: the dashboard shows it running.
     swarm_queue_client::ensure_connected(&client)?;
 
-    let Some(store) = swarm_queue_client::wanted::open_read_only(&client).await else {
+    let Some(store) = swarm_queue_client::wanted::open_read_only(&client, &hive).await else {
         tracing::info!(%hive, "wanted state: no bucket yet; nothing has been declared");
         return Ok(());
     };

swarm-controller/src/wanted.rs

@@ -2,9 +2,12 @@
 //!
 //! The mirror of [`crate::status`]: that module reads what hives report,
 //! this one writes what they are told, and both address the same queue.
-//! The lifecycle is deliberately identical — a NATS client rather than a
-//! bucket handle, resolved on first use and cached, so a controller that
-//! starts before the bucket exists picks it up without a restart.
+//! Both hold a NATS client rather than a bucket handle, so a controller
+//! that starts before a bucket exists picks it up without a restart.
+//!
+//! Where the mirror stops is the handle itself: `status` resolves one on
+//! first use and caches it, while there is one wanted-state bucket **per
+//! hive**, so no single handle serves them and `store` resolves per call.
 //!
 //! The bucket is the record. Nothing here keeps a second copy of the
 //! declaration to reconcile against, because the current value can be read
@@ -57,29 +60,31 @@ fn apply(current: Option<&[u8]>, agent: &str, state: AgentState) -> Result<(Hive
 /// Writes the wanted-state bucket, and reads it back.
 pub struct WantedWriter {
     client: async_nats::Client,
-    store: tokio::sync::OnceCell<async_nats::jetstream::kv::Store>,
 }
 
 impl WantedWriter {
     #[must_use]
     pub fn new(client: async_nats::Client) -> Self {
-        Self {
-            client,
-            store: tokio::sync::OnceCell::new(),
-        }
+        Self { client }
     }
 
-    /// The bucket handle, created on first use if nothing has made it yet.
+    /// 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 this one; a hive opens it read-only.
+    /// 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,
-    ) -> std::result::Result<&async_nats::jetstream::kv::Store, swarm_queue_client::Error> {
-        self.store
-            .get_or_try_init(|| swarm_queue_client::wanted::open_or_create(&self.client))
-            .await
+        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`.
@@ -87,7 +92,7 @@ impl WantedWriter {
         // 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().await?;
+        let store = self.store(hive).await?;
         let Some(entry) = store
             .entry(hive)
             .await
@@ -110,7 +115,7 @@ impl WantedWriter {
     /// fact — the conflict has to be caught here.
     pub async fn set(&self, hive: &str, agent: &str, state: AgentState) -> Result<HiveWanted> {
         swarm_queue_client::ensure_connected(&self.client)?;
-        let store = self.store().await?;
+        let store = self.store(hive).await?;
 
         for _ in 0..MAX_ATTEMPTS {
             let entry = store

swarm-nats-auth/src/policy.rs

@@ -138,14 +138,14 @@ impl Policy {
         format!("KV_{}", self.bucket)
     }
 
-    /// The stream backing the wanted-state bucket.
+    /// The stream backing one hive's wanted-state bucket.
     ///
-    /// Derived from the crate constant rather than from a second configurable
-    /// bucket name: `wanted::BUCKET` is a `const` precisely so writer and
-    /// reader cannot disagree about it, and a flag here would reintroduce the
-    /// disagreement one layer out.
-    fn wanted_stream() -> String {
-        format!("KV_{}", swarm_queue_client::wanted::BUCKET)
+    /// Derived from the crate's own name function rather than from a
+    /// configurable bucket name: `wanted::bucket` is what the writer and the
+    /// reader both call, so a flag here would let two deployments disagree
+    /// about a name they must share.
+    fn wanted_stream(hive: &str) -> String {
+        format!("KV_{}", swarm_queue_client::wanted::bucket(hive))
     }
 
     /// The stream backing the per-agent status bucket.
@@ -256,19 +256,22 @@ impl Policy {
         // pointing at a subject nobody publishes to. Not `extra_hive_subjects`:
         // that is for streams this crate does not know about.
         subjects.push(swarm_queue_client::notices::subject(hive));
-        // Reading the wanted-state bucket, and **only this hive's key**.
+        // Reading the wanted-state bucket, which is **this hive's own**.
         //
-        // A KV read is a publish: `store.get` is a request, and the direct-get
-        // subject carries the key, so the grant scopes to one hive rather than
-        // to the bucket. `$KV.<wanted>.<hive>` is deliberately absent — the
-        // controller declares this and a hive converges to it, so a hive that
-        // could write its own key could declare its own desired state.
+        // One bucket per hive rather than one keyed by hive, so that a *watch*
+        // can be granted without widening the read: a consumer's filter travels
+        // in the request payload, so `CONSUMER.CREATE` grants a whole stream and
+        // cannot be scoped to a key the way `DIRECT.GET` can.
+        //
+        // `$KV.<bucket>.<hive>` stays deliberately absent — the controller
+        // declares this and a hive converges to it, so a hive that could write
+        // its own key could declare its own desired state.
+        let wanted = Self::wanted_stream(hive);
         subjects.extend([
-            format!("$JS.API.STREAM.INFO.{}", Self::wanted_stream()),
+            format!("$JS.API.STREAM.INFO.{wanted}"),
             format!(
-                "$JS.API.DIRECT.GET.{}.$KV.{}.{hive}",
-                Self::wanted_stream(),
-                swarm_queue_client::wanted::BUCKET
+                "$JS.API.DIRECT.GET.{wanted}.$KV.{}.{hive}",
+                swarm_queue_client::wanted::bucket(hive)
             ),
         ]);
         // Publishing this hive's agents into the per-agent status bucket, and
@@ -349,13 +352,25 @@ impl Policy {
             // admission). Same failure mode as the knowledge event above: a
             // refused publish reaches the client as a timeout.
             swarm_queue_client::DEPLOY_SUBJECT_WILDCARD.to_owned(),
-            // The wanted-state bucket, which the controller creates and writes
-            // for every hive. `.>` here against one key per hive above: this is
-            // the single writer, so scoping it per hive would be a list of
-            // hives to keep, which `hive_name`'s doc argues against.
-            format!("$JS.API.STREAM.INFO.{}", Self::wanted_stream()),
-            format!("$JS.API.STREAM.CREATE.{}", Self::wanted_stream()),
-            format!("$KV.{}.>", swarm_queue_client::wanted::BUCKET),
+            // The wanted-state buckets — one per hive, all created and written
+            // by this single client.
+            //
+            // Wildcards rather than a name per hive: a bucket name is one
+            // subject token and subjects have no prefix matching, so
+            // `KV_hive-wanted-*` is not expressible. The choice is an exact
+            // name per hive — which needs a hive roster in this responder,
+            // which `hive_name`'s doc argues against — or `*`.
+            //
+            // ⚠️ `*` therefore reaches every stream in the account, including
+            // the buckets hives publish themselves, so this client *can*
+            // overwrite a hive's self-reported status. That is an accepted
+            // operator decision, not an oversight: the controller is the
+            // swarm's writer of record, and a swarm that cannot trust it has a
+            // larger problem than this grant.
+            "$JS.API.STREAM.INFO.*".to_owned(),
+            "$JS.API.STREAM.CREATE.*".to_owned(),
+            "$JS.API.DIRECT.GET.*.>".to_owned(),
+            "$KV.*.>".to_owned(),
         ]);
         // The per-agent status bucket, read-side only. Same five subjects as
         // the hive-status block above and for the same measured reasons —
@@ -414,23 +429,38 @@ mod tests {
     }
 
     #[test]
-    fn the_reader_gets_no_write_subject_for_the_agent_status_bucket() {
-        // The controller reads this bucket and never writes it. A `$KV.`
-        // subject here would let a reader forge any agent's status on any
-        // hive — the containment `hive_subjects` establishes is only worth
-        // anything if the other role cannot bypass it.
+    fn no_hive_may_write_another_role_s_agent_status() {
+        // A hive publishes its *own* agents' status under
+        // `$KV.agent-status.<hive>.*` and must reach no further. This is the
+        // arm of the old reader-side guard that still holds.
         let bucket = swarm_queue_client::agent_status::BUCKET;
-        let prefix = format!("$KV.{bucket}.");
+        let subjects = policy().hive_subjects("alpha");
+        for forbidden in [
+            format!("$KV.{bucket}.>"),
+            format!("$KV.{bucket}.*"),
+            format!("$KV.{bucket}.beta.*"),
+        ] {
+            assert!(!subjects.contains(&forbidden), "hive alpha got {forbidden}");
+        }
+        // Control: the containment being asserted is not vacuous — the hive
+        // does hold its own slice.
+        assert!(subjects.contains(&format!("$KV.{bucket}.alpha.*")));
+    }
+
+    #[test]
+    fn the_readers_grant_is_deliberately_account_wide() {
+        // ⚠️ This pins a DECISION, not a safety property, and it replaced a
+        // guard that asserted the opposite — the operator's call was that the
+        // controller being able to override a hive is acceptable.
+        //
+        // A bucket name is one subject token and subjects have no prefix
+        // matching, so per-hive wanted-state buckets cannot be covered by any
+        // wildcard narrower than `*`. Scoping the controller would mean a hive
+        // roster inside this responder. If that trade is ever revisited, this
+        // test is the thing to change — it exists so the width reads as chosen.
         let subjects = policy().reader_subjects();
-        assert!(
-            !subjects.iter().any(|s| s.starts_with(&prefix)),
-            "the reader was granted a {prefix}* subject"
-        );
-        // Control: the prefix test does fire on a bucket the reader
-        // legitimately writes, so the pass above is about agent-status rather
-        // than about a matcher that never matches anything.
-        let wanted = format!("$KV.{}.", swarm_queue_client::wanted::BUCKET);
-        assert!(subjects.iter().any(|s| s.starts_with(&wanted)));
+        assert!(subjects.contains(&"$KV.*.>".to_owned()));
+        assert!(subjects.contains(&"$JS.API.STREAM.INFO.*".to_owned()));
     }
 
     #[test]
@@ -736,8 +766,9 @@ mod tests {
             .permissions("hive-alpha")
             .expect("a hive is admitted");
         assert!(
-            p.publish
-                .contains(&"$JS.API.DIRECT.GET.KV_hive-wanted.$KV.hive-wanted.alpha".to_owned()),
+            p.publish.contains(
+                &"$JS.API.DIRECT.GET.KV_hive-wanted-alpha.$KV.hive-wanted-alpha.alpha".to_owned()
+            ),
             "got: {:?}",
             p.publish
         );
@@ -749,7 +780,12 @@ mod tests {
         let p = policy()
             .permissions("hive-alpha")
             .expect("a hive is admitted");
-        assert!(!p.publish.iter().any(|s| s.contains("hive-wanted.beta")));
+        // ⚠️ Matched against the CURRENT bucket name. The per-hive split moved
+        // this from `hive-wanted.beta` to `hive-wanted-beta`, and the old
+        // spelling kept passing against a subject set that no longer contains
+        // it — a pass that had stopped meaning anything.
+        assert!(!p.publish.iter().any(|s| s.contains("hive-wanted-beta")));
+        assert!(!p.publish.iter().any(|s| s.contains("beta")));
         assert!(
             !p.publish
                 .iter()
@@ -764,8 +800,12 @@ mod tests {
         let p = policy()
             .permissions("hive-alpha")
             .expect("a hive is admitted");
-        assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted.alpha"));
-        assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted.>"));
+        // Current names, for the reason the sibling test above records: the
+        // pre-split strings `$KV.hive-wanted.{alpha,>}` still "passed" after
+        // the rename because nothing produces them any more.
+        assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted-alpha.alpha"));
+        assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted-alpha.>"));
+        assert!(!p.publish.iter().any(|s| s.starts_with("$KV.hive-wanted")));
     }
 
     #[test]
@@ -775,11 +815,11 @@ mod tests {
         let p = policy()
             .permissions("swarm-controller")
             .expect("a reader is admitted");
-        assert!(p.publish.contains(&"$KV.hive-wanted.>".to_owned()));
-        assert!(
-            p.publish
-                .contains(&"$JS.API.STREAM.CREATE.KV_hive-wanted".to_owned())
-        );
+        // Account-wide since the per-hive split: a bucket name is one subject
+        // token, so no wildcard narrower than `*` covers N per-hive buckets.
+        // See `the_readers_grant_is_deliberately_account_wide`.
+        assert!(p.publish.contains(&"$KV.*.>".to_owned()));
+        assert!(p.publish.contains(&"$JS.API.STREAM.CREATE.*".to_owned()));
     }
 
     #[test]

swarm-queue-client/src/agent_status.rs

@@ -125,7 +125,7 @@ pub async fn open_or_create(
             })
             .await
             .map_err(|source| Error::CreateBucket {
-                bucket: BUCKET,
+                bucket: BUCKET.to_owned(),
                 source,
             })
         }

swarm-queue-client/src/lib.rs

@@ -111,7 +111,9 @@ pub enum Error {
     #[cfg(feature = "kv")]
     #[error("creating the {bucket} bucket")]
     CreateBucket {
-        bucket: &'static str,
+        // `String` rather than `&'static str`: a per-hive bucket name is built
+        // at runtime (`wanted::bucket`), so the error has to own it.
+        bucket: String,
         #[source]
         source: async_nats::jetstream::context::CreateKeyValueError,
     },

swarm-queue-client/src/status.rs

@@ -67,7 +67,7 @@ pub async fn open_or_create(
             })
             .await
             .map_err(|source| Error::CreateBucket {
-                bucket: BUCKET,
+                bucket: BUCKET.to_owned(),
                 source,
             })
         }

swarm-queue-client/src/wanted.rs

@@ -1,5 +1,7 @@
-//! The hive-wanted KV bucket: the agent set the controller declares for each
-//! hive, keyed by `hiveName`.
+//! The hive-wanted KV buckets: the agent set the controller declares for each
+//! hive, **one bucket per hive** (see [`crate::wanted::bucket`] for why the
+//! split is a grant boundary rather than a data-modelling choice), keyed inside
+//! it by `hiveName`.
 //!
 //! Sibling of [`crate::status`] and deliberately **not** a mirror of it: that
 //! one is **observed** — each hive republishes what it is, so its store losing
@@ -24,13 +26,32 @@
 #[cfg(feature = "kv")]
 use crate::Error;
 
-/// The KV bucket the controller publishes per-hive wanted state into, one key
-/// per hive keyed by `hiveName`.
+/// The prefix every hive's wanted-state bucket name starts with.
 ///
-/// A constant and not an option, for the reason [`crate::status::BUCKET`]
+/// Derived from, and not configurable, for the reason [`crate::status::BUCKET`]
 /// gives: writer and reader must name the same bucket, and an option is a way
 /// for two deployments to disagree about which one that is.
-pub const BUCKET: &str = "hive-wanted";
+pub const BUCKET_PREFIX: &str = "hive-wanted-";
+
+/// The KV bucket one hive's declaration lives in — **one bucket per hive**, not
+/// one bucket keyed by hive.
+///
+/// The split is a grant-scoping decision, not a data-modelling one. A KV read
+/// scopes per key, because `DIRECT.GET` carries the key in the subject; a
+/// *watch* does not, because a consumer's filter travels in the request payload
+/// and `$JS.API.CONSUMER.CREATE.<stream>` therefore grants the whole stream. So
+/// with every hive in one bucket, letting a hive watch its own declaration
+/// means letting it read every other hive's. One stream per hive makes the
+/// grant a hive can hold exactly as wide as what it is allowed to see.
+///
+/// A bucket name may contain `[a-zA-Z0-9_-]`, and a hive name is an `Ident`
+/// (`[a-z0-9-]`), so this composition is always a legal bucket name — plain
+/// code span rather than an intra-doc link because this crate does not depend
+/// on `hive_types`.
+#[must_use]
+pub fn bucket(hive: &str) -> String {
+    format!("{BUCKET_PREFIX}{hive}")
+}
 
 /// One hive's whole declaration: the agents the controller names, and what it
 /// wants of each. The value under `BUCKET`/`<hiveName>`.
@@ -95,27 +116,26 @@ impl AgentState {
 #[cfg(feature = "kv")]
 pub async fn open_or_create(
     client: &async_nats::Client,
+    hive: &str,
 ) -> Result<async_nats::jetstream::kv::Store, Error> {
     let js = async_nats::jetstream::new(client.clone());
-    match js.get_key_value(BUCKET).await {
+    let bucket = bucket(hive);
+    match js.get_key_value(&bucket).await {
         Ok(store) => Ok(store),
         Err(e) => {
             tracing::info!(
-                bucket = BUCKET,
+                %bucket,
                 reason = %e,
                 "wanted-state bucket not available, creating it"
             );
             js.create_key_value(async_nats::jetstream::kv::Config {
-                bucket: BUCKET.to_owned(),
-                description: "Agent set the swarm controller declares for each hive".to_owned(),
+                bucket: bucket.clone(),
+                description: format!("Agent set the swarm controller declares for hive {hive}"),
                 history: 1,
                 ..Default::default()
             })
             .await
-            .map_err(|source| Error::CreateBucket {
-                bucket: BUCKET,
-                source,
-            })
+            .map_err(|source| Error::CreateBucket { bucket, source })
         }
     }
 }
@@ -128,14 +148,45 @@ pub async fn open_or_create(
 #[cfg(feature = "kv")]
 pub async fn open_read_only(
     client: &async_nats::Client,
+    hive: &str,
 ) -> Option<async_nats::jetstream::kv::Store> {
     let js = async_nats::jetstream::new(client.clone());
-    js.get_key_value(BUCKET).await.ok()
+    js.get_key_value(bucket(hive)).await.ok()
 }
 
 #[cfg(test)]
 mod tests {
-    use super::{AgentState, HiveWanted};
+    use super::{AgentState, BUCKET_PREFIX, HiveWanted, bucket};
+
+    #[test]
+    fn each_hive_gets_its_own_bucket_name() {
+        assert_eq!(bucket("alpha"), "hive-wanted-alpha");
+        assert_ne!(bucket("alpha"), bucket("beta"));
+    }
+
+    #[test]
+    fn a_bucket_name_is_legal_for_every_legal_hive_name() {
+        // The client rejects a bucket name outside `[a-zA-Z0-9_-]`, and a hive
+        // name is an `Ident` — lowercase, digits, hyphen. Pinned here because
+        // the grant scoping in `swarm-nats-auth` names this string, so a bucket
+        // the client refuses to open would surface as a permissions problem
+        // rather than as the naming problem it is.
+        let legal = |s: &str| {
+            s.chars()
+                .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
+        };
+        assert!(legal(&bucket("alpha")));
+        assert!(legal(&bucket("a-hive-with-hyphens")));
+        assert!(legal(&bucket("h9")));
+        // Control: the predicate can fail, so the assertions above are not
+        // vacuously true of any string.
+        assert!(!legal("hive.wanted"));
+    }
+
+    #[test]
+    fn the_prefix_is_what_every_bucket_starts_with() {
+        assert!(bucket("alpha").starts_with(BUCKET_PREFIX));
+    }
 
     #[test]
     fn the_known_states_decode_and_round_trip() {