refactor: convert extension install to async (#3815)

src/dfx-core/Cargo.toml

@@ -10,6 +10,7 @@ rust-version.workspace = true
 [dependencies]
 aes-gcm.workspace = true
 argon2.workspace = true
+backoff.workspace = true
 bip32 = "0.4.0"
 byte-unit = { workspace = true, features = ["serde"] }
 bytes.workspace = true

src/dfx-core/src/error/extension.rs

@@ -1,4 +1,5 @@
 #![allow(dead_code)]
+use crate::error::reqwest::WrappedReqwestError;
 use crate::error::structured_file::StructuredFileError;
 use thiserror::Error;
 
@@ -82,8 +83,8 @@ pub enum NewExtensionManagerError {
 
 #[derive(Error, Debug)]
 pub enum DownloadAndInstallExtensionToTempdirError {
-    #[error("Downloading extension from '{0}' failed")]
-    ExtensionDownloadFailed(url::Url, #[source] reqwest::Error),
+    #[error(transparent)]
+    ExtensionDownloadFailed(WrappedReqwestError),
 
     #[error("Cannot get extensions directory")]
     EnsureExtensionDirExistsFailed(#[source] crate::error::fs::FsError),

src/dfx-core/src/error/mod.rs

@@ -17,6 +17,7 @@ pub mod load_dfx_config;
 pub mod load_networks_config;
 pub mod network_config;
 pub mod process;
+pub mod reqwest;
 pub mod root_key;
 pub mod socket_addr_conversion;
 pub mod structured_file;

src/dfx-core/src/error/reqwest.rs

@@ -0,0 +1,28 @@
+use crate::http::retryable::Retryable;
+use reqwest::StatusCode;
+use thiserror::Error;
+
+// reqwest::Error's fmt::Display appends the error descriptions of all sources.
+// For this reason, it is not marked as #[source] here, so that we don't
+// display the error descriptions of all sources repeatedly.
+#[derive(Error, Debug)]
+#[error("{}", .0)]
+pub struct WrappedReqwestError(pub reqwest::Error);
+
+impl Retryable for WrappedReqwestError {
+    fn is_retryable(&self) -> bool {
+        let err = &self.0;
+        err.is_timeout()
+            || err.is_connect()
+            || matches!(
+                err.status(),
+                Some(
+                    StatusCode::INTERNAL_SERVER_ERROR
+                        | StatusCode::BAD_GATEWAY
+                        | StatusCode::SERVICE_UNAVAILABLE
+                        | StatusCode::GATEWAY_TIMEOUT
+                        | StatusCode::TOO_MANY_REQUESTS
+                )
+            )
+    }
+}

src/dfx-core/src/extension/manager/install.rs

@@ -3,21 +3,25 @@ use crate::error::extension::{
     FindLatestExtensionCompatibleVersionError, GetExtensionArchiveNameError,
     GetExtensionDownloadUrlError, InstallExtensionError,
 };
+use crate::error::reqwest::WrappedReqwestError;
 use crate::extension::{manager::ExtensionManager, manifest::ExtensionCompatibilityMatrix};
+use crate::http::get::get_with_retries;
+use backoff::exponential::ExponentialBackoff;
 use flate2::read::GzDecoder;
 use reqwest::Url;
 use semver::{BuildMetadata, Prerelease, Version};
 use std::io::Cursor;
 #[cfg(unix)]
 use std::os::unix::fs::PermissionsExt;
+use std::time::Duration;
 use tar::Archive;
 use tempfile::{tempdir_in, TempDir};
 
 const DFINITY_DFX_EXTENSIONS_RELEASES_URL: &str =
     "https://github.com/dfinity/dfx-extensions/releases/download";
 
 impl ExtensionManager {
-    pub fn install_extension(
+    pub async fn install_extension(
         &self,
         extension_name: &str,
         install_as: Option<&str>,
@@ -36,13 +40,13 @@ impl ExtensionManager {
 
         let extension_version = match version {
             Some(version) => version.clone(),
-            None => self.get_extension_compatible_version(extension_name)?,
+            None => self.get_highest_compatible_version(extension_name).await?,
         };
         let github_release_tag = get_git_release_tag(extension_name, &extension_version);
         let extension_archive = get_extension_archive_name(extension_name)?;
         let url = get_extension_download_url(&github_release_tag, &extension_archive)?;
 
-        let temp_dir = self.download_and_unpack_extension_to_tempdir(url)?;
+        let temp_dir = self.download_and_unpack_extension_to_tempdir(url).await?;
 
         self.finalize_installation(
             extension_name,
@@ -64,31 +68,31 @@ impl ExtensionManager {
         dfx_version
     }
 
-    fn get_extension_compatible_version(
+    async fn get_highest_compatible_version(
         &self,
         extension_name: &str,
     ) -> Result<Version, FindLatestExtensionCompatibleVersionError> {
-        let manifest = ExtensionCompatibilityMatrix::fetch()?;
+        let manifest = ExtensionCompatibilityMatrix::fetch().await?;
         let dfx_version = self.dfx_version_strip_semver();
         manifest.find_latest_compatible_extension_version(extension_name, &dfx_version)
     }
 
-    fn download_and_unpack_extension_to_tempdir(
+    async fn download_and_unpack_extension_to_tempdir(
         &self,
         download_url: Url,
     ) -> Result<TempDir, DownloadAndInstallExtensionToTempdirError> {
-        let response = reqwest::blocking::get(download_url.clone()).map_err(|e| {
-            DownloadAndInstallExtensionToTempdirError::ExtensionDownloadFailed(
-                download_url.clone(),
-                e,
-            )
-        })?;
+        let retry_policy = ExponentialBackoff {
+            max_elapsed_time: Some(Duration::from_secs(60)),
+            ..Default::default()
+        };
+        let response = get_with_retries(download_url.clone(), retry_policy)
+            .await
+            .map_err(DownloadAndInstallExtensionToTempdirError::ExtensionDownloadFailed)?;
 
-        let bytes = response.bytes().map_err(|e| {
-            DownloadAndInstallExtensionToTempdirError::ExtensionDownloadFailed(
-                download_url.clone(),
+        let bytes = response.bytes().await.map_err(|e| {
+            DownloadAndInstallExtensionToTempdirError::ExtensionDownloadFailed(WrappedReqwestError(
                 e,
-            )
+            ))
         })?;
 
         crate::fs::composite::ensure_dir_exists(&self.dir)

src/dfx-core/src/extension/manifest/compatibility_matrix.rs

@@ -31,12 +31,14 @@ pub struct ExtensionCompatibleVersions {
 }
 
 impl ExtensionCompatibilityMatrix {
-    pub fn fetch() -> Result<Self, FetchExtensionCompatibilityMatrixError> {
-        let resp = reqwest::blocking::get(COMMON_EXTENSIONS_MANIFEST_LOCATION).map_err(|e| {
-            CompatibilityMatrixFetchError(COMMON_EXTENSIONS_MANIFEST_LOCATION.to_string(), e)
-        })?;
+    pub async fn fetch() -> Result<Self, FetchExtensionCompatibilityMatrixError> {
+        let resp = reqwest::get(COMMON_EXTENSIONS_MANIFEST_LOCATION)
+            .await
+            .map_err(|e| {
+                CompatibilityMatrixFetchError(COMMON_EXTENSIONS_MANIFEST_LOCATION.to_string(), e)
+            })?;
 
-        resp.json().map_err(MalformedCompatibilityMatrix)
+        resp.json().await.map_err(MalformedCompatibilityMatrix)
     }
 
     pub fn find_latest_compatible_extension_version(

src/dfx-core/src/http/get.rs

@@ -0,0 +1,25 @@
+use crate::error::reqwest::WrappedReqwestError;
+use crate::http::retryable::Retryable;
+use backoff::exponential::ExponentialBackoff;
+use backoff::future::retry;
+use backoff::SystemClock;
+use reqwest::Response;
+use url::Url;
+
+pub async fn get_with_retries(
+    url: Url,
+    retry_policy: ExponentialBackoff<SystemClock>,
+) -> Result<Response, WrappedReqwestError> {
+    let operation = || async {
+        let response = reqwest::get(url.clone())
+            .await
+            .and_then(|resp| resp.error_for_status())
+            .map_err(WrappedReqwestError);
+        match response {
+            Ok(doc) => Ok(doc),
+            Err(e) if e.is_retryable() => Err(backoff::Error::transient(e)),
+            Err(e) => Err(backoff::Error::permanent(e)),
+        }
+    };
+    retry(retry_policy, operation).await
+}

src/dfx-core/src/http/mod.rs

@@ -0,0 +1,2 @@
+pub mod get;
+pub mod retryable;

src/dfx-core/src/http/retryable.rs

@@ -0,0 +1,3 @@
+pub trait Retryable {
+    fn is_retryable(&self) -> bool;
+}

src/dfx-core/src/lib.rs

@@ -5,6 +5,7 @@ pub mod error;
 pub mod extension;
 pub mod foundation;
 pub mod fs;
+pub mod http;
 pub mod identity;
 pub mod interface;
 pub mod json;

src/dfx/src/commands/extension/install.rs

@@ -6,6 +6,7 @@ use anyhow::bail;
 use clap::Parser;
 use clap::Subcommand;
 use semver::Version;
+use tokio::runtime::Runtime;
 
 #[derive(Parser)]
 pub struct InstallOpts {
@@ -30,11 +31,16 @@ pub fn exec(env: &dyn Environment, opts: InstallOpts) -> DfxResult<()> {
         bail!("Extension '{}' cannot be installed because it conflicts with an existing command. Consider using '--install-as' flag to install this extension under different name.", opts.name)
     }
 
-    mgr.install_extension(
-        &opts.name,
-        opts.install_as.as_deref(),
-        opts.version.as_ref(),
-    )?;
+    let runtime = Runtime::new().expect("Unable to create a runtime");
+
+    runtime.block_on(async {
+        mgr.install_extension(
+            &opts.name,
+            opts.install_as.as_deref(),
+            opts.version.as_ref(),
+        )
+        .await
+    })?;
     spinner.finish_with_message(
         format!(
             "Extension '{}' installed successfully{}",