diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index a1c9084f..68440cc2 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", "notices"] } +swarm-queue-client = { workspace = true, features = ["kv"] } 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 321fee42..5d55b9cf 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -326,13 +326,16 @@ 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"); } - crate::swarm_notices::notify( - "core", - Some(format!("destroyed:{agent}")), - format!("agent '{agent}' destroyed"), - None, - ) - .await; + let _ = coord + .push_todo( + hive_sh4re::manager::MANAGER_AGENT, + "core", + Some(format!("destroyed:{agent}")), + format!("agent '{agent}' destroyed"), + None, + false, + ) + .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). @@ -355,7 +358,16 @@ 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); - crate::swarm_notices::notify("core", Some(format!("rebuilt:{agent}")), summary, None).await; + let _ = coord + .push_todo( + hive_sh4re::manager::MANAGER_AGENT, + "core", + Some(format!("rebuilt:{agent}")), + summary, + None, + false, + ) + .await; } /// Write the agent's durable power intent — the DAG-node form of the old @@ -610,13 +622,16 @@ async fn run_stop(coord: &Arc, name: &str) -> Result<()> { // own kind now. crate::lifecycle::kill(name).await?; coord.unregister_agent(name); - crate::swarm_notices::notify( - "core", - Some(format!("killed:{name}")), - format!("agent '{name}' killed"), - None, - ) - .await; + let _ = coord + .push_todo( + hive_sh4re::manager::MANAGER_AGENT, + "core", + Some(format!("killed:{name}")), + format!("agent '{name}' killed"), + None, + false, + ) + .await; coord.rescan_containers_and_emit().await; Ok(()) } diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index f0274ce5..5e5d6f06 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -35,8 +35,6 @@ 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 36049d75..753e67c3 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -296,26 +296,32 @@ async fn handle_spawn(coord: &Arc, name: &str) -> Result { // Spawn failed: register_agent was never called, so there is - // 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; + // 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; return Err(e); } } diff --git a/hive-c0re/src/socket_server/lifecycle_handlers.rs b/hive-c0re/src/socket_server/lifecycle_handlers.rs index e7b497b8..8feda850 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 swarm. The caller +/// `Kill` — kill a container, unregister it, notify the manager. 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,13 +138,16 @@ pub(super) async fn handle_kill(coord: &Arc, agent: &str, name: &st .await; match result { Ok(()) => { - crate::swarm_notices::notify( - "core", - Some(format!("killed:{name}")), - format!("agent '{name}' killed"), - None, - ) - .await; + let _ = coord + .push_todo( + hive_sh4re::manager::MANAGER_AGENT, + "core", + Some(format!("killed:{name}")), + format!("agent '{name}' killed"), + None, + false, + ) + .await; Response::Ok } Err(e) => Response::Err { diff --git a/hive-c0re/src/swarm_notices.rs b/hive-c0re/src/swarm_notices.rs deleted file mode 100644 index d2b9ceb4..00000000 --- a/hive-c0re/src/swarm_notices.rs +++ /dev/null @@ -1,140 +0,0 @@ -//! 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 deleted file mode 100644 index a79a069e..00000000 --- a/hive-c0re/src/swarm_queue.rs +++ /dev/null @@ -1,72 +0,0 @@ -//! 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 6f71fc03..d4310b91 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -47,6 +47,10 @@ 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 @@ -55,15 +59,43 @@ const FAILURES_BEFORE_BANNER: u32 = 3; /// Start the publish loop, if this deployment wired up a swarm queue. /// -/// 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. +/// 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. 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", @@ -75,11 +107,26 @@ pub fn spawn( }; tokio::spawn(async move { - 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; + // `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; + } }; // 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 1dd67f34..cc15e6f3 100644 --- a/hive-c0re/src/workers/crash_watch.rs +++ b/hive-c0re/src/workers/crash_watch.rs @@ -51,6 +51,7 @@ 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, @@ -127,6 +128,7 @@ fn is_deliberate_stop(active: Option, recently_cleared: Option) -> b } async fn emit_login_transitions( + coord: &Coordinator, prev: &HashSet, current: &HashSet, sub_agents: &[String], @@ -134,13 +136,16 @@ async fn emit_login_transitions( ) { for agent in current.difference(prev) { tracing::info!(%agent, "agent logged in"); - crate::swarm_notices::notify( - "core", - Some(format!("logged_in:{agent}")), - format!("agent '{agent}' logged in"), - None, - ) - .await; + let _ = coord + .push_todo( + hive_sh4re::manager::MANAGER_AGENT, + "core", + Some(format!("logged_in:{agent}")), + format!("agent '{agent}' logged in"), + None, + false, + ) + .await; } // Detect transitions into "needs login": an agent that was previously // logged-in goes unsigned (credentials deleted), OR a brand-new agent @@ -163,13 +168,16 @@ async fn emit_login_transitions( .collect(); for agent in current_needs.difference(&prev_needs) { tracing::info!(%agent, "agent needs login"); - crate::swarm_notices::notify( - "core", - Some(format!("needs_login:{agent}")), - format!("agent '{agent}' needs login"), - None, - ) - .await; + let _ = coord + .push_todo( + hive_sh4re::manager::MANAGER_AGENT, + "core", + Some(format!("needs_login:{agent}")), + format!("agent '{agent}' needs login"), + None, + false, + ) + .await; } } diff --git a/swarm-nats-auth/Cargo.toml b/swarm-nats-auth/Cargo.toml index 20a52c4f..60b63ba5 100644 --- a/swarm-nats-auth/Cargo.toml +++ b/swarm-nats-auth/Cargo.toml @@ -22,12 +22,11 @@ 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` 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"] } +# 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 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 8c1fea29..1eb3e083 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -180,36 +180,6 @@ 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. /// @@ -232,7 +202,6 @@ impl Policy { self.create(), format!("$KV.{}.{hive}", self.bucket), ]); - subjects.extend(Self::notices_subjects()); subjects.extend( self.extra_hive_subjects .iter() @@ -480,67 +449,6 @@ 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 b0b63588..0c25927d 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -17,19 +17,6 @@ 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` 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 diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 5e5d956e..c48f981b 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -115,14 +115,6 @@ 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. @@ -173,12 +165,6 @@ 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 deleted file mode 100644 index ca1a8b35..00000000 --- a/swarm-queue-client/src/notices.rs +++ /dev/null @@ -1,92 +0,0 @@ -//! 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"; - -/// 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 `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!("{STREAM}.{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!("{STREAM}.>")], - max_age: std::time::Duration::from_hours(30 * 24), - ..Default::default() - }) - .await - .map_err(|source| Error::CreateStream { - stream: STREAM, - source, - }) - } - } -}