feat(ext/net): add signal option to Deno.connect() (#27113)

Ref #26819

An optional **timeout** parameter has been added to the
**Deno.connect()** interface. This parameter allows specifying a timeout
(in milliseconds) within which the application must establish a
connection. If the timeout is exceeded without successfully connecting,
the operation is automatically aborted with an error. If the parameter
is not provided, the default behavior remains unchanged (no timeout).
Currently, the timeout functionality is implemented only for TCP
connections. Other connection types are not affected by this change.

---------

Co-authored-by: Yoshiya Hinosawa <stibium121@gmail.com>

ext/net/01_net.js

@@ -623,16 +623,36 @@ function createListenDatagram(udpOpFn, unixOpFn) {
 async function connect(args) {
   switch (args.transport ?? "tcp") {
     case "tcp": {
+      let cancelRid;
+      let abortHandler;
+      if (args?.signal) {
+        args.signal.throwIfAborted();
+        cancelRid = createCancelHandle();
+        abortHandler = () => core.tryClose(cancelRid);
+        args.signal[abortSignal.add](abortHandler);
+      }
       const port = validatePort(args.port);
-      const { 0: rid, 1: localAddr, 2: remoteAddr } = await op_net_connect_tcp(
-        {
-          hostname: args.hostname ?? "127.0.0.1",
-          port,
-        },
-      );
-      localAddr.transport = "tcp";
-      remoteAddr.transport = "tcp";
-      return new TcpConn(rid, remoteAddr, localAddr);
+
+      try {
+        const { 0: rid, 1: localAddr, 2: remoteAddr } =
+          await op_net_connect_tcp(
+            {
+              hostname: args.hostname ?? "127.0.0.1",
+              port,
+            },
+            undefined,
+            cancelRid,
+          );
+        localAddr.transport = "tcp";
+        remoteAddr.transport = "tcp";
+
+        return new TcpConn(rid, remoteAddr, localAddr);
+      } finally {
+        if (args?.signal) {
+          args.signal[abortSignal.remove](abortHandler);
+          args.signal.throwIfAborted();
+        }
+      }
     }
     case "unix": {
       const { 0: rid, 1: localAddr, 2: remoteAddr } = await op_net_connect_unix(

ext/net/ops.rs

@@ -422,11 +422,13 @@ pub async fn op_net_connect_tcp<NP>(
   state: Rc<RefCell<OpState>>,
   #[serde] addr: IpAddr,
   #[cppgc] net_perm_token: Option<&NetPermToken>,
+  #[smi] resource_abort_id: Option<ResourceId>,
 ) -> Result<(ResourceId, IpAddr, IpAddr), NetError>
 where
   NP: NetPermissions + 'static,
 {
-  op_net_connect_tcp_inner::<NP>(state, addr, net_perm_token).await
+  op_net_connect_tcp_inner::<NP>(state, addr, net_perm_token, resource_abort_id)
+    .await
 }
 
 #[inline]
@@ -434,6 +436,7 @@ pub async fn op_net_connect_tcp_inner<NP>(
   state: Rc<RefCell<OpState>>,
   addr: IpAddr,
   net_perm_token: Option<&NetPermToken>,
+  resource_abort_id: Option<ResourceId>,
 ) -> Result<(ResourceId, IpAddr, IpAddr), NetError>
 where
   NP: NetPermissions + 'static,
@@ -455,7 +458,32 @@ where
     .await?
     .next()
     .ok_or_else(|| NetError::NoResolvedAddress)?;
-  let tcp_stream = TcpStream::connect(&addr).await?;
+
+  let cancel_handle = resource_abort_id.and_then(|rid| {
+    state
+      .borrow_mut()
+      .resource_table
+      .get::<CancelHandle>(rid)
+      .ok()
+  });
+
+  let tcp_stream_result = if let Some(cancel_handle) = &cancel_handle {
+    TcpStream::connect(&addr).or_cancel(cancel_handle).await?
+  } else {
+    TcpStream::connect(&addr).await
+  };
+
+  if let Some(cancel_rid) = resource_abort_id {
+    if let Ok(res) = state.borrow_mut().resource_table.take_any(cancel_rid) {
+      res.close();
+    }
+  }
+
+  let tcp_stream = match tcp_stream_result {
+    Ok(tcp_stream) => tcp_stream,
+    Err(e) => return Err(NetError::Io(e)),
+  };
+
   let local_addr = tcp_stream.local_addr()?;
   let remote_addr = tcp_stream.peer_addr()?;
 
@@ -1363,9 +1391,10 @@ mod tests {
       port: server_addr[1].parse().unwrap(),
     };
 
-    let mut connect_fut =
-      op_net_connect_tcp_inner::<TestPermission>(conn_state, ip_addr, None)
-        .boxed_local();
+    let mut connect_fut = op_net_connect_tcp_inner::<TestPermission>(
+      conn_state, ip_addr, None, None,
+    )
+    .boxed_local();
     let mut rid = None;
 
     tokio::select! {