Verify that bin_files exist in resolve stage (#382)

* Refactor: Extract new fn `BinFile::check_source_exists`
* Impl new async fn `AutoAbortJoinHandle::flattened_join`
* Impl new fn `Fetcher::fetcher_name`
* Verify that `bin_files` exist in `resolve` stage
   To ensure that the installation stage won't fail because of missing
   binaries.
* Rm unused `MultiFecther`
* Simplify `Future` impl for `AutoAbortJoinHandle`
* Add new variant `BinstallError::CargoTomlMissingPackage`
* Replace `unwrap` in `resolve_inner` with proper error handling
* Make `Fetcher::new` as a regular function
   instead of an `async` function.
* Ret `Arc<dyn Fetcher>` in trait fn `Fetcher::new`
* Refactor `resolve_inner`

Signed-off-by: Jiahao XU <Jiahao_XU@outlook.com>

crates/binstalk/src/bins.rs

@@ -83,10 +83,17 @@ impl BinFile {
         )
     }
 
-    pub fn install_bin(&self) -> Result<(), BinstallError> {
+    /// Return `Ok` if the source exists, otherwise `Err`.
+    pub fn check_source_exists(&self) -> Result<(), BinstallError> {
         if !self.source.try_exists()? {
-            return Err(BinstallError::BinFileNotFound(self.source.clone()));
+            Err(BinstallError::BinFileNotFound(self.source.clone()))
+        } else {
+            Ok(())
         }
+    }
+
+    pub fn install_bin(&self) -> Result<(), BinstallError> {
+        self.check_source_exists()?;
 
         debug!(
             "Atomically install file from '{}' to '{}'",

crates/binstalk/src/errors.rs

@@ -276,6 +276,14 @@ pub enum BinstallError {
     #[diagnostic(severity(error), code(binstall::binfile))]
     BinFileNotFound(PathBuf),
 
+    /// `Cargo.toml` of the crate does not have section "Package".
+    ///
+    /// - Code: `binstall::cargo_manifest`
+    /// - Exit: 89
+    #[error("Cargo.toml of crate {0} does not have section \"Package\"")]
+    #[diagnostic(severity(error), code(binstall::cargo_manifest))]
+    CargoTomlMissingPackage(CompactString),
+
     /// A wrapped error providing the context of which crate the error is about.
     #[error("for crate {crate_name}")]
     CrateContext {
@@ -310,6 +318,7 @@ impl BinstallError {
             UnspecifiedBinaries => 86,
             NoViableTargets => 87,
             BinFileNotFound(_) => 88,
+            CargoTomlMissingPackage(_) => 89,
             CrateContext { error, .. } => error.exit_number(),
         };
 

crates/binstalk/src/fetchers.rs

@@ -8,7 +8,6 @@ use reqwest::Client;
 
 use crate::{
     errors::BinstallError,
-    helpers::tasks::AutoAbortJoinHandle,
     manifests::cargo_toml_binstall::{PkgFmt, PkgMeta},
 };
 
@@ -18,7 +17,8 @@ mod quickinstall;
 #[async_trait::async_trait]
 pub trait Fetcher: Send + Sync {
     /// Create a new fetcher from some data
-    async fn new(client: &Client, data: &Arc<Data>) -> Arc<Self>
+    #[allow(clippy::new_ret_no_self)]
+    fn new(client: &Client, data: &Arc<Data>) -> Arc<dyn Fetcher>
     where
         Self: Sized;
 
@@ -44,6 +44,13 @@ pub trait Fetcher: Send + Sync {
     /// A short human-readable name or descriptor for the package source
     fn source_name(&self) -> CompactString;
 
+    /// A short human-readable name, must contains only characters
+    /// and numbers and it also must be unique.
+    ///
+    /// It is used to create a temporary dir where it is used for
+    /// [`Fetcher::fetch_and_extract`].
+    fn fetcher_name(&self) -> &'static str;
+
     /// Should return true if the remote is from a third-party source
     fn is_third_party(&self) -> bool;
 
@@ -60,45 +67,3 @@ pub struct Data {
     pub repo: Option<String>,
     pub meta: PkgMeta,
 }
-
-type FetcherJoinHandle = AutoAbortJoinHandle<Result<bool, BinstallError>>;
-
-pub struct MultiFetcher(Vec<(Arc<dyn Fetcher>, FetcherJoinHandle)>);
-
-impl MultiFetcher {
-    pub fn with_capacity(n: usize) -> Self {
-        Self(Vec::with_capacity(n))
-    }
-
-    pub fn add(&mut self, fetcher: Arc<dyn Fetcher>) {
-        self.0.push((
-            fetcher.clone(),
-            AutoAbortJoinHandle::spawn(async move { fetcher.find().await }),
-        ));
-    }
-
-    pub async fn first_available(self) -> Option<Arc<dyn Fetcher>> {
-        for (fetcher, handle) in self.0 {
-            match handle.await {
-                Ok(Ok(true)) => return Some(fetcher),
-                Ok(Ok(false)) => (),
-                Ok(Err(err)) => {
-                    debug!(
-                        "Error while checking fetcher {}: {}",
-                        fetcher.source_name(),
-                        err
-                    );
-                }
-                Err(join_err) => {
-                    debug!(
-                        "Error while joining the task that checks the fetcher {}: {}",
-                        fetcher.source_name(),
-                        join_err
-                    );
-                }
-            }
-        }
-
-        None
-    }
-}

crates/binstalk/src/fetchers/gh_crate_meta.rs

@@ -71,7 +71,7 @@ impl GhCrateMeta {
 
 #[async_trait::async_trait]
 impl super::Fetcher for GhCrateMeta {
-    async fn new(client: &Client, data: &Arc<Data>) -> Arc<Self> {
+    fn new(client: &Client, data: &Arc<Data>) -> Arc<dyn super::Fetcher> {
         Arc::new(Self {
             client: client.clone(),
             data: data.clone(),
@@ -175,6 +175,10 @@ impl super::Fetcher for GhCrateMeta {
             .unwrap_or_else(|| "invalid url".into())
     }
 
+    fn fetcher_name(&self) -> &'static str {
+        "GhCrateMeta"
+    }
+
     fn is_third_party(&self) -> bool {
         false
     }

crates/binstalk/src/fetchers/quickinstall.rs

@@ -27,7 +27,7 @@ pub struct QuickInstall {
 
 #[async_trait::async_trait]
 impl super::Fetcher for QuickInstall {
-    async fn new(client: &Client, data: &Arc<Data>) -> Arc<Self> {
+    fn new(client: &Client, data: &Arc<Data>) -> Arc<dyn super::Fetcher> {
         let crate_name = &data.name;
         let version = &data.version;
         let target = data.target.clone();
@@ -68,6 +68,10 @@ impl super::Fetcher for QuickInstall {
         CompactString::from("QuickInstall")
     }
 
+    fn fetcher_name(&self) -> &'static str {
+        "QuickInstall"
+    }
+
     fn is_third_party(&self) -> bool {
         true
     }

crates/binstalk/src/helpers/tasks.rs

@@ -56,6 +56,12 @@ impl<T> Future for AutoAbortJoinHandle<T> {
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         Pin::new(&mut Pin::into_inner(self).0)
             .poll(cx)
-            .map(|res| res.map_err(BinstallError::TaskJoinError))
+            .map_err(BinstallError::TaskJoinError)
+    }
+}
+
+impl<T> AutoAbortJoinHandle<Result<T, BinstallError>> {
+    pub async fn flattened_join(self) -> Result<T, BinstallError> {
+        self.await?
     }
 }

crates/binstalk/src/ops/install.rs

@@ -1,4 +1,4 @@
-use std::{path::PathBuf, sync::Arc};
+use std::sync::Arc;
 
 use cargo_toml::Package;
 use compact_str::CompactString;
@@ -9,7 +9,6 @@ use super::{resolve::Resolution, Options};
 use crate::{
     bins,
     errors::BinstallError,
-    fetchers::Fetcher,
     helpers::jobserver_client::LazyJobserverClient,
     manifests::{
         cargo_toml_binstall::Meta,
@@ -29,7 +28,6 @@ pub async fn install(
             package,
             name,
             version_req,
-            bin_path,
             bin_files,
         } => {
             let current_version =
@@ -42,19 +40,17 @@ pub async fn install(
                     })?;
             let target = fetcher.target().into();
 
-            install_from_package(fetcher, opts, bin_path, bin_files)
-                .await
-                .map(|option| {
-                    option.map(|bins| CrateInfo {
-                        name,
-                        version_req,
-                        current_version,
-                        source: CrateSource::cratesio_registry(),
-                        target,
-                        bins,
-                        other: Default::default(),
-                    })
+            install_from_package(opts, bin_files).await.map(|option| {
+                option.map(|bins| CrateInfo {
+                    name,
+                    version_req,
+                    current_version,
+                    source: CrateSource::cratesio_registry(),
+                    target,
+                    bins,
+                    other: Default::default(),
                 })
+            })
         }
         Resolution::InstallFromSource { package } => {
             let desired_targets = opts.desired_targets.get().await;
@@ -63,7 +59,7 @@ pub async fn install(
                 .ok_or(BinstallError::NoViableTargets)?;
 
             if !opts.dry_run {
-                install_from_source(package, target, jobserver_client, opts.quiet, opts.force)
+                install_from_source(&package, target, jobserver_client, opts.quiet, opts.force)
                     .await
                     .map(|_| None)
             } else {
@@ -78,42 +74,9 @@ pub async fn install(
 }
 
 async fn install_from_package(
-    fetcher: Arc<dyn Fetcher>,
     opts: Arc<Options>,
-    bin_path: PathBuf,
     bin_files: Vec<bins::BinFile>,
 ) -> Result<Option<Vec<CompactString>>, BinstallError> {
-    // Download package
-    if opts.dry_run {
-        info!("Dry run, not downloading package");
-    } else {
-        fetcher.fetch_and_extract(&bin_path).await?;
-    }
-
-    #[cfg(incomplete)]
-    {
-        // Fetch and check package signature if available
-        if let Some(pub_key) = meta.as_ref().map(|m| m.pub_key.clone()).flatten() {
-            debug!("Found public key: {pub_key}");
-
-            // Generate signature file URL
-            let mut sig_ctx = ctx.clone();
-            sig_ctx.format = "sig".to_string();
-            let sig_url = sig_ctx.render(&pkg_url)?;
-
-            debug!("Fetching signature file: {sig_url}");
-
-            // Download signature file
-            let sig_path = temp_dir.join(format!("{pkg_name}.sig"));
-            download(&sig_url, &sig_path).await?;
-
-            // TODO: do the signature check
-            unimplemented!()
-        } else {
-            warn!("No public key found, package signature could not be validated");
-        }
-    }
-
     if opts.dry_run {
         info!("Dry run, not proceeding");
         return Ok(None);
@@ -139,7 +102,7 @@ async fn install_from_package(
 }
 
 async fn install_from_source(
-    package: Package<Meta>,
+    package: &Package<Meta>,
     target: &str,
     lazy_jobserver_client: LazyJobserverClient,
     quiet: bool,
@@ -155,9 +118,9 @@ async fn install_from_source(
     let mut cmd = Command::new("cargo");
 
     cmd.arg("install")
-        .arg(package.name)
+        .arg(&package.name)
         .arg("--version")
-        .arg(package.version)
+        .arg(&package.version)
         .arg("--target")
         .arg(target);
 

crates/binstalk/src/ops/resolve.rs

@@ -5,6 +5,7 @@ use std::{
 
 use cargo_toml::{Manifest, Package, Product};
 use compact_str::{CompactString, ToCompactString};
+use itertools::Itertools;
 use log::{debug, info, warn};
 use reqwest::Client;
 use semver::{Version, VersionReq};
@@ -15,7 +16,8 @@ use crate::{
     bins,
     drivers::fetch_crate_cratesio,
     errors::BinstallError,
-    fetchers::{Data, Fetcher, GhCrateMeta, MultiFetcher, QuickInstall},
+    fetchers::{Data, Fetcher, GhCrateMeta, QuickInstall},
+    helpers::tasks::AutoAbortJoinHandle,
     manifests::cargo_toml_binstall::{Meta, PkgMeta},
 };
 
@@ -32,7 +34,6 @@ pub enum Resolution {
         package: Package<Meta>,
         name: CompactString,
         version_req: CompactString,
-        bin_path: PathBuf,
         bin_files: Vec<bins::BinFile>,
     },
     InstallFromSource {
@@ -95,8 +96,8 @@ pub async fn resolve(
     crates_io_api_client: crates_io_api::AsyncClient,
 ) -> Result<Resolution, BinstallError> {
     let crate_name_name = crate_name.name.clone();
-    resolve_inner(
-        opts,
+    let resolution = resolve_inner(
+        &opts,
         crate_name,
         curr_version,
         temp_dir,
@@ -105,11 +106,15 @@ pub async fn resolve(
         crates_io_api_client,
     )
     .await
-    .map_err(|err| err.crate_context(crate_name_name))
+    .map_err(|err| err.crate_context(crate_name_name))?;
+
+    resolution.print(&opts);
+
+    Ok(resolution)
 }
 
 async fn resolve_inner(
-    opts: Arc<Options>,
+    opts: &Options,
     crate_name: CrateName,
     curr_version: Option<Version>,
     temp_dir: Arc<Path>,
@@ -142,7 +147,9 @@ async fn resolve_inner(
         }
     };
 
-    let package = manifest.package.unwrap();
+    let package = manifest
+        .package
+        .ok_or_else(|| BinstallError::CargoTomlMissingPackage(crate_name.name.clone()))?;
 
     if let Some(curr_version) = curr_version {
         let new_version =
@@ -171,65 +178,150 @@ async fn resolve_inner(
 
     let desired_targets = opts.desired_targets.get().await;
 
-    let mut fetchers = MultiFetcher::with_capacity(desired_targets.len() * 2);
+    let mut handles: Vec<(Arc<dyn Fetcher>, _)> = Vec::with_capacity(desired_targets.len() * 2);
 
-    for target in desired_targets {
-        debug!("Building metadata for target: {target}");
-        let mut target_meta = meta.clone_without_overrides();
+    handles.extend(
+        desired_targets
+            .iter()
+            .map(|target| {
+                debug!("Building metadata for target: {target}");
+                let mut target_meta = meta.clone_without_overrides();
 
-        // Merge any overrides
-        if let Some(o) = meta.overrides.get(target) {
-            target_meta.merge(o);
-        }
+                // Merge any overrides
+                if let Some(o) = meta.overrides.get(target) {
+                    target_meta.merge(o);
+                }
 
-        target_meta.merge(&opts.cli_overrides);
-        debug!("Found metadata: {target_meta:?}");
+                target_meta.merge(&opts.cli_overrides);
+                debug!("Found metadata: {target_meta:?}");
 
-        let fetcher_data = Arc::new(Data {
-            name: package.name.clone(),
-            target: target.clone(),
-            version: package.version.clone(),
-            repo: package.repository.clone(),
-            meta: target_meta,
-        });
+                Arc::new(Data {
+                    name: package.name.clone(),
+                    target: target.clone(),
+                    version: package.version.clone(),
+                    repo: package.repository.clone(),
+                    meta: target_meta,
+                })
+            })
+            .cartesian_product([GhCrateMeta::new, QuickInstall::new])
+            .map(|(fetcher_data, f)| {
+                let fetcher = f(&client, &fetcher_data);
+                (
+                    fetcher.clone(),
+                    AutoAbortJoinHandle::spawn(async move { fetcher.find().await }),
+                )
+            }),
+    );
 
-        fetchers.add(GhCrateMeta::new(&client, &fetcher_data).await);
-        fetchers.add(QuickInstall::new(&client, &fetcher_data).await);
-    }
+    for (fetcher, handle) in handles {
+        match handle.flattened_join().await {
+            Ok(true) => {
+                // Generate temporary binary path
+                let bin_path = temp_dir.join(format!(
+                    "bin-{}-{}-{}",
+                    crate_name.name,
+                    fetcher.target(),
+                    fetcher.fetcher_name()
+                ));
 
-    let resolution = match fetchers.first_available().await {
-        Some(fetcher) => {
-            // Build final metadata
-            let meta = fetcher.target_meta();
-
-            // Generate temporary binary path
-            let bin_path = temp_dir.join(format!("bin-{}", crate_name.name));
-            debug!("Using temporary binary path: {}", bin_path.display());
-
-            let bin_files = collect_bin_files(
-                fetcher.as_ref(),
-                &package,
-                meta,
-                binaries,
-                bin_path.clone(),
-                install_path.to_path_buf(),
-            )?;
-
-            Resolution::Fetch {
-                fetcher,
-                package,
-                name: crate_name.name,
-                version_req: version_req.to_compact_string(),
-                bin_path,
-                bin_files,
+                match download_extract_and_verify(
+                    fetcher.as_ref(),
+                    &bin_path,
+                    &package,
+                    &install_path,
+                    binaries.clone(),
+                )
+                .await
+                {
+                    Ok(bin_files) => {
+                        return Ok(Resolution::Fetch {
+                            fetcher,
+                            package,
+                            name: crate_name.name,
+                            version_req: version_req.to_compact_string(),
+                            bin_files,
+                        })
+                    }
+                    Err(err) => {
+                        warn!(
+                            "Error while downloading and extracting from fetcher {}: {}",
+                            fetcher.source_name(),
+                            err
+                        );
+                    }
+                }
+            }
+            Ok(false) => (),
+            Err(err) => {
+                warn!(
+                    "Error while checking fetcher {}: {}",
+                    fetcher.source_name(),
+                    err
+                );
             }
         }
-        None => Resolution::InstallFromSource { package },
-    };
+    }
 
-    resolution.print(&opts);
+    Ok(Resolution::InstallFromSource { package })
+}
 
-    Ok(resolution)
+///  * `fetcher` - `fetcher.find()` must return `Ok(true)`.
+async fn download_extract_and_verify(
+    fetcher: &dyn Fetcher,
+    bin_path: &Path,
+    package: &Package<Meta>,
+    install_path: &Path,
+    // TODO: Use &[Product]
+    binaries: Vec<Product>,
+) -> Result<Vec<bins::BinFile>, BinstallError> {
+    // Build final metadata
+    let meta = fetcher.target_meta();
+
+    let bin_files = collect_bin_files(
+        fetcher,
+        package,
+        meta,
+        binaries,
+        bin_path.to_path_buf(),
+        install_path.to_path_buf(),
+    )?;
+
+    // Download and extract it.
+    // If that fails, then ignore this fetcher.
+    fetcher.fetch_and_extract(bin_path).await?;
+
+    #[cfg(incomplete)]
+    {
+        // Fetch and check package signature if available
+        if let Some(pub_key) = meta.as_ref().map(|m| m.pub_key.clone()).flatten() {
+            debug!("Found public key: {pub_key}");
+
+            // Generate signature file URL
+            let mut sig_ctx = ctx.clone();
+            sig_ctx.format = "sig".to_string();
+            let sig_url = sig_ctx.render(&pkg_url)?;
+
+            debug!("Fetching signature file: {sig_url}");
+
+            // Download signature file
+            let sig_path = temp_dir.join(format!("{pkg_name}.sig"));
+            download(&sig_url, &sig_path).await?;
+
+            // TODO: do the signature check
+            unimplemented!()
+        } else {
+            warn!("No public key found, package signature could not be validated");
+        }
+    }
+
+    // Verify that all the bin_files exist
+    block_in_place(|| {
+        for bin_file in bin_files.iter() {
+            bin_file.check_source_exists()?;
+        }
+
+        Ok(bin_files)
+    })
 }
 
 fn collect_bin_files(