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

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