streams: pipeline error if any stream is destroyed

Fixes: nodejs#36674
Refs: nodejs#29227 (comment)

Author: kalenikalexander

lib/internal/streams/pipeline.js

@@ -274,6 +274,11 @@ function pipeline(...streams) {
       throw new ERR_INVALID_ARG_TYPE(
         name, ['Stream', 'Function'], stream);
     }
+
+    if (isStream(ret) && ret.destroyed) {
+      finish(new ERR_STREAM_DESTROYED(reading ? 'pipe' : 'write'));
+      break;
+    }
   }
 
   // TODO(ronag): Consider returning a Duplex proxy if the first argument

test/parallel/test-stream-pipeline.js

@@ -1301,3 +1301,32 @@ const net = require('net');
     assert.strictEqual(res, '123');
   }));
 }
+
+{
+  const r = new Readable();
+  const t = new PassThrough();
+  const w = new Writable();
+
+  w.destroy();
+
+  pipeline([r, t, w], common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    assert.strictEqual(err.message, 'Cannot call write after a stream was destroyed');
+    assert.strictEqual(r.destroyed, true);
+    assert.strictEqual(t.destroyed, true);
+  }));
+}
+
+{
+  const r = new Readable();
+  const t = new PassThrough();
+  const w = new Writable();
+
+  t.destroy();
+
+  pipeline([r, t, w], common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    assert.strictEqual(err.message, 'Cannot call pipe after a stream was destroyed');
+    assert.strictEqual(r.destroyed, true);
+  }));
+}