fix(storage): destroy writable stream on read error and reset read stream override in tests

Author: thiyaguk09

handwritten/storage/src/bucket.ts

@@ -4509,6 +4509,7 @@ class Bucket extends ServiceObject<Bucket, BucketMetadata> {
             readStream
               .on('error', err => {
                 readStream.destroy();
+                writable.destroy();
                 bail(err);
               })
               .pipe(writable)

handwritten/storage/test/bucket.ts

@@ -241,6 +241,7 @@ describe('Bucket', () => {
 
   beforeEach(() => {
     fsStatOverride = null;
+    fsCreateReadStreamOverride = null;
     pLimitOverride = null;
     bucket = new Bucket(STORAGE, BUCKET_NAME);
   });
@@ -3265,6 +3266,7 @@ describe('Bucket', () => {
       bucket.upload(textfilepath, options, (err: Error) => {
         try {
           assert.strictEqual(err.message, 'write error');
+          assert.ok(readStream);
           assert.ok(readStream.destroyed);
           done();
         } catch (e) {