fix: Reset the retry counter to 0 when receiving data

gax/src/streamingCalls/streaming.ts

@@ -271,7 +271,7 @@ export class StreamProxy extends duplexify implements GRPCCallResult {
    */
   streamHandoffHelper(stream: CancellableStream, retry: RetryOptions): void {
     let enteredError = false;
-    const eventsToForward = ['metadata', 'response', 'status', 'data'];
+    const eventsToForward = ['metadata', 'response', 'status'];
 
     eventsToForward.forEach(event => {
       stream.on(event, this.emit.bind(this, event));
@@ -282,6 +282,11 @@ export class StreamProxy extends duplexify implements GRPCCallResult {
       this.streamHandoffErrorHandler(stream, retry, error);
     });
 
+    stream.on('data', (data: ResponseType) => {
+      this.retries = 0;
+      this.emit.bind(this, 'data')(data);
+    });
+
     stream.on('end', () => {
       if (!enteredError) {
         enteredError = true;

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

@@ -80,6 +80,8 @@ async function testShowcase() {
   const restClient = new EchoClient(restClientOpts);
   const restClientCompat = new EchoClient(restClientOptsCompat);
 
+  await testResetRetriesToZero(grpcSequenceClientWithServerStreamingRetries);
+
   // assuming gRPC server is started locally
   await testEcho(grpcClient);
   await testEchoError(grpcClient);
@@ -714,6 +716,83 @@ async function testServerStreamingRetrieswithRetryRequestOptions(
   });
 }
 
+// When the stream recieves data then the retry count should be set to 0
+async function testResetRetriesToZero(client: SequenceServiceClient) {
+  const finalData: string[] = [];
+  const shouldRetryFn = (error: GoogleError) => {
+    return [4, 5, 6, 7].includes(error!.code!);
+  };
+  const backoffSettings = createBackoffSettings(
+    10000,
+    2.5,
+    1000,
+    null,
+    1.5,
+    3000,
+    null
+  );
+  // intentionally set maxRetries to a value less than
+  // the number of errors in the sequence
+  backoffSettings.maxRetries = 2;
+  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.NOT_FOUND,
+      Status.ALREADY_EXISTS,
+      Status.PERMISSION_DENIED,
+      Status.OK,
+    ],
+    [0.1, 0.1, 0.1, 0.1, 0.1],
+    [1, 2, 3, 4, 5],
+    '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', (response: {content: string}) => {
+      finalData.push(response.content);
+    });
+    attemptStream.on('error', error => {
+      reject(error);
+    });
+    attemptStream.on('end', () => {
+      attemptStream.end();
+      resolve();
+    });
+  }).then(() => {
+    assert.deepStrictEqual(
+      finalData.join(' '),
+      'This This is This is testing This is testing the This is testing the brand'
+    );
+  });
+}
+
 // When maxRetries are set to 2 then on the third error from the server gax
 // should throw an error that says the retry count has been exceeded.
 async function testShouldFailOnThirdError(client: SequenceServiceClient) {

gax/test/unit/streaming.ts

@@ -1316,7 +1316,6 @@ describe('handles server streaming retries in gax when gaxStreamingRetries is en
       const s = new PassThrough({
         objectMode: true,
       });
-      s.push('hello');
       setImmediate(() => {
         s.emit('metadata');
       });