swarm-queue-based lifecycle notices, replacing push_todo(MANAGER_AGENT)

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<Mutex<SweepHealth>> = 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<String>, summary: String, source: Option<String>) {
+    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(())
+}