http2: Fix client async storage persistence

Create and store an AsyncResource for each stream, following a similar
approach as used in HttpAgent.

Fixes: #55376

Author: orgads

lib/internal/http2/core.js

@@ -173,6 +173,7 @@ const {
 const { kTimeout } = require('internal/timers');
 const { isArrayBufferView } = require('internal/util/types');
 const { format } = require('internal/util/inspect');
+const { AsyncResource } = require('async_hooks');
 
 const { FileHandle } = internalBinding('fs');
 const binding = internalBinding('http2');
@@ -241,6 +242,7 @@ const kPendingRequestCalls = Symbol('kPendingRequestCalls');
 const kProceed = Symbol('proceed');
 const kProtocol = Symbol('protocol');
 const kRemoteSettings = Symbol('remote-settings');
+const kRequestAsyncResource = Symbol('requestAsyncResource');
 const kSelectPadding = Symbol('select-padding');
 const kSentHeaders = Symbol('sent-headers');
 const kSentTrailers = Symbol('sent-trailers');
@@ -408,7 +410,11 @@ function onSessionHeaders(handle, id, cat, flags, headers, sensitiveHeaders) {
       originSet.delete(stream[kOrigin]);
     }
     debugStream(id, type, "emitting stream '%s' event", event);
-    process.nextTick(emit, stream, event, obj, flags, headers);
+    const reqAsync = stream[kRequestAsyncResource];
+    if (reqAsync)
+      reqAsync.runInAsyncScope(process.nextTick, null, emit, stream, event, obj, flags, headers);
+    else
+      process.nextTick(emit, stream, event, obj, flags, headers);
   }
   if (endOfStream) {
     stream.push(null);
@@ -1089,7 +1095,11 @@ function setupHandle(socket, type, options) {
     ReflectApply(this.origin, this, options.origins);
   }
 
-  process.nextTick(emit, this, 'connect', this, socket);
+  const reqAsync = this[kRequestAsyncResource];
+  if (reqAsync)
+    reqAsync.runInAsyncScope(process.nextTick, null, emit, this, 'connect', this, socket);
+  else
+    process.nextTick(emit, this, 'connect', this, socket);
 }
 
 // Emits a close event followed by an error event if err is truthy. Used
@@ -1797,6 +1807,8 @@ class ClientHttp2Session extends Http2Session {
     stream[kSentHeaders] = headers;
     stream[kOrigin] = `${headers[HTTP2_HEADER_SCHEME]}://` +
                       `${getAuthority(headers)}`;
+    const asyncRes = new AsyncResource('PendingRequest');
+    stream[kRequestAsyncResource] = asyncRes;
 
     // Close the writable side of the stream if options.endStream is set.
     if (options.endStream)
@@ -1819,7 +1831,7 @@ class ClientHttp2Session extends Http2Session {
       }
     }
 
-    const onConnect = requestOnConnect.bind(stream, headersList, options);
+    const onConnect = asyncRes.bind(requestOnConnect.bind(stream, headersList, options));
     if (this.connecting) {
       if (this[kPendingRequestCalls] !== null) {
         this[kPendingRequestCalls].push(onConnect);

test/parallel/test-http2-async-local-storage.js

@@ -0,0 +1,55 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const async_hooks = require('async_hooks');
+
+const storage = new async_hooks.AsyncLocalStorage();
+
+const {
+  HTTP2_HEADER_CONTENT_TYPE,
+  HTTP2_HEADER_PATH,
+  HTTP2_HEADER_STATUS,
+} = http2.constants;
+
+const server = http2.createServer();
+server.on('stream', (stream) => {
+  stream.respond({
+    [HTTP2_HEADER_CONTENT_TYPE]: 'text/plain; charset=utf-8',
+    [HTTP2_HEADER_STATUS]: 200
+  });
+  stream.on('error', common.mustNotCall());
+  stream.end('data');
+});
+
+server.listen(0, async () => {
+  const client = storage.run({ id: 0 }, () => http2.connect(`http://localhost:${server.address().port}`));
+
+  async function doReq(id) {
+    const req = client.request({ [HTTP2_HEADER_PATH]: '/' });
+
+    req.on('response', common.mustCall((headers) => {
+      assert.strictEqual(headers[HTTP2_HEADER_STATUS], 200);
+      assert.equal(id, storage.getStore().id);
+    }));
+    req.on('data', common.mustCall((data) => {
+      assert.equal(data, 'data');
+      assert.equal(id, storage.getStore().id);
+    }));
+    req.on('end', common.mustCall(() => {
+      assert.equal(id, storage.getStore().id);
+      server.close();
+      client.close();
+    }));
+  }
+
+  function doReqWith(id) {
+    storage.run({ id }, () => doReq(id));
+  }
+
+  doReqWith(1);
+  doReqWith(2);
+});