From eaddb030550c9579ee93f173e09b9b027761e757 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sat, 13 Nov 2021 18:31:05 +0200
Subject: [PATCH] stream: destroying stream without error is abort

If an autoDestroy stream is destroyed by user without
an error we automatically convert it to an AbortError in
order to avoid a weird state.
---
 doc/api/stream.md                       | 16 ++++++++++++++
 lib/internal/streams/destroy.js         | 12 +++++++++-
 lib/internal/streams/readable.js        |  2 ++
 lib/internal/streams/writable.js        |  2 ++
 lib/net.js                              |  1 +
 test/parallel/test-stream-auto-abort.js | 29 +++++++++++++++++++++++++
 6 files changed, 61 insertions(+), 1 deletion(-)
 create mode 100644 test/parallel/test-stream-auto-abort.js

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 66436eff9cb6d4..732f0de4a685a5 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -421,6 +421,10 @@ This is a destructive and immediate way to destroy a stream. Previous calls to
 Use `end()` instead of destroy if data should flush before close, or wait for
 the `'drain'` event before destroying the stream.
 
+If `.destroy()` is called without an `error` and `autoDestroy` is
+enabled, then if the stream has not completed it will be
+automatically destroyed with an `AbortError`.
+
 ```cjs
 const { Writable } = require('stream');
 
@@ -1101,6 +1105,10 @@ further errors except from `_destroy()` may be emitted as `'error'`.
 Implementors should not override this method, but instead implement
 [`readable._destroy()`][readable-_destroy].
 
+If `.destroy()` is called without an `error` and `autoDestroy` is
+enabled, then if the stream has not completed it will be
+automatically destroyed with an `AbortError`.
+
 ##### `readable.closed`
 
 <!-- YAML
@@ -1805,6 +1813,10 @@ unless `emitClose` is set in false.
 Once `destroy()` has been called, any further calls will be a no-op and no
 further errors except from `_destroy()` may be emitted as `'error'`.
 
+If `.destroy()` is called without an `error` and `autoDestroy` is
+enabled, then if the stream has not completed it will be
+automatically destroyed with an `AbortError`.
+
 ### `stream.finished(stream[, options], callback)`
 
 <!-- YAML
@@ -2508,6 +2520,8 @@ changes:
     [`stream._construct()`][writable-_construct] method.
   * `autoDestroy` {boolean} Whether this stream should automatically call
     `.destroy()` on itself after ending. **Default:** `true`.
+  * `autoAbort` {boolean} Whether this stream should automatically
+  error if `.destroy()` is called without an error before the stream has emitted `'finish'`.
   * `signal` {AbortSignal} A signal representing possible cancellation.
 
 <!-- eslint-disable no-useless-constructor -->
@@ -2865,6 +2879,8 @@ changes:
     [`stream._construct()`][readable-_construct] method.
   * `autoDestroy` {boolean} Whether this stream should automatically call
     `.destroy()` on itself after ending. **Default:** `true`.
+  * `autoAbort` {boolean} Whether this stream should automatically
+  error if `.destroy()` is called without an error before the stream has emitted `'end'`.
   * `signal` {AbortSignal} A signal representing possible cancellation.
 
 <!-- eslint-disable no-useless-constructor -->
diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
index 7d3657443e6ab5..fd9614f4a97199 100644
--- a/lib/internal/streams/destroy.js
+++ b/lib/internal/streams/destroy.js
@@ -14,7 +14,9 @@ const {
   kDestroyed,
   isDestroyed,
   isFinished,
-  isServerRequest
+  isServerRequest,
+  isReadableFinished,
+  isWritableEnded
 } = require('internal/streams/utils');
 
 const kDestroy = Symbol('kDestroy');
@@ -86,6 +88,14 @@ function _destroy(self, err, cb) {
     const r = self._readableState;
     const w = self._writableState;
 
+    if (!err) {
+      if (r?.autoAbort && !isReadableFinished(self)) {
+        err = new AbortError();
+      } else if (w?.autoAbort && !isWritableEnded(self)) {
+        err = new AbortError();
+      }
+    }
+
     checkError(err, w, r);
 
     if (w) {
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index bd9c288aa71929..ffb9dda0ba1424 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -174,6 +174,8 @@ function ReadableState(options, stream, isDuplex) {
 
   this.dataEmitted = false;
 
+  this.autoAbort = options?.autoAbort ?? false;
+
   this.decoder = null;
   this.encoding = null;
   if (options && options.encoding) {
diff --git a/lib/internal/streams/writable.js b/lib/internal/streams/writable.js
index 92e982e4821586..0a1c88de9a2d8b 100644
--- a/lib/internal/streams/writable.js
+++ b/lib/internal/streams/writable.js
@@ -196,6 +196,8 @@ function WritableState(options, stream, isDuplex) {
   // depending on emitClose.
   this.closeEmitted = false;
 
+  this.autoAbort = options?.autoAbort ?? false;
+
   this[kOnFinished] = [];
 }
 
diff --git a/lib/net.js b/lib/net.js
index 41ff284e1ec027..54a2e24371c308 100644
--- a/lib/net.js
+++ b/lib/net.js
@@ -324,6 +324,7 @@ function Socket(options) {
   // For backwards compat do not emit close on destroy.
   options.emitClose = false;
   options.autoDestroy = true;
+  options.autoAbort = false;
   // Handle strings directly.
   options.decodeStrings = false;
   stream.Duplex.call(this, options);
diff --git a/test/parallel/test-stream-auto-abort.js b/test/parallel/test-stream-auto-abort.js
new file mode 100644
index 00000000000000..9114fedf73ad5f
--- /dev/null
+++ b/test/parallel/test-stream-auto-abort.js
@@ -0,0 +1,29 @@
+'use strict';
+
+const common = require('../common');
+const { Readable, Writable } = require('stream');
+const assert = require('assert');
+
+{
+  const w = new Writable({
+    write() {
+
+    }
+  });
+  w.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.name, 'AbortError');
+  }));
+  w.destroy();
+}
+
+{
+  const r = new Readable({
+    read() {
+
+    }
+  });
+  r.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.name, 'AbortError');
+  }));
+  r.destroy();
+}