Repository: nodejs/undici

Problem statement:
[HTTP/2] Aborted requests for same domain might fail unexpectedly
## Bug Description

As the title, pipelined requests might fail unexpectedly if another request toward the same domain has been aborted:

```
node:assert:399
    throw err;
    ^

AssertionError [ERR_ASSERTION]: The expression evaluated to a falsy value:

  assert(!this.aborted)

    at Request.onHeaders (/home/runner/undici-h2-promise-any/node_modules/undici/lib/core/request.js:235:5)
    at ClientHttp2Stream.<anonymous> (/home/runner/undici-h2-promise-any/node_modules/undici/lib/client.js:1804:17)
    at Object.onceWrapper (node:events:628:26)
    at ClientHttp2Stream.emit (node:events:513:28)
    at emit (node:internal/http2/core:330:3)
    at process.processTicksAndRejections (node:internal/process/task_queues:85:22) {
  generatedMessage: true,
  code: 'ERR_ASSERTION',
  actual: false,
  expected: true,
  operator: '=='
}

Node.js v18.16.1
```

## Reproducible By

https://replit.com/@isukkaw/undici-h2-promise-any

```js
const undici = require('undici');

undici.setGlobalDispatcher(new undici.Agent({
  allowH2: true,
  pipelining: 10
}));

const { fetch } = undici;

const fetchUbuntuXenialReleaseGpg = () => {
  const controller = new AbortController();
  return Promise.any(
    [
      'https://mirror-cdn.xtom.com/ubuntu/dists/xenial/Release.gpg',
      'https://mirrors.xtom.com/ubuntu/dists/xenial/Release.gpg',
      'https://mirrors.xtom.com.hk/ubuntu/dists/xenial/Release.gpg',
      'https://mirrors.xtom.de/ubuntu/dists/xenial/Release.gpg',
      'https://mirrors.xtom.nl/ubuntu/dists/xenial/Release.gpg',
      'https://mirrors.xtom.ee/ubuntu/dists/xenial/Release.gpg',
      'https://mirrors.xtom.jp/ubuntu/dists/xenial/Release.gpg',
      'https://mirrors.xtom.au/ubuntu/dists/xenial/Release.gpg'
    ].map(async url => {
      try {
        const resp = await fetch(url, { signal: controller.signal });
        const text = await resp.text();
        // The body should have been consumed, abort the requests
        controller.abort();
        return text;
      } catch(e) {
        if (e.name !== 'AbortError') {
          console.error('Request Fail:', url)
          throw e;
        }
      }
    })
  );
};

const fetchDebianBookwormReleaseGpg = () => {
  const controller = new AbortController();
  return Promise.any(
    [
      'https://mirror-cdn.xtom.com/debian/dists/bookworm/Release.gpg',
      'https://mirrors.xtom.com/debian/dists/bookworm/Release.gpg',
      'https://mirrors.xtom.com.hk/debian/dists/bookworm/Release.gpg',
      'https://mirrors.xtom.de/debian/dists/bookworm/Release.gpg',
      'https://mirrors.xtom.nl/debian/dists/bookworm/Release.gpg',
      'https://mirrors.xtom.ee/debian/dists/bookworm/Release.gpg',
      'https://mirrors.xtom.jp/debian/dists/bookworm/Release.gpg',
      'https://mirrors.xtom.au/debian/dists/bookworm/Release.gpg'
    ].map(async url => {
      try {
        const resp = await fetch(url, { signal: controller.signal });
        const text = await resp.text();
        // The body should have been consumed, abort the requests
        controller.abort();
        return text;
      } catch(e) {
        if (e.name !== 'AbortError') {
          console.error('Request Fail:', url)
          throw e;
        }
      }
    })
  );
};

(async () => {
  const [ubuntu, debian] = await Promise.all([
    fetchUbuntuXenialReleaseGpg(),
    fetchDebianBookwormReleaseGpg()
  ]);

  console.log({
    ubuntu: ubuntu.length,
    debian: debian.length
  })
})();
```

## Expected Behavior

`AssertionError [ERR_ASSERTION]: The expression evaluated to a falsy value: assert(!this.aborted)` should never happened.

## Logs & Screenshots

```
(node:3251) [UNDICI-H2] Warning: H2 support is experimental, expect them to change at any time.
(Use `node --trace-warnings ...` to show where the warning was created)
node:assert:399
    throw err;
    ^

AssertionError [ERR_ASSERTION]: The expression evaluated to a falsy value:

  assert(!this.aborted)

    at Request.onHeaders (/home/runner/undici-h2-promise-any/node_modules/undici/lib/core/request.js:235:5)
    at ClientHttp2Stream.<anonymous> (/home/runner/undici-h2-promise-any/node_modules/undici/lib/client.js:1804:17)
    at Object.onceWrapper (node:events:628:26)
    at ClientHttp2Stream.emit (node:events:513:28)
    at emit (node:internal/http2/core:330:3)
    at process.processTicksAndRejections (node:internal/process/task_queues:85:22) {
  generatedMessage: true,
  code: 'ERR_ASSERTION',
  actual: false,
  expected: true,
  operator: '=='
}

Node.js v18.16.1
```

## Environment

Node.js 18.16.0 on Ubuntu 22.04.1



Hints:
@metcoder95 
Started doing some research about, and bug is in fact quite complex; it seems that somewhere we are reusing a `request` instance that was already aborted to handle a second request. Haven't found the root cause yet. I'll made more research later on
The repro SukkaW provided still reproduces even with Pipelining disabled, maybe this issue should be re-titled? 

Trying to swap to undici in our software, I'm able to get this stack trace, we only issue one request to a dummy /debug/stream endpoint which pushes one sse per second and abort it after a few seconds. 

but we receive exactly one more chunk afterwards, resulting in this assertion failure

```
[10/25/2023, 6:25:20 PM] [PLATFORM_FETCH] Sending GET: https://localhost:8020/debug/stream undefined (GET /debug/stream YR+xTDLhXJa6jIE13bX3WWaFzDQx) 
undici init path /debug/stream
undici SUCCESS connected {"host":"localhost:8020","hostname":"localhost","protocol":"https:","port":"8020","servername":null,"localAddress":null}
(node:3957036) [UNDICI-H2] Warning: H2 support is experimental, expect them to change at any time.
(Use `node --trace-warnings ...` to show where the warning was created)
[10/25/2023, 6:25:20 PM] [PLATFORM_FETCH] Started response (200) for request GET (https://localhost:8020/debug/stream) undefined (GET /debug/stream YR+xTDLhXJa6jIE13bX3WWaFzDQx) 
data from in  client /debug/stream <Buffer 7b 22 30 22 3a 22 31 36 39 38 32 37 32 37 32 31 34 39 34 22 7d 0a 0a>
data from in  client /debug/stream <Buffer 7b 22 31 22 3a 22 31 36 39 38 32 37 32 37 32 32 34 39 34 22 7d 0a 0a>
data from in  client /debug/stream <Buffer 7b 22 32 22 3a 22 31 36 39 38 32 37 32 37 32 33 34 39 35 22 7d 0a 0a>
aborting
undici request error This operation was aborted /debug/stream
[10/25/2023, 6:25:24 PM] [PLATFORM_FETCH] Finished response (closed stream via abort controller) for request (GET https://localhost:8020/debug/stream YR+xTDLhXJa6jIE13bX3WWaFzDQx)
data from in  client /debug/stream <Buffer 7b 22 33 22 3a 22 31 36 39 38 32 37 32 37 32 34 34 39 34 22 7d 0a 0a>
Trace: Aborted! /debug/stream <Buffer 7b 22 33 22 3a 22 31 36 39 38 32 37 32 37 32 34 34 39 34 22 7d 0a 0a>

node:internal/event_target:912
  process.nextTick(() => { throw err; });
                           ^

AssertionError [ERR_ASSERTION]: The expression evaluated to a falsy value:

  assert2(!this.aborted)

    at Request2.onData (/nix/store/wvxs5i3mvh1lbvay44i13d24yy9rn95k-esbuild_node/depengine_worker.js:4815:9)
    at ClientHttp2Stream.<anonymous> (/nix/store/wvxs5i3mvh1lbvay44i13d24yy9rn95k-esbuild_node/depengine_worker.js:7177:21)
    at ClientHttp2Stream.emit (node:events:527:28)
    at addChunk (node:internal/streams/readable:324:12)
    at readableAddChunk (node:internal/streams/readable:297:9)
    at ClientHttp2Stream.Readable.push (node:internal/streams/readable:234:10)
    at Http2Stream.onStreamRead (node:internal/stream_base_commons:190:23)
Emitted 'error' event on Worker instance at:
    at Worker.[kOnErrorMessage] (node:internal/worker:289:10)
    at Worker.[kOnMessage] (node:internal/worker:300:37)
    at MessagePort.<anonymous> (node:internal/worker:201:57)
    at MessagePort.[nodejs.internal.kHybridDispatch] (node:internal/event_target:643:20)
    at MessagePort.exports.emitMessage (node:internal/per_context/messageport:23:28)

Node.js v17.9.0
```

Patch:
diff --git a/lib/dispatcher/client-h2.js b/lib/dispatcher/client-h2.js
--- a/lib/dispatcher/client-h2.js
+++ b/lib/dispatcher/client-h2.js
@@ -391,6 +391,18 @@ function writeH2 (client, request) {
     const { [HTTP2_HEADER_STATUS]: statusCode, ...realHeaders } = headers
     request.onResponseStarted()
 
+    // Due to the stream nature, it is possible we face a race condition
+    // where the stream has been assigned, but the request has been aborted
+    // the request remains in-flight and headers hasn't been received yet
+    // for those scenarios, best effort is to destroy the stream immediately
+    // as there's no value to keep it open.
+    if (request.aborted || request.completed) {
+      const err = new RequestAbortedError()
+      errorRequest(client, request, err)
+      util.destroy(stream, err)
+      return
+    }
+
     if (request.onHeaders(Number(statusCode), realHeaders, stream.resume.bind(stream), '') === false) {
       stream.pause()
     }
@@ -426,7 +438,6 @@ function writeH2 (client, request) {
 
   stream.once('close', () => {
     session[kOpenStreams] -= 1
-    // TODO(HTTP/2): unref only if current streams count is 0
     if (session[kOpenStreams] === 0) {
       session.unref()
     }


Test patch:
diff --git a/test/http2.js b/test/http2.js
--- a/test/http2.js
+++ b/test/http2.js
@@ -1294,3 +1294,88 @@ test('Should throw informational error on half-closed streams (remote)', async t
     t.strictEqual(err.code, 'UND_ERR_INFO')
   })
 })
+
+test('#2364 - Concurrent aborts', async t => {
+  const server = createSecureServer(pem)
+
+  server.on('stream', (stream, headers, _flags, rawHeaders) => {
+    t.strictEqual(headers['x-my-header'], 'foo')
+    t.strictEqual(headers[':method'], 'GET')
+    setTimeout(() => {
+      stream.respond({
+        'content-type': 'text/plain; charset=utf-8',
+        'x-custom-h2': 'hello',
+        ':status': 200
+      })
+      stream.end('hello h2!')
+    }, 100)
+  })
+
+  server.listen(0)
+  await once(server, 'listening')
+
+  const client = new Client(`https://localhost:${server.address().port}`, {
+    connect: {
+      rejectUnauthorized: false
+    },
+    allowH2: true
+  })
+
+  t = tspl(t, { plan: 18 })
+  after(() => server.close())
+  after(() => client.close())
+  const controller = new AbortController()
+
+  client.request({
+    path: '/',
+    method: 'GET',
+    headers: {
+      'x-my-header': 'foo'
+    }
+  }, (err, response) => {
+    t.ifError(err)
+    t.strictEqual(response.headers['content-type'], 'text/plain; charset=utf-8')
+    t.strictEqual(response.headers['x-custom-h2'], 'hello')
+    t.strictEqual(response.statusCode, 200)
+    response.body.dump()
+  })
+
+  client.request({
+    path: '/',
+    method: 'GET',
+    headers: {
+      'x-my-header': 'foo'
+    },
+    signal: controller.signal
+  }, (err, response) => {
+    t.strictEqual(err.name, 'AbortError')
+  })
+
+  client.request({
+    path: '/',
+    method: 'GET',
+    headers: {
+      'x-my-header': 'foo'
+    }
+  }, (err, response) => {
+    t.ifError(err)
+    t.strictEqual(response.headers['content-type'], 'text/plain; charset=utf-8')
+    t.strictEqual(response.headers['x-custom-h2'], 'hello')
+    t.strictEqual(response.statusCode, 200)
+  })
+
+  client.request({
+    path: '/',
+    method: 'GET',
+    headers: {
+      'x-my-header': 'foo'
+    },
+    signal: controller.signal
+  }, (err, response) => {
+    t.strictEqual(err.name, 'AbortError')
+  })
+
+  controller.abort()
+
+  await t.completed
+})