fix(ext/node): support createConnection option in node:http.request() (#25470)

This commit changes "node:http" module to add support
for the "createConnection" option when the "request()"
API is called.


Closes https://github.com/denoland/deno/issues/19507

---------

Signed-off-by: Yoshiya Hinosawa <stibium121@gmail.com>
Signed-off-by: Satya Rohith <me@satyarohith.com>
Co-authored-by: Yoshiya Hinosawa <stibium121@gmail.com>
Co-authored-by: Bartek Iwańczuk <biwanczuk@gmail.com>
Co-authored-by: crowlkats <crowlkats@toaxl.com>

ext/fetch/lib.rs

@@ -206,9 +206,6 @@ pub enum FetchError {
   RequestBuilderHook(deno_core::error::AnyError),
   #[error(transparent)]
   Io(#[from] std::io::Error),
-  // Only used for node upgrade
-  #[error(transparent)]
-  Hyper(#[from] hyper::Error),
 }
 
 pub type CancelableResponseFuture =

ext/node/lib.rs

@@ -364,9 +364,9 @@ deno_core::extension!(deno_node,
     ops::zlib::brotli::op_create_brotli_decompress,
     ops::zlib::brotli::op_brotli_decompress_stream,
     ops::zlib::brotli::op_brotli_decompress_stream_end,
-    ops::http::op_node_http_request<P>,
     ops::http::op_node_http_fetch_response_upgrade,
-    ops::http::op_node_http_fetch_send,
+    ops::http::op_node_http_request_with_conn<P>,
+    ops::http::op_node_http_await_response,
     ops::http2::op_http2_connect,
     ops::http2::op_http2_poll_client_connection,
     ops::http2::op_http2_client_request,

ext/node/ops/http.rs

@@ -2,18 +2,20 @@
 
 use std::borrow::Cow;
 use std::cell::RefCell;
+use std::fmt::Debug;
 use std::pin::Pin;
 use std::rc::Rc;
 use std::task::Context;
 use std::task::Poll;
 
 use bytes::Bytes;
+use deno_core::error::bad_resource;
+use deno_core::error::type_error;
 use deno_core::futures::stream::Peekable;
 use deno_core::futures::Future;
 use deno_core::futures::FutureExt;
 use deno_core::futures::Stream;
 use deno_core::futures::StreamExt;
-use deno_core::futures::TryFutureExt;
 use deno_core::op2;
 use deno_core::serde::Serialize;
 use deno_core::unsync::spawn;
@@ -25,17 +27,17 @@ use deno_core::ByteString;
 use deno_core::CancelFuture;
 use deno_core::CancelHandle;
 use deno_core::CancelTryFuture;
+use deno_core::Canceled;
 use deno_core::OpState;
 use deno_core::RcRef;
 use deno_core::Resource;
 use deno_core::ResourceId;
-use deno_fetch::get_or_create_client_from_state;
 use deno_fetch::FetchCancelHandle;
-use deno_fetch::FetchError;
-use deno_fetch::FetchRequestResource;
 use deno_fetch::FetchReturn;
-use deno_fetch::HttpClientResource;
 use deno_fetch::ResBody;
+use deno_net::io::TcpStreamResource;
+use deno_net::ops_tls::TlsStreamResource;
+use deno_permissions::PermissionCheckError;
 use http::header::HeaderMap;
 use http::header::HeaderName;
 use http::header::HeaderValue;
@@ -44,41 +46,140 @@ use http::header::CONTENT_LENGTH;
 use http::Method;
 use http_body_util::BodyExt;
 use hyper::body::Frame;
+use hyper::body::Incoming;
 use hyper_util::rt::TokioIo;
 use std::cmp::min;
 use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
 
-#[op2(stack_trace)]
+#[derive(Default, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct NodeHttpResponse {
+  pub status: u16,
+  pub status_text: String,
+  pub headers: Vec<(ByteString, ByteString)>,
+  pub url: String,
+  pub response_rid: ResourceId,
+  pub content_length: Option<u64>,
+  pub remote_addr_ip: Option<String>,
+  pub remote_addr_port: Option<u16>,
+  pub error: Option<String>,
+}
+
+type CancelableResponseResult =
+  Result<Result<http::Response<Incoming>, hyper::Error>, Canceled>;
+
+pub struct NodeHttpClientResponse {
+  response: Pin<Box<dyn Future<Output = CancelableResponseResult>>>,
+  url: String,
+}
+
+impl Debug for NodeHttpClientResponse {
+  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    f.debug_struct("NodeHttpClientResponse")
+      .field("url", &self.url)
+      .finish()
+  }
+}
+
+impl deno_core::Resource for NodeHttpClientResponse {
+  fn name(&self) -> Cow<str> {
+    "nodeHttpClientResponse".into()
+  }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum ConnError {
+  #[error(transparent)]
+  Resource(deno_core::error::AnyError),
+  #[error(transparent)]
+  Permission(#[from] PermissionCheckError),
+  #[error("Invalid URL {0}")]
+  InvalidUrl(Url),
+  #[error(transparent)]
+  InvalidHeaderName(#[from] http::header::InvalidHeaderName),
+  #[error(transparent)]
+  InvalidHeaderValue(#[from] http::header::InvalidHeaderValue),
+  #[error(transparent)]
+  Url(#[from] url::ParseError),
+  #[error(transparent)]
+  Method(#[from] http::method::InvalidMethod),
+  #[error(transparent)]
+  Io(#[from] std::io::Error),
+  #[error("TLS stream is currently in use")]
+  TlsStreamBusy,
+  #[error("TCP stream is currently in use")]
+  TcpStreamBusy,
+  #[error(transparent)]
+  ReuniteTcp(#[from] tokio::net::tcp::ReuniteError),
+  #[error(transparent)]
+  Canceled(#[from] deno_core::Canceled),
+  #[error(transparent)]
+  Hyper(#[from] hyper::Error),
+}
+
+#[op2(async, stack_trace)]
 #[serde]
-pub fn op_node_http_request<P>(
+pub async fn op_node_http_request_with_conn<P>(
-  state: &mut OpState,
+  state: Rc<RefCell<OpState>>,
   #[serde] method: ByteString,
   #[string] url: String,
   #[serde] headers: Vec<(ByteString, ByteString)>,
-  #[smi] client_rid: Option<u32>,
   #[smi] body: Option<ResourceId>,
-) -> Result<FetchReturn, FetchError>
+  #[smi] conn_rid: ResourceId,
+  encrypted: bool,
+) -> Result<FetchReturn, ConnError>
 where
   P: crate::NodePermissions + 'static,
 {
-  let client = if let Some(rid) = client_rid {
+  let (_handle, mut sender) = if encrypted {
-    let r = state
+    let resource_rc = state
+      .borrow_mut()
       .resource_table
-      .get::<HttpClientResource>(rid)
+      .take::<TlsStreamResource>(conn_rid)
-      .map_err(FetchError::Resource)?;
+      .map_err(ConnError::Resource)?;
-    r.client.clone()
+    let resource =
+      Rc::try_unwrap(resource_rc).map_err(|_e| ConnError::TlsStreamBusy)?;
+    let (read_half, write_half) = resource.into_inner();
+    let tcp_stream = read_half.unsplit(write_half);
+    let io = TokioIo::new(tcp_stream);
+    let (sender, conn) = hyper::client::conn::http1::handshake(io).await?;
+    (
+      tokio::task::spawn(async move { conn.with_upgrades().await }),
+      sender,
+    )
   } else {
-    get_or_create_client_from_state(state)?
+    let resource_rc = state
+      .borrow_mut()
+      .resource_table
+      .take::<TcpStreamResource>(conn_rid)
+      .map_err(ConnError::Resource)?;
+    let resource =
+      Rc::try_unwrap(resource_rc).map_err(|_| ConnError::TcpStreamBusy)?;
+    let (read_half, write_half) = resource.into_inner();
+    let tcp_stream = read_half.reunite(write_half)?;
+    let io = TokioIo::new(tcp_stream);
+    let (sender, conn) = hyper::client::conn::http1::handshake(io).await?;
+
+    // Spawn a task to poll the connection, driving the HTTP state
+    (
+      tokio::task::spawn(async move {
+        conn.with_upgrades().await?;
+        Ok::<_, _>(())
+      }),
+      sender,
+    )
   };
 
+  // Create the request.
   let method = Method::from_bytes(&method)?;
-  let mut url = Url::parse(&url)?;
+  let mut url_parsed = Url::parse(&url)?;
-  let maybe_authority = deno_fetch::extract_authority(&mut url);
+  let maybe_authority = deno_fetch::extract_authority(&mut url_parsed);
 
   {
-    let permissions = state.borrow_mut::<P>();
+    let mut state_ = state.borrow_mut();
-    permissions.check_net_url(&url, "ClientRequest")?;
+    let permissions = state_.borrow_mut::<P>();
+    permissions.check_net_url(&url_parsed, "ClientRequest")?;
   }
 
   let mut header_map = HeaderMap::new();
@@ -93,9 +194,10 @@ where
     (
       BodyExt::boxed(NodeHttpResourceToBodyAdapter::new(
         state
+          .borrow_mut()
           .resource_table
           .take_any(body)
-          .map_err(FetchError::Resource)?,
+          .map_err(ConnError::Resource)?,
       )),
       None,
     )
@@ -117,10 +219,13 @@ where
 
   let mut request = http::Request::new(body);
   *request.method_mut() = method.clone();
-  *request.uri_mut() = url
+  let path = url_parsed.path();
-    .as_str()
+  let query = url_parsed.query();
+  *request.uri_mut() = query
+    .map(|q| format!("{}?{}", path, q))
+    .unwrap_or_else(|| path.to_string())
     .parse()
-    .map_err(|_| FetchError::InvalidUrl(url.clone()))?;
+    .map_err(|_| ConnError::InvalidUrl(url_parsed.clone()))?;
   *request.headers_mut() = header_map;
 
   if let Some((username, password)) = maybe_authority {
@@ -136,86 +241,44 @@ where
   let cancel_handle = CancelHandle::new_rc();
   let cancel_handle_ = cancel_handle.clone();
 
-  let fut = async move {
+  let fut =
-    client
+    async move { sender.send_request(request).or_cancel(cancel_handle_).await };
-      .send(request)
-      .map_err(Into::into)
-      .or_cancel(cancel_handle_)
-      .await
-  };
 
-  let request_rid = state.resource_table.add(FetchRequestResource {
+  let rid = state
-    future: Box::pin(fut),
+    .borrow_mut()
-    url,
+    .resource_table
-  });
+    .add(NodeHttpClientResponse {
+      response: Box::pin(fut),
+      url: url.clone(),
+    });
 
-  let cancel_handle_rid =
+  let cancel_handle_rid = state
-    state.resource_table.add(FetchCancelHandle(cancel_handle));
+    .borrow_mut()
+    .resource_table
+    .add(FetchCancelHandle(cancel_handle));
 
   Ok(FetchReturn {
-    request_rid,
+    request_rid: rid,
     cancel_handle_rid: Some(cancel_handle_rid),
   })
 }
 
-#[derive(Default, Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct NodeHttpFetchResponse {
-  pub status: u16,
-  pub status_text: String,
-  pub headers: Vec<(ByteString, ByteString)>,
-  pub url: String,
-  pub response_rid: ResourceId,
-  pub content_length: Option<u64>,
-  pub remote_addr_ip: Option<String>,
-  pub remote_addr_port: Option<u16>,
-  pub error: Option<String>,
-}
-
 #[op2(async)]
 #[serde]
-pub async fn op_node_http_fetch_send(
+pub async fn op_node_http_await_response(
   state: Rc<RefCell<OpState>>,
   #[smi] rid: ResourceId,
-) -> Result<NodeHttpFetchResponse, FetchError> {
+) -> Result<NodeHttpResponse, ConnError> {
-  let request = state
+  let resource = state
     .borrow_mut()
     .resource_table
-    .take::<FetchRequestResource>(rid)
+    .take::<NodeHttpClientResponse>(rid)
-    .map_err(FetchError::Resource)?;
+    .map_err(ConnError::Resource)?;
-
+  let resource = Rc::try_unwrap(resource)
-  let request = Rc::try_unwrap(request)
+    .map_err(|_| ConnError::Resource(bad_resource("NodeHttpClientResponse")))?;
-    .ok()
-    .expect("multiple op_node_http_fetch_send ongoing");
-
-  let res = match request.future.await {
-    Ok(Ok(res)) => res,
-    Ok(Err(err)) => {
-      // We're going to try and rescue the error cause from a stream and return it from this fetch.
-      // If any error in the chain is a hyper body error, return that as a special result we can use to
-      // reconstruct an error chain (eg: `new TypeError(..., { cause: new Error(...) })`).
-      // TODO(mmastrac): it would be a lot easier if we just passed a v8::Global through here instead
-
-      if let FetchError::ClientSend(err_src) = &err {
-        if let Some(client_err) = std::error::Error::source(&err_src.source) {
-          if let Some(err_src) = client_err.downcast_ref::<hyper::Error>() {
-            if let Some(err_src) = std::error::Error::source(err_src) {
-              return Ok(NodeHttpFetchResponse {
-                error: Some(err_src.to_string()),
-                ..Default::default()
-              });
-            }
-          }
-        }
-      }
-
-      return Err(err);
-    }
-    Err(_) => return Err(FetchError::RequestCanceled),
-  };
 
+  let res = resource.response.await??;
   let status = res.status();
-  let url = request.url.into();
   let mut res_headers = Vec::new();
   for (key, val) in res.headers().iter() {
     res_headers.push((key.as_str().into(), val.as_bytes().into()));
@@ -232,16 +295,22 @@ pub async fn op_node_http_fetch_send(
     (None, None)
   };
 
+  let (parts, body) = res.into_parts();
+  let body = body.map_err(deno_core::anyhow::Error::from);
+  let body = body.boxed();
+
+  let res = http::Response::from_parts(parts, body);
+
   let response_rid = state
     .borrow_mut()
     .resource_table
-    .add(NodeHttpFetchResponseResource::new(res, content_length));
+    .add(NodeHttpResponseResource::new(res, content_length));
 
-  Ok(NodeHttpFetchResponse {
+  Ok(NodeHttpResponse {
     status: status.as_u16(),
     status_text: status.canonical_reason().unwrap_or("").to_string(),
     headers: res_headers,
-    url,
+    url: resource.url,
     response_rid,
     content_length,
     remote_addr_ip,
@@ -255,12 +324,12 @@ pub async fn op_node_http_fetch_send(
 pub async fn op_node_http_fetch_response_upgrade(
   state: Rc<RefCell<OpState>>,
   #[smi] rid: ResourceId,
-) -> Result<ResourceId, FetchError> {
+) -> Result<ResourceId, ConnError> {
   let raw_response = state
     .borrow_mut()
     .resource_table
-    .take::<NodeHttpFetchResponseResource>(rid)
+    .take::<NodeHttpResponseResource>(rid)
-    .map_err(FetchError::Resource)?;
+    .map_err(ConnError::Resource)?;
   let raw_response = Rc::try_unwrap(raw_response)
     .expect("Someone is holding onto NodeHttpFetchResponseResource");
 
@@ -283,7 +352,7 @@ pub async fn op_node_http_fetch_response_upgrade(
         }
         read_tx.write_all(&buf[..read]).await?;
       }
-      Ok::<_, FetchError>(())
+      Ok::<_, ConnError>(())
     });
     spawn(async move {
       let mut buf = [0; 1024];
@@ -294,7 +363,7 @@ pub async fn op_node_http_fetch_response_upgrade(
         }
         upgraded_tx.write_all(&buf[..read]).await?;
       }
-      Ok::<_, FetchError>(())
+      Ok::<_, ConnError>(())
     });
   }
 
@@ -379,13 +448,13 @@ impl Default for NodeHttpFetchResponseReader {
 }
 
 #[derive(Debug)]
-pub struct NodeHttpFetchResponseResource {
+pub struct NodeHttpResponseResource {
   pub response_reader: AsyncRefCell<NodeHttpFetchResponseReader>,
   pub cancel: CancelHandle,
   pub size: Option<u64>,
 }
 
-impl NodeHttpFetchResponseResource {
+impl NodeHttpResponseResource {
   pub fn new(response: http::Response<ResBody>, size: Option<u64>) -> Self {
     Self {
       response_reader: AsyncRefCell::new(NodeHttpFetchResponseReader::Start(
@@ -400,14 +469,14 @@ impl NodeHttpFetchResponseResource {
     let reader = self.response_reader.into_inner();
     match reader {
       NodeHttpFetchResponseReader::Start(resp) => {
-        Ok(hyper::upgrade::on(resp).await?)
+        hyper::upgrade::on(resp).await
       }
       _ => unreachable!(),
     }
   }
 }
 
-impl Resource for NodeHttpFetchResponseResource {
+impl Resource for NodeHttpResponseResource {
   fn name(&self) -> Cow<str> {
     "fetchResponse".into()
   }
@@ -454,9 +523,7 @@ impl Resource for NodeHttpFetchResponseResource {
             // safely call `await` on it without creating a race condition.
             Some(_) => match reader.as_mut().next().await.unwrap() {
               Ok(chunk) => assert!(chunk.is_empty()),
-              Err(err) => {
+              Err(err) => break Err(type_error(err.to_string())),
-                break Err(deno_core::error::type_error(err.to_string()))
-              }
             },
             None => break Ok(BufView::empty()),
           }
@@ -464,7 +531,7 @@ impl Resource for NodeHttpFetchResponseResource {
       };
 
       let cancel_handle = RcRef::map(self, |r| &r.cancel);
-      fut.try_or_cancel(cancel_handle).await.map_err(Into::into)
+      fut.try_or_cancel(cancel_handle).await
     })
   }
 
@@ -514,8 +581,9 @@ impl Stream for NodeHttpResourceToBodyAdapter {
         Poll::Ready(res) => match res {
           Ok(buf) if buf.is_empty() => Poll::Ready(None),
           Ok(buf) => {
+            let bytes: Bytes = buf.to_vec().into();
             this.1 = Some(this.0.clone().read(64 * 1024));
-            Poll::Ready(Some(Ok(buf.to_vec().into())))
+            Poll::Ready(Some(Ok(bytes)))
           }
           Err(err) => Poll::Ready(Some(Err(err))),
         },

ext/node/polyfills/_http_outgoing.ts

@@ -491,19 +491,53 @@ Object.defineProperties(
       return ret;
     },
 
+    /** Right after socket is ready, we need to writeHeader() to setup the request and
+     *  client. This is invoked by onSocket(). */
+    _flushHeaders() {
+      if (!this._headerSent) {
+        this._headerSent = true;
+        this._writeHeader();
+      }
+    },
+
     // deno-lint-ignore no-explicit-any
     _send(data: any, encoding?: string | null, callback?: () => void) {
-      if (!this._headerSent && this._header !== null) {
+      // if socket is ready, write the data after headers are written.
-        this._writeHeader();
+      // if socket is not ready, buffer data in outputbuffer.
-        this._headerSent = true;
+      if (
+        this.socket && !this.socket.connecting && this.outputData.length === 0
+      ) {
+        if (!this._headerSent) {
+          this._writeHeader();
+          this._headerSent = true;
+        }
+
+        return this._writeRaw(data, encoding, callback);
+      } else {
+        this.outputData.push({ data, encoding, callback });
       }
-      return this._writeRaw(data, encoding, callback);
+      return false;
     },
 
     _writeHeader() {
       throw new ERR_METHOD_NOT_IMPLEMENTED("_writeHeader()");
     },
 
+    _flushBuffer() {
+      const outputLength = this.outputData.length;
+      if (outputLength <= 0 || !this.socket || !this._bodyWriter) {
+        return undefined;
+      }
+
+      const { data, encoding, callback } = this.outputData.shift();
+      const ret = this._writeRaw(data, encoding, callback);
+      if (this.outputData.length > 0) {
+        this.once("drain", this._flushBuffer);
+      }
+
+      return ret;
+    },
+
     _writeRaw(
       // deno-lint-ignore no-explicit-any
       data: any,
@@ -517,11 +551,15 @@ Object.defineProperties(
         data = new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
       }
       if (data.buffer.byteLength > 0) {
-        this._bodyWriter.write(data).then(() => {
+        this._bodyWriter.ready.then(() => {
-          callback?.();
+          if (this._bodyWriter.desiredSize > 0) {
-          this.emit("drain");
+            this._bodyWriter.write(data).then(() => {
-        }).catch((e) => {
+              callback?.();
-          this._requestSendError = e;
+              this.emit("drain");
+            }).catch((e) => {
+              this._requestSendError = e;
+            });
+          }
         });
       }
       return false;
@@ -658,7 +696,6 @@ Object.defineProperties(
 
       const { header } = state;
       this._header = header + "\r\n";
-      this._headerSent = false;
 
       // Wait until the first body chunk, or close(), is sent to flush,
       // UNLESS we're sending Expect: 100-continue.

ext/node/polyfills/_tls_wrap.ts

@@ -154,6 +154,13 @@ export class TLSSocket extends net.Socket {
       const afterConnect = handle.afterConnect;
       handle.afterConnect = async (req: any, status: number) => {
         options.hostname ??= undefined; // coerce to undefined if null, startTls expects hostname to be undefined
+        if (tlssock._isNpmAgent) {
+          // skips the TLS handshake for @npmcli/agent as it's handled by
+          // onSocket handler of ClientRequest object.
+          tlssock.emit("secure");
+          tlssock.removeListener("end", onConnectEnd);
+          return afterConnect.call(handle, req, status);
+        }
 
         try {
           const conn = await Deno.startTls(handle[kStreamBaseField], options);

ext/node/polyfills/http.ts

@@ -5,16 +5,17 @@
 
 import { core, primordials } from "ext:core/mod.js";
 import {
+  op_node_http_await_response,
   op_node_http_fetch_response_upgrade,
-  op_node_http_fetch_send,
+  op_node_http_request_with_conn,
-  op_node_http_request,
+  op_tls_start,
 } from "ext:core/ops";
 
 import { TextEncoder } from "ext:deno_web/08_text_encoding.js";
 import { setTimeout } from "ext:deno_web/02_timers.js";
 import {
   _normalizeArgs,
-  // createConnection,
+  createConnection,
   ListenOptions,
   Socket,
 } from "node:net";
@@ -48,9 +49,10 @@ import { kOutHeaders } from "ext:deno_node/internal/http.ts";
 import { _checkIsHttpToken as checkIsHttpToken } from "node:_http_common";
 import { Agent, globalAgent } from "node:_http_agent";
 import { urlToHttpOptions } from "ext:deno_node/internal/url.ts";
-import { kEmptyObject } from "ext:deno_node/internal/util.mjs";
+import { kEmptyObject, once } from "ext:deno_node/internal/util.mjs";
 import { constants, TCP } from "ext:deno_node/internal_binding/tcp_wrap.ts";
-import { notImplemented, warnNotImplemented } from "ext:deno_node/_utils.ts";
+import { kStreamBaseField } from "ext:deno_node/internal_binding/stream_wrap.ts";
+import { notImplemented } from "ext:deno_node/_utils.ts";
 import {
   connResetException,
   ERR_HTTP_HEADERS_SENT,
@@ -62,7 +64,6 @@ import {
 } from "ext:deno_node/internal/errors.ts";
 import { getTimerDuration } from "ext:deno_node/internal/timers.mjs";
 import { serve, upgradeHttpRaw } from "ext:deno_http/00_serve.ts";
-import { createHttpClient } from "ext:deno_fetch/22_http_client.js";
 import { headersEntries } from "ext:deno_fetch/20_headers.js";
 import { timerId } from "ext:deno_web/03_abort_signal.js";
 import { clearTimeout as webClearTimeout } from "ext:deno_web/02_timers.js";
@@ -148,6 +149,10 @@ class FakeSocket extends EventEmitter {
   }
 }
 
+function emitErrorEvent(request, error) {
+  request.emit("error", error);
+}
+
 /** ClientRequest represents the http(s) request from the client */
 class ClientRequest extends OutgoingMessage {
   defaultProtocol = "http:";
@@ -160,6 +165,8 @@ class ClientRequest extends OutgoingMessage {
   useChunkedEncodingByDefault: boolean;
   path: string;
   _req: { requestRid: number; cancelHandleRid: number | null } | undefined;
+  _encrypted = false;
+  socket: Socket;
 
   constructor(
     input: string | URL,
@@ -382,17 +389,11 @@ class ClientRequest extends OutgoingMessage {
       delete optsWithoutSignal.signal;
     }
 
-    if (options!.createConnection) {
-      warnNotImplemented("ClientRequest.options.createConnection");
-    }
-
     if (options!.lookup) {
       notImplemented("ClientRequest.options.lookup");
     }
 
-    // initiate connection
+    if (this.agent) {
-    // TODO(crowlKats): finish this
-    /*if (this.agent) {
       this.agent.addRequest(this, optsWithoutSignal);
     } else {
       // No agent, default to Connection:close.
@@ -422,8 +423,7 @@ class ClientRequest extends OutgoingMessage {
         debug("CLIENT use net.createConnection", optsWithoutSignal);
         this.onSocket(createConnection(optsWithoutSignal));
       }
-    }*/
+    }
-    this.onSocket(new FakeSocket({ encrypted: this._encrypted }));
   }
 
   _writeHeader() {
@@ -437,9 +437,6 @@ class ClientRequest extends OutgoingMessage {
       }
     }
 
-    const client = this._getClient() ?? createHttpClient({ http2: false });
-    this._client = client;
-
     if (
       this.method === "POST" || this.method === "PATCH" || this.method === "PUT"
     ) {
@@ -455,17 +452,29 @@ class ClientRequest extends OutgoingMessage {
       this._bodyWriteRid = resourceForReadableStream(readable);
     }
 
-    this._req = op_node_http_request(
-      this.method,
-      url,
-      headers,
-      client[internalRidSymbol],
-      this._bodyWriteRid,
-    );
-
     (async () => {
       try {
-        const res = await op_node_http_fetch_send(this._req.requestRid);
+        const parsedUrl = new URL(url);
+        let baseConnRid =
+          this.socket._handle[kStreamBaseField][internalRidSymbol];
+        if (this._encrypted) {
+          [baseConnRid] = op_tls_start({
+            rid: baseConnRid,
+            hostname: parsedUrl.hostname,
+            caCerts: [],
+            alpnProtocols: ["http/1.0", "http/1.1"],
+          });
+        }
+        this._req = await op_node_http_request_with_conn(
+          this.method,
+          url,
+          headers,
+          this._bodyWriteRid,
+          baseConnRid,
+          this._encrypted,
+        );
+        this._flushBuffer();
+        const res = await op_node_http_await_response(this._req!.requestRid);
         if (this._req.cancelHandleRid !== null) {
           core.tryClose(this._req.cancelHandleRid);
         }
@@ -473,7 +482,6 @@ class ClientRequest extends OutgoingMessage {
           this._timeout.removeEventListener("abort", this._timeoutCb);
           webClearTimeout(this._timeout[timerId]);
         }
-        this._client.close();
         const incoming = new IncomingMessageForClient(this.socket);
         incoming.req = this;
         this.res = incoming;
@@ -512,12 +520,9 @@ class ClientRequest extends OutgoingMessage {
           if (this.method === "CONNECT") {
             throw new Error("not implemented CONNECT");
           }
-
           const upgradeRid = await op_node_http_fetch_response_upgrade(
             res.responseRid,
           );
-          assert(typeof res.remoteAddrIp !== "undefined");
-          assert(typeof res.remoteAddrIp !== "undefined");
           const conn = new UpgradedConn(
             upgradeRid,
             {
@@ -543,13 +548,11 @@ class ClientRequest extends OutgoingMessage {
           this._closed = true;
           this.emit("close");
         } else {
-          {
+          incoming._bodyRid = res.responseRid;
-            incoming._bodyRid = res.responseRid;
-          }
           this.emit("response", incoming);
         }
       } catch (err) {
-        if (this._req.cancelHandleRid !== null) {
+        if (this._req && this._req.cancelHandleRid !== null) {
           core.tryClose(this._req.cancelHandleRid);
         }
 
@@ -592,11 +595,54 @@ class ClientRequest extends OutgoingMessage {
     return undefined;
   }
 
-  // TODO(bartlomieju): handle error
+  onSocket(socket, err) {
-  onSocket(socket, _err) {
     nextTick(() => {
-      this.socket = socket;
+      // deno-lint-ignore no-this-alias
-      this.emit("socket", socket);
+      const req = this;
+      if (req.destroyed || err) {
+        req.destroyed = true;
+
+        // deno-lint-ignore no-inner-declarations
+        function _destroy(req, err) {
+          if (!req.aborted && !err) {
+            err = new connResetException("socket hang up");
+          }
+          if (err) {
+            emitErrorEvent(req, err);
+          }
+          req._closed = true;
+          req.emit("close");
+        }
+
+        if (socket) {
+          if (!err && req.agent && !socket.destroyed) {
+            socket.emit("free");
+          } else {
+            finished(socket.destroy(err || req[kError]), (er) => {
+              if (er?.code === "ERR_STREAM_PREMATURE_CLOSE") {
+                er = null;
+              }
+              _destroy(req, er || err);
+            });
+            return;
+          }
+        }
+
+        _destroy(req, err || req[kError]);
+      } else {
+        // Note: this code is specific to deno to initiate a request.
+        const onConnect = () => {
+          // Flush the internal buffers once socket is ready.
+          this._flushHeaders();
+        };
+        this.socket = socket;
+        this.emit("socket", socket);
+        if (socket.readyState === "opening") {
+          socket.on("connect", onConnect);
+        } else {
+          onConnect();
+        }
+      }
     });
   }
 
@@ -618,14 +664,20 @@ class ClientRequest extends OutgoingMessage {
     if (chunk) {
       this.write_(chunk, encoding, null, true);
     } else if (!this._headerSent) {
-      this._contentLength = 0;
+      if (
-      this._implicitHeader();
+        (this.socket && !this.socket.connecting) || // socket is not connecting, or
-      this._send("", "latin1");
+        (!this.socket && this.outputData.length === 0) // no data to send
+      ) {
+        this._contentLength = 0;
+        this._implicitHeader();
+        this._send("", "latin1");
+      }
     }
-    (async () => {
+    const finish = async () => {
       try {
+        await this._bodyWriter.ready;
         await this._bodyWriter?.close();
-      } catch (_) {
+      } catch {
         // The readable stream resource is dropped right after
         // read is complete closing the writable stream resource.
         // If we try to close the writer again, it will result in an
@@ -633,10 +685,20 @@ class ClientRequest extends OutgoingMessage {
       }
       try {
         cb?.();
-      } catch (_) {
+      } catch {
         //
       }
-    })();
+    };
+
+    if (this.socket && this._bodyWriter) {
+      finish();
+    } else {
+      this.on("drain", () => {
+        if (this.outputData.length === 0) {
+          finish();
+        }
+      });
+    }
 
     return this;
   }
@@ -658,11 +720,6 @@ class ClientRequest extends OutgoingMessage {
     }
     this.destroyed = true;
 
-    const rid = this._client?.[internalRidSymbol];
-    if (rid) {
-      core.tryClose(rid);
-    }
-
     // Request might be closed before we actually made it
     if (this._req !== undefined && this._req.cancelHandleRid !== null) {
       core.tryClose(this._req.cancelHandleRid);

ext/node/polyfills/https.ts

@@ -112,7 +112,7 @@ export const globalAgent = new Agent({
 
 /** HttpsClientRequest class loosely follows http.ClientRequest class API. */
 class HttpsClientRequest extends ClientRequest {
-  override _encrypted: true;
+  override _encrypted = true;
   override defaultProtocol = "https:";
   override _getClient(): Deno.HttpClient | undefined {
     if (caCerts === null) {

ext/node/polyfills/internal_binding/cares_wrap.ts

@@ -36,7 +36,6 @@ import {
 } from "ext:deno_node/internal_binding/async_wrap.ts";
 import { ares_strerror } from "ext:deno_node/internal_binding/ares.ts";
 import { notImplemented } from "ext:deno_node/_utils.ts";
-import { isWindows } from "ext:deno_node/_util/os.ts";
 
 interface LookupAddress {
   address: string;
@@ -68,7 +67,7 @@ export function getaddrinfo(
   _hints: number,
   verbatim: boolean,
 ): number {
-  let addresses: string[] = [];
+  const addresses: string[] = [];
 
   // TODO(cmorten): use hints
   // REF: https://nodejs.org/api/dns.html#dns_supported_getaddrinfo_flags
@@ -107,13 +106,6 @@ export function getaddrinfo(
       });
     }
 
-    // TODO(@bartlomieju): Forces IPv4 as a workaround for Deno not
-    // aligning with Node on implicit binding on Windows
-    // REF: https://github.com/denoland/deno/issues/10762
-    if (isWindows && hostname === "localhost") {
-      addresses = addresses.filter((address) => isIPv4(address));
-    }
-
     req.oncomplete(error, addresses);
   })();
 

ext/node/polyfills/net.ts

@@ -986,16 +986,20 @@ function _lookupAndConnect(
         } else {
           self._unrefTimer();
 
-          defaultTriggerAsyncIdScope(
+          defaultTriggerAsyncIdScope(self[asyncIdSymbol], nextTick, () => {
-            self[asyncIdSymbol],
+            if (self.connecting) {
-            _internalConnect,
+              defaultTriggerAsyncIdScope(
-            self,
+                self[asyncIdSymbol],
-            ip,
+                _internalConnect,
-            port,
+                self,
-            addressType,
+                ip,
-            localAddress,
+                port,
-            localPort,
+                addressType,
-          );
+                localAddress,
+                localPort,
+              );
+            }
+          });
         }
       },
     );
@@ -1197,6 +1201,9 @@ export class Socket extends Duplex {
   _host: string | null = null;
   // deno-lint-ignore no-explicit-any
   _parent: any = null;
+  // The flag for detecting if it's called in @npmcli/agent
+  // See discussions in https://github.com/denoland/deno/pull/25470 for more details.
+  _isNpmAgent = false;
   autoSelectFamilyAttemptedAddresses: AddressInfo[] | undefined = undefined;
 
   constructor(options: SocketOptions | number) {
@@ -1217,6 +1224,19 @@ export class Socket extends Duplex {
 
     super(options);
 
+    // Note: If the socket is created from @npmcli/agent, the 'socket' event
+    // on ClientRequest object happens after 'connect' event on Socket object.
+    // That swaps the sequence of op_node_http_request_with_conn() call and
+    // initial socket read. That causes op_node_http_request_with_conn() not
+    // working.
+    // To avoid the above situation, we detect the socket created from
+    // @npmcli/agent and pause the socket (and also skips the startTls call
+    // if it's TLSSocket)
+    this._isNpmAgent = new Error().stack?.includes("@npmcli/agent") || false;
+    if (this._isNpmAgent) {
+      this.pause();
+    }
+
     if (options.handle) {
       this._handle = options.handle;
       this[asyncIdSymbol] = _getNewAsyncId(this._handle);

runtime/errors.rs

@@ -712,7 +712,6 @@ fn get_fetch_error(error: &FetchError) -> &'static str {
     FetchError::ClientSend(_) => "TypeError",
     FetchError::RequestBuilderHook(_) => "TypeError",
     FetchError::Io(e) => get_io_error_class(e),
-    FetchError::Hyper(e) => get_hyper_error_class(e),
   }
 }
 
@@ -1083,6 +1082,7 @@ mod node {
   pub use deno_node::ops::crypto::SignEd25519Error;
   pub use deno_node::ops::crypto::VerifyEd25519Error;
   pub use deno_node::ops::fs::FsError;
+  pub use deno_node::ops::http::ConnError;
   pub use deno_node::ops::http2::Http2Error;
   pub use deno_node::ops::idna::IdnaError;
   pub use deno_node::ops::ipc::IpcError;
@@ -1538,6 +1538,24 @@ mod node {
   pub fn get_verify_ed25519_error(_: &VerifyEd25519Error) -> &'static str {
     "TypeError"
   }
+
+  pub fn get_conn_error(e: &ConnError) -> &'static str {
+    match e {
+      ConnError::Resource(e) => get_error_class_name(e).unwrap_or("Error"),
+      ConnError::Permission(e) => get_permission_check_error_class(e),
+      ConnError::InvalidUrl(_) => "TypeError",
+      ConnError::InvalidHeaderName(_) => "TypeError",
+      ConnError::InvalidHeaderValue(_) => "TypeError",
+      ConnError::Url(e) => get_url_parse_error_class(e),
+      ConnError::Method(_) => "TypeError",
+      ConnError::Io(e) => get_io_error_class(e),
+      ConnError::Hyper(e) => super::get_hyper_error_class(e),
+      ConnError::TlsStreamBusy => "Busy",
+      ConnError::TcpStreamBusy => "Busy",
+      ConnError::ReuniteTcp(_) => "Error",
+      ConnError::Canceled(_) => "Error",
+    }
+  }
 }
 
 fn get_os_error(error: &OsError) -> &'static str {
@@ -1730,6 +1748,10 @@ pub fn get_error_class_name(e: &AnyError) -> Option<&'static str> {
       e.downcast_ref::<node::VerifyEd25519Error>()
         .map(node::get_verify_ed25519_error)
     })
+    .or_else(|| {
+      e.downcast_ref::<node::ConnError>()
+        .map(node::get_conn_error)
+    })
     .or_else(|| e.downcast_ref::<NApiError>().map(get_napi_error_class))
     .or_else(|| e.downcast_ref::<WebError>().map(get_web_error_class))
     .or_else(|| {

tests/node_compat/config.jsonc

@@ -565,9 +565,7 @@
       "test-handle-wrap-close-abort.js",
       "test-http-abort-before-end.js",
       "test-http-addrequest-localaddress.js",
-      "test-http-agent-false.js",
       "test-http-agent-getname.js",
-      "test-http-agent-keepalive-delay.js",
       "test-http-agent-maxtotalsockets.js",
       "test-http-agent-no-protocol.js",
       "test-http-agent-null.js",
@@ -590,7 +588,6 @@
       "test-http-client-race.js",
       "test-http-client-read-in-error.js",
       "test-http-client-reject-unexpected-agent.js",
-      "test-http-client-timeout-connect-listener.js",
       "test-http-client-timeout-with-data.js",
       "test-http-client-unescaped-path.js",
       "test-http-client-upload-buf.js",
@@ -604,7 +601,6 @@
       "test-http-date-header.js",
       "test-http-decoded-auth.js",
       "test-http-default-encoding.js",
-      "test-http-dump-req-when-res-ends.js",
       "test-http-end-throw-socket-handling.js",
       "test-http-eof-on-connect.js",
       "test-http-extra-response.js",
@@ -622,7 +618,6 @@
       "test-http-hex-write.js",
       "test-http-highwatermark.js",
       "test-http-host-headers.js",
-      "test-http-hostname-typechecking.js",
       "test-http-incoming-message-destroy.js",
       "test-http-invalid-path-chars.js",
       "test-http-invalidheaderfield.js",
@@ -1292,10 +1287,7 @@
       "test-buffer-creation-regression.js",
       "test-child-process-exit.js",
       "test-http-server-keep-alive-timeout-slow-server.js",
-      "test-net-better-error-messages-port.js",
-      "test-net-connect-handle-econnrefused.js",
       "test-net-connect-local-error.js",
-      "test-net-reconnect-error.js",
       "test-net-response-size.js",
       "test-net-server-bind.js",
       "test-tls-lookup.js",

tests/node_compat/runner/TODO.md

@@ -1,7 +1,7 @@
 <!-- deno-fmt-ignore-file -->
 # Remaining Node Tests
 
-1163 tests out of 3681 have been ported from Node 20.11.1 (31.59% ported, 68.92% remaining).
+1155 tests out of 3681 have been ported from Node 20.11.1 (31.38% ported, 69.14% remaining).
 
 NOTE: This file should not be manually edited. Please edit `tests/node_compat/config.json` and run `deno task setup` in `tests/node_compat/runner` dir instead.
 
@@ -792,6 +792,8 @@ NOTE: This file should not be manually edited. Please edit `tests/node_compat/co
 - [parallel/test-http-agent-destroyed-socket.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-agent-destroyed-socket.js)
 - [parallel/test-http-agent-domain-reused-gc.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-agent-domain-reused-gc.js)
 - [parallel/test-http-agent-error-on-idle.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-agent-error-on-idle.js)
+- [parallel/test-http-agent-false.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-agent-false.js)
+- [parallel/test-http-agent-keepalive-delay.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-agent-keepalive-delay.js)
 - [parallel/test-http-agent-keepalive.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-agent-keepalive.js)
 - [parallel/test-http-agent-maxsockets-respected.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-agent-maxsockets-respected.js)
 - [parallel/test-http-agent-maxsockets.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-agent-maxsockets.js)
@@ -848,6 +850,7 @@ NOTE: This file should not be manually edited. Please edit `tests/node_compat/co
 - [parallel/test-http-client-set-timeout.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-client-set-timeout.js)
 - [parallel/test-http-client-spurious-aborted.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-client-spurious-aborted.js)
 - [parallel/test-http-client-timeout-agent.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-client-timeout-agent.js)
+- [parallel/test-http-client-timeout-connect-listener.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-client-timeout-connect-listener.js)
 - [parallel/test-http-client-timeout-event.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-client-timeout-event.js)
 - [parallel/test-http-client-timeout-on-connect.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-client-timeout-on-connect.js)
 - [parallel/test-http-client-timeout-option-listeners.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-client-timeout-option-listeners.js)
@@ -865,6 +868,7 @@ NOTE: This file should not be manually edited. Please edit `tests/node_compat/co
 - [parallel/test-http-destroyed-socket-write2.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-destroyed-socket-write2.js)
 - [parallel/test-http-dns-error.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-dns-error.js)
 - [parallel/test-http-double-content-length.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-double-content-length.js)
+- [parallel/test-http-dump-req-when-res-ends.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-dump-req-when-res-ends.js)
 - [parallel/test-http-early-hints-invalid-argument.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-early-hints-invalid-argument.js)
 - [parallel/test-http-early-hints.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-early-hints.js)
 - [parallel/test-http-exceptions.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-exceptions.js)
@@ -876,6 +880,7 @@ NOTE: This file should not be manually edited. Please edit `tests/node_compat/co
 - [parallel/test-http-header-badrequest.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-header-badrequest.js)
 - [parallel/test-http-header-overflow.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-header-overflow.js)
 - [parallel/test-http-host-header-ipv6-fail.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-host-header-ipv6-fail.js)
+- [parallel/test-http-hostname-typechecking.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-hostname-typechecking.js)
 - [parallel/test-http-incoming-matchKnownFields.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-incoming-matchKnownFields.js)
 - [parallel/test-http-incoming-message-connection-setter.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-incoming-message-connection-setter.js)
 - [parallel/test-http-incoming-message-options.js](https://github.com/nodejs/node/tree/v20.11.1/test/parallel/test-http-incoming-message-options.js)
@@ -2508,9 +2513,12 @@ NOTE: This file should not be manually edited. Please edit `tests/node_compat/co
 - [sequential/test-inspector-port-cluster.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-inspector-port-cluster.js)
 - [sequential/test-module-loading.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-module-loading.js)
 - [sequential/test-net-GH-5504.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-net-GH-5504.js)
+- [sequential/test-net-better-error-messages-port.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-net-better-error-messages-port.js)
 - [sequential/test-net-connect-econnrefused.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-net-connect-econnrefused.js)
+- [sequential/test-net-connect-handle-econnrefused.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-net-connect-handle-econnrefused.js)
 - [sequential/test-net-listen-shared-ports.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-net-listen-shared-ports.js)
 - [sequential/test-net-localport.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-net-localport.js)
+- [sequential/test-net-reconnect-error.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-net-reconnect-error.js)
 - [sequential/test-net-server-address.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-net-server-address.js)
 - [sequential/test-next-tick-error-spin.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-next-tick-error-spin.js)
 - [sequential/test-perf-hooks.js](https://github.com/nodejs/node/tree/v20.11.1/test/sequential/test-perf-hooks.js)

tests/node_compat/test/parallel/test-http-agent-false.js

@@ -1,53 +0,0 @@
-// deno-fmt-ignore-file
-// deno-lint-ignore-file
-
-// Copyright Joyent and Node contributors. All rights reserved. MIT license.
-// Taken from Node 20.11.1
-// This file is automatically generated by `tests/node_compat/runner/setup.ts`. Do not modify this file manually.
-
-// Copyright Joyent, Inc. and other Node contributors.
-//
-// Permission is hereby granted, free of charge, to any person obtaining a
-// copy of this software and associated documentation files (the
-// "Software"), to deal in the Software without restriction, including
-// without limitation the rights to use, copy, modify, merge, publish,
-// distribute, sublicense, and/or sell copies of the Software, and to permit
-// persons to whom the Software is furnished to do so, subject to the
-// following conditions:
-//
-// The above copyright notice and this permission notice shall be included
-// in all copies or substantial portions of the Software.
-//
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
-// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
-// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
-// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
-// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
-// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
-// USE OR OTHER DEALINGS IN THE SOFTWARE.
-
-'use strict';
-const common = require('../common');
-const http = require('http');
-
-// Sending `agent: false` when `port: null` is also passed in (i.e. the result
-// of a `url.parse()` call with the default port used, 80 or 443), should not
-// result in an assertion error...
-const opts = {
-  host: '127.0.0.1',
-  port: null,
-  path: '/',
-  method: 'GET',
-  agent: false
-};
-
-// We just want an "error" (no local HTTP server on port 80) or "response"
-// to happen (user happens ot have HTTP server running on port 80).
-// As long as the process doesn't crash from a C++ assertion then we're good.
-const req = http.request(opts);
-
-// Will be called by either the response event or error event, not both
-const oneResponse = common.mustCall();
-req.on('response', oneResponse);
-req.on('error', oneResponse);
-req.end();

tests/node_compat/test/parallel/test-http-agent-keepalive-delay.js

@@ -1,43 +0,0 @@
-// deno-fmt-ignore-file
-// deno-lint-ignore-file
-
-// Copyright Joyent and Node contributors. All rights reserved. MIT license.
-// Taken from Node 20.11.1
-// This file is automatically generated by `tests/node_compat/runner/setup.ts`. Do not modify this file manually.
-
-'use strict';
-
-const common = require('../common');
-const assert = require('assert');
-const http = require('http');
-const { Agent } = require('_http_agent');
-
-const agent = new Agent({
-  keepAlive: true,
-  keepAliveMsecs: 1000,
-});
-
-const server = http.createServer(common.mustCall((req, res) => {
-  res.end('ok');
-}));
-
-server.listen(0, common.mustCall(() => {
-  const createConnection = agent.createConnection;
-  agent.createConnection = (options, ...args) => {
-    assert.strictEqual(options.keepAlive, true);
-    assert.strictEqual(options.keepAliveInitialDelay, agent.keepAliveMsecs);
-    return createConnection.call(agent, options, ...args);
-  };
-  http.get({
-    host: 'localhost',
-    port: server.address().port,
-    agent: agent,
-    path: '/'
-  }, common.mustCall((res) => {
-    // for emit end event
-    res.on('data', () => {});
-    res.on('end', () => {
-      server.close();
-    });
-  }));
-}));

tests/node_compat/test/parallel/test-http-client-timeout-connect-listener.js

@@ -1,49 +0,0 @@
-// deno-fmt-ignore-file
-// deno-lint-ignore-file
-
-// Copyright Joyent and Node contributors. All rights reserved. MIT license.
-// Taken from Node 20.11.1
-// This file is automatically generated by `tests/node_compat/runner/setup.ts`. Do not modify this file manually.
-
-'use strict';
-const common = require('../common');
-
-// This test ensures that `ClientRequest.prototype.setTimeout()` does
-// not add a listener for the `'connect'` event to the socket if the
-// socket is already connected.
-
-const assert = require('assert');
-const http = require('http');
-
-// Maximum allowed value for timeouts.
-const timeout = 2 ** 31 - 1;
-
-const server = http.createServer((req, res) => {
-  res.end();
-});
-
-server.listen(0, common.mustCall(() => {
-  const agent = new http.Agent({ keepAlive: true, maxSockets: 1 });
-  const options = { port: server.address().port, agent: agent };
-
-  doRequest(options, common.mustCall(() => {
-    const req = doRequest(options, common.mustCall(() => {
-      agent.destroy();
-      server.close();
-    }));
-
-    req.on('socket', common.mustCall((socket) => {
-      assert.strictEqual(socket.listenerCount('connect'), 0);
-    }));
-  }));
-}));
-
-function doRequest(options, callback) {
-  const req = http.get(options, (res) => {
-    res.on('end', callback);
-    res.resume();
-  });
-
-  req.setTimeout(timeout);
-  return req;
-}

tests/node_compat/test/parallel/test-http-dump-req-when-res-ends.js

@@ -1,73 +0,0 @@
-// deno-fmt-ignore-file
-// deno-lint-ignore-file
-
-// Copyright Joyent and Node contributors. All rights reserved. MIT license.
-// Taken from Node 20.11.1
-// This file is automatically generated by `tests/node_compat/runner/setup.ts`. Do not modify this file manually.
-
-'use strict';
-
-const { mustCall } = require('../common');
-
-const fs = require('fs');
-const http = require('http');
-const { strictEqual } = require('assert');
-
-const server = http.createServer(mustCall(function(req, res) {
-  strictEqual(req.socket.listenerCount('data'), 1);
-  req.socket.once('data', mustCall(function() {
-    // Ensure that a chunk of data is received before calling `res.end()`.
-    res.end('hello world');
-  }));
-  // This checks if the request gets dumped
-  // resume will be triggered by res.end().
-  req.on('resume', mustCall(function() {
-    // There is no 'data' event handler anymore
-    // it gets automatically removed when dumping the request.
-    strictEqual(req.listenerCount('data'), 0);
-    req.on('data', mustCall());
-  }));
-
-  // We explicitly pause the stream
-  // so that the following on('data') does not cause
-  // a resume.
-  req.pause();
-  req.on('data', function() {});
-
-  // Start sending the response.
-  res.flushHeaders();
-}));
-
-server.listen(0, mustCall(function() {
-  const req = http.request({
-    method: 'POST',
-    port: server.address().port
-  });
-
-  // Send the http request without waiting
-  // for the body.
-  req.flushHeaders();
-
-  req.on('response', mustCall(function(res) {
-    // Pipe the body as soon as we get the headers of the
-    // response back.
-    fs.createReadStream(__filename).pipe(req);
-
-    res.resume();
-
-    // On some platforms the `'end'` event might not be emitted because the
-    // socket could be destroyed by the other peer while data is still being
-    // sent. In this case the 'aborted'` event is emitted instead of `'end'`.
-    // `'close'` is used here because it is always emitted and does not
-    // invalidate the test.
-    res.on('close', function() {
-      server.close();
-    });
-  }));
-
-  req.on('error', function() {
-    // An error can happen if there is some data still
-    // being sent, as the other side is calling .destroy()
-    // this is safe to ignore.
-  });
-}));

tests/node_compat/test/parallel/test-http-hostname-typechecking.js

@@ -1,49 +0,0 @@
-// deno-fmt-ignore-file
-// deno-lint-ignore-file
-
-// Copyright Joyent and Node contributors. All rights reserved. MIT license.
-// Taken from Node 20.11.1
-// This file is automatically generated by `tests/node_compat/runner/setup.ts`. Do not modify this file manually.
-
-'use strict';
-
-const common = require('../common');
-const assert = require('assert');
-const http = require('http');
-
-// All of these values should cause http.request() to throw synchronously
-// when passed as the value of either options.hostname or options.host
-const vals = [{}, [], NaN, Infinity, -Infinity, true, false, 1, 0, new Date()];
-
-vals.forEach((v) => {
-  const received = common.invalidArgTypeHelper(v);
-  assert.throws(
-    () => http.request({ hostname: v }),
-    {
-      code: 'ERR_INVALID_ARG_TYPE',
-      name: 'TypeError',
-      message: 'The "options.hostname" property must be of ' +
-               'type string or one of undefined or null.' +
-               received
-    }
-  );
-
-  assert.throws(
-    () => http.request({ host: v }),
-    {
-      code: 'ERR_INVALID_ARG_TYPE',
-      name: 'TypeError',
-      message: 'The "options.host" property must be of ' +
-               'type string or one of undefined or null.' +
-               received
-    }
-  );
-});
-
-// These values are OK and should not throw synchronously.
-// Only testing for 'hostname' validation so ignore connection errors.
-const dontCare = () => {};
-['', undefined, null].forEach((v) => {
-  http.request({ hostname: v }).on('error', dontCare).end();
-  http.request({ host: v }).on('error', dontCare).end();
-});

tests/node_compat/test/sequential/test-net-better-error-messages-port.js

@@ -1,24 +0,0 @@
-// deno-fmt-ignore-file
-// deno-lint-ignore-file
-
-// Copyright Joyent and Node contributors. All rights reserved. MIT license.
-// Taken from Node 20.11.1
-// This file is automatically generated by `tests/node_compat/runner/setup.ts`. Do not modify this file manually.
-
-'use strict';
-const common = require('../common');
-const net = require('net');
-const assert = require('assert');
-
-const c = net.createConnection(common.PORT);
-
-c.on('connect', common.mustNotCall());
-
-c.on('error', common.mustCall(function(error) {
-  // Family autoselection might be skipped if only a single address is returned by DNS.
-  const failedAttempt = Array.isArray(error.errors) ? error.errors[0] : error;
-
-  assert.strictEqual(failedAttempt.code, 'ECONNREFUSED');
-  assert.strictEqual(failedAttempt.port, common.PORT);
-  assert.match(failedAttempt.address, /^(127\.0\.0\.1|::1)$/);
-}));

tests/node_compat/test/sequential/test-net-connect-handle-econnrefused.js

@@ -1,39 +0,0 @@
-// deno-fmt-ignore-file
-// deno-lint-ignore-file
-
-// Copyright Joyent and Node contributors. All rights reserved. MIT license.
-// Taken from Node 20.11.1
-// This file is automatically generated by `tests/node_compat/runner/setup.ts`. Do not modify this file manually.
-
-// Copyright Joyent, Inc. and other Node contributors.
-//
-// Permission is hereby granted, free of charge, to any person obtaining a
-// copy of this software and associated documentation files (the
-// "Software"), to deal in the Software without restriction, including
-// without limitation the rights to use, copy, modify, merge, publish,
-// distribute, sublicense, and/or sell copies of the Software, and to permit
-// persons to whom the Software is furnished to do so, subject to the
-// following conditions:
-//
-// The above copyright notice and this permission notice shall be included
-// in all copies or substantial portions of the Software.
-//
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
-// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
-// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
-// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
-// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
-// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
-// USE OR OTHER DEALINGS IN THE SOFTWARE.
-
-'use strict';
-const common = require('../common');
-const net = require('net');
-const assert = require('assert');
-
-const c = net.createConnection(common.PORT);
-c.on('connect', common.mustNotCall());
-c.on('error', common.mustCall((e) => {
-  assert.strictEqual(c.connecting, false);
-  assert.strictEqual(e.code, 'ECONNREFUSED');
-}));

tests/node_compat/test/sequential/test-net-reconnect-error.js

@@ -1,50 +0,0 @@
-// deno-fmt-ignore-file
-// deno-lint-ignore-file
-
-// Copyright Joyent and Node contributors. All rights reserved. MIT license.
-// Taken from Node 20.11.1
-// This file is automatically generated by `tests/node_compat/runner/setup.ts`. Do not modify this file manually.
-
-// Copyright Joyent, Inc. and other Node contributors.
-//
-// Permission is hereby granted, free of charge, to any person obtaining a
-// copy of this software and associated documentation files (the
-// "Software"), to deal in the Software without restriction, including
-// without limitation the rights to use, copy, modify, merge, publish,
-// distribute, sublicense, and/or sell copies of the Software, and to permit
-// persons to whom the Software is furnished to do so, subject to the
-// following conditions:
-//
-// The above copyright notice and this permission notice shall be included
-// in all copies or substantial portions of the Software.
-//
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
-// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
-// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
-// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
-// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
-// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
-// USE OR OTHER DEALINGS IN THE SOFTWARE.
-
-'use strict';
-const common = require('../common');
-const net = require('net');
-const assert = require('assert');
-const N = 20;
-let disconnectCount = 0;
-
-const c = net.createConnection(common.PORT);
-
-c.on('connect', common.mustNotCall('client should not have connected'));
-
-c.on('error', common.mustCall((error) => {
-  // Family autoselection might be skipped if only a single address is returned by DNS.
-  const actualError = Array.isArray(error.errors) ? error.errors[0] : error;
-
-  assert.strictEqual(actualError.code, 'ECONNREFUSED');
-}, N + 1));
-
-c.on('close', common.mustCall(() => {
-  if (disconnectCount++ < N)
-    c.connect(common.PORT); // reconnect
-}, N + 1));

tests/unit_node/http_test.ts

@@ -499,7 +499,6 @@ Deno.test("[node/http] send request with non-chunked body", async () => {
     assert(socket.writable);
     assert(socket.readable);
     socket.setKeepAlive();
-    socket.destroy();
     socket.setTimeout(100);
   });
   req.write("hello ");
@@ -512,6 +511,11 @@ Deno.test("[node/http] send request with non-chunked body", async () => {
     // in order to not cause a flaky test sanitizer failure
     await new Promise((resolve) => setTimeout(resolve, 100)),
   ]);
+
+  if (Deno.build.os === "windows") {
+    // FIXME(kt3k): This is necessary for preventing op leak on windows
+    await new Promise((resolve) => setTimeout(resolve, 4000));
+  }
 });
 
 Deno.test("[node/http] send request with chunked body", async () => {
@@ -559,6 +563,11 @@ Deno.test("[node/http] send request with chunked body", async () => {
   req.end();
 
   await servePromise;
+
+  if (Deno.build.os === "windows") {
+    // FIXME(kt3k): This is necessary for preventing op leak on windows
+    await new Promise((resolve) => setTimeout(resolve, 4000));
+  }
 });
 
 Deno.test("[node/http] send request with chunked body as default", async () => {
@@ -604,6 +613,11 @@ Deno.test("[node/http] send request with chunked body as default", async () => {
   req.end();
 
   await servePromise;
+
+  if (Deno.build.os === "windows") {
+    // FIXME(kt3k): This is necessary for preventing op leak on windows
+    await new Promise((resolve) => setTimeout(resolve, 4000));
+  }
 });
 
 Deno.test("[node/http] ServerResponse _implicitHeader", async () => {
@@ -689,7 +703,7 @@ Deno.test("[node/http] ClientRequest handle non-string headers", async () => {
   assertEquals(headers!["1"], "2");
 });
 
-Deno.test("[node/http] ClientRequest uses HTTP/1.1", async () => {
+Deno.test("[node/https] ClientRequest uses HTTP/1.1", async () => {
   let body = "";
   const { promise, resolve, reject } = Promise.withResolvers<void>();
   const req = https.request("https://localhost:5545/http_version", {
@@ -800,8 +814,9 @@ Deno.test("[node/http] ClientRequest search params", async () => {
   let body = "";
   const { promise, resolve, reject } = Promise.withResolvers<void>();
   const req = http.request({
-    host: "localhost:4545",
+    host: "localhost",
-    path: "search_params?foo=bar",
+    port: 4545,
+    path: "/search_params?foo=bar",
   }, (resp) => {
     resp.on("data", (chunk) => {
       body += chunk;
@@ -1011,28 +1026,50 @@ Deno.test(
 
 Deno.test(
   "[node/http] client destroy before sending request should not error",
-  () => {
+  async () => {
+    const { resolve, promise } = Promise.withResolvers<void>();
     const request = http.request("http://localhost:5929/");
     // Calling this would throw
     request.destroy();
+    request.on("error", (e) => {
+      assertEquals(e.message, "socket hang up");
+    });
+    request.on("close", () => resolve());
+    await promise;
+
+    if (Deno.build.os === "windows") {
+      // FIXME(kt3k): This is necessary for preventing op leak on windows
+      await new Promise((resolve) => setTimeout(resolve, 4000));
+    }
   },
 );
 
+const isWindows = Deno.build.os === "windows";
+
 Deno.test(
   "[node/http] destroyed requests should not be sent",
+  { sanitizeResources: !isWindows, sanitizeOps: !isWindows },
   async () => {
     let receivedRequest = false;
-    const server = Deno.serve(() => {
+    const requestClosed = Promise.withResolvers<void>();
+    const ac = new AbortController();
+    const server = Deno.serve({ port: 0, signal: ac.signal }, () => {
       receivedRequest = true;
       return new Response(null);
     });
     const request = http.request(`http://localhost:${server.addr.port}/`);
     request.destroy();
     request.end("hello");
-
+    request.on("error", (err) => {
-    await new Promise((r) => setTimeout(r, 500));
+      assert(err.message.includes("socket hang up"));
+      ac.abort();
+    });
+    request.on("close", () => {
+      requestClosed.resolve();
+    });
+    await requestClosed.promise;
     assertEquals(receivedRequest, false);
-    await server.shutdown();
+    await server.finished;
   },
 );
 
@@ -1060,22 +1097,33 @@ Deno.test("[node/https] node:https exports globalAgent", async () => {
   );
 });
 
-Deno.test("[node/http] node:http request.setHeader(header, null) doesn't throw", () => {
+Deno.test("[node/http] node:http request.setHeader(header, null) doesn't throw", async () => {
   {
-    const req = http.request("http://localhost:4545/");
+    const { promise, resolve } = Promise.withResolvers<void>();
-    req.on("error", () => {});
+    const req = http.request("http://localhost:4545/", (res) => {
+      res.on("data", () => {});
+      res.on("end", () => {
+        resolve();
+      });
+    });
     // @ts-expect-error - null is not a valid header value
     req.setHeader("foo", null);
     req.end();
-    req.destroy();
+    await promise;
   }
   {
-    const req = https.request("https://localhost:4545/");
+    const { promise, resolve } = Promise.withResolvers<void>();
-    req.on("error", () => {});
+    const req = http.request("http://localhost:4545/", (res) => {
+      res.on("data", () => {});
+      res.on("end", () => {
+        resolve();
+      });
+    });
     // @ts-expect-error - null is not a valid header value
     req.setHeader("foo", null);
     req.end();
-    req.destroy();
+
+    await promise;
   }
 });