diff --git a/doc/api/stream.md b/doc/api/stream.md index cf03163218bfbc..b2f121f62604e0 100644 --- a/doc/api/stream.md +++ b/doc/api/stream.md @@ -1525,6 +1525,11 @@ Especially useful in error handling scenarios where a stream is destroyed prematurely (like an aborted HTTP request), and will not emit `'end'` or `'finish'`. +`stream.finished()` will error with `ERR_STREAM_PREMATURE_CLOSE` if: +* `Writable` emits `'close'` before `'finish'` and +[`writable.writableFinished`][]. +* `Readable` emits `'close'` before `'end'` and [`readable.readableEnded`][]. + The `finished` API is promisify-able as well; ```js @@ -1647,6 +1652,10 @@ run().catch(console.error); * `Readable` streams which have emitted `'end'` or `'close'`. * `Writable` streams which have emitted `'finish'` or `'close'`. +If any `Writable` or `Readable` stream emits `'close'` without being able to +fully flush or drain, `stream.pipeline()` will error with +`ERR_STREAM_PREMATURE_CLOSE`. + `stream.pipeline()` leaves dangling event listeners on the streams after the `callback` has been invoked. In the case of reuse of streams after failure, this can cause event listener leaks and swallowed errors. @@ -2865,6 +2874,7 @@ contain multi-byte characters. [`process.stdout`]: process.html#process_process_stdout [`readable._read()`]: #stream_readable_read_size_1 [`readable.push('')`]: #stream_readable_push +[`readable.readableEnded`]: #stream_readable_readableended [`readable.setEncoding()`]: #stream_readable_setencoding_encoding [`stream.Readable.from()`]: #stream_stream_readable_from_iterable_options [`stream.cork()`]: #stream_writable_cork diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js index 6fa3540057be33..73d18b19bb53dc 100644 --- a/lib/_stream_readable.js +++ b/lib/_stream_readable.js @@ -147,6 +147,8 @@ function ReadableState(options, stream, isDuplex) { // Indicates whether the stream has errored. this.errored = false; + this.closed = false; + // Crypto is kind of old and crusty. Historically, its default string // encoding is 'binary' so we have to make this configurable. // Everything else in the universe uses 'utf8', though. diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js index 3f029883705a0a..6d65abebcb030c 100644 --- a/lib/_stream_writable.js +++ b/lib/_stream_writable.js @@ -175,6 +175,8 @@ function WritableState(options, stream, isDuplex) { // is disabled we need a way to tell whether the stream has failed. this.errored = false; + this.closed = false; + // Count buffered requests this.bufferedRequestCount = 0; diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js index 8755e22de0d00f..39ce3bbedd198e 100644 --- a/lib/internal/streams/async_iterator.js +++ b/lib/internal/streams/async_iterator.js @@ -122,16 +122,6 @@ const ReadableStreamAsyncIteratorPrototype = ObjectSetPrototypeOf({ return() { return new Promise((resolve, reject) => { const stream = this[kStream]; - - // TODO(ronag): Remove this check once finished() handles - // already ended and/or destroyed streams. - const ended = stream.destroyed || stream.readableEnded || - (stream._readableState && stream._readableState.endEmitted); - if (ended) { - resolve(createIterResult(undefined, true)); - return; - } - finished(stream, (err) => { if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') { reject(err); diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js index b80fb56a6bf324..3115202ca6bdee 100644 --- a/lib/internal/streams/destroy.js +++ b/lib/internal/streams/destroy.js @@ -73,6 +73,13 @@ function emitCloseNT(self) { const r = self._readableState; const w = self._writableState; + if (w) { + w.closed = true; + } + if (r) { + r.closed = true; + } + if ((w && w.emitClose) || (r && r.emitClose)) { self.emit('close'); } @@ -103,6 +110,7 @@ function undestroy() { if (r) { r.destroyed = false; r.errored = false; + r.closed = false; r.reading = false; r.ended = false; r.endEmitted = false; @@ -112,6 +120,7 @@ function undestroy() { if (w) { w.destroyed = false; w.errored = false; + w.closed = false; w.ended = false; w.ending = false; w.finalCalled = false; diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js index bb01e800becdf3..0388533dac0167 100644 --- a/lib/internal/streams/end-of-stream.js +++ b/lib/internal/streams/end-of-stream.js @@ -13,6 +13,18 @@ function isRequest(stream) { return stream.setHeader && typeof stream.abort === 'function'; } +function isReadable(stream) { + return typeof stream.readable === 'boolean' || + typeof stream.readableEnded === 'boolean' || + !!stream._readableState; +} + +function isWritable(stream) { + return typeof stream.writable === 'boolean' || + typeof stream.writableEnded === 'boolean' || + !!stream._writableState; +} + function eos(stream, opts, callback) { if (arguments.length === 2) { callback = opts; @@ -28,43 +40,72 @@ function eos(stream, opts, callback) { callback = once(callback); - let readable = opts.readable || (opts.readable !== false && stream.readable); - let writable = opts.writable || (opts.writable !== false && stream.writable); + const onerror = (err) => { + callback.call(stream, err); + }; + + const w = stream._writableState; + const r = stream._readableState; + + let writableFinished = stream.writableFinished || (w && w.finished); + let readableEnded = stream.readableEnded || (r && r.endEmitted); + + const errorEmitted = (w && w.errorEmitted) || (r && r.errorEmitted); + const closed = (w && w.closed) || (r && r.closed); + + if (writableFinished || readableEnded || errorEmitted || closed) { + // TODO(ronag): rethrow if errorEmitted? + // TODO(ronag): premature close if closed but not + // errored, finished or ended? + + // Swallow any error past this point. + if (opts.error !== false) stream.on('error', onerror); + + process.nextTick(callback.bind(stream)); + + return () => { + stream.removeListener('error', onerror); + }; + } + + const readable = opts.readable || + (opts.readable !== false && isReadable(stream)); + const writable = opts.writable || + (opts.writable !== false && isWritable(stream)); const onlegacyfinish = () => { if (!stream.writable) onfinish(); }; - let writableEnded = stream._writableState && stream._writableState.finished; const onfinish = () => { - writable = false; - writableEnded = true; - if (!readable) callback.call(stream); + writableFinished = true; + if (!readable || readableEnded) callback.call(stream); }; - let readableEnded = stream.readableEnded || - (stream._readableState && stream._readableState.endEmitted); const onend = () => { - readable = false; readableEnded = true; - if (!writable) callback.call(stream); - }; - - const onerror = (err) => { - callback.call(stream, err); + if (!writable || writableFinished) callback.call(stream); }; const onclose = () => { - let err; - if (readable && !readableEnded) { - if (!stream._readableState || !stream._readableState.ended) - err = new ERR_STREAM_PREMATURE_CLOSE(); - return callback.call(stream, err); - } - if (writable && !writableEnded) { - if (!stream._writableState || !stream._writableState.ended) - err = new ERR_STREAM_PREMATURE_CLOSE(); - return callback.call(stream, err); + if (readable) { + const ended = (stream._readableState && + stream._readableState.endEmitted) || stream.readableEnded; + if (!ended && !readableEnded) { + callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE()); + } else if (!readableEnded) { + // Compat. Stream has ended but haven't emitted 'end'. + callback.call(stream); + } + } else if (writable) { + const finished = (stream._writableState && + stream._writableState.finished) || stream.writableFinished; + if (!finished && !writableFinished) { + callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE()); + } else if (!writableFinished) { + // Compat. Stream has finished but haven't emitted 'finish'. + callback.call(stream); + } } }; diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js index e0834171bfb8fc..9a3521f523b54e 100644 --- a/lib/internal/streams/pipeline.js +++ b/lib/internal/streams/pipeline.js @@ -46,6 +46,29 @@ function destroyer(stream, reading, writing, callback) { if (eos === undefined) eos = require('internal/streams/end-of-stream'); eos(stream, { readable: reading, writable: writing }, (err) => { + if ( + err && + err.code === 'ERR_STREAM_PREMATURE_CLOSE' && + reading && + (stream._readableState && stream._readableState.ended) + ) { + // Some readable streams will emit 'close' before 'end'. However, since + // this is on the readable side 'end' should still be emitted if the + // stream has been ended and no error emitted. This should be allowed in + // favor of backwards compatibility. Since the stream is piped to a + // destination this should not result in any observable difference. + // We don't need to check if this is a writable premature close since + // eos will only fail with premature close on the reading side for + // duplex streams. + stream + .on('end', () => { + closed = true; + callback(); + }) + .on('error', callback); + return; + } + if (err) return callback(err); closed = true; callback(); diff --git a/test/parallel/test-http-client-finished.js b/test/parallel/test-http-client-finished.js index 2d7e5b95b3ca33..8eddf8660b4476 100644 --- a/test/parallel/test-http-client-finished.js +++ b/test/parallel/test-http-client-finished.js @@ -25,3 +25,109 @@ const { finished } = require('stream'); .end(); })); } + +{ + // Test abort before finished. + + const server = http.createServer(function(req, res) { + }); + + server.listen(0, common.mustCall(function() { + const req = http.request({ + port: this.address().port + }, common.mustNotCall()); + req.abort(); + finished(req, common.mustCall(() => { + server.close(); + })); + })); +} + +{ + // Test abort after request. + + const server = http.createServer(function(req, res) { + }); + + server.listen(0, common.mustCall(function() { + const req = http.request({ + port: this.address().port + }).end(); + finished(req, (err) => { + common.expectsError({ + name: 'Error', + code: 'ERR_STREAM_PREMATURE_CLOSE' + })(err); + finished(req, common.mustCall(() => { + server.close(); + })); + }); + req.abort(); + })); +} + +{ + // Test abort before end. + + const server = http.createServer(function(req, res) { + res.write('test'); + }); + + server.listen(0, common.mustCall(function() { + const req = http.request({ + port: this.address().port + }).on('response', common.mustCall((res) => { + req.abort(); + finished(res, common.mustCall(() => { + finished(res, common.mustCall(() => { + server.close(); + })); + })); + })).end(); + })); +} + +{ + // Test destroy before end. + + const server = http.createServer(function(req, res) { + res.write('test'); + }); + + server.listen(0, common.mustCall(function() { + http.request({ + port: this.address().port + }).on('response', common.mustCall((res) => { + // TODO(ronag): Bug? Won't emit 'close' unless read. + res.on('data', () => {}); + res.destroy(); + finished(res, common.mustCall(() => { + finished(res, common.mustCall(() => { + server.close(); + })); + })); + })).end(); + })); +} + +{ + // Test finish after end. + + const server = http.createServer(function(req, res) { + res.end('asd'); + }); + + server.listen(0, common.mustCall(function() { + http.request({ + port: this.address().port + }).on('response', common.mustCall((res) => { + // TODO(ronag): Bug? Won't emit 'close' unless read. + res.on('data', () => {}); + finished(res, common.mustCall(() => { + finished(res, common.mustCall(() => { + server.close(); + })); + })); + })).end(); + })); +} diff --git a/test/parallel/test-stream-finished.js b/test/parallel/test-stream-finished.js index d6361ea303635d..b6db7a867606a4 100644 --- a/test/parallel/test-stream-finished.js +++ b/test/parallel/test-stream-finished.js @@ -97,6 +97,18 @@ const { promisify } = require('util'); })); } +{ + const rs = new Readable(); + + finished(rs, common.mustCall((err) => { + assert(err, 'premature close error'); + })); + + rs.push(null); + rs.emit('close'); + rs.resume(); +} + { const rs = new Readable(); @@ -105,7 +117,9 @@ const { promisify } = require('util'); })); rs.push(null); - rs.emit('close'); // Should not trigger an error + rs.on('end', common.mustCall(() => { + rs.emit('close'); // Should not trigger an error + })); rs.resume(); } @@ -155,8 +169,9 @@ const { promisify } = require('util'); rs.resume(); } -// Test that calling returned function removes listeners { + // Nothing happens if disposed. + const ws = new Writable({ write(data, env, cb) { cb(); @@ -168,6 +183,8 @@ const { promisify } = require('util'); } { + // Nothing happens if disposed. + const rs = new Readable(); const removeListeners = finished(rs, common.mustNotCall()); removeListeners(); @@ -178,9 +195,93 @@ const { promisify } = require('util'); } { + // Completed if readable-like is ended before. + + let ticked = false; const streamLike = new EE(); streamLike.readableEnded = true; streamLike.readable = true; - finished(streamLike, common.mustCall); + finished(streamLike, common.mustCall(() => { + assert.strictEqual(ticked, true); + })); + ticked = true; +} + +{ + // Completed if readable-like is never ended. + + const streamLike = new EE(); + streamLike.readableEnded = false; + streamLike.readable = true; + finished(streamLike, common.expectsError({ + code: 'ERR_STREAM_PREMATURE_CLOSE' + })); + streamLike.emit('close'); +} + +{ + // Completed if streamlike is finished before. + + const streamLike = new EE(); + streamLike.writableFinished = true; + streamLike.writable = true; + finished(streamLike, common.mustCall()); +} + +{ + // Premature close if stream is not finished. + + const streamLike = new EE(); + streamLike.writableFinished = false; + streamLike.writable = true; + finished(streamLike, common.expectsError({ + code: 'ERR_STREAM_PREMATURE_CLOSE' + })); + streamLike.emit('close'); +} + +{ + // No error if stream never emitted 'end' + // even if readableEnded says something else. + + const streamLike = new EE(); + streamLike.writable = true; + finished(streamLike, common.mustCall((err) => { + assert.strictEqual(err, undefined); + })); + streamLike.writableFinished = true; + streamLike.emit('close'); +} + +{ + // No error if stream never emitted 'finished' + // even if writableFinished says something else. + + const streamLike = new EE(); + streamLike.readable = true; + finished(streamLike, common.mustCall((err) => { + assert.strictEqual(err, undefined); + })); + streamLike.readableEnded = true; streamLike.emit('close'); } + +{ + // Completes if already finished. + + const w = new Writable(); + finished(w, common.mustCall(() => { + finished(w, common.mustCall()); + })); + w.destroy(); +} + +{ + // Completes if already ended. + + const r = new Readable(); + finished(r, common.mustCall(() => { + finished(r, common.mustCall()); + })); + r.destroy(); +} diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js index 19fc246e2bf3cd..99876738415c69 100644 --- a/test/parallel/test-stream-pipeline.js +++ b/test/parallel/test-stream-pipeline.js @@ -912,4 +912,35 @@ const { promisify } = require('util'); }, common.mustCall((err) => { assert.strictEqual(err.message, 'kaboom'); })); + // Make sure 'close' before 'end' finishes without error + // if readable has received eof. + // Ref: https://github.com/nodejs/node/issues/29699 + const r = new Readable(); + const w = new Writable({ + write(chunk, encoding, cb) { + cb(); + } + }); + pipeline(r, w, (err) => { + assert.strictEqual(err, undefined); + }); + r.push(null); + r.destroy(); +} + +{ + // Make sure 'close' before 'end' finishes without error + // if readable has received eof. + // Ref: https://github.com/nodejs/node/issues/29699 + const r = new Readable(); + const w = new Writable({ + write(chunk, encoding, cb) { + cb(); + } + }); + pipeline(r, w, (err) => { + assert.strictEqual(err, undefined); + }); + r.push(null); + r.emit('close'); }