Repository: googleapis/nodejs-storage

Issue:
Uncatchable error when upload stream is used in failed pipeline.
### Please make sure you have searched for information in the following guides.

- [X] Search the issues already opened: https://github.com/GoogleCloudPlatform/google-cloud-node/issues
- [X] Search StackOverflow: http://stackoverflow.com/questions/tagged/google-cloud-platform+node.js
- [X] Check our Troubleshooting guide: https://github.com/googleapis/google-cloud-node/blob/main/docs/troubleshooting.md
- [X] Check our FAQ: https://github.com/googleapis/google-cloud-node/blob/main/docs/faq.md
- [X] Check our libraries HOW-TO: https://github.com/googleapis/gax-nodejs/blob/main/client-libraries.md
- [X] Check out our authentication guide: https://github.com/googleapis/google-auth-library-nodejs
- [X] Check out handwritten samples for many of our APIs: https://github.com/GoogleCloudPlatform/nodejs-docs-samples

### A screenshot that you have tested with "Try this API".


N/A. This issue seems to be related to the client library's error handling ability and not the underlying API.

### Link to the code that reproduces this issue. A link to a **public** Github Repository or gist with a minimal reproduction.


https://gist.github.com/thebenperson/c6c1e66653fdc8c05cab52726b2f0664

### A step-by-step description of how to reproduce the issue, based on the linked reproduction.


1. Create a readable input stream.
2. Create a writable output stream by calling File.createWriteStream().
3. Call Stream.promises.pipeline(input, output).
4. Have the input stream emit() an error before it can provide any data.

### A clear and concise description of what the bug is, and what you expected to happen.

Expected behavior: Stream.promises.pipeline() promise is rejected.
Actual behavior: Program ends due to an unhandled "error" event.

I might have found the problem.

The error from the input stream is passed on to the output stream.
But, no data has been provided from the input stream yet.

This means that this callback hasn't been called yet:
https://github.com/googleapis/nodejs-storage/blob/740d30d09e8a2ff5a85c7f12831b03284d483fa6/src/file.ts#L2121

Meaning pipeline() hasn't been called here yet:
https://github.com/googleapis/nodejs-storage/blob/740d30d09e8a2ff5a85c7f12831b03284d483fa6/src/file.ts#L2128

So that means that there is nothing to handle errors when 
emitStream.destroy(e) is called here:
https://github.com/googleapis/nodejs-storage/blob/740d30d09e8a2ff5a85c7f12831b03284d483fa6/src/file.ts#L2075



### A clear and concise description WHY you expect this behavior, i.e., was it a recent change, there is documentation that points to this behavior, etc. **

Other types of writable streams do not cause errors like this one.
The writable stream from this library is interfering with the pipeline() function's ability to handle errors.


Hints from issue comments:
@seanlennaerts: I've encountered this same issue when using `pipeline()` with a GCP storage writer stream. When an error occurs upstream in the pipeline it results in an uncaught error because the GCP storage writer doesn't propagate the error to the pipeline correctly. 

The error propagates correctly with fs.createWriteStream but not with GCP's writer. For comparison:
- Works: pipeline(reader, transformerThatEmitsError, fs.createWriteStream())
- Fails with unhandled error: pipeline(reader, transformerThatEmitsError, GCP storage writer)

Hope this additional context helps resolve the issue!

edit:

So I think @thebenperson is correct about the problem. If you call `writer.write('')` before the pipeline then errors are handled properly

```typescript
const writer = Storage().bucket('bucket').file('file').createWriteStream()

// call write so that the 'writing' event gets emitted and the internal pipeline is inited
writer.write('')

await pipeline(reader, transformerThatEmitsError, writer)
```@surjikal: A while back, I was also trying to make this work with `stream/promises`, but I couldn't. This was on Node 18 so, not sure if that's what the problem was.

But still, in case it helps, here's what ended up working...

```typescript
import Stream from 'node:stream';
import StreamPromises from 'node:stream/promises';

const reader = ...;
const transform = ...;
const writer = Storage().bucket('bucket').file('file').createWriteStream()

// The important bit: don't use the promise version of pipeline
const pipeline = Stream.pipeline(reader, transform, writer, err => {
    err ? pipeline.end() : void 0;
});

// Makes it awaitable. Note: I don't think the options are necessary to make it work.
await StreamPromises.finished(pipeline, { signal, readable: false, writable: true });
```@wvanderdeijl: We also suffer from this issue and it is preventing us from upgrading. We have (sometimes complex) pipelines that end in a GCS write stream. It is quite likely that an error occurs earlier in that pipeline that we loose because of this bug.
We can't reproduce the issue in version 7.9.0 and it start appearing in 7.10.0 and is still present in the (now latest) version 7.16.0@wvanderdeijl: I have created PR #2624 in an attempt to fix this issues as well as unit test for this issue and related (previously fixed) issue #2367@surjikal: I mentioned in my [previous comment](https://github.com/googleapis/nodejs-storage/issues/2560#issuecomment-2763872172) that we were seeing a difference between using the callback vs promise flavors of `pipeline`. I could only get the callback one to work. That said, we were using Node18, and promise pipeline itself could have been buggy I guess.

Still, I wanted to bring this up because, it may be worth adding some tests that cover the promise  version of pipeline too.

Solution patch:
diff --git a/src/file.ts b/src/file.ts
index 7372995..57f0645 100644
--- a/src/file.ts
+++ b/src/file.ts
@@ -2100,6 +2100,10 @@ class File extends ServiceObject<File, FileMetadata> {
 
     const emitStream = new PassThroughShim();
 
+    // If `writeStream` is destroyed before the `writing` event, `emitStream` will not have any listeners. This prevents an unhandled error.
+    const noop = () => {};
+    emitStream.on('error', noop);
+
     let hashCalculatingStream: HashStreamValidator | null = null;
 
     if (crc32c || md5) {
@@ -2138,6 +2142,9 @@ class File extends ServiceObject<File, FileMetadata> {
         this.startResumableUpload_(fileWriteStream, options);
       }
 
+      // remove temporary noop listener as we now create a pipeline that handles the errors
+      emitStream.removeListener('error', noop);
+
       pipeline(
         emitStream,
         ...(transformStreams as [Transform]),



Test patch:
diff --git a/test/file.ts b/test/file.ts
index 9748836..564d10c 100644
--- a/test/file.ts
+++ b/test/file.ts
@@ -23,7 +23,14 @@ import {
 } from '../src/nodejs-common/index.js';
 import {describe, it, before, beforeEach, afterEach} from 'mocha';
 import {PromisifyAllOptions} from '@google-cloud/promisify';
-import {Readable, PassThrough, Stream, Duplex, Transform} from 'stream';
+import {
+  Readable,
+  PassThrough,
+  Stream,
+  Duplex,
+  Transform,
+  pipeline,
+} from 'stream';
 import assert from 'assert';
 import * as crypto from 'crypto';
 import duplexify from 'duplexify';
@@ -2281,6 +2288,67 @@ describe('File', () => {
       writable.end('data');
     });
 
+    it('should close upstream when pipeline fails', done => {
+      const writable: Stream.Writable = file.createWriteStream();
+      const error = new Error('My error');
+      const uploadStream = new PassThrough();
+
+      let receivedBytes = 0;
+      const validateStream = new PassThrough();
+      validateStream.on('data', (chunk: Buffer) => {
+        receivedBytes += chunk.length;
+        if (receivedBytes > 5) {
+          // this aborts the pipeline which should also close the internal pipeline within createWriteStream
+          pLine.destroy(error);
+        }
+      });
+
+      file.startResumableUpload_ = (dup: duplexify.Duplexify) => {
+        dup.setWritable(uploadStream);
+        // Emit an error so the pipeline's error-handling logic is triggered
+        uploadStream.emit('error', error);
+        // Explicitly destroy the stream so that the 'close' event is guaranteed to fire,
+        // even in Node v14 where autoDestroy defaults may prevent automatic closing
+        uploadStream.destroy();
+      };
+
+      let closed = false;
+      uploadStream.on('close', () => {
+        closed = true;
+      });
+
+      const pLine = pipeline(
+        (function* () {
+          yield 'foo'; // write some data
+          yield 'foo'; // write some data
+          yield 'foo'; // write some data
+        })(),
+        validateStream,
+        writable,
+        (e: Error | null) => {
+          assert.strictEqual(e, error);
+          assert.strictEqual(closed, true);
+          done();
+        }
+      );
+    });
+
+    it('should error pipeline if source stream emits error before any data', done => {
+      const writable = file.createWriteStream();
+      const error = new Error('Error before first chunk');
+      pipeline(
+        // eslint-disable-next-line require-yield
+        (function* () {
+          throw error;
+        })(),
+        writable,
+        (e: Error | null) => {
+          assert.strictEqual(e, error);
+          done();
+        }
+      );
+    });
+
     describe('validation', () => {
       const data = 'test';