fix: raise error on retryable error when max retries is 0 (#1605)

* fix: raise error on retryable error when max retries is 0

Co-authored-by: Daniel Bruce <djbruce@google.com>

* add in test

* npm run fix

---------

Co-authored-by: Daniel Bruce <djbruce@google.com>
Co-authored-by: danieljbruce <danieljbruce@users.noreply.github.com>

gax/src/streamingCalls/streaming.ts

@@ -438,6 +438,12 @@ export class StreamProxy extends duplexify implements GRPCCallResult {
           return; // end chunk
         }
       } else {
+        if (maxRetries === 0) {
+          const e = GoogleError.parseGRPCStatusDetails(error);
+          e.note = 'Max retries is set to zero.';
+          this.destroy(e);
+          return; // end chunk
+        }
         return GoogleError.parseGRPCStatusDetails(error);
       }
     });

gax/test/test-application/src/index.ts

@@ -28,6 +28,7 @@ import {
   GoogleAuth,
   Status,
   createBackoffSettings,
+  createMaxRetriesBackoffSettings,
   RetryOptions,
 } from 'google-gax';
 import {RequestType} from 'google-gax/build/src/apitypes';
@@ -108,6 +109,7 @@ async function testShowcase() {
   await testChatThrows(restClientCompat); // REGAPIC does not support bidi streaming
   await testWait(restClientCompat);
   // Testing with gaxServerStreamingRetries being true
+
   await testServerStreamingRetryOptions(
     grpcSequenceClientWithServerStreamingRetries
   );
@@ -144,6 +146,12 @@ async function testShowcase() {
     grpcSequenceClientWithServerStreamingRetries
   );
 
+  await testShouldFailOnThirdError(
+    grpcSequenceClientWithServerStreamingRetries
+  );
+
+  await testErrorMaxRetries0(grpcSequenceClientWithServerStreamingRetries);
+  // ensure legacy tests pass with streaming retries client
   await testEcho(grpcClientWithServerStreamingRetries);
   await testEchoError(grpcClientWithServerStreamingRetries);
   await testExpand(grpcClientWithServerStreamingRetries);
@@ -152,9 +160,6 @@ async function testShowcase() {
   await testCollect(grpcClientWithServerStreamingRetries);
   await testChat(grpcClientWithServerStreamingRetries);
   await testWait(grpcClientWithServerStreamingRetries);
-  await testShouldFailOnThirdError(
-    grpcSequenceClientWithServerStreamingRetries
-  );
 }
 
 function createStreamingSequenceRequestFactory(
@@ -1146,6 +1151,76 @@ async function testServerStreamingThrowsCannotSetTotalTimeoutMillisMaxRetries(
   });
 }
 
+// The test should not retry when the max retries are set to 0
+// and the emitted error should bubble up to the user when it does not retry.
+async function testErrorMaxRetries0(client: SequenceServiceClient) {
+  const finalData: string[] = [];
+  const shouldRetryFn = (error: GoogleError) => {
+    return [4].includes(error!.code!);
+  };
+  const backoffSettings = createMaxRetriesBackoffSettings(
+    10000,
+    2.5,
+    1000,
+    0,
+    1.5,
+    3000,
+    0
+  );
+  const getResumptionRequestFn = (request: RequestType) => {
+    return request;
+  };
+
+  const retryOptions = new RetryOptions(
+    [],
+    backoffSettings,
+    shouldRetryFn,
+    getResumptionRequestFn
+  );
+
+  const settings = {
+    retry: retryOptions,
+  };
+
+  client.initialize();
+
+  const request = createStreamingSequenceRequestFactory(
+    [Status.DEADLINE_EXCEEDED, Status.OK],
+    [0.1, 0.1],
+    [0, 1],
+    'This is testing the brand new and shiny StreamingSequence server 3'
+  );
+  const response = await client.createStreamingSequence(request);
+  await new Promise<void>((resolve, reject) => {
+    const sequence = response[0];
+
+    const attemptRequest =
+      new protos.google.showcase.v1beta1.AttemptStreamingSequenceRequest();
+    attemptRequest.name = sequence.name!;
+    const attemptStream = client.attemptStreamingSequence(
+      attemptRequest,
+      settings
+    );
+    attemptStream.on('data', () => {
+      reject(new GoogleError('The stream should not receive any data'));
+    });
+    attemptStream.on('error', (error: GoogleError) => {
+      try {
+        assert.strictEqual(error.code, 4);
+        assert.strictEqual(error.note, 'Max retries is set to zero.');
+        resolve();
+      } catch (assertionError: unknown) {
+        reject(assertionError);
+      }
+    });
+    attemptStream.on('end', () => {
+      reject(
+        new GoogleError('The stream should not end before it receives an error')
+      );
+    });
+  });
+}
+
 async function main() {
   const showcaseServer = new ShowcaseServer();
   try {