stream: fix multiple Writable.destroy() calls

Calling Writable.destroy() multiple times in the same tick
could cause an assertion error.

Fixes: #38189

PR-URL: #38221
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Nitzan Uziely <linkgoron@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

Author: ronag

lib/internal/streams/writable.js

@@ -770,7 +770,11 @@ ObjectDefineProperties(Writable.prototype, {
 const destroy = destroyImpl.destroy;
 Writable.prototype.destroy = function(err, cb) {
   const state = this._writableState;
-  if (!state.destroyed) {
+
+  // Invoke pending callbacks.
+  if (!state.destroyed &&
+    (state.bufferedIndex < state.buffered.length ||
+      state[kOnFinished].length)) {
     process.nextTick(errorBuffer, state, new ERR_STREAM_DESTROYED('write'));
   }
   destroy.call(this, err, cb);

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

@@ -417,3 +417,57 @@ const assert = require('assert');
   }));
   write.write('asd');
 }
+
+{
+  const ac = new AbortController();
+  const write = addAbortSignal(ac.signal, new Writable({
+    write(chunk, enc, cb) { cb(); }
+  }));
+
+  write.on('error', common.mustCall((e) => {
+    assert.strictEqual(e.name, 'AbortError');
+    assert.strictEqual(write.destroyed, true);
+  }));
+  write.write('asd');
+  ac.abort();
+}
+
+{
+  const ac = new AbortController();
+  const write = new Writable({
+    signal: ac.signal,
+    write(chunk, enc, cb) { cb(); }
+  });
+
+  write.on('error', common.mustCall((e) => {
+    assert.strictEqual(e.name, 'AbortError');
+    assert.strictEqual(write.destroyed, true);
+  }));
+  write.write('asd');
+  ac.abort();
+}
+
+{
+  const signal = AbortSignal.abort();
+
+  const write = new Writable({
+    signal,
+    write(chunk, enc, cb) { cb(); }
+  });
+
+  write.on('error', common.mustCall((e) => {
+    assert.strictEqual(e.name, 'AbortError');
+    assert.strictEqual(write.destroyed, true);
+  }));
+}
+
+{
+  // Destroy twice
+  const write = new Writable({
+    write(chunk, enc, cb) { cb(); }
+  });
+
+  write.end(common.mustCall());
+  write.destroy();
+  write.destroy();
+}