From e44ea9d8d450bc299d8faea276bf43e4f95d7c7f Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 16 Aug 2026 15:58:45 +0200 Subject: [PATCH] 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, + }) + } + } +}