fix(http): wake runtime after direct serve dispatch (#34387)

Fixes #34369.

## Root cause

`Deno.serve` request dispatch now calls the JS request handler directly
from a standalone hyper task. That path performs a V8 microtask
checkpoint after invoking the callback, but it does not otherwise
guarantee that the full `JsRuntime` event loop is polled again.

If the handler reaches an async boundary backed by `node:net` /
uv-compatible I/O, the TCP handle can be marked ready inside the
uv-compat layer, but the outer runtime task still needs to be woken so
the uv I/O phase runs. Previously the mpsc/op-driven request path
provided that scheduling edge implicitly.

This change stores the runtime waker in `ServerCallback` and wakes it
after direct JS dispatch, restoring one full event-loop pass after
request-handler invocation.

## Changes

- Thread the runtime `OpState` waker into `ServerCallback`.
- Wake the runtime after the post-dispatch microtask checkpoint.
- Add a regression test covering `Deno.serve` awaiting `node:net` I/O
from inside the request handler.

## Validation

- `cargo build --profile release-lite --bin deno`
- `./target/release-lite/deno run
--allow-net=127.0.0.1:12477,127.0.0.1:12478
tests/specs/serve/node_net_wake/main.ts`

I also checked the regression against an unfixed release-lite binary: it
timed out after 5s. The fixed binary returned `ok` immediately.

## Performance

Hello-world `Deno.serve`, release-lite builds, `oha -z 15s -c 100`,
alternating baseline/fixed samples:

| Build | Avg req/s | Avg p50 | Avg p99 |
| --- | ---: | ---: | ---: |
| baseline | 135,073 | 0.707 ms | 0.897 ms |
| fixed | 134,452 | 0.709 ms | 0.903 ms |

Throughput delta: -0.46%, within local run noise.

Author: nathanwhit

ext/http/http_next.rs

@@ -1279,7 +1279,12 @@ where
 
   let lifetime = resource.lifetime();
   let callback_global = v8::Global::new(scope, callback);
-  let callback = Rc::new(ServerCallback::new(scope, isolate, callback_global));
+  let callback = Rc::new(ServerCallback::new(
+    scope,
+    isolate,
+    callback_global,
+    state.borrow().waker.clone(),
+  ));
 
   let options = {
     let state = state.borrow();
@@ -1336,7 +1341,12 @@ where
 
   let resource: Rc<HttpJoinHandle> = Rc::new(HttpJoinHandle::new());
   let callback_global = v8::Global::new(scope, callback);
-  let callback = Rc::new(ServerCallback::new(scope, isolate, callback_global));
+  let callback = Rc::new(ServerCallback::new(
+    scope,
+    isolate,
+    callback_global,
+    state.borrow().waker.clone(),
+  ));
 
   let options = {
     let state = state.borrow();

ext/http/service.rs

@@ -9,6 +9,7 @@ use std::future::Future;
 use std::mem::ManuallyDrop;
 use std::pin::Pin;
 use std::rc::Rc;
+use std::sync::Arc;
 use std::sync::OnceLock;
 use std::task::Context;
 use std::task::Poll;
@@ -18,6 +19,7 @@ use std::task::ready;
 use deno_core::BufView;
 use deno_core::OpState;
 use deno_core::ResourceId;
+use deno_core::futures::task::AtomicWaker;
 use deno_core::v8;
 use deno_error::JsErrorBox;
 use http::request::Parts;
@@ -167,13 +169,15 @@ pub struct ServerCallback {
   isolate_ptr: v8::UnsafeRawIsolatePtr,
   context: v8::Global<v8::Context>,
   callback: v8::Global<v8::Function>,
+  runtime_waker: Arc<AtomicWaker>,
 }
 
 impl ServerCallback {
   pub fn new(
     scope: &mut v8::PinScope<'_, '_>,
     isolate: &mut v8::Isolate,
     callback: v8::Global<v8::Function>,
+    runtime_waker: Arc<AtomicWaker>,
   ) -> Self {
     let ctx = scope.get_current_context();
     let context = v8::Global::new(scope, ctx);
@@ -183,6 +187,7 @@ impl ServerCallback {
       isolate_ptr,
       context,
       callback,
+      runtime_waker,
     }
   }
 
@@ -231,6 +236,12 @@ impl ServerCallback {
       // to the first real async-op boundary in user code -- matching
       // how Bun's onRequest invokes its handler.
       pin_scope.perform_microtask_checkpoint();
+
+      // Request arrival is handled by a standalone hyper task, not by
+      // an op future owned by JsRuntime. If the handler reached an
+      // async boundary above, the full event loop must run at least
+      // once to drive timers, ops, and uv-compatible Node I/O.
+      self.runtime_waker.wake();
     }
   }
 }

tests/specs/serve/node_net_wake/__test__.jsonc

@@ -0,0 +1,6 @@
+{
+  "args": "run --allow-net=127.0.0.1:12477,127.0.0.1:12478 main.ts",
+  "output": "main.out",
+  "exitCode": 0,
+  "tempDir": true
+}

tests/specs/serve/node_net_wake/main.out

@@ -0,0 +1 @@
+ok

tests/specs/serve/node_net_wake/main.ts

@@ -0,0 +1,55 @@
+import net from "node:net";
+
+const httpPort = 12477;
+const tcpPort = 12478;
+
+const listener = Deno.listen({ hostname: "127.0.0.1", port: tcpPort });
+
+(async () => {
+  try {
+    for await (const conn of listener) {
+      (async () => {
+        const buf = new Uint8Array(16);
+        await conn.read(buf);
+        await conn.write(new TextEncoder().encode("ok"));
+        conn.close();
+      })();
+    }
+  } catch {
+    // Listener shutdown.
+  }
+})();
+
+const server = Deno.serve({
+  hostname: "127.0.0.1",
+  port: httpPort,
+  onListen() {},
+}, async () => {
+  const body = await new Promise<string>((resolve, reject) => {
+    const socket = net.connect(tcpPort, "127.0.0.1");
+    socket.on("connect", () => socket.write("x"));
+    socket.on("data", (chunk) => {
+      resolve(String(chunk));
+      socket.destroy();
+    });
+    socket.on("error", reject);
+  });
+  return new Response(body);
+});
+
+(async () => {
+  try {
+    const response = await fetch(`http://127.0.0.1:${httpPort}/`, {
+      signal: AbortSignal.timeout(5000),
+    });
+    console.log(await response.text());
+    await server.shutdown();
+    listener.close();
+    Deno.exit(0);
+  } catch (error) {
+    console.error(error);
+    await server.shutdown().catch(() => {});
+    listener.close();
+    Deno.exit(2);
+  }
+})();