feat: rework low level message stream retries, add debugging (#1713)

* feat: rework how subscriber streams are retried and generally managed

* feat: add debug logging from the low level message stream for reconnects

* chore: rename retryBackoff

* chore: update retry max backoff

* feat: clarify 'debug' channel a bit, add explicit debug message class

* chore: push stream retries to full individual retry

Author: feywind

src/debug.ts

@@ -0,0 +1,33 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/**
+ * Represents a debug message the user might want to print out for logging
+ * while debugging or whatnot. These will always come by way of the 'error'
+ * channel on streams or other event emitters. It's completely fine to
+ * ignore them, as some will just be verbose logging info, but they may
+ * help figure out what's going wrong. Support may also ask you to catch
+ * these channels, which you can do like so:
+ *
+ * ```
+ * subscription.on('debug', msg => console.log(msg.message));
+ * ```
+ *
+ * These values are _not_ guaranteed to remain stable, even within a major
+ * version, so don't depend on them for your program logic. Debug outputs
+ * may be added or removed at any time, without warning.
+ */
+export class DebugMessage {
+  constructor(public message: string, public error?: Error) {}
+}

src/exponential-retry.ts

@@ -138,6 +138,17 @@ export class ExponentialRetry<T> {
     this.scheduleRetry();
   }
 
+  /**
+   * Resets an item that was previously retried. This is useful if you have
+   * persistent items that just need to be retried occasionally.
+   *
+   * @private
+   */
+  reset(item: T) {
+    const retried = item as RetriedItem<T>;
+    delete retried.retryInfo;
+  }
+
   // Takes a time delta and adds fuzz.
   private randomizeDelta(durationMs: number): number {
     // The fuzz distance should never exceed one second, but in the

src/index.ts

@@ -174,6 +174,7 @@ export {
   TopicMetadata,
 } from './topic';
 export {Duration, TotalOfUnit, DurationLike} from './temporal';
+export {DebugMessage} from './debug';
 
 if (process.env.DEBUG_GRPC) {
   console.info('gRPC logging set to verbose');

src/message-queues.ts

@@ -32,6 +32,7 @@ import {
 } from './subscriber';
 import {Duration} from './temporal';
 import {addToBucket} from './util';
+import {DebugMessage} from './debug';
 
 /**
  * @private
@@ -65,15 +66,16 @@ export interface BatchOptions {
  * @param {string} message The error message.
  * @param {GoogleError} err The grpc error.
  */
-export class BatchError extends Error {
+export class BatchError extends DebugMessage {
   ackIds: string[];
   code: grpc.status;
   details: string;
   constructor(err: GoogleError, ackIds: string[], rpc: string) {
     super(
       `Failed to "${rpc}" for ${ackIds.length} message(s). Reason: ${
         process.env.DEBUG_GRPC ? err.stack : err.message
-      }`
+      }`,
+      err
     );
 
     this.ackIds = ackIds;
@@ -278,7 +280,9 @@ export abstract class MessageQueue {
       // These queues are used for ack and modAck messages, which should
       // never surface an error to the user level. However, we'll emit
       // them onto this debug channel in case debug info is needed.
-      this._subscriber.emit('debug', e);
+      const err = e as Error;
+      const debugMsg = new DebugMessage(err.message, err);
+      this._subscriber.emit('debug', debugMsg);
     }
 
     this.numInFlightRequests -= batchSize;
@@ -404,10 +408,8 @@ export abstract class MessageQueue {
     const others = toError.get(AckResponses.Other);
     if (others?.length) {
       const otherIds = others.map(e => e.ackId);
-      this._subscriber.emit(
-        'debug',
-        new BatchError(rpcError, otherIds, operation)
-      );
+      const debugMsg = new BatchError(rpcError, otherIds, operation);
+      this._subscriber.emit('debug', debugMsg);
     }
 
     // Take care of following up on all the Promises.
@@ -492,7 +494,8 @@ export class AckQueue extends MessageQueue {
           return results.toRetry;
         } catch (e) {
           // This should only ever happen if there's a code failure.
-          this._subscriber.emit('debug', e);
+          const err = e as Error;
+          this._subscriber.emit('debug', new DebugMessage(err.message, err));
           const exc = new AckError(AckResponses.Other, 'Code error');
           batch.forEach(m => {
             m.responsePromise?.reject(exc);