fix(ext/node): per-request executionAsyncResource() for async_hooks (#34188)

## Summary

`executionAsyncResource()` from `node:async_hooks` previously returned a
fresh
`{}` on every call, so writes never persisted and the standard
CLS-on-async-
hooks pattern blew up with:

```
TypeError: Cannot destructure property 'state' of 'executionAsyncResource(...)[sym]'
as it is undefined.
```

This change tracks the current resource via an `AsyncVariable` (backed
by V8
continuation-preserved embedder data), defaulting to a shared top-level
singleton. The value propagates naturally across promise/await
transitions and
the existing setTimeout async-context capture/restore.

- `AsyncResource.runInAsyncScope` enters `this` as the current resource
for the
  scope.
- The HTTP/1.1 server enters a per-request scope around the
`request`/`checkContinue`/`checkExpectation` emit so concurrent requests
get
isolated resources. Without this, every concurrent request would race on
  anything stashed on the shared top-level resource.

Enables these two Node compat tests that the issue called out:

- `parallel/test-async-hooks-execution-async-resource.js`
- `parallel/test-async-hooks-execution-async-resource-await.js`

Closes denoland/orchid#129

## Test plan

- [x] Both target tests pass under `cargo test --test node_compat`
- [x] Full `async-hooks` and `async-local-storage` suites — no
regressions vs.
      tests already enabled in `tests/node_compat/config.jsonc`
- [x] `parallel::test-http` suite — no new regressions in tests already
in
`config.jsonc` (one pre-existing flaky HTTP/2 test unrelated to this
      change)

Co-authored-by: divybot <divybot@users.noreply.github.com>
Co-authored-by: Divy Srivastava <me@littledivy.com>

Author: divybot

ext/node/polyfills/_http_server.js

@@ -78,6 +78,10 @@ const {
   validateObject,
 } = core.loadExtScript("ext:deno_node/internal/validators.mjs");
 const { nextTick } = core.loadExtScript("ext:deno_node/_next_tick.ts");
+const {
+  enterAsyncResource,
+  exitAsyncResource,
+} = core.loadExtScript("ext:deno_node/internal/async_hooks.ts");
 const { enqueueNodePerformanceEntry } = core.loadExtScript(
   "ext:deno_node/perf_hooks.js",
 );
@@ -865,58 +869,69 @@ function parserOnIncoming(server, socket, state, req, keepAlive) {
     });
   }
 
-  let handled = false;
-
-  if (req.httpVersionMajor === 1 && req.httpVersionMinor === 1) {
-    if (
-      server.requireHostHeader !== false &&
-      req.headers.host === undefined
-    ) {
-      res.writeHead(400, ["Connection", "close"]);
-      res.end();
-      return 0;
-    }
+  // Enter a new async-hooks resource scope for the duration of the request
+  // emission. Each request gets its own resource (the IncomingMessage), so
+  // executionAsyncResource() returns a per-request object that is preserved
+  // across timers, await transitions, etc. Without this, every request would
+  // share the top-level resource and concurrent requests would race on any
+  // state stashed there.
+  const prevAsyncResource = enterAsyncResource(req);
+  try {
+    let handled = false;
+
+    if (req.httpVersionMajor === 1 && req.httpVersionMinor === 1) {
+      if (
+        server.requireHostHeader !== false &&
+        req.headers.host === undefined
+      ) {
+        res.writeHead(400, ["Connection", "close"]);
+        res.end();
+        return 0;
+      }
 
-    const isRequestsLimitSet =
-      typeof server.maxRequestsPerSocket === "number" &&
-      server.maxRequestsPerSocket > 0;
+      const isRequestsLimitSet =
+        typeof server.maxRequestsPerSocket === "number" &&
+        server.maxRequestsPerSocket > 0;
 
-    if (isRequestsLimitSet) {
-      state.requestsCount++;
-      res.maxRequestsOnConnectionReached =
-        server.maxRequestsPerSocket <= state.requestsCount;
-    }
+      if (isRequestsLimitSet) {
+        state.requestsCount++;
+        res.maxRequestsOnConnectionReached =
+          server.maxRequestsPerSocket <= state.requestsCount;
+      }
 
-    if (
-      isRequestsLimitSet &&
-      server.maxRequestsPerSocket < state.requestsCount
-    ) {
-      handled = true;
-      server.emit("dropRequest", req, socket);
-      res.writeHead(503);
-      res.end();
-    } else if (req.headers.expect !== undefined) {
-      handled = true;
-
-      if (continueExpression.test(req.headers.expect)) {
-        res._expect_continue = true;
-        if (server.listenerCount("checkContinue") > 0) {
-          server.emit("checkContinue", req, res);
+      if (
+        isRequestsLimitSet &&
+        server.maxRequestsPerSocket < state.requestsCount
+      ) {
+        handled = true;
+        server.emit("dropRequest", req, socket);
+        res.writeHead(503);
+        res.end();
+      } else if (req.headers.expect !== undefined) {
+        handled = true;
+
+        if (continueExpression.test(req.headers.expect)) {
+          res._expect_continue = true;
+          if (server.listenerCount("checkContinue") > 0) {
+            server.emit("checkContinue", req, res);
+          } else {
+            res.writeContinue();
+            server.emit("request", req, res);
+          }
+        } else if (server.listenerCount("checkExpectation") > 0) {
+          server.emit("checkExpectation", req, res);
         } else {
-          res.writeContinue();
-          server.emit("request", req, res);
+          res.writeHead(417);
+          res.end();
         }
-      } else if (server.listenerCount("checkExpectation") > 0) {
-        server.emit("checkExpectation", req, res);
-      } else {
-        res.writeHead(417);
-        res.end();
       }
     }
-  }
 
-  if (!handled) {
-    server.emit("request", req, res);
+    if (!handled) {
+      server.emit("request", req, res);
+    }
+  } finally {
+    exitAsyncResource(prevAsyncResource);
   }
 
   return 0;

ext/node/polyfills/async_hooks.ts

@@ -16,7 +16,10 @@ const {
   emitBefore,
   emitDestroy: emitDestroyHook,
   emitInit,
+  enterAsyncResource,
+  exitAsyncResource,
   executionAsyncId: internalExecutionAsyncId,
+  executionAsyncResource: internalExecutionAsyncResource,
   newAsyncId,
 } = core.loadExtScript("ext:deno_node/internal/async_hooks.ts");
 
@@ -70,7 +73,14 @@ class AsyncResource {
     emitBefore(this.#asyncId);
     try {
       setAsyncContext(this.#snapshot);
-      return ReflectApply(fn, thisArg, args);
+      // Enter this resource as the current executionAsyncResource() so that
+      // user code inside the scope observes `this` as the active resource.
+      const prevResource = enterAsyncResource(this);
+      try {
+        return ReflectApply(fn, thisArg, args);
+      } finally {
+        exitAsyncResource(prevResource);
+      }
     } finally {
       setAsyncContext(previousContext);
       emitAfter(this.#asyncId);
@@ -198,9 +208,7 @@ function triggerAsyncId() {
   return 0;
 }
 
-function executionAsyncResource() {
-  return {};
-}
+const executionAsyncResource = internalExecutionAsyncResource;
 
 const asyncWrapProviders = ObjectFreeze({
   __proto__: null,

ext/node/polyfills/internal/async_hooks.ts

@@ -23,6 +23,10 @@ const {
   FunctionPrototypeApply,
   Symbol,
 } = primordials;
+const {
+  AsyncVariable,
+  setAsyncContext,
+} = core;
 
 interface ActiveHooks {
   array: AsyncHook[];
@@ -80,6 +84,33 @@ function executionAsyncId(): number {
   return executionAsyncIdStack[executionAsyncIdStack.length - 1] || 0;
 }
 
+// Per-async-context "current resource" tracked via the AsyncVariable
+// machinery (V8 ContinuationPreservedEmbedderData). This propagates across
+// promises and await transitions automatically. The top-level resource is a
+// shared singleton used before any specific resource has been entered.
+// deno-lint-ignore no-explicit-any
+const topLevelResource: any = { __proto__: null };
+// deno-lint-ignore no-explicit-any
+const executionResourceVariable: any = new AsyncVariable();
+
+// deno-lint-ignore no-explicit-any
+function executionAsyncResource(): any {
+  const r = executionResourceVariable.get();
+  return r === undefined ? topLevelResource : r;
+}
+
+// Enter a new "current resource" scope. The returned value is the previous
+// async context snapshot that must be restored by exitAsyncResource.
+// deno-lint-ignore no-explicit-any
+function enterAsyncResource(resource: any): any {
+  return executionResourceVariable.enter(resource);
+}
+
+// deno-lint-ignore no-explicit-any
+function exitAsyncResource(previousContext: any): void {
+  setAsyncContext(previousContext);
+}
+
 // Emit functions that work with the internal hook system
 function emitBefore(asyncId: number): void {
   ArrayPrototypePush(executionAsyncIdStack, asyncId);
@@ -461,6 +492,9 @@ return {
   emitInit: emitInitNative,
   constants,
   executionAsyncId,
+  executionAsyncResource,
+  enterAsyncResource,
+  exitAsyncResource,
   emitBefore,
   emitAfter,
   emitDestroy,

tests/node_compat/config.jsonc

@@ -182,6 +182,8 @@
     "parallel/test-async-hooks-disable-gc-tracking.js": {},
     "parallel/test-async-hooks-enable-disable.js": {},
     "parallel/test-async-hooks-enabledhooksexits.js": {},
+    "parallel/test-async-hooks-execution-async-resource-await.js": {},
+    "parallel/test-async-hooks-execution-async-resource.js": {},
     "parallel/test-async-hooks-http-agent-destroy.js": {},
     "parallel/test-async-hooks-http-agent.js": {},
     "parallel/test-async-hooks-prevent-double-destroy.js": {},