refactor: extract structs for downloading tarballs and npm registry packuments (#24067)

4 files changed, 85 insertions, 50 deletions

diff --git a/cli/npm/managed/resolvers/common.rs b/cli/npm/managed/resolvers/common.rs
index b010bdd7c..2d540accd 100644
--- a/cli/npm/managed/resolvers/common.rs
+++ b/cli/npm/managed/resolvers/common.rs
@@ -12,7 +12,7 @@ use deno_ast::ModuleSpecifier;
 use deno_core::anyhow::Context;
 use deno_core::error::AnyError;
 use deno_core::futures;
-use deno_core::unsync::spawn;
+use deno_core::futures::StreamExt;
 use deno_core::url::Url;
 use deno_npm::NpmPackageCacheFolderId;
 use deno_npm::NpmPackageId;
@@ -21,7 +21,8 @@ use deno_runtime::deno_fs::FileSystem;
 use deno_runtime::deno_node::NodePermissions;
 use deno_runtime::deno_node::NodeResolutionMode;
 
-use super::super::cache::NpmCache;
+use crate::http_util::HttpClient;
+use crate::npm::managed::cache::TarballCache;
 
 /// Part of the resolution that interacts with the file system.
 #[async_trait]
@@ -49,7 +50,10 @@ pub trait NpmPackageFsResolver: Send + Sync {
     specifier: &ModuleSpecifier,
   ) -> Result<Option<NpmPackageCacheFolderId>, AnyError>;
 
-  async fn cache_packages(&self) -> Result<(), AnyError>;
+  async fn cache_packages(
+    &self,
+    http_client: &Arc<HttpClient>,
+  ) -> Result<(), AnyError>;
 
   fn ensure_read_permission(
     &self,
@@ -126,20 +130,20 @@ impl RegistryReadPermissionChecker {
 /// Caches all the packages in parallel.
 pub async fn cache_packages(
   packages: Vec<NpmResolutionPackage>,
-  cache: &Arc<NpmCache>,
+  tarball_cache: &Arc<TarballCache>,
+  http_client: &Arc<HttpClient>,
 ) -> Result<(), AnyError> {
-  let mut handles = Vec::with_capacity(packages.len());
+  let mut futures_unordered = futures::stream::FuturesUnordered::new();
   for package in packages {
-    let cache = cache.clone();
-    let handle = spawn(async move {
-      cache.ensure_package(&package.id.nv, &package.dist).await
+    futures_unordered.push(async move {
+      tarball_cache
+        .ensure_package(&package.id.nv, &package.dist, http_client)
+        .await
     });
-    handles.push(handle);
   }
-  let results = futures::future::join_all(handles).await;
-  for result in results {
+  while let Some(result) = futures_unordered.next().await {
     // surface the first error
-    result??;
+    result?;
   }
   Ok(())
 }
diff --git a/cli/npm/managed/resolvers/global.rs b/cli/npm/managed/resolvers/global.rs
index cfc57e591..4ffcb251f 100644
--- a/cli/npm/managed/resolvers/global.rs
+++ b/cli/npm/managed/resolvers/global.rs
@@ -20,8 +20,11 @@ use deno_runtime::deno_fs::FileSystem;
 use deno_runtime::deno_node::NodePermissions;
 use deno_runtime::deno_node::NodeResolutionMode;
 
+use crate::http_util::HttpClient;
+
 use super::super::super::common::types_package_name;
 use super::super::cache::NpmCache;
+use super::super::cache::TarballCache;
 use super::super::resolution::NpmResolution;
 use super::common::cache_packages;
 use super::common::NpmPackageFsResolver;
@@ -31,6 +34,7 @@ use super::common::RegistryReadPermissionChecker;
 #[derive(Debug)]
 pub struct GlobalNpmPackageResolver {
   cache: Arc<NpmCache>,
+  tarball_cache: Arc<TarballCache>,
   resolution: Arc<NpmResolution>,
   system_info: NpmSystemInfo,
   registry_read_permission_checker: RegistryReadPermissionChecker,
@@ -38,19 +42,21 @@ pub struct GlobalNpmPackageResolver {
 
 impl GlobalNpmPackageResolver {
   pub fn new(
-    fs: Arc<dyn FileSystem>,
     cache: Arc<NpmCache>,
+    fs: Arc<dyn FileSystem>,
+    tarball_cache: Arc<TarballCache>,
     resolution: Arc<NpmResolution>,
     system_info: NpmSystemInfo,
   ) -> Self {
     Self {
-      cache: cache.clone(),
-      resolution,
-      system_info,
       registry_read_permission_checker: RegistryReadPermissionChecker::new(
         fs,
         cache.root_folder(),
       ),
+      cache,
+      tarball_cache,
+      resolution,
+      system_info,
     }
   }
 
@@ -123,12 +129,20 @@ impl NpmPackageFsResolver for GlobalNpmPackageResolver {
     )
   }
 
-  async fn cache_packages(&self) -> Result<(), AnyError> {
+  async fn cache_packages(
+    &self,
+    http_client: &Arc<HttpClient>,
+  ) -> Result<(), AnyError> {
     let package_partitions = self
       .resolution
       .all_system_packages_partitioned(&self.system_info);
 
-    cache_packages(package_partitions.packages, &self.cache).await?;
+    cache_packages(
+      package_partitions.packages,
+      &self.tarball_cache,
+      http_client,
+    )
+    .await?;
 
     // create the copy package folders
     for copy in package_partitions.copy_packages {
diff --git a/cli/npm/managed/resolvers/local.rs b/cli/npm/managed/resolvers/local.rs
index f0c2a3f65..5c3b1f15e 100644
--- a/cli/npm/managed/resolvers/local.rs
+++ b/cli/npm/managed/resolvers/local.rs
@@ -14,6 +14,7 @@ use std::path::PathBuf;
 use std::sync::Arc;
 
 use crate::cache::CACHE_PERM;
+use crate::http_util::HttpClient;
 use crate::npm::cache_dir::mixed_case_package_name_decode;
 use crate::util::fs::atomic_write_file;
 use crate::util::fs::canonicalize_path_maybe_not_exists_with_fs;
@@ -27,16 +28,15 @@ use deno_ast::ModuleSpecifier;
 use deno_core::anyhow::bail;
 use deno_core::anyhow::Context;
 use deno_core::error::AnyError;
+use deno_core::futures::stream::FuturesUnordered;
+use deno_core::futures::StreamExt;
 use deno_core::parking_lot::Mutex;
-use deno_core::unsync::spawn;
-use deno_core::unsync::JoinHandle;
 use deno_core::url::Url;
 use deno_npm::resolution::NpmResolutionSnapshot;
 use deno_npm::NpmPackageCacheFolderId;
 use deno_npm::NpmPackageId;
 use deno_npm::NpmResolutionPackage;
 use deno_npm::NpmSystemInfo;
-use deno_runtime::deno_core::futures;
 use deno_runtime::deno_fs;
 use deno_runtime::deno_node::NodePermissions;
 use deno_runtime::deno_node::NodeResolutionMode;
@@ -48,6 +48,7 @@ use crate::npm::cache_dir::mixed_case_package_name_encode;
 
 use super::super::super::common::types_package_name;
 use super::super::cache::NpmCache;
+use super::super::cache::TarballCache;
 use super::super::resolution::NpmResolution;
 use super::common::NpmPackageFsResolver;
 use super::common::RegistryReadPermissionChecker;
@@ -56,10 +57,11 @@ use super::common::RegistryReadPermissionChecker;
 /// and resolves packages from it.
 #[derive(Debug)]
 pub struct LocalNpmPackageResolver {
-  fs: Arc<dyn deno_fs::FileSystem>,
   cache: Arc<NpmCache>,
+  fs: Arc<dyn deno_fs::FileSystem>,
   progress_bar: ProgressBar,
   resolution: Arc<NpmResolution>,
+  tarball_cache: Arc<TarballCache>,
   root_node_modules_path: PathBuf,
   root_node_modules_url: Url,
   system_info: NpmSystemInfo,
@@ -68,26 +70,28 @@ pub struct LocalNpmPackageResolver {
 
 impl LocalNpmPackageResolver {
   pub fn new(
-    fs: Arc<dyn deno_fs::FileSystem>,
     cache: Arc<NpmCache>,
+    fs: Arc<dyn deno_fs::FileSystem>,
     progress_bar: ProgressBar,
-    node_modules_folder: PathBuf,
     resolution: Arc<NpmResolution>,
+    tarball_cache: Arc<TarballCache>,
+    node_modules_folder: PathBuf,
     system_info: NpmSystemInfo,
   ) -> Self {
     Self {
-      fs: fs.clone(),
       cache,
+      fs: fs.clone(),
       progress_bar,
       resolution,
-      root_node_modules_url: Url::from_directory_path(&node_modules_folder)
-        .unwrap(),
-      root_node_modules_path: node_modules_folder.clone(),
-      system_info,
+      tarball_cache,
       registry_read_permission_checker: RegistryReadPermissionChecker::new(
         fs,
-        node_modules_folder,
+        node_modules_folder.clone(),
       ),
+      root_node_modules_url: Url::from_directory_path(&node_modules_folder)
+        .unwrap(),
+      root_node_modules_path: node_modules_folder,
+      system_info,
     }
   }
 
@@ -225,11 +229,16 @@ impl NpmPackageFsResolver for LocalNpmPackageResolver {
     Ok(get_package_folder_id_from_folder_name(&folder_name))
   }
 
-  async fn cache_packages(&self) -> Result<(), AnyError> {
+  async fn cache_packages(
+    &self,
+    http_client: &Arc<HttpClient>,
+  ) -> Result<(), AnyError> {
     sync_resolution_with_fs(
       &self.resolution.snapshot(),
       &self.cache,
+      http_client,
       &self.progress_bar,
+      &self.tarball_cache,
       &self.root_node_modules_path,
       &self.system_info,
     )
@@ -251,7 +260,9 @@ impl NpmPackageFsResolver for LocalNpmPackageResolver {
 async fn sync_resolution_with_fs(
   snapshot: &NpmResolutionSnapshot,
   cache: &Arc<NpmCache>,
+  http_client: &Arc<HttpClient>,
   progress_bar: &ProgressBar,
+  tarball_cache: &Arc<TarballCache>,
   root_node_modules_dir_path: &Path,
   system_info: &NpmSystemInfo,
 ) -> Result<(), AnyError> {
@@ -288,8 +299,7 @@ async fn sync_resolution_with_fs(
   // node_modules/.deno/<package_folder_id_folder_name>/node_modules/<package_name>
   let package_partitions =
     snapshot.all_system_packages_partitioned(system_info);
-  let mut handles: Vec<JoinHandle<Result<(), AnyError>>> =
-    Vec::with_capacity(package_partitions.packages.len());
+  let mut cache_futures = FuturesUnordered::new();
   let mut newest_packages_by_name: HashMap<&String, &NpmResolutionPackage> =
     HashMap::with_capacity(package_partitions.packages.len());
   let bin_entries = Arc::new(Mutex::new(bin_entries::BinEntries::new()));
@@ -317,21 +327,19 @@ async fn sync_resolution_with_fs(
       // are forced to be recreated
       setup_cache.remove_dep(&package_folder_name);
 
-      let pb = progress_bar.clone();
-      let cache = cache.clone();
-      let package = package.clone();
       let bin_entries_to_setup = bin_entries.clone();
-      let handle = spawn(async move {
-        cache.ensure_package(&package.id.nv, &package.dist).await?;
-        let pb_guard = pb.update_with_prompt(
+      cache_futures.push(async move {
+        tarball_cache
+          .ensure_package(&package.id.nv, &package.dist, http_client)
+          .await?;
+        let pb_guard = progress_bar.update_with_prompt(
           ProgressMessagePrompt::Initialize,
           &package.id.nv.to_string(),
         );
         let sub_node_modules = folder_path.join("node_modules");
         let package_path =
           join_package_name(&sub_node_modules, &package.id.nv.name);
-        let cache_folder =
-          cache.package_folder_for_name_and_version(&package.id.nv);
+        let cache_folder = cache.package_folder_for_nv(&package.id.nv);
 
         deno_core::unsync::spawn_blocking({
           let package_path = package_path.clone();
@@ -353,15 +361,13 @@ async fn sync_resolution_with_fs(
 
         // finally stop showing the progress bar
         drop(pb_guard); // explicit for clarity
-        Ok(())
+        Ok::<_, AnyError>(())
       });
-      handles.push(handle);
     }
   }
 
-  let results = futures::future::join_all(handles).await;
-  for result in results {
-    result??; // surface the first error
+  while let Some(result) = cache_futures.next().await {
+    result?; // surface the first error
   }
 
   // 2. Create any "copy" packages, which are used for peer dependencies
diff --git a/cli/npm/managed/resolvers/mod.rs b/cli/npm/managed/resolvers/mod.rs
index d5472344a..5f0343805 100644
--- a/cli/npm/managed/resolvers/mod.rs
+++ b/cli/npm/managed/resolvers/mod.rs
@@ -7,6 +7,7 @@ mod local;
 use std::path::PathBuf;
 use std::sync::Arc;
 
+use deno_npm::npm_rc::ResolvedNpmRc;
 use deno_npm::NpmSystemInfo;
 use deno_runtime::deno_fs::FileSystem;
 
@@ -18,28 +19,38 @@ use self::global::GlobalNpmPackageResolver;
 use self::local::LocalNpmPackageResolver;
 
 use super::cache::NpmCache;
+use super::cache::TarballCache;
 use super::resolution::NpmResolution;
 
 pub fn create_npm_fs_resolver(
   fs: Arc<dyn FileSystem>,
-  cache: Arc<NpmCache>,
+  npm_cache: Arc<NpmCache>,
+  npm_rc: Arc<ResolvedNpmRc>,
   progress_bar: &ProgressBar,
   resolution: Arc<NpmResolution>,
   maybe_node_modules_path: Option<PathBuf>,
   system_info: NpmSystemInfo,
 ) -> Arc<dyn NpmPackageFsResolver> {
+  let tarball_cache = Arc::new(TarballCache::new(
+    npm_cache.clone(),
+    fs.clone(),
+    npm_rc,
+    progress_bar.clone(),
+  ));
   match maybe_node_modules_path {
     Some(node_modules_folder) => Arc::new(LocalNpmPackageResolver::new(
+      npm_cache,
       fs,
-      cache,
       progress_bar.clone(),
-      node_modules_folder,
       resolution,
+      tarball_cache,
+      node_modules_folder,
       system_info,
     )),
     None => Arc::new(GlobalNpmPackageResolver::new(
+      npm_cache,
       fs,
-      cache,
+      tarball_cache,
       resolution,
       system_info,
     )),