From e44ea9d8d450bc299d8faea276bf43e4f95d7c7f Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 16 Aug 2026 15:58:45 +0200 Subject: [PATCH 1/3] swarm-queue-based lifecycle notices, replacing push_todo(MANAGER_AGENT) --- hive-c0re/Cargo.toml | 2 +- hive-c0re/src/job_queue/exec.rs | 45 ++---- hive-c0re/src/main.rs | 2 + hive-c0re/src/server.rs | 36 ++--- .../src/socket_server/lifecycle_handlers.rs | 19 +-- hive-c0re/src/swarm_notices.rs | 140 ++++++++++++++++++ hive-c0re/src/swarm_queue.rs | 72 +++++++++ hive-c0re/src/swarm_status.rs | 67 ++------- hive-c0re/src/workers/crash_watch.rs | 36 ++--- swarm-queue-client/Cargo.toml | 6 + swarm-queue-client/src/lib.rs | 14 ++ swarm-queue-client/src/notices.rs | 91 ++++++++++++ 12 files changed, 388 insertions(+), 142 deletions(-) create mode 100644 hive-c0re/src/swarm_notices.rs create mode 100644 hive-c0re/src/swarm_queue.rs create mode 100644 swarm-queue-client/src/notices.rs diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 68440cc2..a1c9084f 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -53,7 +53,7 @@ serde_json.workspace = true # Offering this hive's status to the swarm (`swarm_status`). The same crate # the swarm controller reads it with, and `kv` for the same reason: the # bucket's name and creation config belong to neither end of it alone. -swarm-queue-client = { workspace = true, features = ["kv"] } +swarm-queue-client = { workspace = true, features = ["kv", "notices"] } tokio.workspace = true tokio-stream.workspace = true tracing.workspace = true diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 5d55b9cf..321fee42 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -326,16 +326,13 @@ async fn run_destroy_bookkeeping(coord: &Arc, agent: &str, purge: b if let Err(e) = coord.power.remove(agent) { tracing::warn!(%agent, error = ?e, "agent_power: remove on destroy failed"); } - let _ = coord - .push_todo( - hive_sh4re::manager::MANAGER_AGENT, - "core", - Some(format!("destroyed:{agent}")), - format!("agent '{agent}' destroyed"), - None, - false, - ) - .await; + crate::swarm_notices::notify( + "core", + Some(format!("destroyed:{agent}")), + format!("agent '{agent}' destroyed"), + None, + ) + .await; // Container row disappeared — rescan so the dashboard fires // `ContainerRemoved` for the gone row, then emit the tombstones snapshot // (gained one on destroy, lost one on purge — recompute either way). @@ -358,16 +355,7 @@ async fn run_emit_rebuilt(coord: &Arc, agent: &str, dag_id: Option< .then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag))) .flatten(); let summary = crate::coordinator::rebuilt_todo_summary(agent, ok, note.as_deref(), None, None); - let _ = coord - .push_todo( - hive_sh4re::manager::MANAGER_AGENT, - "core", - Some(format!("rebuilt:{agent}")), - summary, - None, - false, - ) - .await; + crate::swarm_notices::notify("core", Some(format!("rebuilt:{agent}")), summary, None).await; } /// Write the agent's durable power intent — the DAG-node form of the old @@ -622,16 +610,13 @@ async fn run_stop(coord: &Arc, name: &str) -> Result<()> { // own kind now. crate::lifecycle::kill(name).await?; coord.unregister_agent(name); - let _ = coord - .push_todo( - hive_sh4re::manager::MANAGER_AGENT, - "core", - Some(format!("killed:{name}")), - format!("agent '{name}' killed"), - None, - false, - ) - .await; + crate::swarm_notices::notify( + "core", + Some(format!("killed:{name}")), + format!("agent '{name}' killed"), + None, + ) + .await; coord.rescan_containers_and_emit().await; Ok(()) } diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 5e5d6f06..f0274ce5 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -35,6 +35,8 @@ mod snapshot_push; mod socket_server; mod stats; mod stores; +mod swarm_notices; +mod swarm_queue; mod swarm_status; #[cfg(test)] mod test_env; diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 753e67c3..36049d75 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -296,32 +296,26 @@ async fn handle_spawn(coord: &Arc, name: &str) -> Result { // Spawn failed: register_agent was never called, so there is - // nothing to unregister. Notify the manager and propagate. - let _ = coord - .push_todo( - hive_sh4re::manager::MANAGER_AGENT, - "core", - Some(format!("spawned:{name}")), - format!("agent '{name}' spawn FAILED: {e:#}"), - None, - false, - ) - .await; + // nothing to unregister. Notify the swarm and propagate. + crate::swarm_notices::notify( + "core", + Some(format!("spawned:{name}")), + format!("agent '{name}' spawn FAILED: {e:#}"), + None, + ) + .await; return Err(e); } } diff --git a/hive-c0re/src/socket_server/lifecycle_handlers.rs b/hive-c0re/src/socket_server/lifecycle_handlers.rs index 8feda850..e7b497b8 100644 --- a/hive-c0re/src/socket_server/lifecycle_handlers.rs +++ b/hive-c0re/src/socket_server/lifecycle_handlers.rs @@ -118,7 +118,7 @@ async fn handle_restart_infra( } } -/// `Kill` — kill a container, unregister it, notify the manager. The caller +/// `Kill` — kill a container, unregister it, notify the swarm. The caller /// must be an ancestor of `name` in the topology. pub(super) async fn handle_kill(coord: &Arc, agent: &str, name: &str) -> Response { if let Some(err) = require_descendant(agent, name, "kill") { @@ -138,16 +138,13 @@ pub(super) async fn handle_kill(coord: &Arc, agent: &str, name: &st .await; match result { Ok(()) => { - let _ = coord - .push_todo( - hive_sh4re::manager::MANAGER_AGENT, - "core", - Some(format!("killed:{name}")), - format!("agent '{name}' killed"), - None, - false, - ) - .await; + crate::swarm_notices::notify( + "core", + Some(format!("killed:{name}")), + format!("agent '{name}' killed"), + None, + ) + .await; Response::Ok } Err(e) => Response::Err { diff --git a/hive-c0re/src/swarm_notices.rs b/hive-c0re/src/swarm_notices.rs new file mode 100644 index 00000000..d2b9ceb4 --- /dev/null +++ b/hive-c0re/src/swarm_notices.rs @@ -0,0 +1,140 @@ +//! Publishing lifecycle notices onto the swarm queue. +//! +//! Replaces the old `push_todo(MANAGER_AGENT, ...)` fallback the +//! lifecycle-notice call sites used to reach for when there was nobody +//! else to tell. Every hive is swarm-controlled now, so there is no case +//! left that needs a manager-agent recipient — this module has no such +//! fallback, on purpose, not by omission. +//! +//! **Why a stream and not the [`crate::swarm_status`] KV bucket shape**: +//! a status snapshot has a current value a late reader can always ask +//! for; a lifecycle notice ("container crashed at 04:12") does not — miss +//! it and there is nothing left to read later that says it happened. See +//! [`swarm_queue_client::notices`] for the stream this publishes into. +//! +//! **Best-effort, never fatal to the caller.** A hive with no queue +//! configured is a silent no-op (the ordinary case). A hive whose queue +//! is unreachable loses the swarm's visibility of the notice, not the +//! host's — `warn!` fires on every failed attempt regardless, and the +//! dashboard banners only after [`FAILURES_BEFORE_BANNER`] consecutive +//! misses, the same debounce shape [`crate::swarm_status`] uses and for +//! the same reason: a `warn!` that fires every call for three weeks is +//! indistinguishable from silence in practice. + +use anyhow::{Context as _, Result}; +use tokio::sync::{Mutex, OnceCell}; + +use crate::stats::sweep_health::{self, SweepHealth}; + +/// Consecutive failed publishes before the dashboard banners — same +/// value [`crate::swarm_status`] uses and for the same reason: a debounce +/// against one blip flapping a banner an operator learns to ignore. +const FAILURES_BEFORE_BANNER: u32 = 3; + +static HEALTH: OnceCell> = OnceCell::const_new(); + +async fn health() -> tokio::sync::MutexGuard<'static, SweepHealth> { + HEALTH + .get_or_init(|| async { + Mutex::new(SweepHealth::new( + "swarm_notices_publish", + "warn", + FAILURES_BEFORE_BANNER, + )) + }) + .await + .lock() + .await +} + +/// Publish one lifecycle notice for this hive. +/// +/// `subsystem`/`key`/`summary`/`source` carry the same meaning and the +/// same owned-`String` shape they did as `push_todo` arguments — a +/// drop-in replacement for that call, minus the recipient (there is +/// none) and `reopen_if_acked` (an inbox-todo concept with no equivalent +/// on an append-only stream). +pub async fn notify(subsystem: &str, key: Option, summary: String, source: Option) { + let Some(client) = crate::swarm_queue::client().await else { + return; + }; + // Same absent-name condition `swarm_status` bails on — that module + // already banners it under `swarm_status_config` the first time + // either of us hits it; nothing more to add from here. + let Some(hive) = crate::container_view::hive_swarm_names().0 else { + return; + }; + + match publish( + &client, + &hive, + subsystem, + key.as_deref(), + &summary, + source.as_deref(), + ) + .await + { + Ok(()) => health().await.record_ok(), + Err(e) => { + tracing::warn!(error = ?e, subsystem, summary, "swarm notice: publish failed"); + let err = format!("{e:#}"); + health().await.record_err(|ctx| { + let age = ctx.since_last_ok.map_or_else( + || "no success this session".to_owned(), + |d| format!("last ok {} ago", sweep_health::fmt_age(d)), + ); + format!( + "swarm notice publishing is failing ({} consecutive, {age}) \ + — notices are being lost, not just delayed: {err}", + ctx.consecutive + ) + }); + } + } +} + +#[derive(serde::Serialize)] +struct Notice<'a> { + subsystem: &'a str, + key: Option<&'a str>, + summary: &'a str, + source: Option<&'a str>, +} + +async fn publish( + client: &async_nats::Client, + hive: &str, + subsystem: &str, + key: Option<&str>, + summary: &str, + source: Option<&str>, +) -> Result<()> { + // An unconnected client does not fail a JetStream request, it hangs + // on it — see `ensure_connected`'s own doc comment for why this has + // to run before every such request, not just the first one. + swarm_queue_client::ensure_connected(client)?; + + // Ensures the stream exists; the handle itself is unused below — + // `Context::publish` routes by subject, it does not need the + // `Stream` object in hand. + swarm_queue_client::notices::open_or_create(client) + .await + .context("opening the notices stream")?; + + let payload = serde_json::to_vec(&Notice { + subsystem, + key, + summary, + source, + }) + .context("serialising the notice")?; + + let js = async_nats::jetstream::new(client.clone()); + js.publish(swarm_queue_client::notices::subject(hive), payload.into()) + .await + .context("publishing the notice")? + .await + .context("awaiting the notice's ack")?; + Ok(()) +} diff --git a/hive-c0re/src/swarm_queue.rs b/hive-c0re/src/swarm_queue.rs new file mode 100644 index 00000000..a79a069e --- /dev/null +++ b/hive-c0re/src/swarm_queue.rs @@ -0,0 +1,72 @@ +//! One swarm-queue connection, shared by every consumer in this process. +//! +//! [`swarm_status`](crate::swarm_status) and +//! [`swarm_notices`](crate::swarm_notices) both need the swarm queue, and +//! both authenticate as the *same* identity (`hive-`, minted for +//! this hive — see `swarm-authelia.nix`). Two independent `connect()` +//! calls would be two token mints and two live connections for one +//! identity, not two different credentials — the same shape that turned +//! `swarm-controller`'s own connect into the shared `swarm-queue-client` +//! crate in the first place, one layer up. This module is that same move +//! made again, this time between two consumers *inside* one process. +//! +//! Connects lazily on first use rather than at boot — nothing here +//! blocks `hive-c0re` starting up on hosts with no queue configured, +//! which is the ordinary case. + +use tokio::sync::OnceCell; + +/// Env var prefix for this daemon's swarm-queue credentials — see +/// [`swarm_queue_client::QueueConfig::from_env`]. All four or none. +const ENV_PREFIX: &str = "HIVE_C0RE"; + +static CLIENT: OnceCell> = OnceCell::const_new(); + +/// The shared swarm-queue client, connecting on first call and memoized +/// for the rest of the process's life. +/// +/// `None` covers both "no queue configured" (the ordinary case, logged +/// once at `info`) and "config present but connecting failed" (bannered +/// once via [`crate::warnings::set_boot_warning`] the first time this is +/// called) — either way, a caller with `None` should just skip whatever +/// it was about to publish. No caller needs to distinguish the two: both +/// mean "this hive is not offering anything to the swarm right now." +pub async fn client() -> Option { + CLIENT.get_or_init(connect_once).await.clone() +} + +async fn connect_once() -> Option { + let cfg = match swarm_queue_client::QueueConfig::from_env(ENV_PREFIX) { + Ok(Some(cfg)) => cfg, + Ok(None) => { + tracing::info!("no swarm queue configured; this hive offers nothing upward"); + return None; + } + Err(e) => { + // A one-shot startup step with no later retry to clear it — + // exactly what `set_boot_warning` is for. `chain`, not + // `{:#}`: this is `swarm_queue_client::Error`, whose + // `Display` ignores the alternate flag (see `chain`'s own + // doc comment), so `{:#}` would drop which env vars are + // actually missing. + crate::warnings::set_boot_warning( + "swarm_queue_config", + "warn", + format!("swarm queue is off: {}", swarm_queue_client::chain(&e)), + ); + return None; + } + }; + + match swarm_queue_client::connect(cfg).await { + Ok(client) => Some(client), + Err(e) => { + crate::warnings::set_boot_warning( + "swarm_queue_config", + "warn", + format!("swarm queue is off: {}", swarm_queue_client::chain(&e)), + ); + None + } + } +} diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs index d4310b91..6f71fc03 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -47,10 +47,6 @@ use crate::stats::sweep_health::{self, SweepHealth}; /// other silently re-tunes the swarm's definition of "quiet". pub const PUBLISH_INTERVAL: Duration = Duration::from_mins(1); -/// Env var prefix for this daemon's swarm-queue credentials — see -/// [`swarm_queue_client::QueueConfig::from_env`]. All four or none. -const ENV_PREFIX: &str = "HIVE_C0RE"; - /// Consecutive failed publishes before the dashboard banners. /// /// At [`PUBLISH_INTERVAL`] this is ~3 minutes of genuine failure, so a @@ -59,43 +55,15 @@ const FAILURES_BEFORE_BANNER: u32 = 3; /// Start the publish loop, if this deployment wired up a swarm queue. /// -/// Absent queue config is the ordinary case — most hives are not in a -/// swarm — so it is an `info` and not a warning. A *half*-set environment -/// is a different thing entirely and [`swarm_queue_client::QueueConfig::from_env`] -/// makes it a hard error; it is bannered here rather than swallowed, -/// because the failure it otherwise produces is a hive that looks fine -/// and silently never reports. +/// The connect itself is shared with every other swarm-queue consumer in +/// this process — see [`crate::swarm_queue`] for why one connection and +/// not one per consumer, and for where "no queue configured" vs. "queue +/// configured but unreachable" gets bannered. This function only decides +/// whether *status* has anything to offer once a client exists. pub fn spawn( coord: std::sync::Arc, mut shutdown: tokio::sync::watch::Receiver, ) { - let cfg = match swarm_queue_client::QueueConfig::from_env(ENV_PREFIX) { - Ok(Some(cfg)) => cfg, - Ok(None) => { - tracing::info!("no swarm queue configured; this hive offers no status upward"); - return; - } - Err(e) => { - // A one-shot startup step with no later retry to clear it — - // exactly what `set_boot_warning` is for. The fix is a - // redeploy, which restarts this process anyway. - // - // `chain`, not `{:#}`: this is the queue client's own error - // type, whose Display ignores the alternate flag, so `{:#}` - // would show only "swarm queue is half-configured" and drop - // which variables are missing. - crate::warnings::set_boot_warning( - "swarm_status_config", - "warn", - format!( - "swarm status publishing is off: {}", - swarm_queue_client::chain(&e) - ), - ); - return; - } - }; - let Some(hive) = crate::container_view::hive_swarm_names().0 else { crate::warnings::set_boot_warning( "swarm_status_config", @@ -107,26 +75,11 @@ pub fn spawn( }; tokio::spawn(async move { - // `retry_on_initial_connect` inside, so this returns a client - // that may not be connected yet rather than failing on a queue - // that comes up second. The publish below is what discovers that, - // and it is already the thing that reports it. - let client = match swarm_queue_client::connect(cfg).await { - Ok(client) => client, - Err(e) => { - // `chain` for the same reason as above: without it this - // banner reads "connecting to the swarm queue at " - // and drops the nats error that says why. - crate::warnings::set_boot_warning( - "swarm_status_config", - "warn", - format!( - "swarm status publishing is off: {}", - swarm_queue_client::chain(&e) - ), - ); - return; - } + let Some(client) = crate::swarm_queue::client().await else { + // Absent or failed — either way already handled (an `info` + // log or a `swarm_queue_config` banner) by the shared + // connector; nothing left to report here. + return; }; // The hive's ONE queue connection, now serving both directions: diff --git a/hive-c0re/src/workers/crash_watch.rs b/hive-c0re/src/workers/crash_watch.rs index cc15e6f3..1dd67f34 100644 --- a/hive-c0re/src/workers/crash_watch.rs +++ b/hive-c0re/src/workers/crash_watch.rs @@ -51,7 +51,6 @@ pub fn spawn(coord: Arc) { if seeded { emit_crash_transitions(&coord, &prev_running, ¤t_running); emit_login_transitions( - &coord, &prev_logged_in, ¤t_logged_in, &sub_agents, @@ -128,7 +127,6 @@ fn is_deliberate_stop(active: Option, recently_cleared: Option) -> b } async fn emit_login_transitions( - coord: &Coordinator, prev: &HashSet, current: &HashSet, sub_agents: &[String], @@ -136,16 +134,13 @@ async fn emit_login_transitions( ) { for agent in current.difference(prev) { tracing::info!(%agent, "agent logged in"); - let _ = coord - .push_todo( - hive_sh4re::manager::MANAGER_AGENT, - "core", - Some(format!("logged_in:{agent}")), - format!("agent '{agent}' logged in"), - None, - false, - ) - .await; + crate::swarm_notices::notify( + "core", + Some(format!("logged_in:{agent}")), + format!("agent '{agent}' logged in"), + None, + ) + .await; } // Detect transitions into "needs login": an agent that was previously // logged-in goes unsigned (credentials deleted), OR a brand-new agent @@ -168,16 +163,13 @@ async fn emit_login_transitions( .collect(); for agent in current_needs.difference(&prev_needs) { tracing::info!(%agent, "agent needs login"); - let _ = coord - .push_todo( - hive_sh4re::manager::MANAGER_AGENT, - "core", - Some(format!("needs_login:{agent}")), - format!("agent '{agent}' needs login"), - None, - false, - ) - .await; + crate::swarm_notices::notify( + "core", + Some(format!("needs_login:{agent}")), + format!("agent '{agent}' needs login"), + None, + ) + .await; } } diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 0c25927d..dc3aa56e 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -17,6 +17,12 @@ edition.workspace = true # to live in one of them, and neither end of that bucket is senior to the # other. kv = ["async-nats/kv"] +# `jetstream` (streams, publish, consumers) is already in async-nats's +# default feature set — `kv` above only adds the KV-specific type +# surface on top of it. This feature exists purely to keep `notices.rs` +# out of a consumer's compiled surface unless it asks for it, matching +# `kv`'s organizational role rather than gating a real async-nats flag. +notices = [] [dependencies] # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index c48f981b..5e5d956e 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -115,6 +115,14 @@ pub enum Error { #[source] source: async_nats::jetstream::context::CreateKeyValueError, }, + + #[cfg(feature = "notices")] + #[error("creating the {stream} stream")] + CreateStream { + stream: &'static str, + #[source] + source: async_nats::jetstream::context::CreateStreamError, + }, } /// Render an error and its source chain on one line. @@ -165,6 +173,12 @@ pub mod status; /// permitted at all — speaks neither `jetstream` nor `kv`. pub const KNOWLEDGE_SUBJECT: &str = "$SWARM.knowledge"; +/// The hive-notices stream, shared by the hive that publishes and +/// whatever eventually consumes it. Behind the `notices` feature, same +/// reason `status` is behind `kv` — see the module doc. +#[cfg(feature = "notices")] +pub mod notices; + /// Only the fields this needs; authelia returns several. #[derive(serde::Deserialize)] struct TokenResponse { diff --git a/swarm-queue-client/src/notices.rs b/swarm-queue-client/src/notices.rs new file mode 100644 index 00000000..6d2f2952 --- /dev/null +++ b/swarm-queue-client/src/notices.rs @@ -0,0 +1,91 @@ +//! The hive-notices stream: its name, subject shape, and how a hive +//! opens it to publish. +//! +//! Same reason [`crate::status`] exists rather than a bare `const` on +//! whichever side happens to need one first: a hive that publishes and a +//! swarm-level reader that eventually consumes live in different crates, +//! and a literal name repeated across both is an agreement nothing +//! checks. +//! +//! **This is a stream, not a bucket, and that is a real design choice — +//! not the same shape as [`crate::status`] wearing a different name.** +//! [`crate::status`]'s hive-status snapshot has a current value: a +//! reconnecting reader can always ask "what does this hive say *now*" +//! and get the true answer, so a KV bucket (last value per key) is the +//! right shape. A lifecycle notice ("container crashed at 04:12") has no +//! such steady state — miss the message and there is nothing left to +//! read later that would tell you it happened. That needs durable +//! delivery (a JetStream stream a consumer acks against), which is what +//! this module opens instead. +//! +//! Feature-gated (`notices`) for the same reason [`crate::status`] is +//! gated behind `kv`: the crate's other consumers (the auth-callout +//! responder, a hive that only publishes status) should not compile +//! against a stream shape they never touch. + +use crate::Error; + +/// The stream a hive publishes lifecycle notices into. +/// +/// A constant and not an option, matching [`crate::status::BUCKET`]: +/// reader and writer must name the same stream, and letting either side +/// pick its own name is how two deployments end up disagreeing about +/// which stream a notice actually landed in. +pub const STREAM: &str = "hive-notices"; + +/// Every hive's notices land under this subject prefix, one subject per +/// hive: `hive-notices.`. +/// +/// Not one subject per notice *kind* — a consumer that wants a specific +/// hive's notices subscribes to `notices_subject(hive)`; one that wants +/// the whole swarm's subscribes to `{PREFIX}.>`. The kind travels inside +/// the message payload instead, so adding a new notice kind is never a +/// subject-design change. +const PREFIX: &str = "hive-notices"; + +/// Build the subject a given hive's notices publish to. +#[must_use] +pub fn subject(hive: &str) -> String { + format!("{PREFIX}.{hive}") +} + +/// Open the notices stream, creating it if nothing has yet. +/// +/// **Retention is time-bounded (30 days), not unbounded.** A notice +/// this old has long since been superseded by whatever the hive is +/// doing now — keeping it forever buys nothing but disk, the same +/// argument `hive-forge`'s own bash-task retention makes elsewhere in +/// this workspace. +/// +/// Creating rather than requiring a provisioning step is the same call +/// [`crate::status::open_or_create`] makes and for the same reason: a +/// hive and a swarm-level consumer come up in no particular order, and +/// a stream that must pre-exist turns "deployed in the wrong order" +/// into a permanent, silent absence of data. +pub async fn open_or_create( + client: &async_nats::Client, +) -> Result { + let js = async_nats::jetstream::new(client.clone()); + match js.get_stream(STREAM).await { + Ok(stream) => Ok(stream), + Err(e) => { + tracing::info!( + stream = STREAM, + reason = %e, + "notices stream not available, creating it" + ); + js.create_stream(async_nats::jetstream::stream::Config { + name: STREAM.to_owned(), + description: Some("Lifecycle notices offered by each hive".to_owned()), + subjects: vec![format!("{PREFIX}.>")], + max_age: std::time::Duration::from_hours(30 * 24), + ..Default::default() + }) + .await + .map_err(|source| Error::CreateStream { + stream: STREAM, + source, + }) + } + } +} From 044e4020fab06d4d5896b328e2cd23b9c0885867 Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 17 Aug 2026 09:50:41 +0200 Subject: [PATCH 2/3] swarm-queue-client: drop notices::PREFIX, reuse STREAM --- swarm-queue-client/src/notices.rs | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/swarm-queue-client/src/notices.rs b/swarm-queue-client/src/notices.rs index 6d2f2952..ca1a8b35 100644 --- a/swarm-queue-client/src/notices.rs +++ b/swarm-queue-client/src/notices.rs @@ -33,20 +33,21 @@ use crate::Error; /// which stream a notice actually landed in. pub const STREAM: &str = "hive-notices"; -/// Every hive's notices land under this subject prefix, one subject per -/// hive: `hive-notices.`. +/// Build the subject a given hive's notices publish to. +/// +/// Every hive's notices land under `STREAM`, one subject per hive: +/// `hive-notices.`. /// /// Not one subject per notice *kind* — a consumer that wants a specific -/// hive's notices subscribes to `notices_subject(hive)`; one that wants -/// the whole swarm's subscribes to `{PREFIX}.>`. The kind travels inside -/// the message payload instead, so adding a new notice kind is never a -/// subject-design change. -const PREFIX: &str = "hive-notices"; - -/// Build the subject a given hive's notices publish to. +/// hive's notices subscribes to `subject(hive)`; one that wants the whole +/// swarm's subscribes to `{STREAM}.>`. The kind travels inside the message +/// payload instead, so adding a new notice kind is never a subject-design +/// change. Deliberately reuses `STREAM` rather than a second `PREFIX` +/// constant with the same value — one name for one fact, same reasoning +/// the module doc above gives for a shared `const` over a repeated literal. #[must_use] pub fn subject(hive: &str) -> String { - format!("{PREFIX}.{hive}") + format!("{STREAM}.{hive}") } /// Open the notices stream, creating it if nothing has yet. @@ -77,7 +78,7 @@ pub async fn open_or_create( js.create_stream(async_nats::jetstream::stream::Config { name: STREAM.to_owned(), description: Some("Lifecycle notices offered by each hive".to_owned()), - subjects: vec![format!("{PREFIX}.>")], + subjects: vec![format!("{STREAM}.>")], max_age: std::time::Duration::from_hours(30 * 24), ..Default::default() }) From 81adf2b6b4dbe2a86c867682f041a72af6759dcb Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 24 Aug 2026 12:38:21 +0200 Subject: [PATCH 3/3] swarm-nats-auth: grant every hive the shared hive-notices stream subjects --- swarm-nats-auth/Cargo.toml | 11 +++-- swarm-nats-auth/src/policy.rs | 92 +++++++++++++++++++++++++++++++++++ swarm-queue-client/Cargo.toml | 19 +++++--- 3 files changed, 111 insertions(+), 11 deletions(-) diff --git a/swarm-nats-auth/Cargo.toml b/swarm-nats-auth/Cargo.toml index 60b63ba5..20a52c4f 100644 --- a/swarm-nats-auth/Cargo.toml +++ b/swarm-nats-auth/Cargo.toml @@ -22,11 +22,12 @@ serde.workspace = true serde_json.workspace = true # The jti digest: base32hex(sha256(claims)) over every JWT this crate signs. sha2.workspace = true -# For `status::BUCKET` alone - the subjects a hive may publish to are derived -# from the bucket name, and the reader, the writer and this responder must -# name the same one. Deliberately WITHOUT the `kv` feature: this crate derives -# subject strings, it never opens the bucket. -swarm-queue-client.workspace = true +# For `status::BUCKET` and `notices::STREAM` - the subjects a hive may +# publish to are derived from these names, and every end that touches them +# must agree on the same one. Deliberately WITHOUT the `kv` feature: this +# crate derives subject strings, it never opens the bucket. `notices` +# is name-only too (no `jetstream`/`kv` surface), same reason. +swarm-queue-client = { workspace = true, features = ["notices"] } tokio.workspace = true tracing.workspace = true tracing-subscriber.workspace = true diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 1eb3e083..8c1fea29 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -180,6 +180,36 @@ impl Policy { format!("$JS.API.STREAM.CREATE.{}", self.stream()) } + /// What every hive needs to open the shared lifecycle-notices stream: + /// look it up, and create it if this hive is the first to arrive. + /// + /// **Not per-hive templated, unlike `extra_hive_subjects`.** The KV + /// bucket case above namespaces a shared resource *within* itself + /// (`$KV..`, one key per hive); the notices stream has + /// no such per-hive split at the `STREAM.INFO`/`STREAM.CREATE` layer + /// — the stream itself, not a slice of it, is what every hive's + /// `open_or_create` needs to reach before it can publish to its own + /// `hive-notices.` subject (which `extra_hive_subjects` already + /// covers, since that one *is* per-hive). Granting these two subjects + /// to every hive is therefore correct, not a widening: it is + /// `CREATE`/`INFO` on one named stream, the same shape already + /// measured safe for the hive-status bucket in [`Self::create`] — + /// create-if-absent, never `STREAM.UPDATE`, so no hive can reshape a + /// stream another hive (or the swarm-controller, once it reads from + /// this stream) already created. + fn notices_subjects() -> [String; 2] { + // Unlike `Self::stream()` above, no `KV_` prefix: the notices + // stream is a plain `JetStream` stream, not a KV bucket, so its + // NATS stream name *is* `swarm_queue_client::notices::STREAM` + // verbatim — the `KV_` prefix is `create_key_value`'s own + // convention, not something every stream carries. + let stream = swarm_queue_client::notices::STREAM; + [ + format!("$JS.API.STREAM.INFO.{stream}"), + format!("$JS.API.STREAM.CREATE.{stream}"), + ] + } + /// What one hive may publish: the account minimum, the bucket lookup, /// creation, and its **own** key. /// @@ -202,6 +232,7 @@ impl Policy { self.create(), format!("$KV.{}.{hive}", self.bucket), ]); + subjects.extend(Self::notices_subjects()); subjects.extend( self.extra_hive_subjects .iter() @@ -449,6 +480,67 @@ mod tests { assert_eq!(policy().permissions("hive-"), None); } + #[test] + fn a_hive_may_open_the_shared_notices_stream() { + // 🩸 The actual defect this closes: `open_or_create`'s `get_stream` + // call needs these two before a hive ever reaches its own + // `hive-notices.` publish subject, and neither has a + // `{hive}` to go through `extra_hive_subjects`. + let p = policy() + .permissions("hive-alpha") + .expect("a hive is admitted"); + assert!( + p.publish + .contains(&"$JS.API.STREAM.INFO.hive-notices".to_owned()) + ); + assert!( + p.publish + .contains(&"$JS.API.STREAM.CREATE.hive-notices".to_owned()) + ); + } + + #[test] + fn the_notices_stream_grant_is_identical_across_hives() { + // Unlike `$KV..` or an `extra_hive_subjects` + // template, these two subjects name the stream itself, not a + // per-hive slice of it — every hive gets the exact same two + // strings, and that is the correct shape, not an oversight. + let alpha = policy() + .permissions("hive-alpha") + .expect("a hive is admitted"); + let beta = policy() + .permissions("hive-beta") + .expect("a hive is admitted"); + for subject in [ + "$JS.API.STREAM.INFO.hive-notices", + "$JS.API.STREAM.CREATE.hive-notices", + ] { + assert!(alpha.publish.contains(&subject.to_owned())); + assert!(beta.publish.contains(&subject.to_owned())); + } + } + + #[test] + fn the_notices_grant_never_includes_stream_update_or_delete() { + // Same invariant `a_hive_grant_never_includes_the_jetstream_wildcard` + // holds for the status bucket, restated for the stream this fold + // grants CREATE/INFO on: create-if-absent must not become + // reshape-or-destroy. + let p = policy() + .permissions("hive-alpha") + .expect("a hive is admitted"); + assert!( + !p.publish + .iter() + .any(|s| s.contains("hive-notices") && s.contains("STREAM.UPDATE")) + ); + assert!( + !p.publish + .iter() + .any(|s| s.contains("hive-notices") && s.contains("STREAM.DELETE")) + ); + } + #[test] fn extra_subjects_are_scoped_to_the_hive_that_publishes_them() { // The extension point: a second stream (lifecycle notices, say) is diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index dc3aa56e..b0b63588 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -17,12 +17,19 @@ edition.workspace = true # to live in one of them, and neither end of that bucket is senior to the # other. kv = ["async-nats/kv"] -# `jetstream` (streams, publish, consumers) is already in async-nats's -# default feature set — `kv` above only adds the KV-specific type -# surface on top of it. This feature exists purely to keep `notices.rs` -# out of a consumer's compiled surface unless it asks for it, matching -# `kv`'s organizational role rather than gating a real async-nats flag. -notices = [] +# 🩸 `jetstream` is NOT in async-nats's default feature set here — the +# workspace-level dependency turns default features off entirely (see +# root `Cargo.toml`: `server_2_14`/`nkeys`/`ring` only). `kv` above works +# standalone only because async-nats's own `kv` feature pulls `jetstream` +# in transitively; `notices.rs` uses `async_nats::jetstream` directly and +# needs the same request explicitly, or it only compiles by accident when +# something else in the same build happens to also enable `kv` (which is +# exactly how this went unnoticed: `cargo test` at the workspace level +# unifies features across every crate being built, so `hive-c0re`'s own +# `kv` request silently carried `notices.rs` until a single-crate +# `cargo check -p swarm-nats-auth` — no `kv` anywhere in that build — +# surfaced it as `cannot find jetstream in async_nats`). +notices = ["async-nats/jetstream"] [dependencies] # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature