push green meta builds to the attic cache and record cached

swarm-builder/src/main.rs

@@ -16,6 +16,8 @@
 //! and an `internal/meta` commit, and the answer is the store path of that
 //! agent's toplevel, built at the flake's own lock. They take the build slot
 //! PR checks take. The controller records the path; the builder writes nothing.
+//! When a cache is set (see [`push`]), a green meta build is pushed to it before
+//! the reply, and the reply says whether that worked.
 //!
 //! The store identity comes from the `BAO_*` environment. The queue client
 //! secret is read at start and again when authelia refuses it, the forge token
@@ -39,6 +41,7 @@ use swarm_wire::builder::{
 
 mod archetypes;
 mod nix_build;
+mod push;
 
 /// The cert-auth role this service logs in to the store as. `swarm-bao.nix`
 /// writes it, matching the leaf's subject.
@@ -64,6 +67,17 @@ const MAX_SLOT_WAIT: Duration = REPLY_DEADLINE
     .saturating_sub(BUILD_DEADLINE)
     .saturating_sub(SLOT_MARGIN);
 
+/// What is held back from a meta build's reply deadline for publishing the reply
+/// after its push.
+const REPLY_MARGIN: Duration = Duration::from_secs(10);
+
+/// How long the push of a meta build that ends at now may run: at most
+/// [`push::PUSH_DEADLINE`], and never past `reply_by`, so a push cannot make the
+/// reply later than the controller waits.
+fn push_budget(reply_by: tokio::time::Instant) -> Duration {
+    push::PUSH_DEADLINE.min(reply_by.saturating_duration_since(tokio::time::Instant::now()))
+}
+
 #[tokio::main]
 async fn main() -> Result<()> {
     hive_log::init();
@@ -83,6 +97,7 @@ async fn main() -> Result<()> {
         deadline: BUILD_DEADLINE,
         lock_deadline: LOCK_DEADLINE,
         system: var("SYSTEM")?,
+        cache: push_target()?,
     });
     let secret =
         swarm_secret_client::read_with_startup_retry("reading the queue client secret", || {
@@ -300,7 +315,7 @@ async fn answer_meta_builds(
                         &slot,
                         received,
                         || busy_meta(&job),
-                        handle_meta(&job, &builder),
+                        handle_meta(&job, &builder, received + REPLY_DEADLINE - REPLY_MARGIN),
                     )
                     .await
                     else {
@@ -383,19 +398,36 @@ fn busy_meta(job: &BuildFromMeta) -> FromMetaReply {
 }
 
 /// Build one meta request and log the build.
-async fn handle_meta(job: &BuildFromMeta, builder: &nix_build::Builder) -> FromMetaReply {
+async fn handle_meta(
+    job: &BuildFromMeta,
+    builder: &nix_build::Builder,
+    reply_by: tokio::time::Instant,
+) -> FromMetaReply {
     let reply = match within(TOKEN_DEADLINE, forge_token()).await {
         Ok(token) => builder.build_from_meta(job, &token).await,
         Err(e) => FromMetaReply::Unavailable {
             reason: format!("{e:#}"),
         },
     };
+    let reply = push::cache_built(
+        reply,
+        builder.cache.as_ref(),
+        attic_push_token(),
+        push_budget(reply_by),
+        &job.agent,
+    )
+    .await;
     match &reply {
-        FromMetaReply::Built { out_path, system } => tracing::info!(
+        FromMetaReply::Built {
+            out_path,
+            system,
+            cached,
+        } => tracing::info!(
             agent = %job.agent,
             meta_rev = %job.meta_rev,
             %out_path,
             %system,
+            cached,
             "agent meta build built"
         ),
         FromMetaReply::Failed { error_tail } => tracing::warn!(
@@ -458,6 +490,23 @@ fn log_build(job: &BuildAgent, reply: &Reply) {
     }
 }
 
+/// The token the builder pushes closures to the cache with, read on every push so
+/// a rotation takes effect on the next one.
+async fn attic_push_token() -> Result<String> {
+    let path = swarm_secret_client::builder::attic_push_token_path()?;
+    let store = swarm_secret_client::SecretStore::from_env(CERT_ROLE)
+        .await
+        .context("logging in to the secret store")?;
+    let stored: Option<swarm_secret_client::builder::AtticPushToken> = store
+        .read_optional(&path)
+        .await
+        .with_context(|| format!("reading {path}"))?;
+    match stored {
+        Some(t) if !t.value.trim().is_empty() => Ok(t.value.trim().to_owned()),
+        _ => bail!("nothing at {path}: the cache push token has not been written"),
+    }
+}
+
 /// The forge token the builder fetches config repos with, read on every
 /// build so a rotation takes effect on the next one.
 async fn forge_token() -> Result<String> {
@@ -495,6 +544,17 @@ async fn client_secret(client_id: &str) -> Result<String> {
     }
 }
 
+/// The cache the unit's environment names, if any. systemd sets
+/// `RUNTIME_DIRECTORY` for the unit's runtime directory.
+fn push_target() -> Result<Option<push::Remote>> {
+    let var = |name: &str| std::env::var(name).ok();
+    push::Remote::from_settings(
+        var(&format!("{PREFIX}_ATTIC_DOMAIN")),
+        var(&format!("{PREFIX}_ATTIC_CACHE")),
+        var("RUNTIME_DIRECTORY"),
+    )
+}
+
 /// The forge's base URL, which must be `https`: the credential helper offers
 /// the token to `https` requests for its host alone.
 fn forge_url() -> Result<String> {
@@ -580,6 +640,7 @@ mod tests {
             FromMetaReply::Built {
                 out_path: "/nix/store/x".to_owned(),
                 system: "s".to_owned(),
+                cached: false,
             }
         })
         .await;
@@ -594,6 +655,7 @@ mod tests {
             FromMetaReply::Built {
                 out_path: "/nix/store/x".to_owned(),
                 system: "s".to_owned(),
+                cached: false,
             }
         })
         .await;
@@ -616,6 +678,20 @@ mod tests {
         assert_eq!(prompt.expect("read"), "t");
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn a_push_gets_the_deadline_unless_the_reply_is_due_sooner() {
+        let now = Instant::now();
+        assert_eq!(
+            push_budget(now + Duration::from_hours(1)),
+            push::PUSH_DEADLINE
+        );
+        assert_eq!(
+            push_budget(now + Duration::from_secs(90)),
+            Duration::from_secs(90)
+        );
+        assert_eq!(push_budget(now), Duration::ZERO);
+    }
+
     #[test]
     fn a_build_started_at_the_slot_limit_ends_inside_the_reply_deadline() {
         assert!(MAX_SLOT_WAIT > Duration::ZERO);

swarm-builder/src/nix_build.rs

@@ -59,6 +59,9 @@ pub struct Builder {
     pub lock_deadline: Duration,
     /// The platform this host builds for, reported with each meta build.
     pub system: String,
+    /// Where a meta build's closure is pushed after it succeeds; `None` pushes
+    /// nothing.
+    pub cache: Option<crate::push::Remote>,
 }
 
 impl Builder {
@@ -88,6 +91,7 @@ impl Builder {
                 Reply::Built { out_path } => FromMetaReply::Built {
                     out_path,
                     system: self.system.clone(),
+                    cached: false,
                 },
                 Reply::Failed { error_tail } => FromMetaReply::Failed { error_tail },
             },
@@ -437,6 +441,7 @@ mod tests {
             deadline: BUILD_DEADLINE,
             lock_deadline: LOCK_DEADLINE,
             system: "x86_64-linux".to_owned(),
+            cache: None,
         }
     }
 
@@ -514,6 +519,7 @@ mod tests {
             FromMetaReply::Built {
                 out_path: format!("{} {}", BUILD_FLAGS.join(" "), b.meta_attr(&meta_job())),
                 system: "x86_64-linux".to_owned(),
+                cached: false,
             }
         );
     }

swarm-builder/src/push.rs

@@ -0,0 +1,437 @@
+//! The push of a built closure to the binary cache, with `attic push`.
+//!
+//! The cache is an attic instance outside the swarm. Its push token is read
+//! from the secret store per push and written to an `attic` config file in a
+//! directory only the builder's unit can enter, which `attic` finds through
+//! `XDG_CONFIG_HOME`. `attic login <name> <url> <token>` is not used: it puts
+//! the token in the process list. `attic` verifies the server's certificate
+//! against the system roots.
+//!
+//! A push that does not happen or does not finish leaves the build green with
+//! `cached` false: the closure is still in the host's store, and the next build
+//! of the agent pushes again.
+
+use std::future::Future;
+use std::path::{Path, PathBuf};
+use std::process::Stdio;
+use std::time::Duration;
+
+use anyhow::{Result, bail};
+use swarm_wire::builder::FromMetaReply;
+
+/// The longest one push may run, when the reply deadline leaves that much.
+pub const PUSH_DEADLINE: Duration = Duration::from_mins(10);
+
+/// The server name in the config file, and in the cache argument of the push.
+const SERVER: &str = "swarm";
+
+/// Where the builder pushes: one cache on one attic server.
+pub struct Remote {
+    /// The `attic` binary: a name found in the unit's path in production.
+    pub attic: PathBuf,
+    /// The server's name, without scheme: `host` or `host:port`.
+    pub domain: String,
+    /// The cache's name on that server.
+    pub cache: String,
+    /// A directory only the builder can enter, which the config files go in.
+    pub state_dir: PathBuf,
+}
+
+impl Remote {
+    /// The cache the environment names: `None` when neither value is set, so
+    /// the builder pushes nothing.
+    ///
+    /// # Errors
+    /// When only one of the two is set, or a value would bend the config file
+    /// or the command.
+    pub fn from_settings(
+        domain: Option<String>,
+        cache: Option<String>,
+        state_dir: Option<String>,
+    ) -> Result<Option<Self>> {
+        let (domain, cache) = match (domain, cache) {
+            (None, None) => return Ok(None),
+            (Some(domain), Some(cache)) => (domain, cache),
+            _ => bail!("the cache domain and the cache name must be set together"),
+        };
+        let plain = |s: &str| {
+            !s.is_empty()
+                && s.bytes()
+                    .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'-' | b'_' | b':'))
+        };
+        if !plain(&domain) || !plain(&cache) {
+            bail!("the cache domain {domain:?} and name {cache:?} must be plain names");
+        }
+        let Some(state_dir) = state_dir else {
+            bail!("a cache is set, but the unit has no runtime directory to hold its config");
+        };
+        Ok(Some(Self {
+            attic: PathBuf::from("attic"),
+            domain,
+            cache,
+            state_dir: PathBuf::from(state_dir),
+        }))
+    }
+
+    /// Push `out_path` with `token`. The result is what `attic` printed.
+    ///
+    /// # Errors
+    /// Why the push did not finish: the config could not be written, `attic`
+    /// could not start, or it exited with an error.
+    pub async fn push(&self, token: &str, out_path: &str) -> Result<String, String> {
+        if !out_path.starts_with("/nix/store/") || out_path.contains(char::is_whitespace) {
+            return Err(format!("{out_path:?} is not a store path"));
+        }
+        let home = ConfigHome::write(&self.state_dir, &self.domain, token)
+            .map_err(|e| format!("writing the attic config: {e}"))?;
+        let output = self
+            .command(&home.0, out_path)
+            .output()
+            .await
+            .map_err(|e| format!("starting attic: {e}"))?;
+        let said = |bytes: &[u8]| {
+            crate::nix_build::last_lines(&String::from_utf8_lossy(bytes))
+                .unwrap_or_default()
+                .replace(token, "<redacted>")
+        };
+        if output.status.success() {
+            Ok(said(&output.stdout))
+        } else {
+            Err(format!(
+                "attic exited with {}: {}",
+                output.status,
+                said(&output.stderr)
+            ))
+        }
+    }
+
+    /// `attic push` of `out_path`, reading its config from `config_home`. The
+    /// token is in the file there and in no argument or variable.
+    fn command(&self, config_home: &Path, out_path: &str) -> tokio::process::Command {
+        let mut cmd = tokio::process::Command::new(&self.attic);
+        cmd.arg("push")
+            .arg(format!("{SERVER}:{}", self.cache))
+            .arg(out_path)
+            .env("XDG_CONFIG_HOME", config_home)
+            .stdin(Stdio::null())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .kill_on_drop(true);
+        cmd
+    }
+}
+
+/// Numbers the config directories of one process's pushes.
+static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+
+/// The config directory of one push, removed with the token in it on drop.
+struct ConfigHome(PathBuf);
+
+impl ConfigHome {
+    /// `<parent>/push-<n>/attic/config.toml`, readable by this user alone.
+    fn write(parent: &Path, domain: &str, token: &str) -> std::io::Result<Self> {
+        use std::io::Write as _;
+        use std::os::unix::fs::{DirBuilderExt as _, OpenOptionsExt as _};
+        // A quote or a control character would end the TOML string early.
+        if token.is_empty() || token.bytes().any(|b| b == b'"' || b == b'\\' || b < b' ') {
+            return Err(std::io::Error::other(
+                "the push token is empty or holds a character the config cannot",
+            ));
+        }
+        let home = Self(parent.join(format!(
+            "push-{}-{}",
+            std::process::id(),
+            NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+        )));
+        let dir = home.0.join("attic");
+        std::fs::DirBuilder::new()
+            .recursive(true)
+            .mode(0o700)
+            .create(&dir)?;
+        let mut file = std::fs::OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .mode(0o600)
+            .open(dir.join("config.toml"))?;
+        write!(
+            file,
+            "default-server = \"{SERVER}\"\n\n[servers.{SERVER}]\n\
+             endpoint = \"https://{domain}\"\ntoken = \"{token}\"\n"
+        )?;
+        Ok(home)
+    }
+}
+
+impl Drop for ConfigHome {
+    fn drop(&mut self) {
+        let _ = std::fs::remove_dir_all(&self.0);
+    }
+}
+
+/// `reply` with `cached` set by pushing its closure, when it is a built one and
+/// a cache is set. `read_token` is not polled without a cache. The token read
+/// and the push together get `budget`.
+///
+/// Whatever happens to the push, a built reply stays built: a failure is logged
+/// and leaves `cached` false.
+pub async fn cache_built(
+    reply: FromMetaReply,
+    cache: Option<&Remote>,
+    read_token: impl Future<Output = Result<String>>,
+    budget: Duration,
+    agent: &str,
+) -> FromMetaReply {
+    let (
+        Some(cache),
+        FromMetaReply::Built {
+            out_path, system, ..
+        },
+    ) = (cache, &reply)
+    else {
+        return reply;
+    };
+    let pushed = tokio::time::timeout(budget, async {
+        let token = read_token
+            .await
+            .map_err(|e| format!("reading the push token: {e:#}"))?;
+        cache.push(&token, out_path).await
+    })
+    .await
+    .unwrap_or_else(|_| {
+        Err(format!(
+            "the push did not finish within {} s",
+            budget.as_secs()
+        ))
+    });
+    match pushed {
+        Ok(_) => {
+            tracing::info!(%agent, %out_path, cache = %cache.cache, "closure pushed to the cache");
+            FromMetaReply::Built {
+                out_path: out_path.clone(),
+                system: system.clone(),
+                cached: true,
+            }
+        }
+        Err(reason) => {
+            tracing::warn!(%agent, %out_path, cache = %cache.cache, %reason, "closure not pushed to the cache");
+            reply
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::os::unix::fs::PermissionsExt as _;
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    use super::*;
+
+    const TOKEN: &str = "eyJ0b2tlbiI6InNlY3JldCJ9.c2lnbmVk";
+    const OUT: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas";
+
+    fn built() -> FromMetaReply {
+        FromMetaReply::Built {
+            out_path: OUT.to_owned(),
+            system: "x86_64-linux".to_owned(),
+            cached: false,
+        }
+    }
+
+    fn cached(reply: &FromMetaReply) -> bool {
+        match reply {
+            FromMetaReply::Built { cached, .. } => *cached,
+            other => panic!("not built: {other:?}"),
+        }
+    }
+
+    /// A cache whose `attic` is the program `attic`: `true` and `false` stand
+    /// in for a push that works and one that fails, and `echo` for one that
+    /// says what it was run with.
+    fn cache_with(attic: &str) -> Remote {
+        Remote {
+            attic: PathBuf::from(attic),
+            domain: "cache.example.org".to_owned(),
+            cache: "grid".to_owned(),
+            state_dir: std::env::temp_dir(),
+        }
+    }
+
+    fn token() -> impl Future<Output = Result<String>> {
+        std::future::ready(Ok(TOKEN.to_owned()))
+    }
+
+    #[tokio::test]
+    async fn a_push_that_works_marks_the_build_cached() {
+        let cache = cache_with("true");
+        let reply = cache_built(built(), Some(&cache), token(), PUSH_DEADLINE, "atlas").await;
+        assert!(cached(&reply), "{reply:?}");
+    }
+
+    #[tokio::test]
+    async fn a_push_that_fails_leaves_the_build_green_and_not_cached() {
+        let cache = cache_with("false");
+        let reply = cache_built(built(), Some(&cache), token(), PUSH_DEADLINE, "atlas").await;
+        assert_eq!(reply, built());
+    }
+
+    #[tokio::test]
+    async fn a_push_that_cannot_start_leaves_the_build_green_and_not_cached() {
+        let cache = cache_with("/nonexistent/attic");
+        let reply = cache_built(built(), Some(&cache), token(), PUSH_DEADLINE, "atlas").await;
+        assert_eq!(reply, built());
+    }
+
+    #[tokio::test]
+    async fn an_unreadable_token_leaves_the_build_green_and_not_cached() {
+        let cache = cache_with("true");
+        let reply = cache_built(
+            built(),
+            Some(&cache),
+            async { anyhow::bail!("nothing at the path") },
+            PUSH_DEADLINE,
+            "atlas",
+        )
+        .await;
+        assert_eq!(reply, built());
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn a_token_read_that_outlasts_the_budget_leaves_the_build_green() {
+        let cache = cache_with("true");
+        let started = tokio::time::Instant::now();
+        let reply = cache_built(
+            built(),
+            Some(&cache),
+            std::future::pending(),
+            Duration::from_secs(5),
+            "atlas",
+        )
+        .await;
+        assert_eq!(reply, built());
+        assert_eq!(started.elapsed(), Duration::from_secs(5));
+    }
+
+    #[tokio::test]
+    async fn without_a_cache_nothing_is_read_or_pushed() {
+        let polled = std::sync::Arc::new(AtomicBool::new(false));
+        let tracked = |polled: &std::sync::Arc<AtomicBool>| {
+            let polled = std::sync::Arc::clone(polled);
+            async move {
+                polled.store(true, Ordering::SeqCst);
+                token().await
+            }
+        };
+        let reply = cache_built(built(), None, tracked(&polled), PUSH_DEADLINE, "atlas").await;
+        assert_eq!(reply, built());
+        assert!(!polled.load(Ordering::SeqCst), "the token was read");
+
+        // Control: with a cache the same future runs.
+        let cache = cache_with("true");
+        cache_built(
+            built(),
+            Some(&cache),
+            tracked(&polled),
+            PUSH_DEADLINE,
+            "atlas",
+        )
+        .await;
+        assert!(polled.load(Ordering::SeqCst));
+    }
+
+    #[tokio::test]
+    async fn a_build_that_is_not_built_is_not_pushed() {
+        let cache = cache_with("true");
+        let failed = FromMetaReply::Failed {
+            error_tail: "e".to_owned(),
+        };
+        let reply = cache_built(
+            failed.clone(),
+            Some(&cache),
+            token(),
+            PUSH_DEADLINE,
+            "atlas",
+        )
+        .await;
+        assert_eq!(reply, failed);
+    }
+
+    #[tokio::test]
+    async fn the_push_command_has_no_token_in_it() {
+        // `echo` prints the arguments it is given.
+        let cache = cache_with("echo");
+        let said = cache.push(TOKEN, OUT).await.expect("echo succeeds");
+        assert_eq!(said, format!("push swarm:grid {OUT}"));
+        assert!(!said.contains(TOKEN));
+
+        let cmd = cache.command(Path::new("/run/home"), OUT);
+        let cmd = cmd.as_std();
+        let args: Vec<_> = cmd.get_args().collect();
+        assert_eq!(args, ["push", "swarm:grid", OUT]);
+        let env: Vec<_> = cmd.get_envs().collect();
+        assert_eq!(
+            env,
+            [(
+                std::ffi::OsStr::new("XDG_CONFIG_HOME"),
+                Some(std::ffi::OsStr::new("/run/home"))
+            )]
+        );
+    }
+
+    #[test]
+    fn the_token_is_in_a_private_config_file_that_is_removed_after() {
+        let parent = std::env::temp_dir().join(format!("push-test-{}", std::process::id()));
+        std::fs::create_dir_all(&parent).expect("parent");
+        let home = ConfigHome::write(&parent, "cache.example.org", TOKEN).expect("written");
+        let file = home.0.join("attic/config.toml");
+        let text = std::fs::read_to_string(&file).expect("reads");
+        assert!(text.contains(&format!("token = \"{TOKEN}\"")), "{text}");
+        assert!(
+            text.contains("endpoint = \"https://cache.example.org\""),
+            "{text}"
+        );
+        let mode = |p: &Path| std::fs::metadata(p).expect("stat").permissions().mode() & 0o777;
+        assert_eq!(mode(&file), 0o600);
+        assert_eq!(mode(&home.0), 0o700);
+        let root = home.0.clone();
+        drop(home);
+        assert!(!root.exists());
+        std::fs::remove_dir_all(&parent).expect("cleanup");
+    }
+
+    #[test]
+    fn a_token_that_would_end_the_config_string_is_refused() {
+        let parent = std::env::temp_dir();
+        for bad in ["", "a\"b", "a\\b", "a\nb"] {
+            assert!(ConfigHome::write(&parent, "h", bad).is_err(), "{bad:?}");
+        }
+    }
+
+    #[tokio::test]
+    async fn a_path_that_is_not_a_store_path_is_not_pushed() {
+        let cache = cache_with("true");
+        assert!(cache.push(TOKEN, "--help").await.is_err());
+        assert!(cache.push(TOKEN, "/nix/store/x y").await.is_err());
+        assert!(cache.push(TOKEN, OUT).await.is_ok());
+    }
+
+    #[test]
+    fn the_settings_name_a_cache_only_when_both_are_set() {
+        let state = || Some("/run/swarm-builder".to_owned());
+        let some = |s: &str| Some(s.to_owned());
+        assert!(
+            Remote::from_settings(None, None, None)
+                .expect("ok")
+                .is_none()
+        );
+        let cache = Remote::from_settings(some("cache.example.org"), some("grid"), state())
+            .expect("ok")
+            .expect("set");
+        assert_eq!(cache.domain, "cache.example.org");
+        assert_eq!(cache.cache, "grid");
+        assert!(Remote::from_settings(some("cache.example.org"), None, state()).is_err());
+        assert!(Remote::from_settings(None, some("grid"), state()).is_err());
+        assert!(Remote::from_settings(some("https://cache"), some("grid"), state()).is_err());
+        assert!(Remote::from_settings(some("cache"), some("a\"b"), state()).is_err());
+        assert!(Remote::from_settings(some("cache"), some("grid"), None).is_err());
+    }
+}