Repository: denoland/deno
Pull request: feat(ext/ops): add signal option to `Deno.connect()`

Related to #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.

Example usage:
```JS
try {
  const conn = await Deno.connect({
    hostname: "example.com",
    port: 5000,
    timeout: 5000, // 5 second timeout
  });

  console.log("Connection established successfully");  
  conn.close();
} catch (error) {
  if (error instanceof Deno.errors.TimedOut) {
    console.error("Connection timed out:", error.message);
  } else {
    console.error("Failed to connect:", error.message);
  }
}

```

If this solution is approved, the [documentation](https://docs.deno.com/api/deno/~/Deno.connect) will need to be updated to include details about the new timeout parameter.

Changed files:
filename: cli/tsc/dts/lib.deno_net.d.ts
patch: @@ -339,7 +339,10 @@ declare namespace Deno {
      *
      * @default {"127.0.0.1"} */
     hostname?: string;
+    /** The transport layer protocol to use. */
     transport?: "tcp";
+    /** An {@linkcode AbortSignal} to close the tcp connection. */
+    signal?: AbortSignal;
   }
 
   /**

filename: ext/net/01_net.js
patch: @@ -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(

filename: ext/net/ops.rs
patch: @@ -422,18 +422,21 @@ 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]
 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! {

filename: tests/unit/net_test.ts
patch: @@ -1277,6 +1277,26 @@ Deno.test({
   }
 });
 
+Deno.test(
+  { permissions: { net: true } },
+  async function netTcpWithAbortSignal() {
+    const controller = new AbortController();
+    setTimeout(() => controller.abort(), 100);
+    const error = await assertRejects(
+      async () => {
+        await Deno.connect({
+          hostname: "deno.com",
+          port: 50000,
+          transport: "tcp",
+          signal: controller.signal,
+        });
+      },
+    );
+    assert(error instanceof DOMException);
+    assertEquals(error.name, "AbortError");
+  },
+);
+
 Deno.test({
   ignore: Deno.build.os === "linux",
   permissions: { net: true },