From 8c94e340b862ae8c36950a6cde6c27762a543638 Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 2 Sep 2026 20:43:57 +0200 Subject: [PATCH] =?UTF-8?q?wip(#4006):=20per-hive=20wanted-state=20buckets?= =?UTF-8?q?=20=E2=80=94=20client=20crate=20only?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CHECKPOINT, NOT A PROPOSAL. mara paused the plan ("i dont quite understand the plan") before any behaviour-changing edit; this commit exists so the work survives a container stop, not because it is ready. wanted.rs: BUCKET -> BUCKET_PREFIX + bucket(hive) producing hive-wanted-; open_or_create/open_read_only take the hive. lib.rs: Error::CreateBucket.bucket becomes String, since a per-hive name is built at runtime. Nothing else is touched, so no caller compiles against the new signatures yet and no deployed behaviour changes. Remaining, if she approves: the two other CreateBucket sites, swarm-controller's OnceCell (one store -> N), hive-c0re's pull, policy.rs per-hive streams + the reader roster, swarm-nats.nix. --- swarm-queue-client/src/lib.rs | 4 +- swarm-queue-client/src/wanted.rs | 79 +++++++++++++++++++++++++------- 2 files changed, 66 insertions(+), 17 deletions(-) diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index ebe213d3..f1098326 100644 --- a/swarm-queue-client/src/lib.rs +++ b/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, }, diff --git a/swarm-queue-client/src/wanted.rs b/swarm-queue-client/src/wanted.rs index f9d6a4ed..13df750e 100644 --- a/swarm-queue-client/src/wanted.rs +++ b/swarm-queue-client/src/wanted.rs @@ -1,5 +1,6 @@ -//! 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 [`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 +25,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.` 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`/``. @@ -95,27 +115,26 @@ impl AgentState { #[cfg(feature = "kv")] pub async fn open_or_create( client: &async_nats::Client, + hive: &str, ) -> Result { 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 +147,42 @@ pub async fn open_or_create( #[cfg(feature = "kv")] pub async fn open_read_only( client: &async_nats::Client, + hive: &str, ) -> Option { 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() {