boulder: Cache upstreams fetched via recipe new & update

Author: tarkah

boulder/Cargo.toml

@@ -48,6 +48,7 @@ mailparse.workspace = true
 walkdir = "2.5.0"
 zstd.workspace = true
 astr.workspace = true
+tempfile.workspace = true
 
 [dev-dependencies]
 tempfile.workspace = true

boulder/src/build.rs

@@ -33,7 +33,7 @@ pub mod git;
 pub mod job;
 pub mod pgo;
 mod root;
-mod upstream;
+pub mod upstream;
 
 pub struct Builder {
     pub targets: Vec<Target>,

boulder/src/cli/recipe.rs

@@ -9,15 +9,16 @@ use std::{
 
 use boulder::{
     Env, Macros, architecture,
-    draft::{self, Drafter},
+    draft::{self, Drafter, upstream::fetched_upstream_cache_path},
     macros, recipe,
 };
 use clap::Parser;
-use fs_err as fs;
+use fs_err::{self as fs};
 use futures_util::StreamExt;
 use itertools::Itertools;
-use moss::{request, runtime};
+use moss::{request, runtime, util};
 use sha2::{Digest, Sha256};
+use tempfile::NamedTempFile;
 use thiserror::Error;
 use tokio::io::AsyncWriteExt;
 use tui::{
@@ -107,14 +108,14 @@ fn parse_upstream(s: &str) -> Result<Upstream, String> {
 pub fn handle(command: Command, env: Env) -> Result<(), Error> {
     match command.subcommand {
         Subcommand::Bump { recipe, release } => bump(recipe, release),
-        Subcommand::New { output, upstreams } => new(output, upstreams, env),
+        Subcommand::New { output, upstreams } => new(env, output, upstreams),
         Subcommand::Update {
             recipe,
             overwrite,
             version,
             upstreams,
             no_bump,
-        } => update(recipe, overwrite, version, upstreams, no_bump),
+        } => update(env, recipe, overwrite, version, upstreams, no_bump),
         Subcommand::Macros { _macro } => macros(_macro, env),
     }
 }
@@ -145,11 +146,11 @@ fn bump(recipe: PathBuf, release: Option<u64>) -> Result<(), Error> {
     Ok(())
 }
 
-fn new(output: PathBuf, upstreams: Vec<Url>, env: Env) -> Result<(), Error> {
+fn new(env: Env, output: PathBuf, upstreams: Vec<Url>) -> Result<(), Error> {
     const RECIPE_FILE: &str = "stone.yaml";
     const MONITORING_FILE: &str = "monitoring.yaml";
 
-    let drafter = Drafter::new(upstreams, env.data_dir);
+    let drafter = Drafter::new(env, upstreams);
     let draft = drafter.run()?;
 
     if !output.is_dir() {
@@ -165,6 +166,7 @@ fn new(output: PathBuf, upstreams: Vec<Url>, env: Env) -> Result<(), Error> {
 }
 
 fn update(
+    env: Env,
     recipe: Option<PathBuf>,
     overwrite: bool,
     version: String,
@@ -245,7 +247,7 @@ fn update(
                 updater.update_value(version, |root| root / "version");
             }
             Update::PlainUpstream(i, key, new_uri) => {
-                let hash = runtime::block_on(fetch_hash(new_uri.clone(), &mpb))?;
+                let hash = runtime::block_on(fetch_and_cache_upstream(&env, new_uri.clone(), &mpb))?;
 
                 let path = |root| root / "upstreams" / i / key.as_str().unwrap_or_default();
 
@@ -281,7 +283,13 @@ fn update(
     Ok(())
 }
 
-async fn fetch_hash(uri: Url, mpb: &MultiProgress) -> Result<String, Error> {
+/// Fetches the upstream at `uri` and caches it so it doesn't need to be refetched
+/// when this recipe is finally built.
+///
+/// Returns the sha256 hash of the fetched upstream
+async fn fetch_and_cache_upstream(env: &Env, uri: Url, mpb: &MultiProgress) -> Result<String, Error> {
+    use fs_err::tokio::{self as fs, File};
+
     let pb = mpb.add(
         ProgressBar::new(u64::MAX)
             .with_message(format!("{} {}", "Fetching".blue(), uri.as_str().bold()))
@@ -293,11 +301,13 @@ async fn fetch_hash(uri: Url, mpb: &MultiProgress) -> Result<String, Error> {
     );
     pb.enable_steady_tick(Duration::from_millis(150));
 
-    let mut stream = request::stream(uri).await?;
+    let mut stream = request::stream(uri.clone()).await?;
 
+    let (temp_file, temp_file_path) = NamedTempFile::with_prefix("boulder-")
+        .map_err(Error::CreateTempFile)?
+        .into_parts();
     let mut hasher = Sha256::new();
-    // Discard bytes
-    let mut out = tokio::io::sink();
+    let mut out = File::from_std(fs_err::File::from_parts(temp_file, &temp_file_path));
 
     while let Some(chunk) = stream.next().await {
         let bytes = &chunk?;
@@ -312,6 +322,22 @@ async fn fetch_hash(uri: Url, mpb: &MultiProgress) -> Result<String, Error> {
 
     let hash = hex::encode(hasher.finalize());
 
+    // Move fetched asset to cache dir so we don't need to refetch it
+    // when the user finally builds this new recipe
+    {
+        let cache_path = fetched_upstream_cache_path(env, &uri, &hash);
+
+        if let Some(parent) = cache_path.parent() {
+            fs::create_dir_all(parent).await.map_err(Error::CreateDir)?;
+        }
+
+        util::async_hardlink_or_copy(&temp_file_path, &cache_path)
+            .await
+            .map_err(Error::MoveTempFile)?;
+
+        drop(temp_file_path);
+    }
+
     pb.finish();
     mpb.remove(&pb);
 
@@ -419,6 +445,10 @@ pub enum Error {
     CreateDir(#[source] io::Error),
     #[error("deserializing recipe")]
     Deser(#[from] serde_yaml::Error),
+    #[error("create temp file")]
+    CreateTempFile(#[source] io::Error),
+    #[error("move temp file")]
+    MoveTempFile(#[source] io::Error),
     #[error("fetch upstream")]
     Fetch(#[from] request::Error),
     #[error("fetch upstream")]

boulder/src/draft.rs

@@ -5,14 +5,15 @@
 use std::path::Path;
 use std::{io, path::PathBuf};
 
-use fs_err as fs;
 use itertools::Itertools;
 use licenses::match_licences;
 use moss::{Dependency, util};
 use thiserror::Error;
 use tui::Styled;
 use url::Url;
 
+use crate::Env;
+
 use self::metadata::Metadata;
 use self::monitoring::Monitoring;
 use self::upstream::Upstream;
@@ -21,11 +22,11 @@ mod build;
 mod licenses;
 mod metadata;
 mod monitoring;
-mod upstream;
+pub mod upstream;
 
 pub struct Drafter {
+    env: Env,
     upstreams: Vec<Url>,
-    datadir: PathBuf,
 }
 
 pub struct Draft {
@@ -34,16 +35,16 @@ pub struct Draft {
 }
 
 impl Drafter {
-    pub fn new(upstreams: Vec<Url>, datadir: PathBuf) -> Self {
-        Self { upstreams, datadir }
+    pub fn new(env: Env, upstreams: Vec<Url>) -> Self {
+        Self { env, upstreams }
     }
 
     pub fn run(&self) -> Result<Draft, Error> {
-        // TODO: Use tempdir
-        let extract_root = PathBuf::from("/tmp/boulder-new");
+        let temp_dir = tempfile::tempdir()?;
+        let extract_root = temp_dir.as_ref();
 
         // Fetch and extract all upstreams
-        let extracted = upstream::fetch_and_extract(&self.upstreams, &extract_root)?;
+        let extracted = upstream::fetch_and_extract(&self.env, &self.upstreams, extract_root)?;
 
         // Build metadata from extracted upstreams
         let metadata = Metadata::new(extracted);
@@ -52,23 +53,20 @@ impl Drafter {
         let monitoring_result = monitoring.run()?;
 
         // Enumerate all extracted files
-        let files = util::enumerate_files(&extract_root, |_| true)?
+        let files = util::enumerate_files(extract_root, |_| true)?
             .into_iter()
-            .map(|path| File {
-                path,
-                extract_root: &extract_root,
-            })
+            .map(|path| File { path, extract_root })
             .collect::<Vec<_>>();
 
         // Analyze files to determine build system / collect deps
         let build = build::analyze(&files).map_err(Error::AnalyzeBuildSystem)?;
 
-        let licences_dir = &self.datadir.join("licenses");
+        let licences_dir = &self.env.data_dir.join("licenses");
 
-        let licenses = format_licenses(match_licences(&extract_root, licences_dir).unwrap_or_default());
+        let licenses = format_licenses(match_licences(extract_root, licences_dir).unwrap_or_default());
 
         // Remove temp extract dir
-        fs::remove_dir_all(extract_root)?;
+        drop(temp_dir);
 
         let build_system = build.detected_system.unwrap_or_else(|| {
             println!(

boulder/src/draft/upstream.rs

@@ -9,27 +9,28 @@ use fs_err::tokio::{self as fs, File};
 use futures_util::{StreamExt, TryStreamExt, stream};
 use moss::{environment, request, runtime, util};
 use sha2::{Digest, Sha256};
+use tempfile::NamedTempFile;
 use thiserror::Error;
 use tokio::{io::AsyncWriteExt, process::Command};
 use tui::{MultiProgress, ProgressBar, ProgressStyle, Styled};
 use url::Url;
 
+use crate::Env;
+
 pub struct Upstream {
     pub uri: Url,
     pub hash: String,
 }
 
 /// Fetch and extract the provided upstreams under `extract_root`
-pub fn fetch_and_extract(upstreams: &[Url], extract_root: &Path) -> Result<Vec<Upstream>, Error> {
-    util::recreate_dir(extract_root)?;
-
+pub fn fetch_and_extract(env: &Env, upstreams: &[Url], extract_root: &Path) -> Result<Vec<Upstream>, Error> {
     let mpb = MultiProgress::new();
 
     let ret = runtime::block_on(
         stream::iter(upstreams)
             .map(|uri| async {
-                let name = util::uri_file_name(uri);
-                let archive_path = extract_root.join(name);
+                let (temp_file, temp_path) = NamedTempFile::with_prefix("boulder-")?.into_parts();
+                let archive_file = File::from_std(fs_err::File::from_parts(temp_file, &temp_path));
 
                 let pb = mpb.add(
                     ProgressBar::new_spinner()
@@ -42,13 +43,26 @@ pub fn fetch_and_extract(upstreams: &[Url], extract_root: &Path) -> Result<Vec<U
                 );
                 pb.enable_steady_tick(Duration::from_millis(150));
 
-                let hash = fetch(uri, &archive_path).await?;
+                let hash = fetch(uri, archive_file).await?;
+
+                // Hardlink or copy fetched asset to cache dir so we don't need
+                // to refetch it when the user finally builds this new recipe
+                {
+                    let cache_path = fetched_upstream_cache_path(env, uri, &hash);
+
+                    if let Some(parent) = cache_path.parent() {
+                        fs::create_dir_all(parent).await?;
+                    }
+
+                    util::async_hardlink_or_copy(&temp_path, &cache_path).await?;
+                }
 
                 pb.set_message(format!("{} {}", "Extracting".yellow(), *uri));
 
-                extract(&archive_path, extract_root).await?;
+                extract(&temp_path, extract_root).await?;
 
-                fs::remove_file(archive_path).await?;
+                // Cleanup temp path
+                drop(temp_path);
 
                 pb.suspend(|| println!("{} {}", "Fetched".green(), *uri));
 
@@ -63,11 +77,9 @@ pub fn fetch_and_extract(upstreams: &[Url], extract_root: &Path) -> Result<Vec<U
     ret
 }
 
-async fn fetch(url: &Url, output: &Path) -> Result<String, Error> {
+async fn fetch(url: &Url, mut file: File) -> Result<String, Error> {
     let mut stream = request::stream(url.clone()).await?;
 
-    let mut file = File::create(&output).await?;
-
     let mut hasher = Sha256::new();
 
     while let Some(bytes) = stream.next().await {
@@ -100,6 +112,22 @@ async fn extract(archive: &Path, destination: &Path) -> Result<(), Error> {
     }
 }
 
+pub fn fetched_upstream_cache_path(env: &Env, uri: &Url, hash: &str) -> PathBuf {
+    let mut hasher = Sha256::new();
+    hasher.update(uri.as_str());
+    hasher.update(hash);
+
+    let hash = hex::encode(hasher.finalize());
+
+    env.cache_dir
+        .join("upstreams")
+        .join("fetched")
+        // Type safe guaranteed to be >= 5 bytes
+        .join(&hash[..5])
+        .join(&hash[hash.len() - 5..])
+        .join(hash)
+}
+
 #[derive(Debug, Error)]
 pub enum Error {
     #[error("failed to run `bsdtar`")]

moss/src/util.rs

@@ -106,6 +106,15 @@ pub fn hardlink_or_copy(from: &Path, to: &Path) -> io::Result<()> {
     Ok(())
 }
 
+pub async fn async_hardlink_or_copy(from: &Path, to: &Path) -> io::Result<()> {
+    let from = from.to_owned();
+    let to = to.to_owned();
+
+    tokio::task::spawn_blocking(move || hardlink_or_copy(&from, &to))
+        .await
+        .expect("join handle")
+}
+
 pub fn uri_file_name(uri: &Url) -> &str {
     let path = uri.path();