diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js index f310cd6936fb..3b7d346527c9 100644 --- a/lib/internal/streams/end-of-stream.js +++ b/lib/internal/streams/end-of-stream.js @@ -155,7 +155,13 @@ function eos(stream, options, callback) { * Error: an error occurred */ let immediateResult; + // An already closed stream can still have data sitting in its readable + // buffer. Whether that counts as a premature close depends on whether the + // buffer is consumed before the callback runs, so the result has to be + // recomputed at that point rather than captured here. + let recomputeOnCloseError = false; if (isClosed(stream)) { + recomputeOnCloseError = true; immediateResult = getEosOnCloseError( stream, readable, @@ -209,7 +215,12 @@ function eos(stream, options, callback) { } if (immediateResult !== undefined) { - process.nextTick(() => ReflectApply(callback, stream, immediateResult === null ? [] : [immediateResult])); + process.nextTick(() => { + const result = recomputeOnCloseError ? + getEosOnCloseError(stream, readable, readableFinished, writable, writableFinished) : + immediateResult; + ReflectApply(callback, stream, result === null ? [] : [result]); + }); return cleanup; } diff --git a/test/parallel/test-http2-async-iteration-buffered-response.js b/test/parallel/test-http2-async-iteration-buffered-response.js new file mode 100644 index 000000000000..73e651e23e13 --- /dev/null +++ b/test/parallel/test-http2-async-iteration-buffered-response.js @@ -0,0 +1,61 @@ +'use strict'; + +// An HTTP/2 stream that closed normally can still hold its whole response in +// the readable buffer. Starting async iteration at that point must yield the +// buffered body rather than reporting ERR_STREAM_PREMATURE_CLOSE. +// Refs: https://github.com/nodejs/node/issues/65677 + +const common = require('../common'); +if (!common.hasCrypto) + common.skip('missing crypto'); + +const assert = require('assert'); +const http2 = require('http2'); +const { once } = require('events'); +const { setImmediate: setImmediateAsync } = require('timers/promises'); + +const body = 'complete response'; + +const server = http2.createServer(); +server.on('stream', (stream) => { + stream.resume(); + stream.on('end', () => { + stream.respond({ ':status': 200 }, { waitForTrailers: true }); + stream.on('wantTrailers', () => stream.sendTrailers({ 'grpc-status': '0' })); + stream.end(body); + }); +}); + +server.listen(0, '127.0.0.1'); + +(async () => { + await once(server, 'listening'); + const session = http2.connect(`http://127.0.0.1:${server.address().port}`); + + try { + await once(session, 'connect'); + const request = session.request({ ':method': 'POST' }); + const response = once(request, 'response'); + request.end('request'); + await response; + + // Hold off consuming the response until the stream reports a normal close. + // The body is buffered at this point and the stream is not destroyed. + while (!request.closed) + await setImmediateAsync(); + + assert.strictEqual(request.rstCode, 0); + assert.strictEqual(request.destroyed, false); + assert.strictEqual(request.readableLength, Buffer.byteLength(body)); + + let received = ''; + for await (const chunk of request) + received += chunk; + + assert.strictEqual(received, body); + } finally { + session.destroy(); + await once(session, 'close'); + server.close(); + } +})().then(common.mustCall());