fix: limit the max amount of concurrent network requests

crates/cli/src/commands/add.rs

@@ -1,5 +1,5 @@
 use clap::Parser;
-use std::collections::VecDeque;
+use std::{collections::VecDeque, sync::Arc};
 
 use crate::{
     package::{fetch_package_version_directly, find_package_version_from_registry},
@@ -9,6 +9,7 @@ use futures_util::future;
 use pacquet_diagnostics::miette::WrapErr;
 use pacquet_package_json::DependencyGroup;
 use pacquet_registry::PackageVersion;
+use tokio::sync::Semaphore;
 
 #[derive(Parser, Debug)]
 pub struct AddCommandArgs {
@@ -59,13 +60,15 @@ impl PackageManager {
     /// 5. Symlink all dependencies to node_modules/.pacquet/pkg@version/node_modules
     /// 6. Update package.json
     pub async fn add(&mut self, args: &AddCommandArgs) -> Result<(), PackageManagerError> {
+        let semaphore = Arc::new(Semaphore::new(16));
         let latest_version = fetch_package_version_directly(
             &self.tarball_cache,
             self.config,
             &self.http_client,
             &args.package,
             "latest",
             &self.config.modules_dir,
+            &semaphore,
         )
         .await?;
         let package_node_modules_path =
@@ -85,6 +88,7 @@ impl PackageManager {
                     name,
                     version,
                     path,
+                    &semaphore,
                 )
             });
 
@@ -109,6 +113,7 @@ impl PackageManager {
                             name,
                             version,
                             &node_modules_path,
+                            &semaphore,
                         )
                     },
                 );

crates/cli/src/commands/install.rs

@@ -7,6 +7,8 @@ use pacquet_diagnostics::tracing;
 use pacquet_package_json::DependencyGroup;
 use pacquet_registry::PackageVersion;
 use pipe_trait::Pipe;
+use std::sync::Arc;
+use tokio::sync::Semaphore;
 
 #[derive(Parser, Debug)]
 pub struct InstallCommandArgs {
@@ -46,7 +48,7 @@ impl PackageManager {
     ///
     /// This function is used by [`PackageManager::install`].
     #[async_recursion]
-    async fn install_dependencies(&self, package: &PackageVersion) {
+    async fn install_dependencies(&self, package: &PackageVersion, semaphore: &Semaphore) {
         let node_modules_path =
             self.config.virtual_store_dir.join(package.to_store_name()).join("node_modules");
 
@@ -62,10 +64,11 @@ impl PackageManager {
                     name,
                     version,
                     &node_modules_path,
+                    &semaphore,
                 )
                 .await
                 .unwrap();
-                self.install_dependencies(&dependency).await;
+                self.install_dependencies(&dependency, &semaphore).await;
             })
             .pipe(future::join_all)
             .await;
@@ -76,21 +79,26 @@ impl PackageManager {
     /// Jobs of the `install` command.
     pub async fn install(&self, args: &InstallCommandArgs) -> Result<(), PackageManagerError> {
         tracing::info!(target: "pacquet::install", "Start all");
+        let semaphore = Arc::new(Semaphore::new(16));
 
         self.package_json
             .dependencies(args.dependency_groups())
-            .map(|(name, version)| async move {
-                let dependency = find_package_version_from_registry(
-                    &self.tarball_cache,
-                    self.config,
-                    &self.http_client,
-                    name,
-                    version,
-                    &self.config.modules_dir,
-                )
-                .await
-                .unwrap();
-                self.install_dependencies(&dependency).await;
+            .map(|(name, version)| {
+                let semaphore_clone = Arc::clone(&semaphore);
+                async move {
+                    let dependency = find_package_version_from_registry(
+                        &self.tarball_cache,
+                        self.config,
+                        &self.http_client,
+                        name,
+                        version,
+                        &self.config.modules_dir,
+                        &semaphore_clone,
+                    )
+                    .await
+                    .unwrap();
+                    self.install_dependencies(&dependency, &semaphore_clone).await;
+                }
             })
             .pipe(future::join_all)
             .await;

crates/cli/src/package.rs

@@ -5,6 +5,7 @@ use pacquet_registry::{Package, PackageVersion};
 use pacquet_tarball::{download_tarball_to_store, Cache};
 use reqwest::Client;
 use std::path::Path;
+use tokio::sync::Semaphore;
 
 /// This function execute the following and returns the package
 /// - retrieves the package from the registry
@@ -21,10 +22,13 @@ pub async fn find_package_version_from_registry(
     name: &str,
     version: &str,
     symlink_path: &Path,
+    semaphore: &Semaphore,
 ) -> Result<PackageVersion, PackageManagerError> {
-    let package = Package::fetch_from_registry(name, http_client, &config.registry).await?;
+    let package =
+        Package::fetch_from_registry(name, http_client, &config.registry, semaphore).await?;
     let package_version = package.pinned_version(version).unwrap();
-    internal_fetch(tarball_cache, http_client, package_version, config, symlink_path).await?;
+    internal_fetch(tarball_cache, http_client, package_version, config, symlink_path, semaphore)
+        .await?;
     Ok(package_version.to_owned())
 }
 
@@ -35,10 +39,18 @@ pub async fn fetch_package_version_directly(
     name: &str,
     version: &str,
     symlink_path: &Path,
+    semaphore: &Semaphore,
 ) -> Result<PackageVersion, PackageManagerError> {
-    let package_version =
-        PackageVersion::fetch_from_registry(name, version, http_client, &config.registry).await?;
-    internal_fetch(tarball_cache, http_client, &package_version, config, symlink_path).await?;
+    let package_version = PackageVersion::fetch_from_registry(
+        name,
+        version,
+        http_client,
+        &config.registry,
+        semaphore,
+    )
+    .await?;
+    internal_fetch(tarball_cache, http_client, &package_version, config, symlink_path, semaphore)
+        .await?;
     Ok(package_version.to_owned())
 }
 
@@ -48,6 +60,7 @@ async fn internal_fetch(
     package_version: &PackageVersion,
     config: &'static Npmrc,
     symlink_path: &Path,
+    semaphore: &Semaphore,
 ) -> Result<(), PackageManagerError> {
     let store_folder_name = package_version.to_store_name();
 
@@ -59,6 +72,7 @@ async fn internal_fetch(
         package_version.dist.integrity.as_ref().expect("has integrity field"),
         package_version.dist.unpacked_size,
         package_version.as_tarball_url(),
+        semaphore,
     )
     .await?;
 

crates/registry/src/package.rs

@@ -7,6 +7,7 @@ use pipe_trait::Pipe;
 use serde::{Deserialize, Serialize};
 
 use crate::{package_version::PackageVersion, NetworkError, RegistryError};
+use tokio::sync::Semaphore;
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Package {
@@ -30,9 +31,11 @@ impl Package {
         name: &str,
         http_client: &reqwest::Client,
         registry: &str,
+        semaphore: &Semaphore,
     ) -> Result<Self, RegistryError> {
         let url = || format!("{registry}{name}"); // TODO: use reqwest URL directly
         let network_error = |error| NetworkError { error, url: url() };
+        let _permit = semaphore.acquire().await;
         http_client
             .get(url())
             .header("content-type", "application/json")

crates/registry/src/package_version.rs

@@ -4,6 +4,7 @@ use pipe_trait::Pipe;
 use serde::{Deserialize, Serialize};
 
 use crate::{package_distribution::PackageDistribution, NetworkError, RegistryError};
+use tokio::sync::Semaphore;
 
 #[derive(Serialize, Deserialize, Debug, Clone, Eq)]
 #[serde(rename_all = "camelCase")]
@@ -28,9 +29,11 @@ impl PackageVersion {
         version: &str,
         http_client: &reqwest::Client,
         registry: &str,
+        semaphore: &Semaphore,
     ) -> Result<Self, RegistryError> {
         let url = || format!("{registry}{name}/{version}");
         let network_error = |error| NetworkError { error, url: url() };
+        let _permit = semaphore.acquire().await;
 
         http_client
             .get(url())

crates/tarball/src/lib.rs

@@ -16,7 +16,7 @@ use pipe_trait::Pipe;
 use reqwest::Client;
 use ssri::{Integrity, IntegrityChecker};
 use tar::Archive;
-use tokio::sync::{Notify, RwLock};
+use tokio::sync::{Notify, RwLock, Semaphore};
 use zune_inflate::{errors::InflateDecodeErrors, DeflateDecoder, DeflateOptions};
 
 #[derive(Error, Debug, Diagnostic)]
@@ -120,6 +120,7 @@ pub async fn download_tarball_to_store(
     package_integrity: &str,
     package_unpacked_size: Option<usize>,
     package_url: &str,
+    semaphore: &Semaphore,
 ) -> Result<Arc<HashMap<OsString, PathBuf>>, TarballError> {
     if let Some(cache_lock) = cache.get(package_url) {
         let notify;
@@ -154,6 +155,7 @@ pub async fn download_tarball_to_store(
             store_dir,
             package_integrity,
             package_unpacked_size,
+            semaphore,
         )
         .await?;
         let mut cache_write = cache_lock.write().await;
@@ -169,10 +171,12 @@ async fn download_tarball_to_store_uncached(
     store_dir: &'static Path,
     package_integrity: &str,
     package_unpacked_size: Option<usize>,
+    semaphore: &Semaphore,
 ) -> Result<Arc<HashMap<OsString, PathBuf>>, TarballError> {
     tracing::info!(target: "pacquet::download", ?package_url, "New cache");
 
     let network_error = |error| NetworkError { url: package_url.to_string(), error };
+    let permit = semaphore.acquire().await;
     let response = http_client
         .get(package_url)
         .send()
@@ -181,6 +185,7 @@ async fn download_tarball_to_store_uncached(
         .bytes()
         .await
         .map_err(network_error)?;
+    drop(permit);
 
     tracing::info!(target: "pacquet::download", ?package_url, "Download completed");
 

tasks/benchmark/src/main.rs

@@ -1,4 +1,4 @@
-use std::{fs, path::Path};
+use std::{fs, path::Path, sync::Arc};
 
 use clap::Parser;
 use criterion::{Criterion, Throughput};
@@ -8,6 +8,7 @@ use pipe_trait::Pipe;
 use project_root::get_project_root;
 use reqwest::Client;
 use tempfile::tempdir;
+use tokio::sync::Semaphore;
 
 #[derive(Debug, Parser)]
 struct CliArgs {
@@ -30,6 +31,7 @@ fn bench_tarball(c: &mut Criterion, server: &mut ServerGuard, fixtures_folder: &
             // NOTE: the tempdir is being leaked, meaning the cleanup would be postponed until the end of the benchmark
             let dir = tempdir().unwrap().pipe(Box::new).pipe(Box::leak);
             let http_client = Client::new();
+            let semaphore = Arc::new(Semaphore::new(16));
 
             let cas_map =
                 download_tarball_to_store(
@@ -39,6 +41,7 @@ fn bench_tarball(c: &mut Criterion, server: &mut ServerGuard, fixtures_folder: &
                     "sha512-dj7vjIn1Ar8sVXj2yAXiMNCJDmS9MQ9XMlIecX2dIzzhjSHCyKo4DdXjXMs7wKW2kj6yvVRSpuQjOZ3YLrh56w==",
                     Some(16697),
                     url,
+                    &semaphore,
                 ).await.unwrap();
             cas_map.len()
         });