fix: hoist invocationId to ensure persistence across upload retries

Hoists the generation of `persistentInvocationId` to the beginning of
the
upload process in `Bucket.upload` and `File.save`. This ensures that
retried multipart upload attempts reuse the same invocation ID in the
`x-goog-api-client` header, rather than generating a new one for
each attempt.

Author: thiyaguk09

handwritten/storage/src/bucket.ts

@@ -28,6 +28,7 @@ import * as http from 'http';
 import * as path from 'path';
 import {promisify} from 'util';
 import AsyncRetry from 'async-retry';
+import {randomUUID} from 'crypto';
 import {convertObjKeysToSnakeCase, handleContextValidation} from './util.js';
 
 import {Acl, AclMetadata} from './acl.js';
@@ -4442,6 +4443,7 @@ class Bucket extends ServiceObject<Bucket, BucketMetadata> {
     optionsOrCallback?: UploadOptions | UploadCallback,
     callback?: UploadCallback,
   ): Promise<UploadResponse> | void {
+    const persistentInvocationId = randomUUID();
     const upload = (numberOfRetries: number | undefined) => {
       const returnValue = AsyncRetry(
         async (bail: (err: GaxiosError | Error) => void) => {
@@ -4452,7 +4454,10 @@ class Bucket extends ServiceObject<Bucket, BucketMetadata> {
             ) {
               newFile.storage.retryOptions.autoRetry = false;
             }
-            const writable = newFile.createWriteStream(options);
+            const writable = newFile.createWriteStream({
+              ...options,
+              invocationId: persistentInvocationId,
+            });
             if (options.onUploadProgress) {
               writable.on('progress', options.onUploadProgress);
             }

handwritten/storage/src/file.ts

@@ -27,6 +27,7 @@ import * as resumableUpload from './resumable-upload.js';
 import {Writable, Readable, pipeline, Transform, PipelineSource} from 'stream';
 import * as zlib from 'zlib';
 import * as http from 'http';
+import {randomUUID} from 'crypto';
 
 import {
   ExceptionMessages,
@@ -248,6 +249,7 @@ export interface CreateResumableUploadOptions
    * @see {@link CRC32C.from} for possible values.
    */
   resumeCRC32C?: Parameters<(typeof CRC32C)['from']>[0];
+  invocationId?: string;
   preconditionOpts?: PreconditionOptions;
   [GCCL_GCS_CMD_KEY]?: resumableUpload.UploadConfig[typeof GCCL_GCS_CMD_KEY];
 }
@@ -4181,13 +4183,17 @@ class File extends ServiceObject<File, FileMetadata> {
     ) {
       maxRetries = 0;
     }
+    const persistentInvocationId = randomUUID();
     const returnValue = AsyncRetry(
       async (bail: (err: Error) => void) => {
         return new Promise<void>((resolve, reject) => {
           if (maxRetries === 0) {
             this.storage.retryOptions.autoRetry = false;
           }
-          const writable = this.createWriteStream(options);
+          const writable = this.createWriteStream({
+            ...options,
+            invocationId: persistentInvocationId,
+          });
 
           if (options.onUploadProgress) {
             writable.on('progress', options.onUploadProgress);
@@ -4449,6 +4455,7 @@ class File extends ServiceObject<File, FileMetadata> {
       chunkSize: options?.chunkSize,
       highWaterMark: options?.highWaterMark,
       universeDomain: this.bucket.storage.universeDomain,
+      invocationId: options.invocationId,
       [GCCL_GCS_CMD_KEY]: options[GCCL_GCS_CMD_KEY],
     };
 
@@ -4508,6 +4515,7 @@ class File extends ServiceObject<File, FileMetadata> {
         uploadType: 'multipart',
       },
       url,
+      invocationId: options.invocationId,
       [GCCL_GCS_CMD_KEY]: options[GCCL_GCS_CMD_KEY],
       method: 'POST',
       responseType: 'json',