stream: pass ERR_STREAM_WRITE_AFTER_END to .end(cb) after end

`stream.end(cb)` called on a stream that has already been ended did
not propagate ERR_STREAM_WRITE_AFTER_END to the callback. Instead,
the callback was queued for the eventual `finish` event and called
with `null`, which was inconsistent with `.write(chunk, cb)` and
`.end(chunk, cb)` after end, both of which correctly call `cb` with
ERR_STREAM_WRITE_AFTER_END.

In `Writable.prototype.end`, when the stream is already ending and
not yet finished or destroyed and a user callback was supplied,
set the error to ERR_STREAM_WRITE_AFTER_END so the existing
`process.nextTick(cb, err)` path delivers it. The forgiving
behavior of `.end()` without a callback being called multiple
times is preserved (no call to errorOrDestroy).

Fixes: #33684
Signed-off-by: Maruthan G <[email protected]>

Author: maruthang

lib/internal/streams/writable.js

@@ -845,6 +845,14 @@ Writable.prototype.end = function(chunk, encoding, cb) {
     err = new ERR_STREAM_ALREADY_FINISHED('end');
   } else if ((state[kState] & kDestroyed) !== 0) {
     err = new ERR_STREAM_DESTROYED('end');
+  } else if ((state[kState] & kEnding) !== 0 && typeof cb === 'function') {
+    // The stream is already ending (or ended) but not yet finished or
+    // destroyed. For consistency with `.write(chunk, cb)` after end (which
+    // invokes cb with ERR_STREAM_WRITE_AFTER_END), surface the same error
+    // through the user-supplied callback. Note: we do not call
+    // errorOrDestroy here in order to preserve the pre-existing forgiving
+    // behavior of `.end()` (no cb) called multiple times.
+    err = new ERR_STREAM_WRITE_AFTER_END();
   }
 
   if (typeof cb === 'function') {

test/parallel/test-stream-writable-end-cb-after-end.js

@@ -0,0 +1,74 @@
+'use strict';
+
+// Regression test for https://github.com/nodejs/node/issues/33684
+// Calling `.end(cb)` on a stream that has already been ended must invoke
+// the user-supplied callback with ERR_STREAM_WRITE_AFTER_END, mirroring
+// the behavior of `.write(chunk, cb)` after end and `.end(chunk, cb)`
+// after end.
+
+const common = require('../common');
+const assert = require('assert');
+const { Writable } = require('stream');
+
+{
+  // `.end()` followed by `.end(cb)` — cb must receive ERR_STREAM_WRITE_AFTER_END.
+  const w = new Writable({
+    write(chunk, encoding, cb) { cb(); },
+  });
+
+  w.end();
+  w.end(common.mustCall((err) => {
+    assert.ok(err);
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
+  }));
+}
+
+{
+  // `.end()` followed by `.end('chunk', cb)` — cb must receive
+  // ERR_STREAM_WRITE_AFTER_END (existing behavior, included for parity).
+  const w = new Writable({
+    write(chunk, encoding, cb) { cb(); },
+  });
+
+  w.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
+  }));
+
+  w.end();
+  w.end('chunk', common.mustCall((err) => {
+    assert.ok(err);
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
+  }));
+}
+
+{
+  // `.write(chunk, cb)` after end calls cb with ERR_STREAM_WRITE_AFTER_END —
+  // sanity check confirming that `.end(cb)` is now consistent with this path.
+  const w = new Writable({
+    write(chunk, encoding, cb) { cb(); },
+  });
+
+  w.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
+  }));
+
+  w.end();
+  w.write('chunk', common.mustCall((err) => {
+    assert.ok(err);
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
+  }));
+}
+
+{
+  // `.end(null, encoding, cb)` after end — cb must receive
+  // ERR_STREAM_WRITE_AFTER_END.
+  const w = new Writable({
+    write(chunk, encoding, cb) { cb(); },
+  });
+
+  w.end();
+  w.end(null, null, common.mustCall((err) => {
+    assert.ok(err);
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
+  }));
+}