swarm: a deploy goes out only once its closure is cached and the hive has its cache facts

swarm-controller/src/attic_copy.rs

@@ -6,14 +6,19 @@
 //! [`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: the pass never deletes. A swarm with no cache stores none, so the
-//! pass writes nothing.
+//! 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};
 
@@ -94,6 +99,51 @@ async fn copy_hive(
     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<usize> {
+    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
@@ -305,6 +355,54 @@ mod tests {
         );
     }
 
+    /// 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);