//! 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//*` 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 deploy waits for the same copy: the `CopyAtticFacts` node ([`copy_node`]) //! copies to the one hive the deploy goes to, through the same [`copy_hive`] as //! the pass, so a hive is told to deploy only once its facts are in its prefix. //! //! A fact that is not stored is not copied, and a copy whose source is gone //! stays: neither the pass nor the node deletes. A swarm with no cache stores //! none, so both write nothing. use std::sync::Arc; use std::time::Duration; use anyhow::{Context, Result}; use hive_jobq::scheduler::Outcome; 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>; /// 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> { Ok(self.read_optional::(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> { 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 { 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) } /// The stored facts copied to `hive`; how many copies were written. The one /// hive [`copy_node`] serves; [`copy_pass`] reads the sources once for all of /// them instead. /// /// # Errors /// A source or a copy that cannot be read, or a copy that cannot be written. pub(crate) async fn ensure_copied(store: &impl FactStore, hive: &str) -> Result { copy_hive(store, hive, &sources(store).await?).await } /// The `CopyAtticFacts` node's work: [`ensure_copied`] for `hive`, `Failed` /// with the cause when it errors. Written copies are logged; a hive whose /// copies are current is `Done` without a write. pub(crate) async fn copy_node(store: &impl FactStore, hive: &str) -> Outcome { match ensure_copied(store, hive).await { Ok(written) => { if written > 0 { tracing::info!(hive, written, "attic facts: copied for a deploy"); } Outcome::Done } Err(e) => Outcome::Failed(format!( "the cache facts could not be copied to {hive}: {e:#}" )), } } /// [`copy_node`] against the swarm's store, within [`PASS_DEADLINE`]. pub(crate) async fn copy_node_now(hive: &str) -> Outcome { let work = async { match crate::store::connect().await { Ok(store) => copy_node(&store, hive).await, Err(e) => Outcome::Failed(format!("logging in to the swarm secret store: {e}")), } }; tokio::time::timeout(PASS_DEADLINE, work) .await .unwrap_or_else(|_| { Outcome::Failed(format!( "copying the cache facts to {hive} timed out after {}s", PASS_DEADLINE.as_secs() )) }) } /// 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 { 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) { 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>, writes: Mutex>, 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 { self.kv.lock().unwrap().get(path).cloned() } } impl FactStore for Memory { async fn read(&self, path: &str) -> Result> { 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 { 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") ); } /// The node and the pass copy through the same `copy_hive`: for one hive /// they leave the store holding the same values. #[tokio::test] async fn the_node_and_the_pass_leave_the_same_copies() { let by_pass = Memory::with(&SOURCES); copy_pass(&by_pass, &["alpha".to_owned()]) .await .expect("pass"); let by_node = Memory::with(&SOURCES); assert_eq!(copy_node(&by_node, "alpha").await, Outcome::Done); for leaf in Leaf::ALL { let path = hive_path("alpha", leaf).expect("a hive path"); assert!(by_pass.get(&path).is_some(), "the pass wrote {path}"); assert_eq!(by_node.get(&path), by_pass.get(&path), "{path}"); } } /// The node copies for its hive only, and a hive already current is `Done` /// without a write. #[tokio::test] async fn the_node_copies_for_its_own_hive_and_leaves_a_current_copy_alone() { let store = Memory::with(&SOURCES); assert_eq!(copy_node(&store, "alpha").await, Outcome::Done); assert_eq!( store.get("swarm/hives/alpha/attic/pull-token").as_deref(), Some("tok") ); assert!(store.get("swarm/hives/beta/attic/pull-token").is_none()); let writes = store.writes.lock().unwrap().len(); assert_eq!(copy_node(&store, "alpha").await, Outcome::Done); assert_eq!(store.writes.lock().unwrap().len(), writes); } /// A copy that cannot be written fails the node, naming the hive and the /// cause. Control: the same store copies for the hive it does not refuse. #[tokio::test] async fn a_copy_that_cannot_be_written_fails_the_node() { let mut store = Memory::with(&SOURCES); store.refuse = Some("swarm/hives/alpha/"); let Outcome::Failed(error) = copy_node(&store, "alpha").await else { panic!("a refused write is not Done") }; assert!(error.contains("alpha"), "{error}"); assert!(error.contains("denied"), "{error}"); assert_eq!(copy_node(&store, "beta").await, Outcome::Done); } #[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()); } }