fix(h2): complete request stream on 'end' instead of waiting for 'close'

lib/dispatcher/client-h2.js

@@ -678,18 +678,22 @@ function onUpgradeStreamClose () {
   closeStreamSession(this)
 }
 
-function onRequestStreamClose () {
+// Idempotent terminal cleanup, called from both 'end' and 'close': the
+// null-state guard no-ops the later call.
+function completeRequestStream () {
   const state = this[kRequestStreamState]
 
-  if (state) {
-    // Release the stream first so request references are cleared,
-    // then complete the response with trailers if available.
-    releaseRequestStream(this)
+  if (state == null) {
+    return
+  }
 
-    if (state.pendingEnd && !state.request.aborted && !state.request.completed) {
-      state.request.onResponseEnd(state.trailers || {})
-      finalizeRequest(state)
-    }
+  // Release the stream first so request references are cleared,
+  // then complete the response with trailers if available.
+  releaseRequestStream(this)
+
+  if (state.pendingEnd && !state.request.aborted && !state.request.completed) {
+    state.request.onResponseEnd(state.trailers || {})
+    finalizeRequest(state)
   }
 
   closeStreamSession(this)
@@ -1127,7 +1131,7 @@ function writeH2 (client, request) {
   }
 
   stream[kHTTP2Session] = session
-  stream.on('close', onRequestStreamClose)
+  stream.on('close', completeRequestStream)
 
   bindRequestToStream(request, stream, releaseRequestStream)
   if (expectContinue) {
@@ -1270,13 +1274,16 @@ function onEnd () {
 
   stream.off('end', onEnd)
 
-  // If we received a response, this is a normal completion.
-  // Defer actual completion to onRequestStreamClose so that
-  // onTrailers (which may fire after 'end' on Windows) can
-  // store trailers first.
+  // onTrailers (which may fire after 'end' on Windows) has already stored
+  // trailers on the state by now, so completing here still delivers them.
   if (state.responseReceived) {
     if (!request.aborted && !request.completed) {
       state.pendingEnd = true
+
+      // Complete on 'end': a blocked event loop can keep the stream's 'close'
+      // from firing, stranding its buffers until OOM. Idempotent, so a later
+      // 'close' no-ops.
+      completeRequestStream.call(stream)
     }
   } else {
     // Stream ended without receiving a response - this is an error
@@ -1347,7 +1354,7 @@ function onTrailers (trailers) {
     return
   }
 
-  // Store trailers for onRequestStreamClose to use when completing
+  // Store trailers for completeRequestStream to use when completing
   state.trailers = trailers
 }
 

test/http2-late-data.js

@@ -164,6 +164,169 @@ test('Should ignore late http2 data after request completion', async (t) => {
   await t.completed
 })
 
+test('Should complete the response and release the stream on end without a close event', async (t) => {
+  // Repro for the completion-path leak: on a busy, long-lived multiplexed
+  // session the native 'close' event can fail to fire. Cleanup (finalize,
+  // listener removal, session stream-count decrement) is gated on 'close', so
+  // completed-but-not-closed streams pin their request graph and buffers until
+  // OOM. This drives a normal completion through 'end' and never emits 'close'.
+  t = tspl(t, { plan: 7 })
+
+  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
+  })
+
+  let onCompleteCalls = 0
+  let endTrailers = null
+
+  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 request = new Request('https://localhost', {
+    path: '/',
+    method: 'GET',
+    headers: {}
+  }, {
+    onRequestStart () {},
+    onResponseStart () {},
+    onResponseData () {},
+    onResponseEnd (_controller, trailers) {
+      onCompleteCalls++
+      endTrailers = trailers
+    },
+    onResponseError (_controller, err) {
+      t.ifError(err)
+    }
+  })
+
+  client[kQueue].push(request)
+
+  t.ok(context.write(request))
+  t.equal(stream.listenerCount('trailers'), 1)
+
+  stream.emit('response', { ':status': 200 })
+  stream.emit('trailers', { 'x-trailer': 'hello' })
+  stream.emit('data', Buffer.from('body'))
+  stream.emit('end')
+  // Intentionally no stream.emit('close').
+
+  t.strictEqual(onCompleteCalls, 1)
+  t.strictEqual(endTrailers?.['x-trailer'], 'hello')
+  t.equal(stream.listenerCount('trailers'), 0)
+  t.equal(stream.listenerCount('aborted'), 0)
+  t.equal(stream.listenerCount('timeout'), 0)
+
+  await t.completed
+})
+
+test('Should complete only once when both end and a late close fire', async (t) => {
+  // Guards the idempotency of completing on 'end': when 'close' still arrives
+  // afterwards, onRequestStreamClose must be a no-op -- no second onResponseEnd
+  // and no double session open-stream decrement.
+  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
+  })
+
+  let onCompleteCalls = 0
+
+  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 request = new Request('https://localhost', {
+    path: '/',
+    method: 'GET',
+    headers: {}
+  }, {
+    onRequestStart () {},
+    onResponseStart () {},
+    onResponseData () {},
+    onResponseEnd () {
+      onCompleteCalls++
+    },
+    onResponseError (_controller, err) {
+      t.ifError(err)
+    }
+  })
+
+  client[kQueue].push(request)
+
+  t.ok(context.write(request))
+
+  stream.emit('response', { ':status': 200 })
+  stream.emit('end')
+  t.strictEqual(onCompleteCalls, 1)
+
+  // A late 'close' must not complete the request a second time.
+  t.doesNotThrow(() => {
+    stream.emit('close')
+  })
+  t.strictEqual(onCompleteCalls, 1)
+
+  await t.completed
+})
+
 test('Should remove request-owned http2 stream listeners after completion', async (t) => {
   t = tspl(t, { plan: 7 })