refactor(ext/node): wire up native TLSWrap from JS (#33184)

Replace the `kStreamBaseField`-based TLS implementation with a native
`TLSWrap` cppgc object backed by rustls. This eliminates the async
stream-swap pattern where TLS upgrades worked by swapping
`handle[kStreamBaseField]` from a `TcpConn` to a `TlsConn`.

Key changes:
- Rewrite `_tls_wrap.js`: `TLSSocket` uses `tls_wrap.wrap()` to create
  a Rust-backed TLS interceptor between JS and the underlying TCP stream
- TLS server rewritten to use `net.Server` (matching Node.js) instead of
  `Deno.listenTls()` directly
- Add `JSStreamSocket` for TLS over arbitrary Duplex streams
- Certificate errors now use Node-style codes
  (`UNABLE_TO_VERIFY_LEAF_SIGNATURE`) instead of Deno-style
  (`InvalidData: invalid peer certificate: UnknownIssuer`)
- Explicit `ca` in SecureContext takes precedence over global
  `setDefaultCACertificates()`, matching Node.js behavior
- Strip trailing dot from servername before passing to rustls
- Remove all `Deno.*` API usage from `_tls_wrap.js`

Based on #32819 by @nathanwhit, adapted for incremental landing.
---------

Co-authored-by: Nathan Whitaker <user@example.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

ext/node/lib.rs

@@ -444,6 +444,7 @@ deno_core::extension!(deno_node,
     "internal_binding/string_decoder.ts",
     "internal_binding/symbols.ts",
     "internal_binding/tcp_wrap.ts",
+    "internal_binding/tls_wrap.ts",
     "internal_binding/tty_wrap.ts",
     "internal_binding/types.ts",
     "internal_binding/udp_wrap.ts",
@@ -497,6 +498,7 @@ deno_core::extension!(deno_node,
     "internal/http2/util.ts",
     "internal/http2/compat.js",
     "internal/idna.ts",
+    "internal/js_stream_socket.js",
     "internal/mime.ts",
     "internal/net.ts",
     "internal/normalize_encoding.ts",
@@ -527,6 +529,7 @@ deno_core::extension!(deno_node,
     "internal/streams/state.js",
     "internal/streams/utils.js",
     "internal/test/binding.ts",
+    "internal/tls_common.js",
     "internal/timers.mjs",
     "internal/tty.js",
     "internal/url.ts",

ext/node/ops/stream_wrap.rs

@@ -293,11 +293,10 @@ impl LibUvStreamWrap {
     // `StreamHandleData` allocation while the native stream is alive.
     let handle_data = unsafe { handle_data_ptr.as_ref() };
     handle_data.desired_read_interceptor.set(interceptor);
-    if let Some(key) = handle_data.active_read.get() {
-      let _ = handle_data
-        .read_callbacks
-        .borrow_mut()
-        .update_interceptor(key, interceptor);
+    if let Some(key) = handle_data.active_read.get()
+      && let Ok(mut callbacks) = handle_data.read_callbacks.try_borrow_mut()
+    {
+      let _ = callbacks.update_interceptor(key, interceptor);
     }
   }
 
@@ -315,18 +314,22 @@ impl LibUvStreamWrap {
       return 0;
     }
 
-    let key =
-      handle_data
-        .read_callbacks
-        .borrow_mut()
-        .insert(ReadCallbackState {
-          isolate: v8::UnsafeRawIsolatePtr::null(),
-          onread: None,
-          stream_base_state: None,
-          handle: None,
-          bytes_read: handle_data.bytes_read.clone(),
-          read_interceptor: handle_data.desired_read_interceptor.get(),
-        });
+    // Use try_borrow_mut to handle re-entrant calls gracefully.
+    // This can happen when uv_read_start fires on_uv_read synchronously
+    // (data already buffered), whose interceptor calls cycle() which
+    // may call back into read_start.
+    let Ok(mut callbacks) = handle_data.read_callbacks.try_borrow_mut() else {
+      return 0;
+    };
+    let key = callbacks.insert(ReadCallbackState {
+      isolate: v8::UnsafeRawIsolatePtr::null(),
+      onread: None,
+      stream_base_state: None,
+      handle: None,
+      bytes_read: handle_data.bytes_read.clone(),
+      read_interceptor: handle_data.desired_read_interceptor.get(),
+    });
+    drop(callbacks);
     handle_data.active_read.set(Some(key));
 
     // SAFETY: `stream` is a valid libuv stream owned by this wrapper.
@@ -343,7 +346,11 @@ impl LibUvStreamWrap {
     // `StreamHandleData` allocation while the native stream is alive.
     let handle_data = unsafe { handle_data_ptr.as_ref() };
     if let Some(key) = handle_data.active_read.take() {
-      let _ = handle_data.read_callbacks.borrow_mut().remove(key);
+      // Use try_borrow_mut to handle re-entrant calls from interceptor
+      // callbacks that may fire during on_uv_read processing.
+      if let Ok(mut callbacks) = handle_data.read_callbacks.try_borrow_mut() {
+        let _ = callbacks.remove(key);
+      }
     }
     // SAFETY: `stream` is a valid libuv stream owned by this wrapper.
     unsafe { uv_compat::uv_read_stop(stream) }