fix(#2364): concurrent aborts (#3005)

metcoder95 · web-flow · commit fa5bcb4597cf · 2024-03-27T19:13:41.000+01:00
* fix: concurrent aborts

* refactor: wording
diff --git 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()
     }
diff --git 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
+})
