fix(stream): remove request-owned drain listener on writable completion

lib/api/api-stream.js

@@ -164,12 +164,16 @@ class StreamHandler extends AsyncResource {
       throw new InvalidReturnValueError('expected Writable')
     }
 
+    const onDrain = () => controller.resume()
+
     trackWritableLifecycle(res, (err, fromErrorEvent) => {
-      const { callback, res, opaque, trailers, abort } = this
+      const { callback, res: writable, opaque, trailers, abort } = this
+
+      res.removeListener('drain', onDrain)
 
       this.res = null
-      if (err || !res?.readable) {
-        util.destroy(res, fromErrorEvent ? undefined : err)
+      if (err || !writable?.readable) {
+        util.destroy(writable, fromErrorEvent ? undefined : err)
       }
 
       this.callback = null
@@ -180,7 +184,7 @@ class StreamHandler extends AsyncResource {
       }
     })
 
-    res.on('drain', () => controller.resume())
+    res.on('drain', onDrain)
 
     this.res = res
 

test/client-stream-drain-retention.js

@@ -0,0 +1,178 @@
+'use strict'
+
+const assert = require('node:assert/strict')
+const { channel } = require('node:diagnostics_channel')
+const { once } = require('node:events')
+const { createServer, get } = require('node:http')
+const { createSecureServer } = require('node:http2')
+const { Writable } = require('node:stream')
+const { test } = require('node:test')
+const { setImmediate: nextTurn, setTimeout: delay } = require('node:timers/promises')
+
+const pem = require('@metcoder95/https-pem')
+const { Client } = require('..')
+
+test('api-stream.js / stream should release unread H2 buffers when a retained ServerResponse closes', { timeout: 10000, skip: !global.gc && 'requires --expose-gc' }, async t => {
+  let nativeReference
+  let firstSession
+  let reusedSession = false
+  let sessionCount = 0
+  let heldResponse
+  let browser = null
+  let client = null
+  const sessions = new Set()
+  const terminal = Promise.withResolvers()
+  const userDrain = () => {}
+  const created = channel('http2.client.stream.created')
+  const onCreated = ({ stream, headers }) => {
+    if (headers[':path'] === '/cancel') {
+      nativeReference = new WeakRef(stream)
+      firstSession = stream.session
+    } else {
+      reusedSession = stream.session === firstSession
+    }
+  }
+  created.subscribe(onCreated)
+
+  const upstream = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
+  upstream.on('session', session => {
+    sessionCount++
+    sessions.add(session)
+    session.on('error', () => {})
+    session.once('close', () => sessions.delete(session))
+  })
+  upstream.on('stream', (stream, headers) => {
+    stream.on('error', () => {})
+    stream.respond({ ':status': 200 })
+    if (headers[':path'] !== '/cancel') {
+      stream.end('healthy')
+      return
+    }
+    const chunk = Buffer.alloc(16 * 1024)
+    let remaining = 2 * 1024 * 1024
+    const write = () => {
+      while (!stream.destroyed && remaining > 0) {
+        remaining -= chunk.length
+        if (!stream.write(chunk)) {
+          stream.once('drain', write)
+          return
+        }
+      }
+      if (!stream.destroyed) stream.end()
+    }
+    write()
+  })
+
+  const downstream = createServer((request, response) => {
+    // The response deliberately outlives the request, as an application owner may.
+    heldResponse = response
+    // Force backpressure independently of the platform's socket buffer capacity.
+    response.cork()
+    response.on('drain', userDrain)
+    client.stream({ path: '/cancel', method: 'GET' }, () => response)
+      .then(() => terminal.resolve(null), error => terminal.resolve(error.code))
+  })
+
+  t.after(async () => {
+    created.unsubscribe(onCreated)
+    browser?.destroy()
+    heldResponse?.destroy()
+    firstSession?.destroy()
+    for (const session of sessions) session.destroy()
+    await client?.destroy()
+    downstream.closeAllConnections()
+    await Promise.all([closeServer(downstream), closeServer(upstream)])
+  })
+  await once(upstream.listen(0, '127.0.0.1'), 'listening')
+  client = new Client(`https://127.0.0.1:${upstream.address().port}`, {
+    allowH2: true,
+    connect: { rejectUnauthorized: false }
+  })
+  await once(downstream.listen(0, '127.0.0.1'), 'listening')
+  browser = get(`http://127.0.0.1:${downstream.address().port}`)
+  browser.on('error', () => {})
+  browser.on('response', response => {
+    response.on('error', () => {})
+    response.pause()
+  })
+
+  await waitFor(() => heldResponse?.writableNeedDrain && unreadBytes(nativeReference) > 0)
+  const unreadBeforeAbort = unreadBytes(nativeReference)
+  browser.destroy()
+  assert.equal(await bounded(terminal.promise), 'ERR_STREAM_PREMATURE_CLOSE')
+  await collect()
+  const unreadAfterTerminal = unreadBytes(nativeReference)
+  const ownedDrainCount = heldResponse.listeners('drain').filter(listener => listener !== userDrain).length
+
+  // This causal control changes only the retained response's request-owned listener.
+  // It must release the unread native stream without closing the reused H2 session.
+  removeRequestDrain(heldResponse, userDrain)
+  await collect()
+  const unreadAfterListenerRemoval = unreadBytes(nativeReference)
+  assert.equal(unreadAfterListenerRemoval, 0)
+  assert.deepEqual(heldResponse.listeners('drain'), [userDrain])
+
+  let healthyBody = ''
+  const healthy = new Writable({
+    write (chunk, encoding, callback) {
+      healthyBody += chunk.toString()
+      callback()
+    }
+  })
+  healthy.on('drain', userDrain)
+  await bounded(client.stream({ path: '/healthy', method: 'GET' }, () => healthy))
+  assert.equal(healthyBody, 'healthy')
+  assert.equal(reusedSession, true)
+  assert.equal(sessionCount, 1)
+  const healthyOwnedDrainCount = healthy.listeners('drain').filter(listener => listener !== userDrain).length
+
+  t.diagnostic(JSON.stringify({ unreadBeforeAbort, unreadAfterTerminal, unreadAfterListenerRemoval, ownedDrainCount, healthyOwnedDrainCount, sessionCount, reusedSession }))
+  assert.equal(ownedDrainCount, 0, 'a terminal request must remove its own drain listener')
+  assert.equal(unreadAfterTerminal, 0, 'a retained response must not retain unread native buffers')
+  assert.deepEqual(healthy.listeners('drain'), [userDrain])
+  const closed = once(firstSession, 'close')
+  firstSession.close()
+  await bounded(closed)
+})
+
+function unreadBytes (reference) {
+  return reference?.deref()?.readableLength || 0
+}
+
+function removeRequestDrain (response, userDrain) {
+  for (const listener of response.listeners('drain')) {
+    if (listener !== userDrain) response.removeListener('drain', listener)
+  }
+}
+
+async function collect () {
+  for (let i = 0; i < 3; i++) {
+    await nextTurn()
+    global.gc()
+  }
+  await nextTurn()
+}
+
+async function waitFor (predicate) {
+  const deadline = Date.now() + 3000
+  while (!predicate()) {
+    assert.ok(Date.now() < deadline, 'the fixture must reach native unread bytes and downstream backpressure')
+    await delay(10)
+  }
+}
+
+async function bounded (promise) {
+  let timer
+  try {
+    return await Promise.race([promise, new Promise((resolve, reject) => { timer = setTimeout(() => reject(new Error('fixture timed out')), 2000) })])
+  } finally {
+    clearTimeout(timer)
+  }
+}
+
+async function closeServer (server) {
+  if (!server.listening) return
+  const closed = once(server, 'close')
+  server.close()
+  await bounded(closed)
+}

test/http2-abort-retention.js

@@ -0,0 +1,138 @@
+'use strict'
+
+const assert = require('node:assert/strict')
+const { channel } = require('node:diagnostics_channel')
+const { once } = require('node:events')
+const { createSecureServer } = require('node:http2')
+const { Writable } = require('node:stream')
+const { test } = require('node:test')
+const { setTimeout: delay } = require('node:timers/promises')
+
+const pem = require('@metcoder95/https-pem')
+const { Client } = require('..')
+
+test('should release unread H2 response streams and reuse their session when the destination closes under backpressure', { timeout: 8000 }, async t => {
+  const sessions = new Set()
+  let sessionCount = 0
+  let requests = 0
+  let firstNative
+  let firstSession
+  let reusedSession = false
+  const nativeClosed = Promise.withResolvers()
+  const clientSessionClosed = Promise.withResolvers()
+  const wroteResponse = Promise.withResolvers()
+  const bodyBytes = 256 * 1024
+  const observation = { bodyBytes, unreadBeforeAbort: 0, requests: 0, sessionCount: 0, reusedSession: false, abortedCode: null, healthyStatus: null, nativeClosed: false, gracefulClosed: false }
+  const created = channel('http2.client.stream.created')
+  const onCreated = ({ stream, headers }) => {
+    if (headers[':path'] === '/abort-retention') {
+      firstNative = new WeakRef(stream)
+      firstSession = stream.session
+      stream.once('close', () => {
+        observation.nativeClosed = true
+        nativeClosed.resolve()
+      })
+      firstSession.once('close', () => {
+        observation.gracefulClosed = true
+        clientSessionClosed.resolve()
+      })
+    } else if (headers[':path'] === '/healthy-retention') {
+      reusedSession = stream.session === firstSession
+    }
+  }
+  created.subscribe(onCreated)
+
+  const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
+  server.on('error', () => {})
+  server.on('secureConnection', socket => socket.on('error', () => {}))
+  server.on('session', session => {
+    sessionCount++
+    sessions.add(session)
+    session.on('error', () => {})
+    session.once('close', () => sessions.delete(session))
+  })
+  server.on('stream', (stream, headers) => {
+    requests++
+    stream.on('error', () => {})
+    stream.respond({ ':status': 200 })
+    stream.end(headers[':path'] === '/abort-retention' ? Buffer.alloc(bodyBytes) : 'healthy')
+  })
+
+  let client = null
+  t.after(async () => {
+    observation.requests = requests
+    observation.sessionCount = sessionCount
+    observation.reusedSession = reusedSession
+    t.diagnostic(JSON.stringify(observation))
+    created.unsubscribe(onCreated)
+    firstSession?.destroy()
+    for (const session of sessions) session.destroy()
+    client?.destroy().catch(() => {})
+    if (server.listening) {
+      const closed = once(server, 'close')
+      server.close()
+      await bounded(closed, 1000, 'fixture cleanup timed out')
+    }
+  })
+  await once(server.listen(0, '127.0.0.1'), 'listening')
+  client = new Client(`https://127.0.0.1:${server.address().port}`, {
+    allowH2: true,
+    connect: { rejectUnauthorized: false }
+  })
+
+  const destination = new Writable({
+    highWaterMark: 1,
+    write (chunk, encoding, callback) {
+      wroteResponse.resolve()
+      // Leaving this write pending gives the consumer real backpressure.
+    }
+  })
+  const abortedResponse = client.stream({ path: '/abort-retention', method: 'GET' }, () => destination)
+    .then(() => null, error => error.code)
+  await bounded(wroteResponse.promise, 2000, 'response body did not reach destination')
+  const deadline = Date.now() + 2000
+  while ((observation.unreadBeforeAbort = unreadBytes(firstNative)) === 0 && Date.now() < deadline) {
+    await delay(10)
+  }
+  assert.ok(observation.unreadBeforeAbort > 0, 'the native response must contain unread bytes before abort')
+  destination.destroy()
+  observation.abortedCode = await bounded(abortedResponse, 1000, 'destination close did not settle the request')
+  assert.equal(observation.abortedCode, 'ERR_STREAM_PREMATURE_CLOSE')
+
+  let healthyBody = ''
+  const healthy = await bounded(client.stream({ path: '/healthy-retention', method: 'GET' }, ({ statusCode }) => {
+    observation.healthyStatus = statusCode
+    assert.equal(statusCode, 200)
+    return new Writable({
+      write (chunk, encoding, callback) {
+        healthyBody += chunk.toString()
+        callback()
+      }
+    })
+  }), 2000, 'healthy request did not settle')
+  assert.ok(healthy)
+  assert.equal(healthyBody, 'healthy')
+  assert.equal(requests, 2)
+  assert.equal(sessionCount, 1)
+  assert.equal(reusedSession, true)
+
+  await bounded(nativeClosed.promise, 1000, 'aborted native response remained attached after a healthy request')
+  firstSession.close()
+  await bounded(clientSessionClosed.promise, 1000, 'the reused H2 session did not close gracefully')
+})
+
+function unreadBytes (reference) {
+  return reference?.deref()?.readableLength || 0
+}
+
+async function bounded (promise, millis, message) {
+  let timer
+  try {
+    return await Promise.race([
+      promise,
+      new Promise((resolve, reject) => { timer = setTimeout(() => reject(new Error(message)), millis) })
+    ])
+  } finally {
+    clearTimeout(timer)
+  }
+}