hyperhive/hive-c0re/src/swarm_queue.rs

//! 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-<name>`, 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.
//!
//! The queue's address, the token endpoint and the client id come from the
//! environment. The client secret comes from the swarm's secret store, at
//! `swarm_secret_client::queue::hive_client_path`, read under this hive's own
//! certificate, held only in memory and read again when authelia refuses it
//! (`swarm_queue_client::stored_secret`). `swarm-secret-publish`, in authelia's
//! container, puts it there.
//!
//! 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. A failed attempt is retried after a
//! backoff capped at [`MAX_RETRY_DELAY`], so a secret published after this
//! daemon started is picked up without a restart.

use std::future::Future;
use std::path::PathBuf;
use std::time::Duration;

use anyhow::{Context, Result, anyhow, bail};
use swarm_queue_client::stored_secret::StoredSecret;
use swarm_queue_client::{ClientSecret, QueueConfig};
use swarm_secret_client::queue::{self, HiveClientCredential};
use tokio::sync::Mutex;
use tokio::time::Instant;

/// Env var prefix for this daemon's swarm-queue coordinates — see
/// [`coordinates_from`].
const ENV_PREFIX: &str = "HIVE_C0RE";

/// Wait after the first failed attempt; doubled after each further one.
const FIRST_RETRY_DELAY: Duration = Duration::from_secs(5);

/// The longest wait between two attempts.
const MAX_RETRY_DELAY: Duration = Duration::from_mins(5);

/// The banner kind a failed attempt raises, cleared by the next success.
const BANNER: &str = "swarm_queue_config";

static CLIENT: Retrying<async_nats::Client> = Retrying::new(FIRST_RETRY_DELAY, MAX_RETRY_DELAY);

/// The shared swarm-queue client, or `None` when there is none right now.
///
/// `None` covers "no queue configured" (the ordinary case, logged once at
/// `info`), a deployment bug in the coordinates (bannered, and permanent for
/// this process), and an attempt that failed or is still backing off. A
/// caller with `None` should just skip whatever it was about to publish.
/// Never waits out a backoff: see [`wait_for_client`] for that.
pub async fn client() -> Option<async_nats::Client> {
    match CLIENT.get(connect_once).await {
        Outcome::Ready(client) => Some(client),
        Outcome::Off | Outcome::NotBefore(_) => None,
    }
}

/// The shared swarm-queue client, waiting through every backoff until there
/// is one. `None` only when this hive publishes nothing to a queue at all.
pub async fn wait_for_client() -> Option<async_nats::Client> {
    loop {
        match CLIENT.get(connect_once).await {
            Outcome::Ready(client) => return Some(client),
            Outcome::Off => return None,
            Outcome::NotBefore(at) => tokio::time::sleep_until(at).await,
        }
    }
}

/// What one connection attempt came to.
enum Attempt<T> {
    /// Nothing to connect to, for the life of the process.
    Off,
    Ready(T),
    /// Worth trying again later.
    Failed(anyhow::Error),
}

/// What [`Retrying::get`] answers.
enum Outcome<T> {
    Off,
    Ready(T),
    /// The last attempt failed; the next one is not tried before this.
    NotBefore(Instant),
}

enum Slot<T> {
    Off,
    Ready(T),
    /// `failures` consecutive attempts have failed. `banner` holds the
    /// dashboard warning until the slot leaves this state.
    Failing {
        failures: u32,
        not_before: Instant,
        banner: Option<crate::warnings::WarningGuard>,
    },
}

/// A value made by an attempt that can fail: a success or an `Off` is kept
/// for the life of the process, a failure only until its backoff runs out.
struct Retrying<T> {
    slot: Mutex<Option<Slot<T>>>,
    first_delay: Duration,
    max_delay: Duration,
}

impl<T: Clone> Retrying<T> {
    const fn new(first_delay: Duration, max_delay: Duration) -> Self {
        Self {
            slot: Mutex::const_new(None),
            first_delay,
            max_delay,
        }
    }

    /// The kept value, or one new `attempt` if the last one's backoff has run
    /// out. Concurrent callers wait on one attempt rather than each making
    /// their own.
    async fn get<F, Fut>(&self, attempt: F) -> Outcome<T>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = Attempt<T>>,
    {
        let mut slot = self.slot.lock().await;
        let failures = match &*slot {
            None => 0,
            Some(Slot::Off) => return Outcome::Off,
            Some(Slot::Ready(value)) => return Outcome::Ready(value.clone()),
            Some(Slot::Failing {
                failures,
                not_before,
                ..
            }) => {
                if Instant::now() < *not_before {
                    return Outcome::NotBefore(*not_before);
                }
                *failures
            }
        };
        match attempt().await {
            Attempt::Off => {
                *slot = Some(Slot::Off);
                Outcome::Off
            }
            Attempt::Ready(value) => {
                if failures > 0 {
                    tracing::info!(failures, "swarm queue: connected after earlier failures");
                }
                *slot = Some(Slot::Ready(value.clone()));
                Outcome::Ready(value)
            }
            Attempt::Failed(e) => {
                let failures = failures.saturating_add(1);
                let delay = self.delay_after(failures);
                let not_before = Instant::now() + delay;
                let message = format!("swarm queue is off, retrying in {delay:?}: {e:#}");
                tracing::warn!(failures, "{message}");
                // Updating the held guard keeps the banner up without a
                // moment where it is cleared and re-raised.
                let banner = match slot.take() {
                    Some(Slot::Failing {
                        banner: Some(banner),
                        ..
                    }) => {
                        banner.update("warn", message);
                        banner
                    }
                    _ => crate::warnings::set_warning(BANNER, "warn", message),
                };
                *slot = Some(Slot::Failing {
                    failures,
                    not_before,
                    banner: Some(banner),
                });
                Outcome::NotBefore(not_before)
            }
        }
    }

    /// `first_delay` doubled per failure after the first, capped at
    /// `max_delay`.
    fn delay_after(&self, failures: u32) -> Duration {
        let doublings = failures.saturating_sub(1).min(16);
        self.first_delay
            .saturating_mul(1 << doublings)
            .min(self.max_delay)
    }
}

async fn connect_once() -> Attempt<async_nats::Client> {
    let coordinates = match coordinates_from(|name| std::env::var(name).ok()) {
        Ok(Some(c)) => c,
        Ok(None) => {
            tracing::info!("no swarm queue configured; this hive offers nothing upward");
            return Attempt::Off;
        }
        Err(e) => {
            // A deployment bug, which no retry clears.
            crate::warnings::set_boot_warning(BANNER, "warn", format!("swarm queue is off: {e:#}"));
            return Attempt::Off;
        }
    };
    let secret = match fetch_secret().await {
        Ok(secret) => secret,
        Err(e) => return Attempt::Failed(e),
    };
    match swarm_queue_client::connect(coordinates.with_secret(secret)).await {
        Ok(client) => Attempt::Ready(client),
        // `chain`, not `{:#}`: this is `swarm_queue_client::Error`, whose
        // `Display` ignores the alternate flag.
        Err(e) => Attempt::Failed(anyhow!(swarm_queue_client::chain(&e))),
    }
}

/// Everything in a [`QueueConfig`] except the secret.
#[derive(Debug, PartialEq, Eq)]
struct Coordinates {
    url: String,
    token_endpoint: String,
    client_id: String,
    ca_file: Option<PathBuf>,
}

impl Coordinates {
    fn with_secret(self, secret: String) -> QueueConfig {
        let stored = StoredSecret::new(self.client_id.clone(), secret, || async {
            fetch_secret().await.map_err(Into::into)
        });
        QueueConfig {
            url: self.url,
            token_endpoint: self.token_endpoint,
            client_id: self.client_id,
            client_secret: ClientSecret::Stored(stored),
            ca_file: self.ca_file,
        }
    }
}

/// `<ENV_PREFIX>_NATS_URL`, `_OIDC_TOKEN_ENDPOINT` and `_OIDC_CLIENT_ID`, all
/// or none; `_OIDC_CA_FILE` optionally.
fn coordinates_from(var: impl Fn(&str) -> Option<String>) -> Result<Option<Coordinates>> {
    let url = var(&format!("{ENV_PREFIX}_NATS_URL"));
    let token_endpoint = var(&format!("{ENV_PREFIX}_OIDC_TOKEN_ENDPOINT"));
    let client_id = var(&format!("{ENV_PREFIX}_OIDC_CLIENT_ID"));
    let ca_file = var(&format!("{ENV_PREFIX}_OIDC_CA_FILE")).map(PathBuf::from);
    match (url, token_endpoint, client_id) {
        (None, None, None) => Ok(None),
        (Some(url), Some(token_endpoint), Some(client_id)) => Ok(Some(Coordinates {
            url,
            token_endpoint,
            client_id,
            ca_file,
        })),
        _ => bail!(
            "swarm queue is half-configured: {ENV_PREFIX}_NATS_URL, \
             {ENV_PREFIX}_OIDC_TOKEN_ENDPOINT and {ENV_PREFIX}_OIDC_CLIENT_ID must be set \
             together or not at all"
        ),
    }
}

/// One login and one read, under this hive's own certificate. An empty value
/// is refused like an absent one: it would reach authelia as the secret and
/// come back as `invalid_client`, which names neither the store nor the
/// publisher.
async fn fetch_secret() -> Result<String> {
    let Some(hive) = crate::container_view::hive_swarm_names().0 else {
        bail!("HYPERHIVE_HIVE_NAME is unset, so there is no store path for this hive's secret");
    };
    let path = queue::hive_client_path(&hive)?;
    let store = swarm_secret_client::SecretStore::from_env(&hive)
        .await
        .context("logging in to the swarm secret store")?;
    let stored: Option<HiveClientCredential> = store
        .read_optional(&path)
        .await
        .with_context(|| format!("reading {path}"))?;
    match stored {
        Some(c) if !c.value.trim().is_empty() => Ok(c.value.trim().to_owned()),
        _ => bail!(
            "nothing at {path}: swarm-secret-publish, in authelia's container, has not \
             published this hive's client secret"
        ),
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use super::*;

    fn env(pairs: &[(&str, &str)]) -> impl Fn(&str) -> Option<String> {
        let map: HashMap<String, String> = pairs
            .iter()
            .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
            .collect();
        move |name| map.get(name).cloned()
    }

    const FULL: [(&str, &str); 3] = [
        ("HIVE_C0RE_NATS_URL", "tls://nats.example:4222"),
        (
            "HIVE_C0RE_OIDC_TOKEN_ENDPOINT",
            "https://auth.example/api/oidc/token",
        ),
        ("HIVE_C0RE_OIDC_CLIENT_ID", "hive-alpha"),
    ];

    #[test]
    fn no_coordinates_is_no_queue() {
        assert_eq!(
            coordinates_from(env(&[])).expect("absent is not an error"),
            None
        );
    }

    #[test]
    fn a_half_set_environment_is_an_error_naming_the_variables() {
        let err = coordinates_from(env(&FULL[..2])).expect_err("a partial set is not absent");
        assert!(
            format!("{err}").contains("HIVE_C0RE_OIDC_CLIENT_ID"),
            "{err}"
        );
    }

    #[tokio::test]
    async fn a_full_environment_carries_the_stored_secret_in_memory() {
        let cfg = coordinates_from(env(&FULL))
            .expect("complete")
            .expect("configured")
            .with_secret("s3cret".to_owned());
        assert_eq!(cfg.client_id, "hive-alpha");
        assert!(matches!(&cfg.client_secret, ClientSecret::Stored(s) if s.current() == "s3cret"));
    }

    /// The secret is not in the store when this daemon first asks, and is
    /// later: the later value is used, with no restart in between.
    #[tokio::test]
    async fn a_failed_first_read_is_retried_and_its_later_secret_used() {
        let shared = Retrying::<String>::new(Duration::ZERO, Duration::ZERO);

        let first = shared
            .get(|| async { Attempt::Failed(anyhow!("nothing at the path yet")) })
            .await;
        assert!(
            matches!(first, Outcome::NotBefore(_)),
            "the first read failed"
        );

        let second = shared
            .get(|| async { Attempt::Ready("published-later".to_owned()) })
            .await;
        assert!(
            matches!(&second, Outcome::Ready(s) if s == "published-later"),
            "the read after the failure is made and its secret used"
        );

        let kept = shared
            .get(|| async { panic!("a success is kept, not re-read") })
            .await;
        assert!(matches!(&kept, Outcome::Ready(s) if s == "published-later"));
    }

    #[tokio::test(start_paused = true)]
    async fn a_failure_is_not_retried_before_its_backoff_runs_out() {
        let shared = Retrying::<String>::new(Duration::from_secs(5), Duration::from_mins(5));
        let failed = shared
            .get(|| async { Attempt::Failed(anyhow!("down")) })
            .await;
        let Outcome::NotBefore(at) = failed else {
            panic!("a failure backs off");
        };
        assert_eq!(at - Instant::now(), Duration::from_secs(5));

        let early = shared
            .get(|| async { panic!("no attempt inside the backoff") })
            .await;
        assert!(matches!(early, Outcome::NotBefore(t) if t == at));

        tokio::time::advance(Duration::from_secs(5)).await;
        let due = shared
            .get(|| async { Attempt::Ready("up".to_owned()) })
            .await;
        assert!(matches!(&due, Outcome::Ready(s) if s == "up"));
    }

    #[test]
    fn the_backoff_doubles_up_to_its_cap() {
        let shared = Retrying::<String>::new(Duration::from_secs(5), Duration::from_mins(5));
        let delays: Vec<u64> = (1..=8).map(|n| shared.delay_after(n).as_secs()).collect();
        assert_eq!(delays, [5, 10, 20, 40, 80, 160, 300, 300]);
        assert_eq!(shared.delay_after(u32::MAX), Duration::from_mins(5));
    }

    #[tokio::test]
    async fn off_is_kept_for_the_life_of_the_process() {
        let shared = Retrying::<String>::new(Duration::ZERO, Duration::ZERO);
        assert!(matches!(
            shared.get(|| async { Attempt::Off }).await,
            Outcome::Off
        ));
        let again = shared.get(|| async { panic!("off is not re-tried") }).await;
        assert!(matches!(again, Outcome::Off));
    }
}