fix(HTTP/2): handle consumption of aborted request (#2387)

Author: metcoder95

lib/client.js

@@ -1658,19 +1658,6 @@ function writeH2 (client, session, request) {
     return false
   }
 
-  try {
-    // TODO(HTTP/2): Should we call onConnect immediately or on stream ready event?
-    request.onConnect((err) => {
-      if (request.aborted || request.completed) {
-        return
-      }
-
-      errorRequest(client, request, err || new RequestAbortedError())
-    })
-  } catch (err) {
-    errorRequest(client, request, err)
-  }
-
   if (request.aborted) {
     return false
   }
@@ -1682,9 +1669,34 @@ function writeH2 (client, session, request) {
   headers[HTTP2_HEADER_AUTHORITY] = host || client[kHost]
   headers[HTTP2_HEADER_METHOD] = method
 
+  try {
+    // We are already connected, streams are pending.
+    // We can call on connect, and wait for abort
+    request.onConnect((err) => {
+      if (request.aborted || request.completed) {
+        return
+      }
+
+      err = err || new RequestAbortedError()
+
+      if (stream != null) {
+        util.destroy(stream, err)
+
+        h2State.openStreams -= 1
+        if (h2State.openStreams === 0) {
+          session.unref()
+        }
+      }
+
+      errorRequest(client, request, err)
+    })
+  } catch (err) {
+    errorRequest(client, request, err)
+  }
+
   if (method === 'CONNECT') {
     session.ref()
-    // we are already connected, streams are pending, first request
+    // We are already connected, streams are pending, first request
     // will create a new stream. We trigger a request to create the stream and wait until
     // `ready` event is triggered
     // We disabled endStream to allow the user to write to the stream

test/fetch/http2.js

@@ -13,7 +13,7 @@ const { Client, fetch, Headers } = require('../..')
 
 const nodeVersion = Number(process.version.split('v')[1].split('.')[0])
 
-plan(7)
+plan(8)
 
 test('[Fetch] Issue#2311', async t => {
   const expectedBody = 'hello from client!'
@@ -413,3 +413,65 @@ test('Issue#2415', async (t) => {
 
   t.doesNotThrow(() => new Headers(response.headers))
 })
+
+test('Issue #2386', async t => {
+  const server = createSecureServer(pem)
+  const body = Buffer.from('hello')
+  const requestChunks = []
+  const expectedResponseBody = { hello: 'h2' }
+  const controller = new AbortController()
+  const signal = controller.signal
+
+  server.on('stream', async (stream, headers) => {
+    t.equal(headers[':method'], 'PUT')
+    t.equal(headers[':path'], '/')
+    t.equal(headers[':scheme'], 'https')
+
+    stream.on('data', chunk => requestChunks.push(chunk))
+
+    stream.respond({
+      'content-type': 'application/json',
+      'x-custom-h2': headers['x-my-header'],
+      ':status': 200
+    })
+
+    stream.end(JSON.stringify(expectedResponseBody))
+  })
+
+  t.plan(4)
+
+  server.listen(0)
+  await once(server, 'listening')
+
+  const client = new Client(`https://localhost:${server.address().port}`, {
+    connect: {
+      rejectUnauthorized: false
+    },
+    allowH2: true
+  })
+
+  t.teardown(server.close.bind(server))
+  t.teardown(client.close.bind(client))
+
+  try {
+    await fetch(
+      `https://localhost:${server.address().port}/`,
+      // Needs to be passed to disable the reject unauthorized
+      {
+        body,
+        signal,
+        method: 'PUT',
+        dispatcher: client,
+        headers: {
+          'x-my-header': 'foo',
+          'content-type': 'text-plain'
+        }
+      }
+    )
+
+    controller.abort()
+    t.pass()
+  } catch (error) {
+    t.error(error)
+  }
+})