stream: fix writable.end callback behavior

doc/api/stream.md

@@ -412,6 +412,9 @@ Is `true` after [`writable.destroy()`][writable-destroy] has been called.
 <!-- YAML
 added: v0.9.4
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/34101
+    description: The `callback` is invoked before 'finish' or on error.
   - version: v14.0.0
     pr-url: https://github.com/nodejs/node/pull/29747
     description: The `callback` is invoked if 'finish' or 'error' is emitted.
@@ -428,15 +431,13 @@ changes:
   `Uint8Array`. For object mode streams, `chunk` may be any JavaScript value
   other than `null`.
 * `encoding` {string} The encoding if `chunk` is a string
-* `callback` {Function} Optional callback for when the stream finishes
-  or errors
+* `callback` {Function} Callback for when the stream is finished.
 * Returns: {this}
 
 Calling the `writable.end()` method signals that no more data will be written
 to the [`Writable`][]. The optional `chunk` and `encoding` arguments allow one
 final additional chunk of data to be written immediately before closing the
-stream. If provided, the optional `callback` function is attached as a listener
-for the [`'finish'`][] and the `'error'` event.
+stream.
 
 Calling the [`stream.write()`][stream-write] method after calling
 [`stream.end()`][stream-end] will raise an error.
@@ -592,7 +593,7 @@ changes:
   `Uint8Array`. For object mode streams, `chunk` may be any JavaScript value
   other than `null`.
 * `encoding` {string} The encoding, if `chunk` is a string. **Default:** `'utf8'`
-* `callback` {Function} Callback for when this chunk of data is flushed
+* `callback` {Function} Callback for when this chunk of data is flushed.
 * Returns: {boolean} `false` if the stream wishes for the calling code to
   wait for the `'drain'` event to be emitted before continuing to write
   additional data; otherwise `true`.

lib/_stream_writable.js

@@ -64,6 +64,8 @@ ObjectSetPrototypeOf(Writable, Stream);
 
 function nop() {}
 
+const kOnFinished = Symbol('kOnFinished');
+
 function WritableState(options, stream, isDuplex) {
   // Duplex streams are both readable and writable, but share
   // the same options object.
@@ -185,6 +187,8 @@ function WritableState(options, stream, isDuplex) {
   // True if close has been emitted or would have been emitted
   // depending on emitClose.
   this.closeEmitted = false;
+
+  this[kOnFinished] = [];
 }
 
 function resetBuffer(state) {
@@ -411,7 +415,7 @@ function onwriteError(stream, state, er, cb) {
   // not enabled. Passing `er` here doesn't make sense since
   // it's related to one specific write, not to the buffered
   // writes.
-  errorBuffer(state, new ERR_STREAM_DESTROYED('write'));
+  errorBuffer(state);
   // This can emit error, but error must always follow cb.
   errorOrDestroy(stream, er);
 }
@@ -487,14 +491,14 @@ function afterWrite(stream, state, count, cb) {
   }
 
   if (state.destroyed) {
-    errorBuffer(state, new ERR_STREAM_DESTROYED('write'));
+    errorBuffer(state);
   }
 
   finishMaybe(stream, state);
 }
 
 // If there's something in the buffer waiting, then invoke callbacks.
-function errorBuffer(state, err) {
+function errorBuffer(state) {
   if (state.writing) {
     return;
   }
@@ -503,7 +507,11 @@ function errorBuffer(state, err) {
     const { chunk, callback } = state.buffered[n];
     const len = state.objectMode ? 1 : chunk.length;
     state.length -= len;
-    callback(err);
+    callback(new ERR_STREAM_DESTROYED('write'));
+  }
+
+  for (const callback of state[kOnFinished].splice(0)) {
+    callback(new ERR_STREAM_DESTROYED('end'));
   }
 
   resetBuffer(state);
@@ -611,10 +619,11 @@ Writable.prototype.end = function(chunk, encoding, cb) {
   }
 
   if (typeof cb === 'function') {
-    if (err || state.finished)
+    if (err || state.finished) {
       process.nextTick(cb, err);
-    else
-      onFinished(this, cb);
+    } else {
+      state[kOnFinished].push(cb);
+    }
   }
 
   return this;
@@ -636,6 +645,9 @@ function callFinal(stream, state) {
   stream._final((err) => {
     state.pendingcb--;
     if (err) {
+      for (const callback of state[kOnFinished].splice(0)) {
+        callback(err);
+      }
       errorOrDestroy(stream, err, state.sync);
     } else if (needFinish(state)) {
       state.prefinished = true;
@@ -683,6 +695,11 @@ function finish(stream, state) {
     return;
 
   state.finished = true;
+
+  for (const callback of state[kOnFinished].splice(0)) {
+    callback();
+  }
+
   stream.emit('finish');
 
   if (state.autoDestroy) {
@@ -701,26 +718,6 @@ function finish(stream, state) {
   }
 }
 
-// TODO(ronag): Avoid using events to implement internal logic.
-function onFinished(stream, cb) {
-  function onerror(err) {
-    stream.removeListener('finish', onfinish);
-    stream.removeListener('error', onerror);
-    cb(err);
-    if (stream.listenerCount('error') === 0) {
-      stream.emit('error', err);
-    }
-  }
-
-  function onfinish() {
-    stream.removeListener('finish', onfinish);
-    stream.removeListener('error', onerror);
-    cb();
-  }
-  stream.on('finish', onfinish);
-  stream.prependListener('error', onerror);
-}
-
 ObjectDefineProperties(Writable.prototype, {
 
   destroyed: {
@@ -800,7 +797,7 @@ const destroy = destroyImpl.destroy;
 Writable.prototype.destroy = function(err, cb) {
   const state = this._writableState;
   if (!state.destroyed) {
-    process.nextTick(errorBuffer, state, new ERR_STREAM_DESTROYED('write'));
+    process.nextTick(errorBuffer, state);
   }
   destroy.call(this, err, cb);
   return this;

test/parallel/test-stream-transform-final-sync.js

@@ -90,7 +90,7 @@ const t = new stream.Transform({
 t.on('finish', common.mustCall(function() {
   state++;
   // finishListener
-  assert.strictEqual(state, 14);
+  assert.strictEqual(state, 15);
 }, 1));
 t.on('end', common.mustCall(function() {
   state++;
@@ -106,5 +106,5 @@ t.write(4);
 t.end(7, common.mustCall(function() {
   state++;
   // endMethodCallback
-  assert.strictEqual(state, 15);
+  assert.strictEqual(state, 14);
 }, 1));

test/parallel/test-stream-transform-final.js

@@ -92,7 +92,7 @@ const t = new stream.Transform({
 t.on('finish', common.mustCall(function() {
   state++;
   // finishListener
-  assert.strictEqual(state, 14);
+  assert.strictEqual(state, 15);
 }, 1));
 t.on('end', common.mustCall(function() {
   state++;
@@ -108,5 +108,5 @@ t.write(4);
 t.end(7, common.mustCall(function() {
   state++;
   // endMethodCallback
-  assert.strictEqual(state, 15);
+  assert.strictEqual(state, 14);
 }, 1));

test/parallel/test-stream-writable-destroy.js

@@ -354,7 +354,7 @@ const assert = require('assert');
     assert.strictEqual(err.message, 'asd');
   }));
   write.end('asd', common.mustCall((err) => {
-    assert.strictEqual(err.message, 'asd');
+    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
   }));
   write.destroy(new Error('asd'));
 }

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

@@ -17,10 +17,10 @@ const stream = require('stream');
   }));
   writable.write('asd');
   writable.end(common.mustCall((err) => {
-    assert.strictEqual(err.message, 'kaboom');
+    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
   }));
   writable.end(common.mustCall((err) => {
-    assert.strictEqual(err.message, 'kaboom');
+    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
   }));
 }
 

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

@@ -19,5 +19,5 @@ writable._final = (cb) => {
 
 writable.write('asd');
 writable.end(common.mustCall((err) => {
-  assert.strictEqual(err.message, 'kaboom');
+  assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
 }));

test/parallel/test-stream-write-destroy.js

@@ -57,7 +57,7 @@ for (const withPendingData of [ false, true ]) {
     w.destroy();
     assert.strictEqual(chunksWritten, 1);
     callbacks.shift()();
-    assert.strictEqual(chunksWritten, 2);
+    assert.strictEqual(chunksWritten, useEnd && !withPendingData ? 1 : 2);
     assert.strictEqual(callbacks.length, 0);
     assert.strictEqual(drains, 1);