diff --git a/crates/binstalk-fetchers/Cargo.toml b/crates/binstalk-fetchers/Cargo.toml index 877162cf..37bd254e 100644 --- a/crates/binstalk-fetchers/Cargo.toml +++ b/crates/binstalk-fetchers/Cargo.toml @@ -6,7 +6,7 @@ edition = "2021" description = "The binstall fetchers" repository = "https://github.com/cargo-bins/cargo-binstall" documentation = "https://docs.rs/binstalk-fetchers" -rust-version = "1.65.0" +rust-version = "1.70.0" authors = ["Jiahao XU "] license = "GPL-3.0-only" diff --git a/crates/binstalk-fetchers/src/common.rs b/crates/binstalk-fetchers/src/common.rs index b3a491ae..10272f10 100644 --- a/crates/binstalk-fetchers/src/common.rs +++ b/crates/binstalk-fetchers/src/common.rs @@ -1,6 +1,11 @@ -use std::sync::{ - atomic::{AtomicBool, Ordering::Relaxed}, - Once, +#![allow(unused)] + +use std::{ + future::Future, + sync::{ + atomic::{AtomicBool, Ordering::Relaxed}, + Once, + }, }; pub(super) use binstalk_downloader::{ @@ -76,3 +81,33 @@ pub(super) async fn does_url_exist( Ok(Box::pin(client.remote_gettable(url.clone())).await?) } + +#[derive(Debug)] +pub(super) struct AutoAbortJoinHandle(JoinHandle); + +impl AutoAbortJoinHandle +where + T: Send + 'static, +{ + pub(super) fn spawn(future: F) -> Self + where + F: Future + Send + 'static, + { + Self(tokio::spawn(future)) + } +} + +impl Drop for AutoAbortJoinHandle { + fn drop(&mut self) { + self.0.abort(); + } +} + +impl AutoAbortJoinHandle> +where + E: Into, +{ + pub(super) async fn flattened_join(mut self) -> Result { + (&mut self.0).await?.map_err(Into::into) + } +} diff --git a/crates/binstalk-fetchers/src/futures_resolver.rs b/crates/binstalk-fetchers/src/futures_resolver.rs index 0a550519..461ab462 100644 --- a/crates/binstalk-fetchers/src/futures_resolver.rs +++ b/crates/binstalk-fetchers/src/futures_resolver.rs @@ -1,5 +1,7 @@ -use std::{future::Future, pin::Pin}; +use std::{fmt::Debug, future::Future, pin::Pin}; + use tokio::sync::mpsc; +use tracing::warn; /// Given multiple futures with output = `Result, E>`, /// returns the the first one that returns either `Err(_)` or @@ -17,7 +19,7 @@ impl Default for FuturesResolver { } } -impl FuturesResolver { +impl FuturesResolver { /// Insert new future into this resolver, they will start running /// right away. pub fn push(&self, fut: Fut) @@ -67,10 +69,18 @@ impl FuturesResolver { } /// Return the resolution. - pub fn resolve(self) -> impl Future, E>> { + pub fn resolve(self) -> impl Future> { let mut rx = self.rx; drop(self.tx); - async move { rx.recv().await.transpose() } + async move { + loop { + match rx.recv().await { + Some(Ok(ret)) => return Some(ret), + Some(Err(err)) => warn!(?err, "Fail to resolve the future"), + None => return None, + } + } + } } } diff --git a/crates/binstalk-fetchers/src/gh_crate_meta.rs b/crates/binstalk-fetchers/src/gh_crate_meta.rs index 976b6942..51cd012c 100644 --- a/crates/binstalk-fetchers/src/gh_crate_meta.rs +++ b/crates/binstalk-fetchers/src/gh_crate_meta.rs @@ -278,7 +278,7 @@ impl super::Fetcher for GhCrateMeta { } } - if let Some(resolved) = resolver.resolve().await? { + if let Some(resolved) = resolver.resolve().await { debug!(?resolved, "Winning URL found!"); self.resolution .set(resolved) diff --git a/crates/binstalk-fetchers/src/lib.rs b/crates/binstalk-fetchers/src/lib.rs index 33e50944..f589e5fc 100644 --- a/crates/binstalk-fetchers/src/lib.rs +++ b/crates/binstalk-fetchers/src/lib.rs @@ -6,7 +6,7 @@ use binstalk_downloader::{download::DownloadError, remote::Error as RemoteError} use binstalk_git_repo_api::gh_api_client::{GhApiError, GhRepo}; use binstalk_types::cargo_toml_binstall::SigningAlgorithm; use thiserror::Error as ThisError; -use tokio::{sync::OnceCell, time::sleep}; +use tokio::{sync::OnceCell, task::JoinError, time::sleep}; pub use url::ParseError as UrlParseError; mod gh_crate_meta; @@ -70,6 +70,9 @@ pub enum FetchError { #[error("Failed to verify signature")] InvalidSignature, + + #[error("Failed to wait for task: {0}")] + TaskJoinError(#[from] JoinError), } impl From for FetchError { diff --git a/crates/binstalk-fetchers/src/quickinstall.rs b/crates/binstalk-fetchers/src/quickinstall.rs index aa9b402a..2862fb89 100644 --- a/crates/binstalk-fetchers/src/quickinstall.rs +++ b/crates/binstalk-fetchers/src/quickinstall.rs @@ -1,4 +1,8 @@ -use std::{borrow::Cow, path::Path, sync::Arc}; +use std::{ + borrow::Cow, + path::Path, + sync::{Arc, OnceLock}, +}; use binstalk_downloader::remote::Method; use binstalk_types::cargo_toml_binstall::{PkgFmt, PkgMeta, PkgSigning}; @@ -61,6 +65,8 @@ pub struct QuickInstall { signature_policy: SignaturePolicy, target_data: Arc, + + signature_verifier: OnceLock, } impl QuickInstall { @@ -75,6 +81,41 @@ impl QuickInstall { .await .copied() } + + fn download_signature( + self: Arc, + ) -> AutoAbortJoinHandle> { + AutoAbortJoinHandle::spawn(async move { + if self.signature_policy == SignaturePolicy::Ignore { + Ok(SignatureVerifier::Noop) + } else { + debug!(url=%self.signature_url, "Downloading signature"); + match Download::new(self.client.clone(), self.signature_url.clone()) + .into_bytes() + .await + { + Ok(signature) => { + trace!(?signature, "got signature contents"); + let config = PkgSigning { + algorithm: SigningAlgorithm::Minisign, + pubkey: QUICKINSTALL_SIGN_KEY, + file: None, + }; + SignatureVerifier::new(&config, &signature) + } + Err(err) => { + if self.signature_policy == SignaturePolicy::Require { + error!("Failed to download signature: {err}"); + Err(FetchError::MissingSignature) + } else { + debug!("Failed to download signature, skipping verification: {err}"); + Ok(SignatureVerifier::Noop) + } + } + } + } + }) + } } #[async_trait::async_trait] @@ -109,6 +150,8 @@ impl super::Fetcher for QuickInstall { signature_policy, target_data, + + signature_verifier: OnceLock::new(), }) } @@ -118,22 +161,28 @@ impl super::Fetcher for QuickInstall { return Ok(false); } - if self.signature_policy == SignaturePolicy::Require { - does_url_exist( - self.client.clone(), - self.gh_api_client.clone(), - &self.signature_url, - ) - .await - .map_err(|_| FetchError::MissingSignature)?; - } + let download_signature_task = self.clone().download_signature(); - does_url_exist( + let is_found = does_url_exist( self.client.clone(), self.gh_api_client.clone(), &self.package_url, ) - .await + .await?; + + if !is_found { + return Ok(false); + } + + if self + .signature_verifier + .set(download_signature_task.flattened_join().await?) + .is_err() + { + panic!("::find is run twice"); + } + + Ok(true) }) } @@ -160,33 +209,8 @@ by rust officially."#, } async fn fetch_and_extract(&self, dst: &Path) -> Result { - let verifier = if self.signature_policy == SignaturePolicy::Ignore { - SignatureVerifier::Noop - } else { - debug!(url=%self.signature_url, "Downloading signature"); - match Download::new(self.client.clone(), self.signature_url.clone()) - .into_bytes() - .await - { - Ok(signature) => { - trace!(?signature, "got signature contents"); - let config = PkgSigning { - algorithm: SigningAlgorithm::Minisign, - pubkey: QUICKINSTALL_SIGN_KEY, - file: None, - }; - SignatureVerifier::new(&config, &signature)? - } - Err(err) => { - if self.signature_policy == SignaturePolicy::Require { - error!("Failed to download signature: {err}"); - return Err(FetchError::MissingSignature); - } - - debug!("Failed to download signature, skipping verification: {err}"); - SignatureVerifier::Noop - } - } + let Some(verifier) = self.signature_verifier.get() else { + panic!("::find has not been called yet!") }; debug!(url=%self.package_url, "Downloading package"); diff --git a/crates/binstalk/src/ops/resolve.rs b/crates/binstalk/src/ops/resolve.rs index 805bfb4a..b8e02374 100644 --- a/crates/binstalk/src/ops/resolve.rs +++ b/crates/binstalk/src/ops/resolve.rs @@ -97,7 +97,7 @@ async fn resolve_inner( _ => None, }; - let mut handles: Vec<(Arc, _)> = Vec::with_capacity( + let mut handles: Vec> = Vec::with_capacity( desired_targets.len() * resolvers.len() + if binary_name.is_some() { desired_targets.len() @@ -139,8 +139,7 @@ async fn resolve_inner( target_data, opts.signature_policy, ); - filter_fetcher_by_name_predicate(fetcher.fetcher_name()) - .then_some((fetcher.clone(), AutoAbortJoinHandle::new(fetcher.find()))) + filter_fetcher_by_name_predicate(fetcher.fetcher_name()).then_some(fetcher) }), ) }; @@ -165,9 +164,12 @@ async fn resolve_inner( ); } - for (fetcher, handle) in handles { + for fetcher in handles { fetcher.clone().report_to_upstream(); - match handle.flattened_join().await { + match AutoAbortJoinHandle::new(fetcher.clone().find()) + .flattened_join() + .await + { Ok(true) => { // Generate temporary binary path let bin_path = opts.temp_dir.join(format!(