[Phillip9587]

Compressed requests can leave body-parser waiting indefinitely when the client disconnects mid-body. req.pipe() does not propagate premature request closure to the decoder that raw-body reads.

Listen for the request’s close event and notify raw-body when the request closes before its body is fully read. Ignore normal completion so pending decompression can finish, and remove the listener when the decoder closes.

[Copilot]

🟢 Approval recommended

No active review issues remain for the compressed-request abort-handling fix.

No open issues 🎉

What changed in this PR

Fixes body-parser hanging when a client disconnects during a compressed request by forwarding premature closure to the decoder read by raw-body.

Changes:

• Notifies the decoder of aborted requests while allowing normal decompression to finish.
• Removes the request-close listener when the decoder closes.
• Adds regression tests for abort handling, verification behavior, listener cleanup, and single-callback behavior across all four parsers.

File	Description
test/​urlencoded.js	Adds compressed-request lifecycle tests for URL-encoded parsing.
test/​text.js	Adds compressed-request lifecycle tests for text parsing.
test/​raw.js	Adds compressed-request lifecycle tests for raw-body parsing.
test/​json.js	Adds lifecycle tests, including cleanup after decompression errors.
lib/​read.js	Propagates premature request closure to the decoder and cleans up the listener.

_{Review effort: Balanced}

[kilisamemarisaaa]

Independent real-HTTP verification of head 337fca67a4b54772818648064329daa0bbdac5a2 against its merge base b4744d739e1e2e5257f2ddac31eed191aa5152a4 (Windows x64, Node 24.12.0, body-parser 2.4.0).

I specifically checked the other decoders and normal request completion while decompression is still pending:

Real HTTP matrix	Merge base	PR head
Mid-frame disconnects: 4 parsers × gzip/deflate/br/zstd × Content-Length/chunked × verify on/off	64/64 still had zero middleware callbacks 350 ms after the actual request close	64/64 settle once with 400 request.aborted / ECONNABORTED
Complete compressed bodies, same combinations	64/64 pass	64/64 pass
Identity controls: complete/aborted × 4 parsers × 2 framings × verify on/off	32/32 pass	32/32 pass

The 64 complete compressed cases all observed the request's close with complete === true and readableEnded === true before the middleware callback. They then parsed the expected body and, where enabled, verified the exact decompressed buffer. This exercises the normal-close guard rather than only the abort branch. All head cases also preserve a pre-existing close listener, remove temporary listeners, and call middleware exactly once over the observed lifecycle; aborts leave req.body undefined and never call verify.

Separately, npm test passes 298 tests, with the 4 expected "no zstandard support" cases pending on this zstd-capable Node; npm run lint passes. No source modifications. I did not test other platforms/Node versions, HTTP/2, sustained load, or repeated requests on one keep-alive connection. This is additional experimental evidence, not a claim that every lifecycle is covered. Review experiments assisted by Codex.

Reproducible standalone probe

Save as matrix.cjs, install dependencies in the checkout, then run node matrix.cjs /absolute/path/to/checkout. I used the same installed dependency tree for both revisions (NODE_PATH pointed at the head checkout's node_modules for the base), so the comparison isolates the source revision. Exit status is 1 for the base's 64 failures and 0 for the head's 160 passes. The 350 ms bound is a post-close observation window, not a claim about an infinite hang.

'use strict'

const assert = require('node:assert/strict')
const http = require('node:http')
const zlib = require('node:zlib')
const { once } = require('node:events')
const { setTimeout: delay, setImmediate: tick } = require('node:timers/promises')
const bodyParser = require(process.argv[2])

const codecs = {
  identity: data => data,
  gzip: zlib.gzipSync,
  deflate: zlib.deflateSync,
  br: zlib.brotliCompressSync
}
if (zlib.zstdCompressSync) codecs.zstd = zlib.zstdCompressSync
const text = 'preserve-actual-body-\u4e2d\u6587-'.repeat(4096)
const specifications = {
  json: { type: 'application/json', payload: JSON.stringify({ text }), expected: { text } },
  text: { type: 'text/plain; charset=utf-8', payload: text, expected: text },
  raw: { type: 'application/octet-stream', payload: text, expected: Buffer.from(text) },
  urlencoded: { type: 'application/x-www-form-urlencoded', payload: `text=${encodeURIComponent(text)}`, expected: { text } }
}

async function runCase (parserName, codec, framing, verify, mode) {
  const spec = specifications[parserName]
  const plain = Buffer.from(spec.payload)
  const compressed = codecs[codec](plain)
  let calls = 0
  let verified = 0
  let closes = 0
  let incoming
  let error
  let initialListeners
  let client
  let resolveMiddleware
  let resolveClosed
  let closeState
  const middlewareDone = new Promise(resolve => { resolveMiddleware = resolve })
  const requestClosed = new Promise(resolve => { resolveClosed = resolve })
  const parser = bodyParser[parserName]({
    limit: '1mb',
    ...(verify ? {
      verify (req, res, buffer) {
        verified++
        assert.deepEqual(buffer, plain)
      }
    } : {})
  })
  const onExistingClose = () => { closes++ }
  const server = http.createServer((req, res) => {
    incoming = req
    req.on('close', onExistingClose)
    req.once('close', () => {
      closeState = { complete: req.complete, ended: req.readableEnded, callbacks: calls }
      resolveClosed()
    })
    initialListeners = req.listeners('close').filter(listener => listener === onExistingClose)
    if (mode === 'abort') req.once('data', () => { client.destroy() })
    parser(req, res, err => {
      calls++
      error = err
      if (mode === 'success') {
        res.statusCode = err ? 400 : 200
        res.end('done')
      }
      resolveMiddleware()
    })
  })
  let deadline
  try {
    server.listen(0, '127.0.0.1')
    await once(server, 'listening')
    const headers = { 'Content-Type': spec.type, 'Content-Encoding': codec }
    if (framing === 'length') headers['Content-Length'] = compressed.length
    else headers['Transfer-Encoding'] = 'chunked'
    const responseDone = new Promise((resolve, reject) => {
      client = http.request({ host: '127.0.0.1', port: server.address().port, method: 'POST', agent: false, headers }, res => {
        res.resume()
        res.once('end', () => resolve(res.statusCode))
        res.once('error', reject)
      })
      client.on('error', err => {
        if (mode === 'success') reject(err)
      })
    })
    // Prevent an infrastructure failure from keeping the fixture alive indefinitely.
    deadline = setTimeout(() => { client.destroy(new Error('fixture deadline')); server.closeAllConnections() }, 5000)
    if (mode === 'abort') {
      // Send a real incomplete frame and only disconnect after the server receives it.
      client.write(compressed.subarray(0, Math.max(1, Math.floor(compressed.length / 2))))
      await requestClosed
      const completed = await Promise.race([middlewareDone.then(() => true), delay(350).then(() => false)])
      assert.equal(completed, true, 'middleware did not settle within 350ms after request close')
    } else {
      client.end(compressed)
      assert.equal(await responseDone, 200)
      await requestClosed
    }
    await tick()
    await tick()
    assert.equal(calls, 1)
    assert.equal(closes, 1)
    assert.deepEqual(incoming.listeners('close'), initialListeners)
    if (mode === 'abort') {
      assert.equal(error?.type, 'request.aborted')
      assert.equal(error?.status, 400)
      assert.equal(error?.code, 'ECONNABORTED')
      assert.equal(incoming.complete, false)
      assert.equal(incoming.destroyed, true)
      assert.equal(incoming.body, undefined)
      assert.equal(verified, 0)
    } else {
      assert.equal(error, undefined)
      assert.deepEqual(incoming.body, spec.expected)
      assert.equal(verified, verify ? 1 : 0)
      assert.equal(closeState.complete, true)
      assert.equal(closeState.ended, true)
    }
    return { pass: true, closeState }
  } catch (err) {
    return { pass: false, message: err.message, calls, verified, closes, closeState }
  } finally {
    clearTimeout(deadline)
    client?.destroy()
    server.closeAllConnections()
    await new Promise(resolve => server.close(resolve))
  }
}

;(async () => {
  console.log(JSON.stringify({ source: process.argv[2], node: process.version, platform: process.platform, arch: process.arch, codecs: Object.keys(codecs) }))
  const counts = { total: 0, passed: 0, failed: 0, compressedAbortFailed: 0, completeBeforeParse: 0 }
  for (const parser of Object.keys(specifications)) {
    for (const codec of Object.keys(codecs)) {
      for (const framing of ['length', 'chunked']) {
        for (const verify of [false, true]) {
          for (const mode of ['abort', 'success']) {
            const result = await runCase(parser, codec, framing, verify, mode)
            counts.total++
            counts[result.pass ? 'passed' : 'failed']++
            if (!result.pass && codec !== 'identity' && mode === 'abort') counts.compressedAbortFailed++
            if (result.pass && mode === 'success' && result.closeState.callbacks === 0) counts.completeBeforeParse++
            console.log(JSON.stringify({ parser, codec, framing, verify, mode, ...result }))
          }
        }
      }
    }
  }
  console.log(JSON.stringify({ summary: counts }))
  process.exitCode = counts.failed ? 1 : 0
})().catch(err => { console.error(err); process.exitCode = 1 })