wip(#4006): per-hive wanted-state buckets — client crate only
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-<hive>;
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.
This commit is contained in:
parent
d7c171b805
commit
8c94e340b8
2 changed files with 66 additions and 17 deletions
|
|
@ -111,7 +111,9 @@ pub enum Error {
|
||||||
#[cfg(feature = "kv")]
|
#[cfg(feature = "kv")]
|
||||||
#[error("creating the {bucket} bucket")]
|
#[error("creating the {bucket} bucket")]
|
||||||
CreateBucket {
|
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]
|
||||||
source: async_nats::jetstream::context::CreateKeyValueError,
|
source: async_nats::jetstream::context::CreateKeyValueError,
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,6 @@
|
||||||
//! The hive-wanted KV bucket: the agent set the controller declares for each
|
//! The hive-wanted KV buckets: the agent set the controller declares for each
|
||||||
//! hive, keyed by `hiveName`.
|
//! 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
|
//! 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
|
//! one is **observed** — each hive republishes what it is, so its store losing
|
||||||
|
|
@ -24,13 +25,32 @@
|
||||||
#[cfg(feature = "kv")]
|
#[cfg(feature = "kv")]
|
||||||
use crate::Error;
|
use crate::Error;
|
||||||
|
|
||||||
/// The KV bucket the controller publishes per-hive wanted state into, one key
|
/// The prefix every hive's wanted-state bucket name starts with.
|
||||||
/// per hive keyed by `hiveName`.
|
|
||||||
///
|
///
|
||||||
/// 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
|
/// 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.
|
/// 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
|
/// One hive's whole declaration: the agents the controller names, and what it
|
||||||
/// wants of each. The value under `BUCKET`/`<hiveName>`.
|
/// wants of each. The value under `BUCKET`/`<hiveName>`.
|
||||||
|
|
@ -95,27 +115,26 @@ impl AgentState {
|
||||||
#[cfg(feature = "kv")]
|
#[cfg(feature = "kv")]
|
||||||
pub async fn open_or_create(
|
pub async fn open_or_create(
|
||||||
client: &async_nats::Client,
|
client: &async_nats::Client,
|
||||||
|
hive: &str,
|
||||||
) -> Result<async_nats::jetstream::kv::Store, Error> {
|
) -> Result<async_nats::jetstream::kv::Store, Error> {
|
||||||
let js = async_nats::jetstream::new(client.clone());
|
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),
|
Ok(store) => Ok(store),
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
tracing::info!(
|
tracing::info!(
|
||||||
bucket = BUCKET,
|
%bucket,
|
||||||
reason = %e,
|
reason = %e,
|
||||||
"wanted-state bucket not available, creating it"
|
"wanted-state bucket not available, creating it"
|
||||||
);
|
);
|
||||||
js.create_key_value(async_nats::jetstream::kv::Config {
|
js.create_key_value(async_nats::jetstream::kv::Config {
|
||||||
bucket: BUCKET.to_owned(),
|
bucket: bucket.clone(),
|
||||||
description: "Agent set the swarm controller declares for each hive".to_owned(),
|
description: format!("Agent set the swarm controller declares for hive {hive}"),
|
||||||
history: 1,
|
history: 1,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
.map_err(|source| Error::CreateBucket {
|
.map_err(|source| Error::CreateBucket { bucket, source })
|
||||||
bucket: BUCKET,
|
|
||||||
source,
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -128,14 +147,42 @@ pub async fn open_or_create(
|
||||||
#[cfg(feature = "kv")]
|
#[cfg(feature = "kv")]
|
||||||
pub async fn open_read_only(
|
pub async fn open_read_only(
|
||||||
client: &async_nats::Client,
|
client: &async_nats::Client,
|
||||||
|
hive: &str,
|
||||||
) -> Option<async_nats::jetstream::kv::Store> {
|
) -> Option<async_nats::jetstream::kv::Store> {
|
||||||
let js = async_nats::jetstream::new(client.clone());
|
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)]
|
#[cfg(test)]
|
||||||
mod tests {
|
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]
|
#[test]
|
||||||
fn the_known_states_decode_and_round_trip() {
|
fn the_known_states_decode_and_round_trip() {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue