stream: invoke buffered write callbacks on error

Refs: #30596

Buffered write callbacks were only invoked upon
error if `autoDestroy` was invoked.

Backport-PR-URL: #31179
PR-URL: #30596
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

Author: ronag

lib/_stream_writable.js

@@ -459,6 +459,11 @@ function onwriteError(stream, state, er, cb) {
   --state.pendingcb;
 
   cb(er);
+  // Ensure callbacks are invoked even when autoDestroy is
+  // 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'));
   // This can emit error, but error must always follow cb.
   errorOrDestroy(stream, er);
 }
@@ -530,9 +535,29 @@ function afterWrite(stream, state, count, cb) {
     cb();
   }
 
+  if (state.destroyed) {
+    errorBuffer(state, new ERR_STREAM_DESTROYED('write'));
+  }
+
   finishMaybe(stream, state);
 }
 
+// If there's something in the buffer waiting, then invoke callbacks.
+function errorBuffer(state, err) {
+  if (state.writing || !state.bufferedRequest) {
+    return;
+  }
+
+  for (let entry = state.bufferedRequest; entry; entry = entry.next) {
+    const len = state.objectMode ? 1 : entry.chunk.length;
+    state.length -= len;
+    entry.callback(err);
+  }
+  state.bufferedRequest = null;
+  state.lastBufferedRequest = null;
+  state.bufferedRequestCount = 0;
+}
+
 // If there's something in the buffer waiting, then process it
 function clearBuffer(stream, state) {
   state.bufferProcessing = true;
@@ -782,12 +807,7 @@ const destroy = destroyImpl.destroy;
 Writable.prototype.destroy = function(err, cb) {
   const state = this._writableState;
   if (!state.destroyed) {
-    for (let entry = state.bufferedRequest; entry; entry = entry.next) {
-      process.nextTick(entry.callback, new ERR_STREAM_DESTROYED('write'));
-    }
-    state.bufferedRequest = null;
-    state.lastBufferedRequest = null;
-    state.bufferedRequestCount = 0;
+    process.nextTick(errorBuffer, state, new ERR_STREAM_DESTROYED('write'));
   }
   destroy.call(this, err, cb);
   return this;

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

@@ -292,3 +292,46 @@ const assert = require('assert');
   }));
   write.uncork();
 }
+
+{
+  // Call buffered write callback with error
+
+  const write = new Writable({
+    write(chunk, enc, cb) {
+      process.nextTick(cb, new Error('asd'));
+    },
+    autoDestroy: false
+  });
+  write.cork();
+  write.write('asd', common.mustCall((err) => {
+    assert.strictEqual(err.message, 'asd');
+  }));
+  write.write('asd', common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+  }));
+  write.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.message, 'asd');
+  }));
+  write.uncork();
+}
+
+{
+  // Ensure callback order.
+
+  let state = 0;
+  const write = new Writable({
+    write(chunk, enc, cb) {
+      // `setImmediate()` is used on purpose to ensure the callback is called
+      // after `process.nextTick()` callbacks.
+      setImmediate(cb);
+    }
+  });
+  write.write('asd', common.mustCall(() => {
+    assert.strictEqual(state++, 0);
+  }));
+  write.write('asd', common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    assert.strictEqual(state++, 1);
+  }));
+  write.destroy();
+}