fix: destroy pipeline streams when returned stream errors

src/file.ts

@@ -2032,6 +2032,16 @@ class File extends ServiceObject<File, FileMetadata> {
         emitStream.write(chunk, encoding, cb);
       },
     });
+    // If the write stream, which is returned to the caller, catches an error we need to make sure that
+    // at least one of the streams in the pipeline below gets notified so that they
+    // all get cleaned up / destroyed.
+    writeStream.once('error', e => {
+      emitStream.destroy(e);
+    });
+    // If the write stream is closed, cleanup the pipeline below by calling destroy on one of the streams.
+    writeStream.once('close', () => {
+      emitStream.destroy();
+    });
 
     const transformStreams: Transform[] = [];
 

src/nodejs-common/service.ts

@@ -271,9 +271,8 @@ export class Service {
     };
 
     if (reqOpts[GCCL_GCS_CMD_KEY]) {
-      reqOpts.headers[
-        'x-goog-api-client'
-      ] += ` gccl-gcs-cmd/${reqOpts[GCCL_GCS_CMD_KEY]}`;
+      reqOpts.headers['x-goog-api-client'] +=
+        ` gccl-gcs-cmd/${reqOpts[GCCL_GCS_CMD_KEY]}`;
     }
 
     if (reqOpts.shouldReturnStream) {

src/resumable-upload.ts

@@ -933,9 +933,8 @@ export class Upload extends Writable {
       // `Content-Length` for multiple chunk uploads is the size of the chunk,
       // not the overall object
       headers['Content-Length'] = bytesToUpload;
-      headers[
-        'Content-Range'
-      ] = `bytes ${this.offset}-${endingByte}/${totalObjectSize}`;
+      headers['Content-Range'] =
+        `bytes ${this.offset}-${endingByte}/${totalObjectSize}`;
     } else {
       headers['Content-Range'] = `bytes ${this.offset}-*/${this.contentLength}`;
     }

src/transfer-manager.ts

@@ -223,9 +223,8 @@ class XMLMultiPartUploadHelper implements MultiPartUploadHelper {
 
         // Prepend command feature to value, if not already there
         if (!value.includes(GCCL_GCS_CMD_FEATURE.UPLOAD_SHARDED)) {
-          headers[
-            key
-          ] = `${value} gccl-gcs-cmd/${GCCL_GCS_CMD_FEATURE.UPLOAD_SHARDED}`;
+          headers[key] =
+            `${value} gccl-gcs-cmd/${GCCL_GCS_CMD_FEATURE.UPLOAD_SHARDED}`;
         }
       } else if (key.toLocaleLowerCase().trim() === 'user-agent') {
         userAgentFound = true;