fix(h2): release completed requests from queue

lib/dispatcher/client-h2.js

@@ -157,9 +157,10 @@ function completeRequest (client, request, resetPendingIdx = false) {
   const queue = client[kQueue]
   const runningIdx = client[kRunningIdx]
 
-  // In-order completion: advance the running index instead of splicing.
-  // The client's resume loop compacts the queue once the index grows.
+  // In-order completion: clear the request and advance without splicing.
+  // The client's resume loop compacts cleared slots once the index grows.
   if (runningIdx < client[kPendingIdx] && queue[runningIdx] === request) {
+    queue[runningIdx] = null
     client[kRunningIdx] = runningIdx + 1
     return
   }
@@ -686,9 +687,9 @@ function completeRequestStream () {
 
   if (state.pendingEnd && !state.request.aborted && !state.request.completed) {
     state.request.onResponseEnd(state.trailers || {})
-    finalizeRequest(state)
   }
 
+  finalizeRequest(state)
   closeStreamSession(this)
   this[kRequestStreamState] = null
 }

test/fetch/issue-5566.js

@@ -0,0 +1,71 @@
+'use strict'
+/* global WeakRef */
+
+const { once } = require('node:events')
+const { createSecureServer } = require('node:http2')
+const { test } = require('node:test')
+
+const pem = require('@metcoder95/https-pem')
+
+const { Client, fetch } = require('../..')
+const { closeClientAndServerAsPromise } = require('../utils/node-http')
+
+// https://github.com/nodejs/undici/issues/5566
+test('an aborted fetch is released while its HTTP/2 session remains open', { timeout: 30000, skip: global.gc == null }, async (t) => {
+  let sessionCount = 0
+  const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
+  server.on('session', () => {
+    sessionCount++
+  })
+  server.on('stream', (stream, headers) => {
+    stream.on('error', () => {})
+    stream.respond({ ':status': 200 })
+
+    if (headers[':path'] === '/abort') {
+      stream.write('partial')
+    } else {
+      stream.end('ok')
+    }
+  })
+
+  server.listen(0)
+  await once(server, 'listening')
+
+  const origin = `https://localhost:${server.address().port}`
+  const client = new Client(origin, {
+    allowH2: true,
+    connect: { rejectUnauthorized: false }
+  })
+  t.after(closeClientAndServerAsPromise(client, server))
+
+  async function abortFetch () {
+    const reason = { message: 'abort reason' }
+    const reasonRef = new WeakRef(reason)
+    const controller = new AbortController()
+    const response = await fetch(`${origin}/abort`, {
+      dispatcher: client,
+      signal: controller.signal
+    })
+
+    controller.abort(reason)
+    await response.text().catch(() => {})
+
+    return reasonRef
+  }
+
+  const reasonRef = await abortFetch()
+  const response = await fetch(`${origin}/complete`, { dispatcher: client })
+  t.assert.strictEqual(await response.text(), 'ok')
+  t.assert.strictEqual(sessionCount, 1)
+
+  for (let i = 0; i < 20 && reasonRef.deref() !== undefined; i++) {
+    await new Promise(resolve => setImmediate(resolve))
+    global.gc()
+  }
+
+  t.assert.strictEqual(
+    reasonRef.deref(),
+    undefined,
+    'the completed request queue retained the aborted fetch'
+  )
+})

test/http2-late-data.js

@@ -401,3 +401,74 @@ test('Should remove request-owned http2 stream listeners after completion', asyn
 
   await t.completed
 })
+
+test('Should finalize an already-aborted request when its stream closes', async (t) => {
+  t = tspl(t, { plan: 4 })
+
+  const http2 = require('node:http2')
+  const originalConnect = http2.connect
+
+  const stream = new FakeStream()
+  const session = new FakeSession(stream)
+
+  http2.connect = function connectStub () {
+    return session
+  }
+
+  after(() => {
+    http2.connect = originalConnect
+  })
+
+  const client = {
+    [kUrl]: new URL('https://localhost'),
+    [kSocket]: null,
+    [kMaxConcurrentStreams]: 100,
+    [kHTTP2InitialWindowSize]: null,
+    [kHTTP2ConnectionWindowSize]: null,
+    [kBodyTimeout]: 30_000,
+    [kStrictContentLength]: true,
+    [kQueue]: [],
+    [kRunningIdx]: 0,
+    [kPendingIdx]: 0,
+    [kRunning]: 1,
+    [kPingInterval]: 0,
+    [kOnError] (err) {
+      t.ifError(err)
+    },
+    [kResume] () {},
+    emit () {},
+    destroyed: false
+  }
+
+  const context = connectH2(client, new FakeSocket())
+  const expectedError = new Error('aborted')
+  let onCompleteCalls = 0
+
+  const request = new Request('https://localhost', {
+    path: '/',
+    method: 'GET',
+    headers: {}
+  }, {
+    onRequestStart () {},
+    onResponseStart () {},
+    onResponseData () {},
+    onResponseEnd () {
+      onCompleteCalls++
+    },
+    onResponseError (_controller, err) {
+      t.strictEqual(err, expectedError)
+    }
+  })
+
+  client[kQueue].push(request)
+  client[kPendingIdx] = 1
+
+  t.ok(context.write(request))
+  request.onResponseError(expectedError)
+  stream.emit('close')
+
+  t.strictEqual(onCompleteCalls, 0)
+  t.strictEqual(client[kQueue][0], null)
+
+  await t.completed
+})