http2: fix client async storage persistence

Create and store an AsyncResource for each stream, following a similar
approach as used in HttpAgent.
Fixes: #55376
PR-URL: #55460
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Stephen Belanger <admin@stephenbelanger.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Gerhard Stöbich <deb2001-github@yahoo.de>

Author: orgads

lib/internal/http2/core.js

@@ -60,6 +60,8 @@ const {
  owner_symbol,
 },
 }=require('internal/async_hooks');
+const{ AsyncResource }=require('async_hooks');
+
 const{
  AbortError,
  aggregateTwoErrors,
@@ -241,6 +243,7 @@ const kPendingRequestCalls = Symbol('kPendingRequestCalls');
 constkProceed=Symbol('proceed');
 constkProtocol=Symbol('protocol');
 constkRemoteSettings=Symbol('remote-settings');
+constkRequestAsyncResource=Symbol('requestAsyncResource');
 constkSelectPadding=Symbol('select-padding');
 constkSentHeaders=Symbol('sent-headers');
 constkSentTrailers=Symbol('sent-trailers');
@@ -408,7 +411,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);
+constreqAsync=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);
@@ -1800,6 +1807,8 @@ class ClientHttp2Session extends Http2Session {
 stream[kSentHeaders]=headers;
 stream[kOrigin]=`${headers[HTTP2_HEADER_SCHEME]}://`+
 `${getAuthority(headers)}`;
+constreqAsync=newAsyncResource('PendingRequest');
+stream[kRequestAsyncResource]=reqAsync;
 
 // Close the writable side of the stream if options.endStream is set.
 if(options.endStream)
@@ -1822,7 +1831,7 @@ class ClientHttp2Session extends Http2Session {
 }
 }
 
-constonConnect=requestOnConnect.bind(stream,headersList,options);
+constonConnect=reqAsync.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';
+
+constcommon=require('../common');
+if(!common.hasCrypto)
+common.skip('missing crypto');
+constassert=require('assert');
+consthttp2=require('http2');
+constasync_hooks=require('async_hooks');
+
+conststorage=newasync_hooks.AsyncLocalStorage();
+
+const{
+HTTP2_HEADER_CONTENT_TYPE,
+HTTP2_HEADER_PATH,
+HTTP2_HEADER_STATUS,
+}=http2.constants;
+
+constserver=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()=>{
+constclient=storage.run({id: 0},()=>http2.connect(`http://localhost:${server.address().port}`));
+
+asyncfunctiondoReq(id){
+constreq=client.request({[HTTP2_HEADER_PATH]: '/'});
+
+req.on('response',common.mustCall((headers)=>{
+assert.strictEqual(headers[HTTP2_HEADER_STATUS],200);
+assert.strictEqual(id,storage.getStore().id);
+}));
+req.on('data',common.mustCall((data)=>{
+assert.strictEqual(data.toString(),'data');
+assert.strictEqual(id,storage.getStore().id);
+}));
+req.on('end',common.mustCall(()=>{
+assert.strictEqual(id,storage.getStore().id);
+server.close();
+client.close();
+}));
+}
+
+functiondoReqWith(id){
+storage.run({ id },()=>doReq(id));
+}
+
+doReqWith(1);
+doReqWith(2);
+});