hyperhive/hive-c0re/src/swarm_notices.rs

//! Publishing lifecycle notices onto the swarm queue.
//!
//! There is no fallback recipient agent: every hive is swarm-controlled,
//! so a notice always goes to the swarm queue, and no agent receives it by
//! name.
//!
//! **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 opening the stream this publishes into.
//!
//! **Best-effort, never fatal to the caller.** A hive with no queue
//! coordinates is a silent no-op (the ordinary case). A hive that cannot
//! reach the queue loses the swarm's visibility of the notice, not the
//! host's — `warn!` fires on every failed attempt regardless, and the
//! raises a hive warning only after [`FAILURES_BEFORE_WARNING`] 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 sweep raises a hive warning — same
/// value [`crate::swarm_status`] uses and for the same reason: a debounce
/// against one blip flapping a warning an operator learns to ignore.
const FAILURES_BEFORE_WARNING: 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_WARNING,
            ))
        })
        .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 warns 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_wire::notices::subject(hive), payload.into())
        .await
        .context("publishing the notice")?
        .await
        .context("awaiting the notice's ack")?;
    Ok(())
}