http2: Async resource store is not applied on 'response' event

[orgads]

Version

22.9.0

Platform

Linux my-server 6.10.11-amd64 #1 SMP PREEMPT_DYNAMIC Debian 6.10.11-1 (2024-09-22) x86_64 GNU/Linux

Subsystem

http2

What steps will reproduce the bug?

server.mjs:

import { createServer } from 'node:http2';

const server = createServer();

server.on('error', (err) => console.error(err));
server.on('stream', (stream, headers) => {
  stream.respond({
    'content-type': 'text/html; charset=utf-8',
    ':status': 200,
  });
  stream.end('<h1>Hello World</h1>');
});

server.listen(9443);

client.mjs:

import { connect } from 'node:http2';
import { AsyncLocalStorage } from 'async_hooks';
import { setTimeout as sleep } from 'timers/promises';

const asyncLocalStorage = new AsyncLocalStorage();

function log(message) {
  const data = asyncLocalStorage?.getStore();
  console.log(`${JSON.stringify(data)}: ${message}`);
}

const client = connect('http://localhost:9443');

async function doReq() {
  await sleep(100); // <-- No sessions at all without this (??)
  client.on('error', (err) => log(err.message));

  log('sending request');
  const req = client.request({ ':path': '/' });

  req.on('response', (headers, flags) => {
    log(`headers: ${JSON.stringify(headers)}`); // <--- No session here :(
  });

  req.setEncoding('utf8');
  let data = '';
  req.on('data', (chunk) => {
    log(`chunk: ${chunk}`); // Works
    data += chunk;
  });
  req.on('error', (err) => log(`Error: ${err.message}`));
  req.on('end', () => {
    log(data); // Works
    client.close();
  });
  req.end();
}

asyncLocalStorage.run({ session: 1 }, doReq);
asyncLocalStorage.run({ session: 2 }, doReq);

Output:

{"session":1}: sending request
{"session":2}: sending request
undefined: headers: {":status":200,"content-type":"text/html; charset=utf-8","date":"Sun, 13 Oct 2024 15:26:19 GMT"}
undefined: headers: {":status":200,"content-type":"text/html; charset=utf-8","date":"Sun, 13 Oct 2024 15:26:19 GMT"}
{"session":1}: chunk: <h1>Hello World</h1>
{"session":1}: <h1>Hello World</h1>
{"session":2}: chunk: <h1>Hello World</h1>
{"session":2}: <h1>Hello World</h1>

How often does it reproduce? Is there a required condition?

Every time

What is the expected behavior? Why is that the expected behavior?

The async store should be applied also on response

What do you see instead?

It's not.

Additional information

No response

[Mian-Ahmed-Raza]

Try updating the client.mjs file like this:

import { connect } from 'node:http2';
import { AsyncLocalStorage } from 'async_hooks';
import { setTimeout as sleep } from 'timers/promises';

const asyncLocalStorage = new AsyncLocalStorage();

function log(message) {
  const data = asyncLocalStorage?.getStore();
  console.log(`${JSON.stringify(data)}: ${message}`);
}

const client = connect('http://localhost:9443');

async function doReq() {
  await sleep(100); // <-- No sessions at all without this (??)
  client.on('error', (err) => log(err.message));

  log('sending request');
  const req = client.request({ ':path': '/' });

  // Bind the response event to the current async context
  req.on('response', asyncLocalStorage.bind((headers, flags) => {
    log(`headers: ${JSON.stringify(headers)}`);
  }));

  req.setEncoding('utf8');
  let data = '';
  req.on('data', asyncLocalStorage.bind((chunk) => {
    log(`chunk: ${chunk}`);
    data += chunk;
  }));
  req.on('error', asyncLocalStorage.bind((err) => log(`Error: ${err.message}`)));
  req.on('end', asyncLocalStorage.bind(() => {
    log(data);
    client.close();
  }));
  req.end();
}

asyncLocalStorage.run({ session: 1 }, doReq);
asyncLocalStorage.run({ session: 2 }, doReq);

[orgads]

@Mian-Ahmed-Raza Thank you. This is a possible workaround, but it is obviously not a solution.

In my case, I don't call http2 directly, I use grpc-js, which internally uses http2. I have no control over the way the library connects to the events.

[orgads]

By the way, if I initialize the client itself with async store, this value is shown for response:

let client;
asyncLocalStorage.run({ session: 0 }, () => {
  client = connect('http://localhost:9443')
});

Output:

{"session":1}: sending request
{"session":2}: sending request
{"session":0}: headers: {":status":200,"content-type":"text/html; charset=utf-8","date":"Sun, 13 Oct 2024 15:30:24 GMT"}
{"session":0}: headers: {":status":200,"content-type":"text/html; charset=utf-8","date":"Sun, 13 Oct 2024 15:30:24 GMT"}
{"session":1}: chunk: <h1>Hello World</h1>
{"session":1}: <h1>Hello World</h1>
{"session":2}: chunk: <h1>Hello World</h1>
{"session":2}: <h1>Hello World</h1>

[Mian-Ahmed-Raza]

To maintain the correct session context throughout the request-response lifecycle, you need to ensure that both the client creation and the requests are encapsulated within the same asyncLocalStorage.run() call. Here’s how you can do it:

import { connect } from 'node:http2';
import { AsyncLocalStorage } from 'async_hooks';
import { setTimeout as sleep } from 'timers/promises';

const asyncLocalStorage = new AsyncLocalStorage();

function log(message) {
    const data = asyncLocalStorage?.getStore();
    console.log(`${JSON.stringify(data)}: ${message}`);
}

async function makeGrpcCall(sessionId) {
    // Create the client inside the async context
    let client = connect('http://localhost:9443');

    // Bind error handling to the client
    client.on('error', (err) => log(err.message));

    log('sending request');
    const req = client.request({ ':path': '/' });

    // Bind the response event to maintain context
    req.on('response', (headers, flags) => {
        log(`headers: ${JSON.stringify(headers)}`);
    });

    req.setEncoding('utf8');
    let data = '';
    req.on('data', (chunk) => {
        log(`chunk: ${chunk}`);
        data += chunk;
    });
    req.on('error', (err) => log(`Error: ${err.message}`));
    req.on('end', () => {
        log(data);
        client.close();
    });
    req.end();
}

// Create requests within separate async contexts
asyncLocalStorage.run({ session: 1 }, () => makeGrpcCall(1));
asyncLocalStorage.run({ session: 2 }, () => makeGrpcCall(2));

[orgads]

Once again, I'm not using http2 directly. This is the relevant part in grpc-js: https://github.com/grpc/grpc-node/blob/67322057b214e17578717d86216fdf99036ba761/packages/grpc-js/src/subchannel-call.ts#L152

Do you suggest that I hook in it? How?

[Mian-Ahmed-Raza]

I understand that you're using grpc-js, which internally utilizes HTTP/2, making it difficult to control how the asynchronous context is managed within those library internals. Given this, let’s explore alternative approaches to maintain the AsyncLocalStorage context without directly initializing the HTTP/2 client.

import { connect } from 'node:http2'; // Assume grpc-js uses this under the hood
import { AsyncLocalStorage } from 'async_hooks';
import { setTimeout as sleep } from 'timers/promises';

const asyncLocalStorage = new AsyncLocalStorage();

function log(message) {
    const data = asyncLocalStorage?.getStore();
    console.log(`${JSON.stringify(data)}: ${message}`);
}

// Simulate making a gRPC call with a callback
async function makeGrpcCall(sessionId) {
    await sleep(100); // Simulating some async operation before making the call

    const client = connect('http://localhost:9443');
    client.on('error', (err) => log(err.message));

    log('sending request');
    const req = client.request({ ':path': '/' });

    req.on('response', (headers) => {
        log(`headers: ${JSON.stringify(headers)}`);
    });

    req.setEncoding('utf8');
    let data = '';
    req.on('data', (chunk) => {
        log(`chunk: ${chunk}`);
        data += chunk;
    });
    req.on('error', (err) => log(`Error: ${err.message}`));
    req.on('end', () => {
        log(data);
        client.close();
    });
    req.end();
}

// Wrap the gRPC call in asyncLocalStorage.run() to maintain context
asyncLocalStorage.run({ session: 1 }, () => makeGrpcCall(1));
asyncLocalStorage.run({ session: 2 }, () => makeGrpcCall(2));

If the issue persists and you're finding it challenging to maintain the context, it may be worth exploring the source code of grpc-js. This could help you understand how it manages its internal HTTP/2 connections and whether there are hooks or extensions available that could allow for better context handling.

[orgads]

But I need all the google-cloud/google-gax/grpc logic, I'm not going to implement all of it because of a bug in node.

[orgads]

I realize my notes may have come across as offensive, and I truly appreciate your help. I had no intention of causing any offense.

I'll try to debug node and propose a solution.

This issue is similar to #41285.

cc @mcollina @nicolo-ribaudo