stream: disallow stream methods on finished stream

Invoke callback with ERR_STREAM_ALREADY_FINISHED error if `end()` is
called on a finished stream.

PR-URL: https://github.com/nodejs/node/pull/28687
Refs: https://github.com/nodejs/node/issues/28667
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

doc/api/errors.md

@@ -1703,6 +1703,12 @@ An attempt was made to call [`stream.pipe()`][] on a [`Writable`][] stream.
 A stream method was called that cannot complete because the stream was
 destroyed using `stream.destroy()`.
 
+<a id="ERR_STREAM_ALREADY_FINISHED"></a>
+### ERR_STREAM_ALREADY_FINISHED
+
+A stream method was called that cannot complete because the stream was
+finished.
+
 <a id="ERR_STREAM_NULL_VALUES"></a>
 ### ERR_STREAM_NULL_VALUES
 

lib/_http_outgoing.js

@@ -46,6 +46,7 @@ const {
     ERR_INVALID_CHAR,
     ERR_METHOD_NOT_IMPLEMENTED,
     ERR_STREAM_CANNOT_PIPE,
+    ERR_STREAM_ALREADY_FINISHED,
     ERR_STREAM_WRITE_AFTER_END
   },
   hideStackFrames
@@ -704,6 +705,13 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
   }
 
   if (this.finished) {
+    if (typeof callback === 'function') {
+      if (!this.writableFinished) {
+        this.on('finish', callback);
+      } else {
+        callback(new ERR_STREAM_ALREADY_FINISHED('end'));
+      }
+    }
     return this;
   }
 

lib/_stream_writable.js

@@ -41,6 +41,7 @@ const {
   ERR_MULTIPLE_CALLBACK,
   ERR_STREAM_CANNOT_PIPE,
   ERR_STREAM_DESTROYED,
+  ERR_STREAM_ALREADY_FINISHED,
   ERR_STREAM_NULL_VALUES,
   ERR_STREAM_WRITE_AFTER_END,
   ERR_UNKNOWN_ENCODING
@@ -591,6 +592,13 @@ Writable.prototype.end = function(chunk, encoding, cb) {
   // Ignore unnecessary end() calls.
   if (!state.ending)
     endWritable(this, state, cb);
+  else if (typeof cb === 'function') {
+    if (!state.finished) {
+      this.once('finish', cb);
+    } else {
+      cb(new ERR_STREAM_ALREADY_FINISHED('end'));
+    }
+  }
 
   return this;
 };

lib/internal/errors.js

@@ -1128,6 +1128,9 @@ E('ERR_SOCKET_DGRAM_NOT_RUNNING', 'Not running', Error);
 E('ERR_SRI_PARSE',
   'Subresource Integrity string %s had an unexpected at %d',
   SyntaxError);
+E('ERR_STREAM_ALREADY_FINISHED',
+  'Cannot call %s after a stream was finished',
+  Error);
 E('ERR_STREAM_CANNOT_PIPE', 'Cannot pipe, not readable', Error);
 E('ERR_STREAM_DESTROYED', 'Cannot call %s after a stream was destroyed', Error);
 E('ERR_STREAM_NULL_VALUES', 'May not write null values to stream', TypeError);

test/parallel/test-http-outgoing-end-multiple.js

@@ -0,0 +1,27 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const http = require('http');
+
+const server = http.createServer(common.mustCall(function(req, res) {
+  res.end('testing ended state', common.mustCall());
+  res.end(common.mustCall());
+  res.on('finish', common.mustCall(() => {
+    res.end(common.mustCall((err) => {
+      assert.strictEqual(err.code, 'ERR_STREAM_ALREADY_FINISHED');
+      server.close();
+    }));
+  }));
+}));
+
+server.listen(0);
+
+server.on('listening', common.mustCall(function() {
+  http
+    .request({
+      port: server.address().port,
+      method: 'GET',
+      path: '/'
+    })
+    .end();
+}));

test/parallel/test-stream-writable-end-multiple.js

@@ -0,0 +1,20 @@
+'use strict';
+
+const common = require('../common');
+
+const assert = require('assert');
+const stream = require('stream');
+
+const writable = new stream.Writable();
+
+writable._write = (chunk, encoding, cb) => {
+  setTimeout(() => cb(), 10);
+};
+
+writable.end('testing ended state', common.mustCall());
+writable.end(common.mustCall());
+writable.on('finish', common.mustCall(() => {
+  writable.end(common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_ALREADY_FINISHED');
+  }));
+}));