vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

Compare commits

base: vinzenz:c5b86afcb0cff1c6ff3a8c6632f177bae92eb976
Branches Tags
vinzenz:main
..
compare: vinzenz:d7c171b8057298428b8f34ce50ffbe887e27ffd6
Branches Tags
vinzenz:main
7 changed files with 86 additions and 184 deletions
Download patch file Download diff file Expand all files Collapse all files

4
hive-c0re/src/workers/wanted.rs
View file

@ -1,6 +1,6 @@
//! Converging this hive onto the agent set the swarm controller declares.
//!
//! The controller gives each hive its own `hive-wanted-<hive>` 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<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, &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(());
};

37
swarm-controller/src/wanted.rs
View file

@ -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<async_nats::jetstream::kv::Store>,
}
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<async_nats::jetstream::kv::Store, swarm_queue_client::Error> {
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<HiveWanted> {
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

138
swarm-nats-auth/src/policy.rs
View file

@ -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.<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);
// 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.
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.<hive>.*` 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]

2
swarm-queue-client/src/agent_status.rs
View file

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

4
swarm-queue-client/src/lib.rs
View file

@ -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,
},

2
swarm-queue-client/src/status.rs
View file

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

83
swarm-queue-client/src/wanted.rs
View file

@ -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.<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}")
}
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`/`<hiveName>`.
@ -116,26 +95,27 @@ 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());
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<async_nats::jetstream::kv::Store> {
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() {