fix(H2 Client): bind stream 'data' listener only after received 'response' event

[st3ffgv4]

This relates to...

#2808

Rationale

Sporadically when doing massive h2 requests for a certain period of time a breaking assertion is triggered "assert(!this.aborted)" in onHeaders Method [undici/lib/core/request.js] causing application crash

Changes

• lib/dispatcher/client-h2.js

Features

N/A

Bug Fixes

• Sometimes happens that H2 Stream "data" event is triggered some milliseconds before the "response" event, this leads to an uninitialized body. To avoid this breaking error if the onData is called before onHeaders, return false so the stream is paused in client-h2.js

Breaking Changes and Deprecations

Status

• I have read and agreed to the [Developer's Certificate of Origin][cert]
• Tested
• Benchmarked (optional)
• Documented
• Review ready
• In review
• Merge ready

[cert]: https://github.com/nodejs/undici/blob/main/CONTRIBUTING.md #

[ronag]

This need to be fixed somewhere else. Can we defer the onData in the h2 wrapper?

[st3ffgv4]

@ronag like doing this?

in client-h2.js

let isResponseStarted = false

stream.once('response', headers => {
    isResponseStarted = true
    const { [HTTP2_HEADER_STATUS]: statusCode, ...realHeaders } = headers
    request.onResponseStarted()

    if (request.onHeaders(Number(statusCode), realHeaders, stream.resume.bind(stream), '') === false) {
      stream.pause()
    }
  })
  
  stream.on('data', (chunk) => {
    if (!isResponseStarted || request.onData(chunk) === false) {
      stream.pause()
    }
  })

[mcollina]

Thanks for opening a PR! Can you please add a unit test?

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 93.49%. Comparing base (8fce214) to head (9219ee3).
Report is 1 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2985   +/-   ##
=======================================
  Coverage   93.49%   93.49%           
=======================================
  Files          89       89           
  Lines       24214    24216    +2     
=======================================
+ Hits        22638    22640    +2     
  Misses       1576     1576           

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[st3ffgv4]

@mcollina this would maybe be a long term test, because the error can happen at any time, immediately or not, it would be like using the script reported in the issue and wait indefinitely if the scenario occurs

[ronag]

this will throw away the first chunk?

[st3ffgv4]

@ronag thaks, good point, I was focused on seeing that the error no longer occurs.

Is it ok to declare a temp chunks array and if isStreamResponseStarted is false in 'data' event push the chunk and then in 'response' event if the temp chunks array has length > 0 call the request.onData method after the request.onHeaders?

[ronag]

This does kind of feel like an error in node core http, it doesn't make sense to me that data is emitted before response. Do you think you could report it in the node core repo in addition to this workaround PR.

[st3ffgv4]

@ronag before report,
i was checking nodejs documentation about http2

there's one example here: https://nodejs.org/api/http2.html#clienthttp2sessionrequestheaders-options

const http2 = require('node:http2');
const clientSession = http2.connect('https://localhost:1234');
const {
  HTTP2_HEADER_PATH,
  HTTP2_HEADER_STATUS,
} = http2.constants;

const req = clientSession.request({ [HTTP2_HEADER_PATH]: '/' });
req.on('response', (headers) => {
  console.log(headers[HTTP2_HEADER_STATUS]);
  req.on('data', (chunk) => { /* .. */ });
  req.on('end', () => { /* .. */ });
});

the data event listener is binded inside the response listener callback.
So i have moved client-h2.js stream on data event inside stream on response (at the and of the callback) and it seems to be ok now

 stream.once('response', headers => {
    const { [HTTP2_HEADER_STATUS]: statusCode, ...realHeaders } = headers
    request.onResponseStarted()

    if (request.onHeaders(Number(statusCode), realHeaders, stream.resume.bind(stream), '') === false) {
      stream.pause()
    }

    stream.on('data', (chunk) => {
      if (request.onData(chunk) === false) {
        stream.pause()
      }
    })
  })

my test script is:

const Agent = require("./lib/dispatcher/agent");
const { fetch } = require("./lib/web/fetch");

const dispatcher = new Agent({ allowH2: true });

(async () => {
  while (true) {
    //console.log("fetching 1000 requests");
    //console.time();
    await Promise.all(
      Array.from({ length: 1000 }).map(async () => {
        try {
          const response = await fetch("https://http2.pro/api/v1", {
            dispatcher,
          });
          const text = await response.text();
          //console.log(text);
          if(!text) throw new Error("NOT GOOD")
        } catch (e) {
          console.error(e);
        }
      }),
    );
    //console.timeEnd();
  }
})();

This make more sense to you?
Thanks

[ronag]

Make sense, but yea, please open an issue because this seems wrong.

[st3ffgv4]

Issue opened

[metcoder95]

Can we add the link to the issue as comment?

[mcollina]

lgtm