hyperhive/swarm-controller/src/attic_copy.rs

//! The cache facts a hive pulls with, copied to where the hive may read them.
//!
//! The cache's operator keeps four facts under `swarm/services/attic/`
//! ([`swarm_secret_client::attic`]). A hive's store identity reads
//! `swarm/hives/<hive>/*` and nothing under `swarm/services/`, so every
//! [`RECONCILE_INTERVAL`] this writes each fact that is stored to every hive's
//! `attic/` prefix, and leaves a copy that already holds the value alone.
//!
//! A fact that is not stored is not copied, and a copy whose source is gone
//! stays: the pass never deletes. A swarm with no cache stores none, so the
//! pass writes nothing.

use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result};
use swarm_secret_client::SecretStore;
use swarm_secret_client::attic::{Fact, Leaf, hive_path, service_path};

/// How often [`spawn`] copies.
const RECONCILE_INTERVAL: Duration = Duration::from_mins(5);

/// A pass that has not finished by now is given up on; the next tick retries.
const PASS_DEADLINE: Duration = Duration::from_mins(2);

/// The two store calls a pass makes, so a test can stand in for the store.
pub(crate) trait FactStore {
    /// The fact at `path`; `None` when nothing is stored there.
    async fn read(&self, path: &str) -> Result<Option<String>>;
    /// Store `value` at `path`, replacing what is there.
    async fn write(&self, path: &str, value: &str) -> Result<()>;
}

impl FactStore for SecretStore {
    async fn read(&self, path: &str) -> Result<Option<String>> {
        Ok(self.read_optional::<Fact>(path).await?.map(|f| f.value))
    }

    async fn write(&self, path: &str, value: &str) -> Result<()> {
        let fact = Fact {
            value: value.to_owned(),
        };
        Ok(SecretStore::write(self, path, &fact).await?)
    }
}

/// What one pass did.
#[derive(Debug, Default, PartialEq, Eq)]
pub(crate) struct Copied {
    /// Copies written because they were absent or held another value.
    pub written: usize,
    /// Hives whose copies could not all be checked or written.
    pub failed: usize,
}

/// The stored facts, as `(leaf, value)`. An empty value counts as not stored.
async fn sources(store: &impl FactStore) -> Result<Vec<(Leaf, String)>> {
    let mut found = Vec::new();
    for leaf in Leaf::ALL {
        let path = service_path(leaf)?;
        let stored = store
            .read(&path)
            .await
            .with_context(|| format!("reading {path}"))?;
        if let Some(value) = stored.filter(|v| !v.trim().is_empty()) {
            found.push((leaf, value));
        }
    }
    Ok(found)
}

/// Bring `hive`'s copies up to `sources`; how many were written.
async fn copy_hive(
    store: &impl FactStore,
    hive: &str,
    sources: &[(Leaf, String)],
) -> Result<usize> {
    let mut written = 0;
    for (leaf, value) in sources {
        let path = hive_path(hive, *leaf)?;
        let current = store
            .read(&path)
            .await
            .with_context(|| format!("reading {path}"))?;
        if current.as_deref() == Some(value.as_str()) {
            continue;
        }
        store
            .write(&path, value)
            .await
            .with_context(|| format!("writing {path}"))?;
        written += 1;
    }
    Ok(written)
}

/// One pass over `hives`.
///
/// # Errors
/// When a source cannot be read, which stops the pass before any copy: a
/// failed read is not an absent fact. A hive that fails is counted in
/// [`Copied::failed`] and logged, and the others are still copied.
pub(crate) async fn copy_pass(store: &impl FactStore, hives: &[String]) -> Result<Copied> {
    let sources = sources(store).await?;
    let mut done = Copied::default();
    for hive in hives {
        match copy_hive(store, hive, &sources).await {
            Ok(n) => done.written += n,
            Err(e) => {
                tracing::warn!(hive, error = %format!("{e:#}"), "attic facts: copy failed");
                done.failed += 1;
            }
        }
    }
    Ok(done)
}

/// Copy now and every [`RECONCILE_INTERVAL`] after. A pass that fails is
/// logged and retried on the next tick; it never stops the daemon.
pub fn spawn(hives: Vec<String>) {
    let hives = Arc::new(hives);
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
        loop {
            ticker.tick().await;
            let pass = async {
                let store = crate::store::connect()
                    .await
                    .context("logging in to the swarm secret store")?;
                copy_pass(&store, &hives).await
            };
            match tokio::time::timeout(PASS_DEADLINE, pass).await {
                Ok(Ok(done)) if done.written > 0 || done.failed > 0 => tracing::info!(
                    written = done.written,
                    failed = done.failed,
                    "attic facts: copied to the hives"
                ),
                Ok(Ok(_)) => tracing::debug!("attic facts: every hive's copy is current"),
                Ok(Err(e)) => tracing::warn!(
                    error = %format!("{e:#}"),
                    retry_in_s = RECONCILE_INTERVAL.as_secs(),
                    "attic facts: pass failed; retrying next tick"
                ),
                Err(_) => tracing::warn!(
                    deadline_s = PASS_DEADLINE.as_secs(),
                    "attic facts: pass timed out; retrying next tick"
                ),
            }
        }
    });
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;
    use std::sync::Mutex;

    use super::*;

    /// An in-memory store that counts writes and can refuse reads or writes
    /// under a prefix.
    #[derive(Default)]
    struct Memory {
        kv: Mutex<BTreeMap<String, String>>,
        writes: Mutex<Vec<String>>,
        refuse: Option<&'static str>,
    }

    impl Memory {
        fn with(pairs: &[(&str, &str)]) -> Self {
            let m = Self::default();
            for (path, value) in pairs {
                m.kv.lock()
                    .unwrap()
                    .insert((*path).to_owned(), (*value).to_owned());
            }
            m
        }

        fn get(&self, path: &str) -> Option<String> {
            self.kv.lock().unwrap().get(path).cloned()
        }
    }

    impl FactStore for Memory {
        async fn read(&self, path: &str) -> Result<Option<String>> {
            Ok(self.get(path))
        }

        async fn write(&self, path: &str, value: &str) -> Result<()> {
            if self.refuse.is_some_and(|p| path.starts_with(p)) {
                anyhow::bail!("denied");
            }
            self.writes.lock().unwrap().push(path.to_owned());
            self.kv
                .lock()
                .unwrap()
                .insert(path.to_owned(), value.to_owned());
            Ok(())
        }
    }

    const SOURCES: [(&str, &str); 4] = [
        ("swarm/services/attic/pull-token", "tok"),
        ("swarm/services/attic/url", "https://cache.example"),
        ("swarm/services/attic/cache", "swarm"),
        ("swarm/services/attic/public-key", "swarm:AAAA"),
    ];

    fn hives() -> Vec<String> {
        vec!["alpha".to_owned(), "beta".to_owned()]
    }

    #[tokio::test]
    async fn present_facts_reach_every_hive() {
        let store = Memory::with(&SOURCES);
        let done = copy_pass(&store, &hives()).await.expect("pass");
        assert_eq!(
            done,
            Copied {
                written: 8,
                failed: 0
            }
        );
        for hive in hives() {
            assert_eq!(
                store
                    .get(&format!("swarm/hives/{hive}/attic/pull-token"))
                    .as_deref(),
                Some("tok")
            );
            assert_eq!(
                store
                    .get(&format!("swarm/hives/{hive}/attic/public-key"))
                    .as_deref(),
                Some("swarm:AAAA")
            );
        }
        // Presence control for the absence cases below: copies are the only
        // thing a pass writes under a hive.
        assert!(store.get("swarm/hives/alpha/forge/token").is_none());
    }

    #[tokio::test]
    async fn absent_facts_write_nothing() {
        let store = Memory::default();
        let done = copy_pass(&store, &hives()).await.expect("pass");
        assert_eq!(done, Copied::default());
        assert!(store.writes.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn only_the_stored_facts_are_copied_and_an_empty_one_is_not_stored() {
        let store = Memory::with(&[
            ("swarm/services/attic/pull-token", "tok"),
            ("swarm/services/attic/url", " \n"),
        ]);
        copy_pass(&store, &hives()).await.expect("pass");
        assert_eq!(
            store.get("swarm/hives/alpha/attic/pull-token").as_deref(),
            Some("tok")
        );
        assert!(store.get("swarm/hives/alpha/attic/url").is_none());
        assert!(store.get("swarm/hives/alpha/attic/cache").is_none());
    }

    #[tokio::test]
    async fn a_current_copy_is_not_rewritten_and_a_changed_value_is() {
        let store = Memory::with(&SOURCES);
        copy_pass(&store, &hives()).await.expect("first");
        let again = copy_pass(&store, &hives()).await.expect("second");
        assert_eq!(again, Copied::default(), "unchanged values are not written");

        store.kv.lock().unwrap().insert(
            "swarm/services/attic/pull-token".to_owned(),
            "rotated".to_owned(),
        );
        let rotated = copy_pass(&store, &hives()).await.expect("third");
        assert_eq!(
            rotated,
            Copied {
                written: 2,
                failed: 0
            }
        );
        assert_eq!(
            store.get("swarm/hives/beta/attic/pull-token").as_deref(),
            Some("rotated")
        );
    }

    #[tokio::test]
    async fn a_removed_source_leaves_the_copies() {
        let store = Memory::with(&SOURCES);
        copy_pass(&store, &hives()).await.expect("first");
        store
            .kv
            .lock()
            .unwrap()
            .remove("swarm/services/attic/pull-token");
        copy_pass(&store, &hives()).await.expect("second");
        assert_eq!(
            store.get("swarm/hives/alpha/attic/pull-token").as_deref(),
            Some("tok")
        );
    }

    #[tokio::test]
    async fn one_hive_failing_does_not_stop_the_others() {
        let mut store = Memory::with(&SOURCES);
        store.refuse = Some("swarm/hives/alpha/");
        let done = copy_pass(&store, &hives()).await.expect("pass");
        assert_eq!(
            done,
            Copied {
                written: 4,
                failed: 1
            }
        );
        assert!(store.get("swarm/hives/beta/attic/cache").is_some());
        assert!(store.get("swarm/hives/alpha/attic/cache").is_none());
    }
}