Reland "refactor(fetch): reimplement fetch with hyper instead of reqwest" (#24593)

Originally landed in
https://github.com/denoland/deno/commit/f6fd6619e708a515831f707438368d81b0c9aa56.
Reverted in https://github.com/denoland/deno/pull/24574.

This reland contains a fix that sends "Accept: */*" header for calls made
from "FileFetcher". Absence of this header made downloading source code
from JSR broken. This is tested by ensuring this header is present in the
test server that servers JSR packages.

---------

Co-authored-by: Sean McArthur <sean@seanmonstar.com>

10 files changed, 259 insertions, 119 deletions

diff --git a/cli/Cargo.toml b/cli/Cargo.toml
index d02411dc2..d9144afff 100644
--- a/cli/Cargo.toml
+++ b/cli/Cargo.toml
@@ -107,6 +107,10 @@ faster-hex.workspace = true
 flate2.workspace = true
 fs3.workspace = true
 glob = "0.3.1"
+http.workspace = true
+http-body.workspace = true
+http-body-util.workspace = true
+hyper-util.workspace = true
 import_map = { version = "=0.20.0", features = ["ext"] }
 indexmap.workspace = true
 jsonc-parser.workspace = true
@@ -128,7 +132,6 @@ phf.workspace = true
 quick-junit = "^0.3.5"
 rand = { workspace = true, features = ["small_rng"] }
 regex.workspace = true
-reqwest.workspace = true
 ring.workspace = true
 rustyline.workspace = true
 rustyline-derive = "=0.7.0"
diff --git a/cli/http_util.rs b/cli/http_util.rs
index 18c0687bd..a8646c188 100644
--- a/cli/http_util.rs
+++ b/cli/http_util.rs
@@ -12,18 +12,22 @@ use deno_core::error::generic_error;
 use deno_core::error::AnyError;
 use deno_core::futures::StreamExt;
 use deno_core::parking_lot::Mutex;
+use deno_core::serde;
+use deno_core::serde_json;
 use deno_core::url::Url;
+use deno_runtime::deno_fetch;
 use deno_runtime::deno_fetch::create_http_client;
-use deno_runtime::deno_fetch::reqwest;
-use deno_runtime::deno_fetch::reqwest::header::HeaderName;
-use deno_runtime::deno_fetch::reqwest::header::HeaderValue;
-use deno_runtime::deno_fetch::reqwest::header::ACCEPT;
-use deno_runtime::deno_fetch::reqwest::header::AUTHORIZATION;
-use deno_runtime::deno_fetch::reqwest::header::IF_NONE_MATCH;
-use deno_runtime::deno_fetch::reqwest::header::LOCATION;
-use deno_runtime::deno_fetch::reqwest::StatusCode;
 use deno_runtime::deno_fetch::CreateHttpClientOptions;
 use deno_runtime::deno_tls::RootCertStoreProvider;
+use http::header::HeaderName;
+use http::header::HeaderValue;
+use http::header::ACCEPT;
+use http::header::AUTHORIZATION;
+use http::header::IF_NONE_MATCH;
+use http::header::LOCATION;
+use http::StatusCode;
+use http_body_util::BodyExt;
+
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::thread::ThreadId;
@@ -208,8 +212,7 @@ pub struct HttpClientProvider {
   // it's not safe to share a reqwest::Client across tokio runtimes,
   // so we store these Clients keyed by thread id
   // https://github.com/seanmonstar/reqwest/issues/1148#issuecomment-910868788
-  #[allow(clippy::disallowed_types)] // reqwest::Client allowed here
-  clients_by_thread_id: Mutex<HashMap<ThreadId, reqwest::Client>>,
+  clients_by_thread_id: Mutex<HashMap<ThreadId, deno_fetch::Client>>,
 }
 
 impl std::fmt::Debug for HttpClientProvider {
@@ -270,9 +273,15 @@ pub struct BadResponseError {
 #[derive(Debug, Error)]
 pub enum DownloadError {
   #[error(transparent)]
-  Reqwest(#[from] reqwest::Error),
+  Fetch(AnyError),
+  #[error(transparent)]
+  UrlParse(#[from] deno_core::url::ParseError),
+  #[error(transparent)]
+  HttpParse(#[from] http::Error),
+  #[error(transparent)]
+  Json(#[from] serde_json::Error),
   #[error(transparent)]
-  ToStr(#[from] reqwest::header::ToStrError),
+  ToStr(#[from] http::header::ToStrError),
   #[error("Redirection from '{}' did not provide location header", .request_url)]
   NoRedirectHeader { request_url: Url },
   #[error("Too many redirects.")]
@@ -283,8 +292,7 @@ pub enum DownloadError {
 
 #[derive(Debug)]
 pub struct HttpClient {
-  #[allow(clippy::disallowed_types)] // reqwest::Client allowed here
-  client: reqwest::Client,
+  client: deno_fetch::Client,
   // don't allow sending this across threads because then
   // it might be shared accidentally across tokio runtimes
   // which will cause issues
@@ -295,22 +303,56 @@ pub struct HttpClient {
 impl HttpClient {
   // DO NOT make this public. You should always be creating one of these from
   // the HttpClientProvider
-  #[allow(clippy::disallowed_types)] // reqwest::Client allowed here
-  fn new(client: reqwest::Client) -> Self {
+  fn new(client: deno_fetch::Client) -> Self {
     Self {
       client,
       _unsend_marker: deno_core::unsync::UnsendMarker::default(),
     }
   }
 
-  // todo(dsherret): don't expose `reqwest::RequestBuilder` because it
-  // is `Sync` and could accidentally be shared with multiple tokio runtimes
-  pub fn get(&self, url: impl reqwest::IntoUrl) -> reqwest::RequestBuilder {
-    self.client.get(url)
+  pub fn get(&self, url: Url) -> Result<RequestBuilder, http::Error> {
+    let body = http_body_util::Empty::new()
+      .map_err(|never| match never {})
+      .boxed();
+    let mut req = http::Request::new(body);
+    *req.uri_mut() = url.as_str().parse()?;
+    Ok(RequestBuilder {
+      client: self.client.clone(),
+      req,
+    })
+  }
+
+  pub fn post(
+    &self,
+    url: Url,
+    body: deno_fetch::ReqBody,
+  ) -> Result<RequestBuilder, http::Error> {
+    let mut req = http::Request::new(body);
+    *req.method_mut() = http::Method::POST;
+    *req.uri_mut() = url.as_str().parse()?;
+    Ok(RequestBuilder {
+      client: self.client.clone(),
+      req,
+    })
   }
 
-  pub fn post(&self, url: impl reqwest::IntoUrl) -> reqwest::RequestBuilder {
-    self.client.post(url)
+  pub fn post_json<S>(
+    &self,
+    url: Url,
+    ser: &S,
+  ) -> Result<RequestBuilder, DownloadError>
+  where
+    S: serde::Serialize,
+  {
+    let json = deno_core::serde_json::to_vec(ser)?;
+    let body = http_body_util::Full::new(json.into())
+      .map_err(|never| match never {})
+      .boxed();
+    let builder = self.post(url, body)?;
+    Ok(builder.header(
+      http::header::CONTENT_TYPE,
+      "application/json".parse().map_err(http::Error::from)?,
+    ))
   }
 
   /// Asynchronously fetches the given HTTP URL one pass only.
@@ -322,27 +364,35 @@ impl HttpClient {
     &self,
     args: FetchOnceArgs<'a>,
   ) -> Result<FetchOnceResult, AnyError> {
-    let mut request = self.client.get(args.url.clone());
+    let body = http_body_util::Empty::new()
+      .map_err(|never| match never {})
+      .boxed();
+    let mut request = http::Request::new(body);
+    *request.uri_mut() = args.url.as_str().parse()?;
 
     if let Some(etag) = args.maybe_etag {
       let if_none_match_val = HeaderValue::from_str(&etag)?;
-      request = request.header(IF_NONE_MATCH, if_none_match_val);
+      request
+        .headers_mut()
+        .insert(IF_NONE_MATCH, if_none_match_val);
     }
     if let Some(auth_token) = args.maybe_auth_token {
       let authorization_val = HeaderValue::from_str(&auth_token.to_string())?;
-      request = request.header(AUTHORIZATION, authorization_val);
+      request
+        .headers_mut()
+        .insert(AUTHORIZATION, authorization_val);
     }
     if let Some(accept) = args.maybe_accept {
       let accepts_val = HeaderValue::from_str(&accept)?;
-      request = request.header(ACCEPT, accepts_val);
+      request.headers_mut().insert(ACCEPT, accepts_val);
     }
-    let response = match request.send().await {
+    let response = match self.client.clone().send(request).await {
       Ok(resp) => resp,
       Err(err) => {
-        if err.is_connect() || err.is_timeout() {
+        if is_error_connect(&err) {
           return Ok(FetchOnceResult::RequestError(err.to_string()));
         }
-        return Err(err.into());
+        return Err(err);
       }
     };
 
@@ -406,18 +456,12 @@ impl HttpClient {
     Ok(FetchOnceResult::Code(body, result_headers))
   }
 
-  pub async fn download_text(
-    &self,
-    url: impl reqwest::IntoUrl,
-  ) -> Result<String, AnyError> {
+  pub async fn download_text(&self, url: Url) -> Result<String, AnyError> {
     let bytes = self.download(url).await?;
     Ok(String::from_utf8(bytes)?)
   }
 
-  pub async fn download(
-    &self,
-    url: impl reqwest::IntoUrl,
-  ) -> Result<Vec<u8>, AnyError> {
+  pub async fn download(&self, url: Url) -> Result<Vec<u8>, AnyError> {
     let maybe_bytes = self.download_inner(url, None, None).await?;
     match maybe_bytes {
       Some(bytes) => Ok(bytes),
@@ -427,7 +471,7 @@ impl HttpClient {
 
   pub async fn download_with_progress(
     &self,
-    url: impl reqwest::IntoUrl,
+    url: Url,
     maybe_header: Option<(HeaderName, HeaderValue)>,
     progress_guard: &UpdateGuard,
   ) -> Result<Option<Vec<u8>>, DownloadError> {
@@ -438,26 +482,26 @@ impl HttpClient {
 
   pub async fn get_redirected_url(
     &self,
-    url: impl reqwest::IntoUrl,
+    url: Url,
     maybe_header: Option<(HeaderName, HeaderValue)>,
   ) -> Result<Url, AnyError> {
-    let response = self.get_redirected_response(url, maybe_header).await?;
-    Ok(response.url().clone())
+    let (_, url) = self.get_redirected_response(url, maybe_header).await?;
+    Ok(url)
   }
 
   async fn download_inner(
     &self,
-    url: impl reqwest::IntoUrl,
+    url: Url,
     maybe_header: Option<(HeaderName, HeaderValue)>,
     progress_guard: Option<&UpdateGuard>,
   ) -> Result<Option<Vec<u8>>, DownloadError> {
-    let response = self.get_redirected_response(url, maybe_header).await?;
+    let (response, _) = self.get_redirected_response(url, maybe_header).await?;
 
     if response.status() == 404 {
       return Ok(None);
     } else if !response.status().is_success() {
       let status = response.status();
-      let maybe_response_text = response.text().await.ok();
+      let maybe_response_text = body_to_string(response).await.ok();
       return Err(DownloadError::BadResponse(BadResponseError {
         status_code: status,
         response_text: maybe_response_text
@@ -469,60 +513,77 @@ impl HttpClient {
     get_response_body_with_progress(response, progress_guard)
       .await
       .map(Some)
-      .map_err(Into::into)
+      .map_err(DownloadError::Fetch)
   }
 
   async fn get_redirected_response(
     &self,
-    url: impl reqwest::IntoUrl,
+    mut url: Url,
     mut maybe_header: Option<(HeaderName, HeaderValue)>,
-  ) -> Result<reqwest::Response, DownloadError> {
-    let mut url = url.into_url()?;
-    let mut builder = self.get(url.clone());
+  ) -> Result<(http::Response<deno_fetch::ResBody>, Url), DownloadError> {
+    let mut req = self.get(url.clone())?.build();
     if let Some((header_name, header_value)) = maybe_header.as_ref() {
-      builder = builder.header(header_name, header_value);
+      req.headers_mut().append(header_name, header_value.clone());
     }
-    let mut response = builder.send().await?;
+    let mut response = self
+      .client
+      .clone()
+      .send(req)
+      .await
+      .map_err(DownloadError::Fetch)?;
     let status = response.status();
     if status.is_redirection() {
       for _ in 0..5 {
         let new_url = resolve_redirect_from_response(&url, &response)?;
-        let mut builder = self.get(new_url.clone());
+        let mut req = self.get(new_url.clone())?.build();
 
         if new_url.origin() == url.origin() {
           if let Some((header_name, header_value)) = maybe_header.as_ref() {
-            builder = builder.header(header_name, header_value);
+            req.headers_mut().append(header_name, header_value.clone());
           }
         } else {
           maybe_header = None;
         }
 
-        let new_response = builder.send().await?;
+        let new_response = self
+          .client
+          .clone()
+          .send(req)
+          .await
+          .map_err(DownloadError::Fetch)?;
         let status = new_response.status();
         if status.is_redirection() {
           response = new_response;
           url = new_url;
         } else {
-          return Ok(new_response);
+          return Ok((new_response, new_url));
         }
       }
       Err(DownloadError::TooManyRedirects)
     } else {
-      Ok(response)
+      Ok((response, url))
     }
   }
 }
 
+fn is_error_connect(err: &AnyError) -> bool {
+  err
+    .downcast_ref::<hyper_util::client::legacy::Error>()
+    .map(|err| err.is_connect())
+    .unwrap_or(false)
+}
+
 async fn get_response_body_with_progress(
-  response: reqwest::Response,
+  response: http::Response<deno_fetch::ResBody>,
   progress_guard: Option<&UpdateGuard>,
-) -> Result<Vec<u8>, reqwest::Error> {
+) -> Result<Vec<u8>, AnyError> {
+  use http_body::Body as _;
   if let Some(progress_guard) = progress_guard {
-    if let Some(total_size) = response.content_length() {
+    if let Some(total_size) = response.body().size_hint().exact() {
       progress_guard.set_total_size(total_size);
       let mut current_size = 0;
       let mut data = Vec::with_capacity(total_size as usize);
-      let mut stream = response.bytes_stream();
+      let mut stream = response.into_body().into_data_stream();
       while let Some(item) = stream.next().await {
         let bytes = item?;
         current_size += bytes.len() as u64;
@@ -532,7 +593,7 @@ async fn get_response_body_with_progress(
       return Ok(data);
     }
   }
-  let bytes = response.bytes().await?;
+  let bytes = response.collect().await?.to_bytes();
   Ok(bytes.into())
 }
 
@@ -563,9 +624,9 @@ fn resolve_url_from_location(base_url: &Url, location: &str) -> Url {
   }
 }
 
-fn resolve_redirect_from_response(
+fn resolve_redirect_from_response<B>(
   request_url: &Url,
-  response: &reqwest::Response,
+  response: &http::Response<B>,
 ) -> Result<Url, DownloadError> {
   debug_assert!(response.status().is_redirection());
   if let Some(location) = response.headers().get(LOCATION) {
@@ -580,6 +641,49 @@ fn resolve_redirect_from_response(
   }
 }
 
+pub async fn body_to_string<B>(body: B) -> Result<String, AnyError>
+where
+  B: http_body::Body,
+  AnyError: From<B::Error>,
+{
+  let bytes = body.collect().await?.to_bytes();
+  let s = std::str::from_utf8(&bytes)?;
+  Ok(s.into())
+}
+
+pub async fn body_to_json<B, D>(body: B) -> Result<D, AnyError>
+where
+  B: http_body::Body,
+  AnyError: From<B::Error>,
+  D: serde::de::DeserializeOwned,
+{
+  let bytes = body.collect().await?.to_bytes();
+  let val = deno_core::serde_json::from_slice(&bytes)?;
+  Ok(val)
+}
+
+pub struct RequestBuilder {
+  client: deno_fetch::Client,
+  req: http::Request<deno_fetch::ReqBody>,
+}
+
+impl RequestBuilder {
+  pub fn header(mut self, name: HeaderName, value: HeaderValue) -> Self {
+    self.req.headers_mut().append(name, value);
+    self
+  }
+
+  pub async fn send(
+    self,
+  ) -> Result<http::Response<deno_fetch::ResBody>, AnyError> {
+    self.client.send(self.req).await
+  }
+
+  pub fn build(self) -> http::Request<deno_fetch::ReqBody> {
+    self.req
+  }
+}
+
 #[allow(clippy::print_stdout)]
 #[allow(clippy::print_stderr)]
 #[cfg(test)]
@@ -600,14 +704,20 @@ mod test {
 
     // make a request to the redirect server
     let text = client
-      .download_text("http://localhost:4546/subdir/redirects/redirect1.js")
+      .download_text(
+        Url::parse("http://localhost:4546/subdir/redirects/redirect1.js")
+          .unwrap(),
+      )
       .await
       .unwrap();
     assert_eq!(text, "export const redirect = 1;\n");
 
     // now make one to the infinite redirects server
     let err = client
-      .download_text("http://localhost:4549/subdir/redirects/redirect1.js")
+      .download_text(
+        Url::parse("http://localhost:4549/subdir/redirects/redirect1.js")
+          .unwrap(),
+      )
       .await
       .err()
       .unwrap();
diff --git a/cli/npm/common.rs b/cli/npm/common.rs
index c55f73cd5..34835216c 100644
--- a/cli/npm/common.rs
+++ b/cli/npm/common.rs
@@ -1,7 +1,7 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
 use deno_npm::npm_rc::RegistryConfig;
-use reqwest::header;
+use http::header;
 
 // TODO(bartlomieju): support more auth methods besides token and basic auth
 pub fn maybe_auth_header_for_npm_registry(
diff --git a/cli/npm/managed/cache/tarball.rs b/cli/npm/managed/cache/tarball.rs
index 46186b87c..eec890bed 100644
--- a/cli/npm/managed/cache/tarball.rs
+++ b/cli/npm/managed/cache/tarball.rs
@@ -11,12 +11,12 @@ use deno_core::error::AnyError;
 use deno_core::futures::future::LocalBoxFuture;
 use deno_core::futures::FutureExt;
 use deno_core::parking_lot::Mutex;
+use deno_core::url::Url;
 use deno_npm::npm_rc::ResolvedNpmRc;
 use deno_npm::registry::NpmPackageVersionDistInfo;
 use deno_runtime::deno_fs::FileSystem;
 use deno_semver::package::PackageNv;
-use reqwest::StatusCode;
-use reqwest::Url;
+use http::StatusCode;
 
 use crate::args::CacheSetting;
 use crate::http_util::DownloadError;
diff --git a/cli/standalone/binary.rs b/cli/standalone/binary.rs
index 10a762093..342c637d5 100644
--- a/cli/standalone/binary.rs
+++ b/cli/standalone/binary.rs
@@ -495,7 +495,7 @@ impl<'a> DenoCompileBinaryWriter<'a> {
       self
         .http_client_provider
         .get_or_create()?
-        .download_with_progress(download_url, None, &progress)
+        .download_with_progress(download_url.parse()?, None, &progress)
         .await?
     };
     let bytes = match maybe_bytes {
diff --git a/cli/tools/registry/api.rs b/cli/tools/registry/api.rs
index ee9579a19..519800660 100644
--- a/cli/tools/registry/api.rs
+++ b/cli/tools/registry/api.rs
@@ -1,8 +1,9 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
+use crate::http_util;
 use deno_core::error::AnyError;
 use deno_core::serde_json;
-use deno_runtime::deno_fetch::reqwest;
+use deno_runtime::deno_fetch;
 use lsp_types::Url;
 use serde::de::DeserializeOwned;
 
@@ -82,7 +83,7 @@ impl std::fmt::Debug for ApiError {
 impl std::error::Error for ApiError {}
 
 pub async fn parse_response<T: DeserializeOwned>(
-  response: reqwest::Response,
+  response: http::Response<deno_fetch::ResBody>,
 ) -> Result<T, ApiError> {
   let status = response.status();
   let x_deno_ray = response
@@ -90,7 +91,7 @@ pub async fn parse_response<T: DeserializeOwned>(
     .get("x-deno-ray")
     .and_then(|value| value.to_str().ok())
     .map(|s| s.to_string());
-  let text = response.text().await.unwrap();
+  let text = http_util::body_to_string(response).await.unwrap();
 
   if !status.is_success() {
     match serde_json::from_str::<ApiError>(&text) {
@@ -122,9 +123,9 @@ pub async fn get_scope(
   client: &HttpClient,
   registry_api_url: &Url,
   scope: &str,
-) -> Result<reqwest::Response, AnyError> {
+) -> Result<http::Response<deno_fetch::ResBody>, AnyError> {
   let scope_url = format!("{}scopes/{}", registry_api_url, scope);
-  let response = client.get(&scope_url).send().await?;
+  let response = client.get(scope_url.parse()?)?.send().await?;
   Ok(response)
 }
 
@@ -141,9 +142,9 @@ pub async fn get_package(
   registry_api_url: &Url,
   scope: &str,
   package: &str,
-) -> Result<reqwest::Response, AnyError> {
+) -> Result<http::Response<deno_fetch::ResBody>, AnyError> {
   let package_url = get_package_api_url(registry_api_url, scope, package);
-  let response = client.get(&package_url).send().await?;
+  let response = client.get(package_url.parse()?)?.send().await?;
   Ok(response)
 }
 
diff --git a/cli/tools/registry/mod.rs b/cli/tools/registry/mod.rs
index 8e4d97897..a22384a52 100644
--- a/cli/tools/registry/mod.rs
+++ b/cli/tools/registry/mod.rs
@@ -23,8 +23,8 @@ use deno_core::futures::StreamExt;
 use deno_core::serde_json;
 use deno_core::serde_json::json;
 use deno_core::serde_json::Value;
-use deno_runtime::deno_fetch::reqwest;
 use deno_terminal::colors;
+use http_body_util::BodyExt;
 use lsp_types::Url;
 use serde::Deserialize;
 use serde::Serialize;
@@ -539,11 +539,13 @@ async fn get_auth_headers(
       let challenge = BASE64_STANDARD.encode(sha2::Sha256::digest(&verifier));
 
       let response = client
-        .post(format!("{}authorizations", registry_url))
-        .json(&serde_json::json!({
-          "challenge": challenge,
-          "permissions": permissions,
-        }))
+        .post_json(
+          format!("{}authorizations", registry_url).parse()?,
+          &serde_json::json!({
+            "challenge": challenge,
+            "permissions": permissions,
+          }),
+        )?
         .send()
         .await
         .context("Failed to create interactive authorization")?;
@@ -573,11 +575,13 @@ async fn get_auth_headers(
       loop {
         tokio::time::sleep(interval).await;
         let response = client
-          .post(format!("{}authorizations/exchange", registry_url))
-          .json(&serde_json::json!({
-            "exchangeToken": auth.exchange_token,
-            "verifier": verifier,
-          }))
+          .post_json(
+            format!("{}authorizations/exchange", registry_url).parse()?,
+            &serde_json::json!({
+              "exchangeToken": auth.exchange_token,
+              "verifier": verifier,
+            }),
+          )?
           .send()
           .await
           .context("Failed to exchange authorization")?;
@@ -634,15 +638,20 @@ async fn get_auth_headers(
         );
 
         let response = client
-          .get(url)
-          .bearer_auth(&oidc_config.token)
+          .get(url.parse()?)?
+          .header(
+            http::header::AUTHORIZATION,
+            format!("Bearer {}", oidc_config.token).parse()?,
+          )
           .send()
           .await
           .context("Failed to get OIDC token")?;
         let status = response.status();
-        let text = response.text().await.with_context(|| {
-          format!("Failed to get OIDC token: status {}", status)
-        })?;
+        let text = crate::http_util::body_to_string(response)
+          .await
+          .with_context(|| {
+            format!("Failed to get OIDC token: status {}", status)
+          })?;
         if !status.is_success() {
           bail!(
             "Failed to get OIDC token: status {}, response: '{}'",
@@ -770,7 +779,7 @@ async fn ensure_scopes_and_packages_exist(
 
     loop {
       tokio::time::sleep(std::time::Duration::from_secs(3)).await;
-      let response = client.get(&package_api_url).send().await?;
+      let response = client.get(package_api_url.parse()?)?.send().await?;
       if response.status() == 200 {
         let name = format!("@{}/{}", package.scope, package.package);
         log::info!("Package {} created", colors::green(name));
@@ -894,11 +903,19 @@ async fn publish_package(
     package.config
   );
 
+  let body = http_body_util::Full::new(package.tarball.bytes.clone())
+    .map_err(|never| match never {})
+    .boxed();
   let response = http_client
-    .post(url)
-    .header(reqwest::header::AUTHORIZATION, authorization)
-    .header(reqwest::header::CONTENT_ENCODING, "gzip")
-    .body(package.tarball.bytes.clone())
+    .post(url.parse()?, body)?
+    .header(
+      http::header::AUTHORIZATION,
+      authorization.parse().map_err(http::Error::from)?,
+    )
+    .header(
+      http::header::CONTENT_ENCODING,
+      "gzip".parse().map_err(http::Error::from)?,
+    )
     .send()
     .await?;
 
@@ -943,7 +960,7 @@ async fn publish_package(
   while task.status != "success" && task.status != "failure" {
     tokio::time::sleep(interval).await;
     let resp = http_client
-      .get(format!("{}publish_status/{}", registry_api_url, task.id))
+      .get(format!("{}publish_status/{}", registry_api_url, task.id).parse()?)?
       .send()
       .await
       .with_context(|| {
@@ -992,7 +1009,8 @@ async fn publish_package(
       package.scope, package.package, package.version
     ))?;
 
-    let meta_bytes = http_client.get(meta_url).send().await?.bytes().await?;
+    let resp = http_client.get(meta_url)?.send().await?;
+    let meta_bytes = resp.collect().await?.to_bytes();
 
     if std::env::var("DISABLE_JSR_MANIFEST_VERIFICATION_FOR_TESTING").is_err() {
       verify_version_manifest(&meta_bytes, &package)?;
@@ -1023,9 +1041,8 @@ async fn publish_package(
       registry_api_url, package.scope, package.package, package.version
     );
     http_client
-      .post(provenance_url)
-      .header(reqwest::header::AUTHORIZATION, authorization)
-      .json(&json!({ "bundle": bundle }))
+      .post_json(provenance_url.parse()?, &json!({ "bundle": bundle }))?
+      .header(http::header::AUTHORIZATION, authorization.parse()?)
       .send()
       .await?;
   }
diff --git a/cli/tools/registry/provenance.rs b/cli/tools/registry/provenance.rs
index 622e483d6..ce3d6ff8a 100644
--- a/cli/tools/registry/provenance.rs
+++ b/cli/tools/registry/provenance.rs
@@ -1,5 +1,6 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
+use crate::http_util;
 use crate::http_util::HttpClient;
 
 use super::api::OidcTokenResponse;
@@ -12,6 +13,8 @@ use deno_core::anyhow;
 use deno_core::anyhow::bail;
 use deno_core::error::AnyError;
 use deno_core::serde_json;
+use deno_core::url::Url;
+use http_body_util::BodyExt;
 use once_cell::sync::Lazy;
 use p256::elliptic_curve;
 use p256::pkcs8::AssociatedOid;
@@ -504,12 +507,12 @@ impl<'a> FulcioSigner<'a> {
 
     let response = self
       .http_client
-      .post(url)
-      .json(&request_body)
+      .post_json(url.parse()?, &request_body)?
       .send()
       .await?;
 
-    let body: SigningCertificateResponse = response.json().await?;
+    let body: SigningCertificateResponse =
+      http_util::body_to_json(response).await?;
 
     let key = body
       .signed_certificate_embedded_sct
@@ -527,15 +530,23 @@ impl<'a> FulcioSigner<'a> {
       bail!("No OIDC token available");
     };
 
-    let res = self
+    let mut url = req_url.parse::<Url>()?;
+    url.query_pairs_mut().append_pair("audience", aud);
+    let res_bytes = self
       .http_client
-      .get(&req_url)
-      .bearer_auth(token)
-      .query(&[("audience", aud)])
+      .get(url)?
+      .header(
+        http::header::AUTHORIZATION,
+        format!("Bearer {}", token)
+          .parse()
+          .map_err(http::Error::from)?,
+      )
       .send()
       .await?
-      .json::<OidcTokenResponse>()
-      .await?;
+      .collect()
+      .await?
+      .to_bytes();
+    let res: OidcTokenResponse = serde_json::from_slice(&res_bytes)?;
     Ok(res.value)
   }
 }
@@ -685,11 +696,10 @@ async fn testify(
 
   let url = format!("{}/api/v1/log/entries", *DEFAULT_REKOR_URL);
   let res = http_client
-    .post(&url)
-    .json(&proposed_intoto_entry)
+    .post_json(url.parse()?, &proposed_intoto_entry)?
     .send()
     .await?;
-  let body: RekorEntry = res.json().await?;
+  let body: RekorEntry = http_util::body_to_json(res).await?;
 
   Ok(body)
 }
diff --git a/cli/tools/test/mod.rs b/cli/tools/test/mod.rs
index 81dc36a89..587b737d6 100644
--- a/cli/tools/test/mod.rs
+++ b/cli/tools/test/mod.rs
@@ -881,12 +881,11 @@ async fn run_tests_for_worker_inner(
     // failing. If we don't do this, a connection to a test server we just tore down might be re-used in
     // the next test.
     // TODO(mmastrac): this should be some sort of callback that we can implement for any subsystem
-    #[allow(clippy::disallowed_types)] // allow using reqwest::Client here
     worker
       .js_runtime
       .op_state()
       .borrow_mut()
-      .try_take::<deno_runtime::deno_fetch::reqwest::Client>();
+      .try_take::<deno_runtime::deno_fetch::Client>();
 
     if desc.ignore {
       send_test_event(
diff --git a/cli/tools/upgrade.rs b/cli/tools/upgrade.rs
index 2afeffc92..fd8394efa 100644
--- a/cli/tools/upgrade.rs
+++ b/cli/tools/upgrade.rs
@@ -571,7 +571,7 @@ async fn get_latest_version(
   check_kind: UpgradeCheckKind,
 ) -> Result<String, AnyError> {
   let url = get_url(release_kind, env!("TARGET"), check_kind);
-  let text = client.download_text(url).await?;
+  let text = client.download_text(url.parse()?).await?;
   Ok(normalize_version_from_server(release_kind, &text))
 }
 
@@ -624,7 +624,7 @@ async fn download_package(
     // text above which will stay alive after the progress bars are complete
     let progress = progress_bar.update("");
     client
-      .download_with_progress(download_url, None, &progress)
+      .download_with_progress(download_url.parse()?, None, &progress)
       .await?
   };
   match maybe_bytes {