fix: tunnels in processes that don't listen

Cargo.toml

@@ -131,7 +131,7 @@ napi_sym = { version = "0.141.0", path = "./ext/napi/sym" }
 node_resolver = { version = "0.49.0", path = "./libs/node_resolver" }
 test_util = { package = "test_server", path = "./tests/util/server" }
 
-deno_tunnel = "0.5.0"
+deno_tunnel = "0.7.0"
 
 # widely used libraries
 anyhow = "1.0.57"

cli/main.rs

@@ -627,7 +627,11 @@ async fn resolve_flags_and_init(
     .or_else(|| env::var("DENO_CONNECTED").ok())
   {
     if let Err(err) = initialize_tunnel(&host, &flags).await {
-      exit_for_error(AnyError::from(err))
+      exit_for_error(err.context("Failed to start with --connected"));
+    }
+    // SAFETY: We're doing this before any threads are created.
+    unsafe {
+      std::env::set_var("DENO_CONNECTED", &host);
     }
   }
 

ext/net/tunnel.rs

@@ -4,6 +4,8 @@ use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::rc::Rc;
 use std::sync::OnceLock;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 
 use deno_core::AsyncMutFuture;
 use deno_core::AsyncRefCell;
@@ -27,14 +29,25 @@ use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
 
 static TUNNEL: OnceLock<TunnelConnection> = OnceLock::new();
+static RUN_BEFORE_EXIT: AtomicBool = AtomicBool::new(true);
 
 pub fn set_tunnel(tunnel: TunnelConnection) {
   if TUNNEL.set(tunnel).is_ok() {
-    deno_signals::before_exit(before_exit);
+    deno_signals::before_exit(before_exit_internal);
   }
 }
 
-fn before_exit() {
+fn before_exit_internal() {
+  if RUN_BEFORE_EXIT.load(Ordering::Relaxed) {
+    before_exit();
+  }
+}
+
+pub fn disable_before_exit() {
+  RUN_BEFORE_EXIT.store(false, Ordering::Relaxed);
+}
+
+pub fn before_exit() {
   if let Some(tunnel) = get_tunnel() {
     log::trace!("deno_net::tunnel::before_exit >");
 

ext/telemetry/lib.rs

@@ -928,6 +928,7 @@ pub fn init(
     .map_err(|_| deno_core::anyhow::anyhow!("failed to set otel globals"))?;
 
   deno_signals::before_exit(before_exit);
+  deno_net::tunnel::disable_before_exit();
 
   Ok(())
 }
@@ -953,6 +954,8 @@ fn before_exit() {
 
   let r = meter_provider.shutdown();
   log::trace!("meters={:?}", r);
+
+  deno_net::tunnel::before_exit();
 }
 
 pub fn handle_log(record: &log::Record) {

tests/specs/run/tunnel/test.out

@@ -1,3 +1,4 @@
+You are connected to https://localhost:[WILDLINE]!
 HTTP/1.1 200 OK
 content-type: application/json
 vary: Accept-Encoding

tests/specs/run/tunnel/test.ts

@@ -1,4 +1,4 @@
-const VERSION = 1;
+const VERSION = 2;
 
 const server = new Deno.QuicEndpoint({
   hostname: "localhost",
@@ -31,6 +31,7 @@ const child = new Deno.Command(Deno.execPath(), {
 }).spawn();
 
 setTimeout(() => {
+  console.error("timeout, killing");
   child.kill("SIGKILL");
   Deno.exit(1);
 }, 5000);
@@ -48,19 +49,19 @@ async function handleConnection(conn: Deno.QuicConn) {
     const reader = bi.readable.getReader({ mode: "byob" });
     const version = await readUint32LE(reader);
     if (version !== VERSION) {
-      conn.close();
+      conn.close({ closeCode: 1, reason: "invalid version" });
       return;
     }
     const writer = bi.writable.getWriter();
     await writeUint32LE(writer, VERSION);
     const header = await readStreamHeader(reader);
     if (header.headerType !== "Control") {
-      conn.close();
+      conn.close({ closeCode: 1, reason: "expected Control" });
       return;
     }
     const auth = await readStreamHeader(reader);
     if (auth.headerType !== "AuthenticateApp") {
-      conn.close();
+      conn.close({ closeCode: 1, reason: "expected AuthenticateApp" });
       return;
     }
     await writeStreamHeader(writer, {
@@ -75,6 +76,16 @@ async function handleConnection(conn: Deno.QuicConn) {
       metadata: {},
     });
 
+    const next = await readStreamHeader(reader);
+    if (next.headerType !== "Listening") {
+      conn.close({ closeCode: 1, reason: "expected Listening" });
+      return;
+    }
+
+    await writeStreamHeader(writer, {
+      headerType: "Routed",
+    });
+
     reader.releaseLock();
     writer.releaseLock();
   }