4 files changed, 114 insertions, 101 deletions

diff --git a/cli/tools/registry/api.rs b/cli/tools/registry/api.rs
index de9b4a333..c7097267d 100644
--- a/cli/tools/registry/api.rs
+++ b/cli/tools/registry/api.rs
@@ -6,6 +6,8 @@ use deno_runtime::deno_fetch::reqwest;
 use lsp_types::Url;
 use serde::de::DeserializeOwned;
 
+use crate::http_util::HttpClient;
+
 #[derive(serde::Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct CreateAuthorizationResponse {
@@ -116,8 +118,8 @@ pub async fn parse_response<T: DeserializeOwned>(
 }
 
 pub async fn get_scope(
-  client: &reqwest::Client,
-  registry_api_url: &str,
+  client: &HttpClient,
+  registry_api_url: &Url,
   scope: &str,
 ) -> Result<reqwest::Response, AnyError> {
   let scope_url = format!("{}scopes/{}", registry_api_url, scope);
@@ -126,7 +128,7 @@ pub async fn get_scope(
 }
 
 pub fn get_package_api_url(
-  registry_api_url: &str,
+  registry_api_url: &Url,
   scope: &str,
   package: &str,
 ) -> String {
@@ -134,8 +136,8 @@ pub fn get_package_api_url(
 }
 
 pub async fn get_package(
-  client: &reqwest::Client,
-  registry_api_url: &str,
+  client: &HttpClient,
+  registry_api_url: &Url,
   scope: &str,
   package: &str,
 ) -> Result<reqwest::Response, AnyError> {
diff --git a/cli/tools/registry/mod.rs b/cli/tools/registry/mod.rs
index 23e8f4313..d300e5eaf 100644
--- a/cli/tools/registry/mod.rs
+++ b/cli/tools/registry/mod.rs
@@ -17,11 +17,13 @@ use deno_config::WorkspaceMemberConfig;
 use deno_core::anyhow::bail;
 use deno_core::anyhow::Context;
 use deno_core::error::AnyError;
+use deno_core::futures::future::LocalBoxFuture;
+use deno_core::futures::stream::FuturesUnordered;
 use deno_core::futures::FutureExt;
+use deno_core::futures::StreamExt;
 use deno_core::serde_json;
 use deno_core::serde_json::json;
 use deno_core::serde_json::Value;
-use deno_core::unsync::JoinSet;
 use deno_runtime::deno_fetch::reqwest;
 use deno_runtime::deno_fs::FileSystem;
 use deno_terminal::colors;
@@ -154,7 +156,7 @@ pub async fn publish(
   }
 
   perform_publish(
-    cli_factory.http_client(),
+    &cli_factory.http_client_provider().get_or_create()?,
     prepared_data.publish_order_graph,
     prepared_data.package_by_name,
     auth_method,
@@ -523,9 +525,9 @@ pub enum Permission<'s> {
 }
 
 async fn get_auth_headers(
-  client: &reqwest::Client,
-  registry_url: String,
-  packages: Vec<Rc<PreparedPublishPackage>>,
+  client: &HttpClient,
+  registry_url: &Url,
+  packages: &[Rc<PreparedPublishPackage>],
   auth_method: AuthMethod,
 ) -> Result<HashMap<(String, String, String), Rc<str>>, AnyError> {
   let permissions = packages
@@ -600,7 +602,7 @@ async fn get_auth_headers(
               colors::cyan(res.user.name)
             );
             let authorization: Rc<str> = format!("Bearer {}", res.token).into();
-            for pkg in &packages {
+            for pkg in packages {
               authorizations.insert(
                 (pkg.scope.clone(), pkg.package.clone(), pkg.version.clone()),
                 authorization.clone(),
@@ -620,7 +622,7 @@ async fn get_auth_headers(
     }
     AuthMethod::Token(token) => {
       let authorization: Rc<str> = format!("Bearer {}", token).into();
-      for pkg in &packages {
+      for pkg in packages {
         authorizations.insert(
           (pkg.scope.clone(), pkg.package.clone(), pkg.version.clone()),
           authorization.clone(),
@@ -682,9 +684,9 @@ async fn get_auth_headers(
 /// Check if both `scope` and `package` already exist, if not return
 /// a URL to the management panel to create them.
 async fn check_if_scope_and_package_exist(
-  client: &reqwest::Client,
-  registry_api_url: &str,
-  registry_manage_url: &str,
+  client: &HttpClient,
+  registry_api_url: &Url,
+  registry_manage_url: &Url,
   scope: &str,
   package: &str,
 ) -> Result<Option<String>, AnyError> {
@@ -714,18 +716,18 @@ async fn check_if_scope_and_package_exist(
 }
 
 async fn ensure_scopes_and_packages_exist(
-  client: &reqwest::Client,
-  registry_api_url: String,
-  registry_manage_url: String,
-  packages: Vec<Rc<PreparedPublishPackage>>,
+  client: &HttpClient,
+  registry_api_url: &Url,
+  registry_manage_url: &Url,
+  packages: &[Rc<PreparedPublishPackage>],
 ) -> Result<(), AnyError> {
   if !std::io::stdin().is_terminal() {
     let mut missing_packages_lines = vec![];
     for package in packages {
       let maybe_create_package_url = check_if_scope_and_package_exist(
         client,
-        &registry_api_url,
-        &registry_manage_url,
+        registry_api_url,
+        registry_manage_url,
         &package.scope,
         &package.package,
       )
@@ -748,8 +750,8 @@ async fn ensure_scopes_and_packages_exist(
   for package in packages {
     let maybe_create_package_url = check_if_scope_and_package_exist(
       client,
-      &registry_api_url,
-      &registry_manage_url,
+      registry_api_url,
+      registry_manage_url,
       &package.scope,
       &package.package,
     )
@@ -770,7 +772,7 @@ async fn ensure_scopes_and_packages_exist(
     let _ = open::that_detached(&create_package_url);
 
     let package_api_url = api::get_package_api_url(
-      &registry_api_url,
+      registry_api_url,
       &package.scope,
       &package.package,
     );
@@ -790,15 +792,14 @@ async fn ensure_scopes_and_packages_exist(
 }
 
 async fn perform_publish(
-  http_client: &Arc<HttpClient>,
+  http_client: &HttpClient,
   mut publish_order_graph: PublishOrderGraph,
   mut prepared_package_by_name: HashMap<String, Rc<PreparedPublishPackage>>,
   auth_method: AuthMethod,
   provenance: bool,
 ) -> Result<(), AnyError> {
-  let client = http_client.client()?;
-  let registry_api_url = jsr_api_url().to_string();
-  let registry_url = jsr_url().to_string();
+  let registry_api_url = jsr_api_url();
+  let registry_url = jsr_url();
 
   let packages = prepared_package_by_name
     .values()
@@ -806,19 +807,20 @@ async fn perform_publish(
     .collect::<Vec<_>>();
 
   ensure_scopes_and_packages_exist(
-    client,
-    registry_api_url.clone(),
-    registry_url.clone(),
-    packages.clone(),
+    http_client,
+    registry_api_url,
+    registry_url,
+    &packages,
   )
   .await?;
 
   let mut authorizations =
-    get_auth_headers(client, registry_api_url.clone(), packages, auth_method)
+    get_auth_headers(http_client, registry_api_url, &packages, auth_method)
       .await?;
 
   assert_eq!(prepared_package_by_name.len(), authorizations.len());
-  let mut futures: JoinSet<Result<String, AnyError>> = JoinSet::default();
+  let mut futures: FuturesUnordered<LocalBoxFuture<Result<String, AnyError>>> =
+    Default::default();
   loop {
     let next_batch = publish_order_graph.next();
 
@@ -844,32 +846,32 @@ async fn perform_publish(
           package.version.clone(),
         ))
         .unwrap();
-      let registry_api_url = registry_api_url.clone();
-      let registry_url = registry_url.clone();
-      let http_client = http_client.clone();
-      futures.spawn(async move {
-        let display_name = package.display_name();
-        publish_package(
-          &http_client,
-          package,
-          &registry_api_url,
-          &registry_url,
-          &authorization,
-          provenance,
-        )
-        .await
-        .with_context(|| format!("Failed to publish {}", display_name))?;
-        Ok(package_name)
-      });
+      futures.push(
+        async move {
+          let display_name = package.display_name();
+          publish_package(
+            http_client,
+            package,
+            registry_api_url,
+            registry_url,
+            &authorization,
+            provenance,
+          )
+          .await
+          .with_context(|| format!("Failed to publish {}", display_name))?;
+          Ok(package_name)
+        }
+        .boxed_local(),
+      );
     }
 
-    let Some(result) = futures.join_next().await else {
+    let Some(result) = futures.next().await else {
       // done, ensure no circular dependency
       publish_order_graph.ensure_no_pending()?;
       break;
     };
 
-    let package_name = result??;
+    let package_name = result?;
     publish_order_graph.finish_package(&package_name);
   }
 
@@ -879,12 +881,11 @@ async fn perform_publish(
 async fn publish_package(
   http_client: &HttpClient,
   package: Rc<PreparedPublishPackage>,
-  registry_api_url: &str,
-  registry_url: &str,
+  registry_api_url: &Url,
+  registry_url: &Url,
   authorization: &str,
   provenance: bool,
 ) -> Result<(), AnyError> {
-  let client = http_client.client()?;
   log::info!(
     "{} @{}/{}@{} ...",
     colors::intense_blue("Publishing"),
@@ -902,7 +903,7 @@ async fn publish_package(
     package.config
   );
 
-  let response = client
+  let response = http_client
     .post(url)
     .header(reqwest::header::AUTHORIZATION, authorization)
     .header(reqwest::header::CONTENT_ENCODING, "gzip")
@@ -950,7 +951,7 @@ async fn publish_package(
   let interval = std::time::Duration::from_secs(2);
   while task.status != "success" && task.status != "failure" {
     tokio::time::sleep(interval).await;
-    let resp = client
+    let resp = http_client
       .get(format!("{}publish_status/{}", registry_api_url, task.id))
       .send()
       .await
@@ -1000,7 +1001,7 @@ async fn publish_package(
       package.scope, package.package, package.version
     ))?;
 
-    let meta_bytes = client.get(meta_url).send().await?.bytes().await?;
+    let meta_bytes = http_client.get(meta_url).send().await?.bytes().await?;
 
     if std::env::var("DISABLE_JSR_MANIFEST_VERIFICATION_FOR_TESTING").is_err() {
       verify_version_manifest(&meta_bytes, &package)?;
@@ -1015,7 +1016,7 @@ async fn publish_package(
         sha256: faster_hex::hex_string(&sha2::Sha256::digest(&meta_bytes)),
       },
     };
-    let bundle = provenance::generate_provenance(subject).await?;
+    let bundle = provenance::generate_provenance(http_client, subject).await?;
 
     let tlog_entry = &bundle.verification_material.tlog_entries[0];
     log::info!("{}",
@@ -1030,7 +1031,7 @@ async fn publish_package(
       "{}scopes/{}/packages/{}/versions/{}/provenance",
       registry_api_url, package.scope, package.package, package.version
     );
-    client
+    http_client
       .post(provenance_url)
       .header(reqwest::header::AUTHORIZATION, authorization)
       .json(&json!({ "bundle": bundle }))
diff --git a/cli/tools/registry/pm.rs b/cli/tools/registry/pm.rs
index e37ee3d82..62d0f604a 100644
--- a/cli/tools/registry/pm.rs
+++ b/cli/tools/registry/pm.rs
@@ -188,7 +188,7 @@ pub async fn add(flags: Flags, add_flags: AddFlags) -> Result<(), AnyError> {
   }
   let config_file_path = config_specifier.to_file_path().unwrap();
 
-  let http_client = cli_factory.http_client();
+  let http_client = cli_factory.http_client_provider();
 
   let mut selected_packages = Vec::with_capacity(add_flags.packages.len());
   let mut package_reqs = Vec::with_capacity(add_flags.packages.len());
@@ -227,6 +227,7 @@ pub async fn add(flags: Flags, add_flags: AddFlags) -> Result<(), AnyError> {
     None,
   );
   deps_file_fetcher.set_download_log_level(log::Level::Trace);
+  let deps_file_fetcher = Arc::new(deps_file_fetcher);
   let jsr_resolver = Arc::new(JsrFetchResolver::new(deps_file_fetcher.clone()));
   let npm_resolver = Arc::new(NpmFetchResolver::new(deps_file_fetcher));
 
diff --git a/cli/tools/registry/provenance.rs b/cli/tools/registry/provenance.rs
index 69926372e..7fa2be381 100644
--- a/cli/tools/registry/provenance.rs
+++ b/cli/tools/registry/provenance.rs
@@ -1,5 +1,7 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
+use crate::http_util::HttpClient;
+
 use super::api::OidcTokenResponse;
 use super::auth::gha_oidc_token;
 use super::auth::is_gha;
@@ -13,7 +15,6 @@ use deno_core::serde_json;
 use once_cell::sync::Lazy;
 use p256::elliptic_curve;
 use p256::pkcs8::AssociatedOid;
-use reqwest::Client;
 use ring::rand::SystemRandom;
 use ring::signature::EcdsaKeyPair;
 use ring::signature::KeyPair;
@@ -291,6 +292,7 @@ pub struct ProvenanceBundle {
 }
 
 pub async fn generate_provenance(
+  http_client: &HttpClient,
   subject: Subject,
 ) -> Result<ProvenanceBundle, AnyError> {
   if !is_gha() {
@@ -306,19 +308,20 @@ pub async fn generate_provenance(
   let slsa = ProvenanceAttestation::new_github_actions(subject);
 
   let attestation = serde_json::to_string(&slsa)?;
-  let bundle = attest(&attestation, INTOTO_PAYLOAD_TYPE).await?;
+  let bundle = attest(http_client, &attestation, INTOTO_PAYLOAD_TYPE).await?;
 
   Ok(bundle)
 }
 
 pub async fn attest(
+  http_client: &HttpClient,
   data: &str,
   type_: &str,
 ) -> Result<ProvenanceBundle, AnyError> {
   // DSSE Pre-Auth Encoding (PAE) payload
   let pae = pre_auth_encoding(type_, data);
 
-  let signer = FulcioSigner::new()?;
+  let signer = FulcioSigner::new(http_client)?;
   let (signature, key_material) = signer.sign(&pae).await?;
 
   let content = SignatureBundle {
@@ -332,7 +335,8 @@ pub async fn attest(
       }],
     },
   };
-  let transparency_logs = testify(&content, &key_material.certificate).await?;
+  let transparency_logs =
+    testify(http_client, &content, &key_material.certificate).await?;
 
   // First log entry is the one we're interested in
   let (_, log_entry) = transparency_logs.iter().next().unwrap();
@@ -363,13 +367,6 @@ static DEFAULT_FULCIO_URL: Lazy<String> = Lazy::new(|| {
     .unwrap_or_else(|_| "https://fulcio.sigstore.dev".to_string())
 });
 
-struct FulcioSigner {
-  // The ephemeral key pair used to sign.
-  ephemeral_signer: EcdsaKeyPair,
-  rng: SystemRandom,
-  client: Client,
-}
-
 static ALGORITHM: &ring::signature::EcdsaSigningAlgorithm =
   &ring::signature::ECDSA_P256_SHA256_ASN1_SIGNING;
 
@@ -424,8 +421,15 @@ struct SigningCertificateResponse {
   signed_certificate_detached_sct: Option<SignedCertificate>,
 }
 
-impl FulcioSigner {
-  pub fn new() -> Result<Self, AnyError> {
+struct FulcioSigner<'a> {
+  // The ephemeral key pair used to sign.
+  ephemeral_signer: EcdsaKeyPair,
+  rng: SystemRandom,
+  http_client: &'a HttpClient,
+}
+
+impl<'a> FulcioSigner<'a> {
+  pub fn new(http_client: &'a HttpClient) -> Result<Self, AnyError> {
     let rng = SystemRandom::new();
     let document = EcdsaKeyPair::generate_pkcs8(ALGORITHM, &rng)?;
     let ephemeral_signer =
@@ -434,7 +438,7 @@ impl FulcioSigner {
     Ok(Self {
       ephemeral_signer,
       rng,
-      client: Client::new(),
+      http_client,
     })
   }
 
@@ -443,7 +447,7 @@ impl FulcioSigner {
     data: &[u8],
   ) -> Result<(ring::signature::Signature, KeyMaterial), AnyError> {
     // Request token from GitHub Actions for audience "sigstore"
-    let token = gha_request_token("sigstore").await?;
+    let token = self.gha_request_token("sigstore").await?;
     // Extract the subject from the token
     let subject = extract_jwt_subject(&token)?;
 
@@ -498,7 +502,12 @@ impl FulcioSigner {
       },
     };
 
-    let response = self.client.post(url).json(&request_body).send().await?;
+    let response = self
+      .http_client
+      .post(url)
+      .json(&request_body)
+      .send()
+      .await?;
 
     let body: SigningCertificateResponse = response.json().await?;
 
@@ -508,6 +517,27 @@ impl FulcioSigner {
       .ok_or_else(|| anyhow::anyhow!("No certificate chain returned"))?;
     Ok(key.chain.certificates)
   }
+
+  async fn gha_request_token(&self, aud: &str) -> Result<String, AnyError> {
+    let Ok(req_url) = env::var("ACTIONS_ID_TOKEN_REQUEST_URL") else {
+      bail!("Not running in GitHub Actions");
+    };
+
+    let Some(token) = gha_oidc_token() else {
+      bail!("No OIDC token available");
+    };
+
+    let res = self
+      .http_client
+      .get(&req_url)
+      .bearer_auth(token)
+      .query(&[("audience", aud)])
+      .send()
+      .await?
+      .json::<OidcTokenResponse>()
+      .await?;
+    Ok(res.value)
+  }
 }
 
 #[derive(Deserialize)]
@@ -532,27 +562,6 @@ fn extract_jwt_subject(token: &str) -> Result<String, AnyError> {
   }
 }
 
-async fn gha_request_token(aud: &str) -> Result<String, AnyError> {
-  let Ok(req_url) = env::var("ACTIONS_ID_TOKEN_REQUEST_URL") else {
-    bail!("Not running in GitHub Actions");
-  };
-
-  let Some(token) = gha_oidc_token() else {
-    bail!("No OIDC token available");
-  };
-
-  let client = Client::new();
-  let res = client
-    .get(&req_url)
-    .bearer_auth(token)
-    .query(&[("audience", aud)])
-    .send()
-    .await?
-    .json::<OidcTokenResponse>()
-    .await?;
-  Ok(res.value)
-}
-
 static DEFAULT_REKOR_URL: Lazy<String> = Lazy::new(|| {
   env::var("REKOR_URL")
     .unwrap_or_else(|_| "https://rekor.sigstore.dev".to_string())
@@ -616,6 +625,7 @@ struct ProposedIntotoEntryHash {
 
 // Rekor witness
 async fn testify(
+  http_client: &HttpClient,
   content: &SignatureBundle,
   public_key: &str,
 ) -> Result<RekorEntry, AnyError> {
@@ -672,9 +682,8 @@ async fn testify(
     },
   };
 
-  let client = Client::new();
   let url = format!("{}/api/v1/log/entries", *DEFAULT_REKOR_URL);
-  let res = client
+  let res = http_client
     .post(&url)
     .json(&proposed_intoto_entry)
     .send()