diff --git a/hive-c0re/src/workers/wanted.rs b/hive-c0re/src/workers/wanted.rs index 6fd39393..4ade198a 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 writes one key per hive into the `hive-wanted` bucket (see +//! The controller gives each hive its own `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).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(()); }; diff --git a/swarm-controller/src/wanted.rs b/swarm-controller/src/wanted.rs index b5d2c7ba..4e25d80a 100644 --- a/swarm-controller/src/wanted.rs +++ b/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, } 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 { + 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 { 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 diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index f58fc1ab..4a0b5520 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 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..` 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..` 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..*` 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] diff --git a/swarm-queue-client/src/agent_status.rs b/swarm-queue-client/src/agent_status.rs index 9311fc6b..65104bc5 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, + bucket: BUCKET.to_owned(), source, }) } 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/status.rs b/swarm-queue-client/src/status.rs index 1f0e69c2..3e019e86 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, + bucket: BUCKET.to_owned(), source, }) } diff --git a/swarm-queue-client/src/wanted.rs b/swarm-queue-client/src/wanted.rs index f9d6a4ed..5af35c63 100644 --- a/swarm-queue-client/src/wanted.rs +++ b/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.` 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 +116,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 +148,45 @@ 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() {