Repository: denoland/deno
Pull request: fix(node): preserve AsyncLocalStorage context in stream.finished callback

## Summary
- Snapshots the async context when `eos()` (the implementation behind `stream.finished`) is called and restores it around the callback invocation
- In Node.js this context propagation happens automatically through the native `AsyncWrap`/`InternalCallbackScope` C++ layer (`src/api/callback.cc`), but Deno's ops don't flow through that path, so `AsyncLocalStorage.getStore()` returned `undefined` inside `stream.finished` callbacks
- Adds a unit test verifying the fix with an HTTP server scenario matching the issue reproduction

Closes #32327

## Test plan
- [x] Added `asyncLocalStoragePreservedInStreamFinished` unit test in `tests/unit_node/async_hooks_test.ts`
- [x] Verified the exact PoC from the issue now prints `OK: got expected value`
- [x] Existing `unit_node::async_hooks_test` suite passes

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Changed files:
filename: ext/node/polyfills/internal/streams/end-of-stream.js
patch: @@ -3,6 +3,7 @@
 
 import process from "node:process";
 import { primordials } from "ext:core/mod.js";
+import { core } from "ext:core/mod.js";
 import imported1 from "ext:deno_node/internal/errors.ts";
 import { kEmptyObject, once } from "ext:deno_node/internal/util.mjs";
 import {
@@ -70,6 +71,22 @@ function eos(stream, options, callback) {
   validateFunction(callback, "callback");
   validateAbortSignal(options.signal, "options.signal");
 
+  // Capture the current async context so that the callback runs in the
+  // same AsyncLocalStorage scope that was active when eos() was called.
+  // In Node.js this happens automatically through the native AsyncWrap
+  // layer, but Deno's ops don't propagate Node-style async context.
+  const snapshot = core.getAsyncContext();
+  const originalCallback = callback;
+  callback = function (...args) {
+    const previousContext = core.getAsyncContext();
+    try {
+      core.setAsyncContext(snapshot);
+      return originalCallback.apply(this, args);
+    } finally {
+      core.setAsyncContext(previousContext);
+    }
+  };
+
   callback = once(callback);
 
   if (isReadableStream(stream) || isWritableStream(stream)) {

filename: tests/unit_node/async_hooks_test.ts
patch: @@ -172,3 +172,38 @@ Deno.test(async function worksWithDynamicImports() {
     assertEquals(data, "data");
   });
 });
+
+Deno.test(async function asyncLocalStoragePreservedInStreamFinished() {
+  const http = await import("node:http");
+  const { finished } = await import("node:stream");
+  const als = new AsyncLocalStorage();
+  const store = { foo: "bar" };
+
+  const { promise, resolve, reject } = Promise.withResolvers<void>();
+
+  const server = http.createServer((_req, res) => {
+    als.run(store, () => {
+      finished(res, () => {
+        try {
+          assertEquals(als.getStore(), store);
+          resolve();
+        } catch (e) {
+          reject(e);
+        }
+      });
+    });
+
+    setTimeout(() => res.end(), 0);
+  });
+
+  server.listen(0, () => {
+    const addr = server.address()!;
+    const port = typeof addr === "string" ? addr : addr.port;
+    http.get(`http://127.0.0.1:${port}`, (res) => {
+      res.resume();
+      res.on("end", () => server.close());
+    });
+  });
+
+  await promise;
+});