fix(storage): Invocation ID is not retained on multipart upload retries (googleapis#8190)

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.

Thank you for opening a Pull Request! Before submitting your PR, there
are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a
[bug/issue](https://github.com/googleapis/{{metadata['repo']['name']}}/issues)
before writing your code! That way we can discuss the change, evaluate
designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> 🦕

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: thiyaguk09

handwritten/storage/src/bucket.ts

@@ -29,6 +29,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';
@@ -38,6 +39,7 @@ import {
   FileOptions,
   CreateResumableUploadOptions,
   CreateWriteStreamOptions,
+  CreateWriteStreamOptionsInternal,
   FileMetadata,
   ContextValue,
 } from './file.js';
@@ -4511,6 +4513,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) => {
@@ -4521,7 +4524,10 @@ class Bucket extends ServiceObject<Bucket, BucketMetadata> {
             ) {
               newFile.storage.retryOptions.autoRetry = false;
             }
-            const writable = newFile.createWriteStream(options);
+            const writable = newFile.createWriteStream({
+              ...options,
+              invocationId: persistentInvocationId,
+            } as CreateWriteStreamOptionsInternal);
             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,
@@ -266,6 +267,14 @@ export interface CreateWriteStreamOptions extends CreateResumableUploadOptions {
   validation?: string | boolean;
 }
 
+/**
+ * @internal
+ */
+export interface CreateWriteStreamOptionsInternal
+  extends CreateWriteStreamOptions {
+  invocationId?: string;
+}
+
 export interface MakeFilePrivateOptions {
   metadata?: FileMetadata;
   strict?: boolean;
@@ -2212,7 +2221,10 @@ class File extends ServiceObject<File, FileMetadata> {
 
     writeStream.once('writing', async () => {
       if (options.resumable === false) {
-        await this.startSimpleUpload_(fileWriteStream, options);
+        await this.startSimpleUpload_(
+          fileWriteStream,
+          options as CreateWriteStreamOptionsInternal,
+        );
       } else {
         await this.startResumableUpload_(fileWriteStream, options);
       }
@@ -4218,13 +4230,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,
+          } as CreateWriteStreamOptionsInternal);
 
           if (options.onUploadProgress) {
             writable.on('progress', options.onUploadProgress);
@@ -4531,7 +4547,7 @@ class File extends ServiceObject<File, FileMetadata> {
    */
   startSimpleUpload_(
     dup: Duplexify,
-    options: CreateWriteStreamOptions = {},
+    options: CreateWriteStreamOptionsInternal = {},
   ): void {
     options.metadata ??= {};
 
@@ -4545,6 +4561,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',

handwritten/storage/src/storage-transport.ts

@@ -49,6 +49,7 @@ export interface StorageQueryParameters extends StandardStorageQueryParams {
 
 export interface StorageRequestOptions extends GaxiosOptions {
   [GCCL_GCS_CMD_KEY]?: string;
+  invocationId?: string;
   interceptors?: GaxiosInterceptor<GaxiosOptionsPrepared>[];
   autoPaginate?: boolean;
   autoPaginateVal?: boolean;
@@ -254,7 +255,10 @@ export class StorageTransport {
   }
 
   #prepareHeaders(reqOpts: StorageRequestOptions): Record<string, string> {
-    const headersObj = this.#buildRequestHeaders(reqOpts.headers);
+    const headersObj = this.#buildRequestHeaders(
+      reqOpts.headers,
+      reqOpts.invocationId,
+    );
 
     if (reqOpts[GCCL_GCS_CMD_KEY]) {
       const current = headersObj.get('x-goog-api-client') || '';
@@ -299,12 +303,16 @@ export class StorageTransport {
     return searchParams.toString();
   };
 
-  #buildRequestHeaders(requestHeaders = {}) {
-    const headers = new Headers(requestHeaders);
+  #buildRequestHeaders(
+    reqHeaders?: GaxiosOptions['headers'],
+    invocationId?: string,
+  ) {
+    const headers = new Headers(reqHeaders);
     headers.set('User-Agent', this.#getUserAgentString());
+    const finalInvocationId = invocationId || randomUUID();
     headers.set(
       'x-goog-api-client',
-      `${getRuntimeTrackingString()} gccl/${this.packageJson.version}-${getModuleFormat()} gccl-invocation-id/${randomUUID()}`,
+      `${getRuntimeTrackingString()} gccl/${this.packageJson.version}-${getModuleFormat()} gccl-invocation-id/${finalInvocationId}`,
     );
     return headers;
   }

handwritten/storage/system-test/storage.ts

@@ -3260,6 +3260,42 @@ describe('storage', function () {
 
       assert.strictEqual(called, true);
     });
+
+    it('should maintain the same invocationId across the upload lifecycle', async () => {
+      const invocationIds: string[] = [];
+
+      const originalRequest = bucket.storageTransport.authClient.request.bind(
+        bucket.storageTransport.authClient,
+      );
+
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      bucket.storageTransport.authClient.request = async (config: any) => {
+        const headers = config.headers || {};
+        const apiHeaderKey = Object.keys(headers).find(
+          key => key.toLowerCase() === 'x-goog-api-client',
+        );
+
+        if (apiHeaderKey) {
+          const val = headers[apiHeaderKey];
+          const match = val.match(/gccl-invocation-id\/([a-f0-9-]+)/);
+          if (match) {
+            invocationIds.push(match[1]);
+          }
+        }
+        return originalRequest(config);
+      };
+
+      try {
+        const destination = `test-id-${Date.now()}.txt`;
+        await bucket.upload(FILES.big.path, {destination, resumable: false});
+
+        assert.ok(invocationIds.length >= 1);
+        const uniqueIds = [...new Set(invocationIds)];
+        assert.strictEqual(uniqueIds.length, 1);
+      } finally {
+        bucket.storageTransport.authClient.request = originalRequest;
+      }
+    });
   });
 
   describe('channels', () => {