Skip to content

Commit

Permalink
http2: Fix client async storage persistence
Browse files Browse the repository at this point in the history
Create and store an AsyncResource for each stream, following a similar
approach as used in HttpAgent.

Fixes nodejs#55376
  • Loading branch information
orgads committed Oct 19, 2024
1 parent 99c6e4e commit 2db3762
Show file tree
Hide file tree
Showing 2 changed files with 70 additions and 3 deletions.
18 changes: 15 additions & 3 deletions lib/internal/http2/core.js
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand Down Expand Up @@ -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');
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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);
Expand Down
55 changes: 55 additions & 0 deletions test/parallel/test-http2-async-local-storage.js
Original file line number Diff line number Diff line change
@@ -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);
});

0 comments on commit 2db3762

Please sign in to comment.