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

lib/internal/streams/writable.js

@@ -845,6 +845,16 @@ 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' &&
+             (state[kState] & (kErrored | kWriting | kBuffered)) === 0 &&
+             state.length === 0) {
+    // The stream is already ending and has no in-flight write, buffered
+    // data, or stored error that would surface a real error to the user.
+    // Synthesize ERR_STREAM_WRITE_AFTER_END so the user-supplied callback
+    // fires consistently with `.write(chunk, cb)` after end. When pending
+    // work or a stored error exists, fall through so the cb is queued on
+    // kOnFinished and receives that real error instead.
+    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');
+  }));
+}