stream: preserve AsyncLocalStorage context in finished()

PR-URL: #57865
Reviewed-By: Raz Luvaton <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Chengzhong Wu <[email protected]>
Reviewed-By: Ruben Bridgewater <[email protected]>
Reviewed-By: James M Snell <[email protected]>

Author: gurgunday

lib/internal/streams/end-of-stream.js

@@ -43,6 +43,9 @@ const {
   willEmitClose: _willEmitClose,
   kIsClosedPromise,
 } = require('internal/streams/utils');
+
+// Lazy load
+let AsyncLocalStorage;
 let addAbortListener;
 
 function isRequest(stream) {
@@ -63,7 +66,8 @@ function eos(stream, options, callback) {
   validateFunction(callback, 'callback');
   validateAbortSignal(options.signal, 'options.signal');
 
-  callback = once(callback);
+  AsyncLocalStorage ??= require('async_hooks').AsyncLocalStorage;
+  callback = once(AsyncLocalStorage.bind(callback));
 
   if (isReadableStream(stream) || isWritableStream(stream)) {
     return eosWeb(stream, options, callback);

test/async-hooks/test-async-local-storage-stream-finished.js

@@ -0,0 +1,20 @@
+'use strict';
+
+const common = require('../common');
+const { Readable, finished } = require('stream');
+const { AsyncLocalStorage } = require('async_hooks');
+const { strictEqual } = require('assert');
+
+// This test verifies that AsyncLocalStorage context is maintained
+// when using stream.finished()
+
+const readable = new Readable();
+const als = new AsyncLocalStorage();
+
+als.run(321, () => {
+  finished(readable, common.mustCall(() => {
+    strictEqual(als.getStore(), 321);
+  }));
+});
+
+readable.destroy();