diff --git a/hive-c0re/src/workers/wanted.rs b/hive-c0re/src/workers/wanted.rs index 4ade198a..6fd39393 100644 --- a/hive-c0re/src/workers/wanted.rs +++ b/hive-c0re/src/workers/wanted.rs @@ -1,6 +1,6 @@ //! Converging this hive onto the agent set the swarm controller declares. //! -//! The controller gives each hive its own `hive-wanted-` bucket (see +//! The controller writes one key per hive into the `hive-wanted` 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) -> 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, &hive).await else { + let Some(store) = swarm_queue_client::wanted::open_read_only(&client).await else { tracing::info!(%hive, "wanted state: no bucket yet; nothing has been declared"); return Ok(()); }; diff --git a/swarm-controller/src/wanted.rs b/swarm-controller/src/wanted.rs index 4e25d80a..b5d2c7ba 100644 --- a/swarm-controller/src/wanted.rs +++ b/swarm-controller/src/wanted.rs @@ -2,12 +2,9 @@ //! //! The mirror of [`crate::status`]: that module reads what hives report, //! this one writes what they are told, and both address the same queue. -//! 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 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. //! //! The bucket is the record. Nothing here keeps a second copy of the //! declaration to reconcile against, because the current value can be read @@ -60,31 +57,29 @@ 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, } impl WantedWriter { #[must_use] pub fn new(client: async_nats::Client) -> Self { - Self { client } + Self { + client, + store: tokio::sync::OnceCell::new(), + } } - /// One hive's bucket handle, created on first use if nothing has made it - /// yet. + /// The 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. + /// controller creates this one; a hive opens it read-only. async fn store( &self, - hive: &str, - ) -> std::result::Result { - swarm_queue_client::wanted::open_or_create(&self.client, hive).await + ) -> 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 } /// The declaration currently published for `hive`, or `None`. @@ -92,7 +87,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(hive).await?; + let store = self.store().await?; let Some(entry) = store .entry(hive) .await @@ -115,7 +110,7 @@ impl WantedWriter { /// fact — the conflict has to be caught here. 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 = self.store().await?; for _ in 0..MAX_ATTEMPTS { let entry = store diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 4a0b5520..f58fc1ab 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -138,14 +138,14 @@ impl Policy { format!("KV_{}", self.bucket) } - /// The stream backing one hive's wanted-state bucket. + /// The stream backing the wanted-state 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)) + /// 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) } /// The stream backing the per-agent status bucket. @@ -256,22 +256,19 @@ 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, which is **this hive's own**. + // Reading the wanted-state bucket, and **only this hive's key**. // - // 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..` 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); + // 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..` 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. subjects.extend([ - format!("$JS.API.STREAM.INFO.{wanted}"), + format!("$JS.API.STREAM.INFO.{}", Self::wanted_stream()), format!( - "$JS.API.DIRECT.GET.{wanted}.$KV.{}.{hive}", - swarm_queue_client::wanted::bucket(hive) + "$JS.API.DIRECT.GET.{}.$KV.{}.{hive}", + Self::wanted_stream(), + swarm_queue_client::wanted::BUCKET ), ]); // Publishing this hive's agents into the per-agent status bucket, and @@ -352,25 +349,13 @@ 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 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 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 per-agent status bucket, read-side only. Same five subjects as // the hive-status block above and for the same measured reasons — @@ -429,38 +414,23 @@ mod tests { } #[test] - fn no_hive_may_write_another_role_s_agent_status() { - // A hive publishes its *own* agents' status under - // `$KV.agent-status..*` and must reach no further. This is the - // arm of the old reader-side guard that still holds. + 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. let bucket = swarm_queue_client::agent_status::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 prefix = format!("$KV.{bucket}."); let subjects = policy().reader_subjects(); - assert!(subjects.contains(&"$KV.*.>".to_owned())); - assert!(subjects.contains(&"$JS.API.STREAM.INFO.*".to_owned())); + 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))); } #[test] @@ -766,9 +736,8 @@ mod tests { .permissions("hive-alpha") .expect("a hive is admitted"); assert!( - p.publish.contains( - &"$JS.API.DIRECT.GET.KV_hive-wanted-alpha.$KV.hive-wanted-alpha.alpha".to_owned() - ), + p.publish + .contains(&"$JS.API.DIRECT.GET.KV_hive-wanted.$KV.hive-wanted.alpha".to_owned()), "got: {:?}", p.publish ); @@ -780,12 +749,7 @@ mod tests { let p = policy() .permissions("hive-alpha") .expect("a hive is admitted"); - // ⚠️ 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().any(|s| s.contains("hive-wanted.beta"))); assert!( !p.publish .iter() @@ -800,12 +764,8 @@ mod tests { let p = policy() .permissions("hive-alpha") .expect("a hive is admitted"); - // 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"))); + assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted.alpha")); + assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted.>")); } #[test] @@ -815,11 +775,11 @@ mod tests { let p = policy() .permissions("swarm-controller") .expect("a reader is admitted"); - // 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())); + assert!(p.publish.contains(&"$KV.hive-wanted.>".to_owned())); + assert!( + p.publish + .contains(&"$JS.API.STREAM.CREATE.KV_hive-wanted".to_owned()) + ); } #[test] diff --git a/swarm-queue-client/src/agent_status.rs b/swarm-queue-client/src/agent_status.rs index 65104bc5..9311fc6b 100644 --- a/swarm-queue-client/src/agent_status.rs +++ b/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.to_owned(), + bucket: BUCKET, source, }) } diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index f1098326..ebe213d3 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -111,9 +111,7 @@ pub enum Error { #[cfg(feature = "kv")] #[error("creating the {bucket} bucket")] CreateBucket { - // `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, + bucket: &'static str, #[source] source: async_nats::jetstream::context::CreateKeyValueError, }, diff --git a/swarm-queue-client/src/status.rs b/swarm-queue-client/src/status.rs index 3e019e86..1f0e69c2 100644 --- a/swarm-queue-client/src/status.rs +++ b/swarm-queue-client/src/status.rs @@ -67,7 +67,7 @@ pub async fn open_or_create( }) .await .map_err(|source| Error::CreateBucket { - bucket: BUCKET.to_owned(), + bucket: BUCKET, source, }) } diff --git a/swarm-queue-client/src/wanted.rs b/swarm-queue-client/src/wanted.rs index 5af35c63..f9d6a4ed 100644 --- a/swarm-queue-client/src/wanted.rs +++ b/swarm-queue-client/src/wanted.rs @@ -1,7 +1,5 @@ -//! 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`. +//! The hive-wanted KV bucket: the agent set the controller declares for each +//! hive, keyed 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 @@ -26,32 +24,13 @@ #[cfg(feature = "kv")] use crate::Error; -/// The prefix every hive's wanted-state bucket name starts with. +/// The KV bucket the controller publishes per-hive wanted state into, one key +/// per hive keyed by `hiveName`. /// -/// Derived from, and not configurable, for the reason [`crate::status::BUCKET`] +/// A constant and not an option, 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_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}") -} +pub const BUCKET: &str = "hive-wanted"; /// One hive's whole declaration: the agents the controller names, and what it /// wants of each. The value under `BUCKET`/``. @@ -116,26 +95,27 @@ 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()); - let bucket = bucket(hive); - match js.get_key_value(&bucket).await { + 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.clone(), - description: format!("Agent set the swarm controller declares for hive {hive}"), + bucket: BUCKET.to_owned(), + description: "Agent set the swarm controller declares for each hive".to_owned(), history: 1, ..Default::default() }) .await - .map_err(|source| Error::CreateBucket { bucket, source }) + .map_err(|source| Error::CreateBucket { + bucket: BUCKET, + source, + }) } } } @@ -148,45 +128,14 @@ 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(hive)).await.ok() + js.get_key_value(BUCKET).await.ok() } #[cfg(test)] mod tests { - 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)); - } + use super::{AgentState, HiveWanted}; #[test] fn the_known_states_decode_and_round_trip() {