Improve error handling, introduce retry logic for GCP PubSub

packages/gcp-pubsub/README.md

@@ -1490,6 +1490,92 @@ When a message cannot be processed (invalid format, schema validation failure, h
 
 **Best Practice:** Always configure a DLQ in production to capture and analyze failed messages.
 
+### Subscription-Level Error Handling
+
+The consumer provides a higher-level error recovery mechanism that complements the SDK's built-in gRPC retry logic. While the `@google-cloud/pubsub` SDK automatically retries some transient errors at the gRPC level, there are scenarios where the SDK does not recover automatically:
+
+1. **Eventual consistency errors** (`NOT_FOUND`, `PERMISSION_DENIED`) are not in the SDK's default retry codes
+2. **Subscription stream disconnections** may not automatically reconnect in all cases
+3. **Infrastructure changes** (e.g., after Terraform deployments) may require full subscription reinitialization
+
+#### Initialization Retry
+
+When calling `start()`, the consumer will automatically retry initialization if it encounters retryable errors. This is particularly useful when:
+
+- Using `locatorConfig` and the subscription doesn't exist yet due to eventual consistency
+- Services start in parallel and the subscription is being created by another process
+- Terraform deployments are still propagating
+
+The retry logic handles errors containing:
+- `does not exist` - Resource not yet visible
+- `NOT_FOUND` - gRPC error code 5
+- `PERMISSION_DENIED` - gRPC error code 7 (IAM propagation delay)
+
+#### Runtime Reconnection
+
+**Retryable Error Codes:**
+
+*Errors the consumer handles via reinitialization during runtime:*
+- `DEADLINE_EXCEEDED` (4): Request timeout that SDK retry couldn't resolve
+- `NOT_FOUND` (5): Subscription may not be propagated yet (eventual consistency)
+- `PERMISSION_DENIED` (7): IAM permissions may not be propagated yet (eventual consistency)
+- `RESOURCE_EXHAUSTED` (8): Quota exceeded, retry with backoff
+- `INTERNAL` (13): Server error, should be transient
+- `UNAVAILABLE` (14): Service temporarily unable to process
+
+When these errors reach the `subscription.on('error')` handler (meaning SDK's built-in retry couldn't resolve them), the consumer will:
+1. Log a warning with error details
+2. Close the existing subscription and remove event listeners
+3. Reinitialize the subscription with exponential backoff
+4. Reattach event handlers and continue consuming
+
+**Why `NOT_FOUND` and `PERMISSION_DENIED`?**
+
+After Terraform deployments, GCP resources and IAM permissions can take several minutes to propagate across GCP's distributed infrastructure. During this window, the subscription may report these errors even though the configuration is correct. The consumer retries with exponential backoff to handle this eventual consistency.
+
+**Note:** For most transient errors, the SDK's built-in retry will handle recovery automatically. The consumer's reinitialization logic is a safety net for cases where SDK retry is exhausted or not applicable.
+
+**Configuration:**
+
+The same retry options apply to both initialization and runtime reconnection:
+
+```typescript
+class MyConsumer extends AbstractPubSubConsumer<MyMessage, ExecutionContext> {
+  constructor(dependencies: PubSubConsumerDependencies) {
+    super(
+      dependencies,
+      {
+        // ... other options ...
+
+        // Optional: Configure retry behavior for both init and runtime errors
+        subscriptionRetryOptions: {
+          maxRetries: 5,           // Maximum retry attempts (default: 5)
+          baseRetryDelayMs: 1000,  // Base delay for exponential backoff (default: 1000ms)
+          maxRetryDelayMs: 30000,  // Maximum delay between retries (default: 30000ms)
+        },
+      },
+      executionContext,
+    )
+  }
+}
+```
+
+**Exponential Backoff Formula:**
+```
+delay = min(baseRetryDelayMs * 2^(attempt-1), maxRetryDelayMs)
+```
+
+With default settings, delays are: 1s, 2s, 4s, 8s, 16s (capped at 30s).
+
+**Unexpected Subscription Closure:**
+
+The consumer also handles unexpected subscription closures (e.g., network issues, GCP service restarts). If the subscription closes while the consumer is still supposed to be consuming, it will automatically attempt reinitialization.
+
+**References:**
+- [GCP Pub/Sub Error Codes](https://cloud.google.com/pubsub/docs/reference/error-codes)
+- [GCP Pub/Sub Troubleshooting](https://cloud.google.com/pubsub/docs/troubleshooting)
+- [Node.js Pub/Sub Subscription Reconnection Issues](https://github.com/googleapis/nodejs-pubsub/issues/979)
+
 ### Error Resolver
 
 ```typescript

packages/gcp-pubsub/lib/errors/SubscriptionDoesNotExistError.ts

@@ -0,0 +1,22 @@
+export class SubscriptionDoesNotExistError extends Error {
+  public readonly subscriptionName: string
+
+  constructor(subscriptionName: string) {
+    super(`Subscription ${subscriptionName} does not exist`)
+    this.name = 'SubscriptionDoesNotExistError'
+    this.subscriptionName = subscriptionName
+  }
+}
+
+export function isSubscriptionDoesNotExistError(
+  error: unknown,
+): error is SubscriptionDoesNotExistError {
+  return (
+    typeof error === 'object' &&
+    error !== null &&
+    'name' in error &&
+    'message' in error &&
+    error.name === 'SubscriptionDoesNotExistError' &&
+    typeof error.message === 'string'
+  )
+}

packages/gcp-pubsub/lib/errors/grpcErrors.ts

@@ -0,0 +1,85 @@
+import { status as GrpcStatus } from '@grpc/grpc-js'
+
+/**
+ * gRPC status codes for which subscription operations should be retried.
+ *
+ * Includes both:
+ * 1. GCP-documented retryable errors (DEADLINE_EXCEEDED, RESOURCE_EXHAUSTED, INTERNAL, UNAVAILABLE)
+ * 2. Eventual consistency errors common after Terraform deployments (NOT_FOUND, PERMISSION_DENIED)
+ *
+ * **Why PERMISSION_DENIED is included:**
+ * After Terraform deployments, IAM permissions can take several minutes to propagate across
+ * GCP's distributed infrastructure. During this window, the subscription may report
+ * PERMISSION_DENIED even though permissions are correctly configured.
+ *
+ * **Why NOT_FOUND is included:**
+ * Similar to PERMISSION_DENIED, newly created subscriptions may not be immediately visible
+ * across all GCP endpoints due to eventual consistency.
+ *
+ * @see https://cloud.google.com/pubsub/docs/reference/error-codes
+ * @see https://github.com/googleapis/nodejs-pubsub/issues/979
+ */
+export const RETRYABLE_GRPC_STATUS_CODES = [
+  GrpcStatus.DEADLINE_EXCEEDED,
+  GrpcStatus.NOT_FOUND,
+  GrpcStatus.PERMISSION_DENIED,
+  GrpcStatus.RESOURCE_EXHAUSTED,
+  GrpcStatus.INTERNAL,
+  GrpcStatus.UNAVAILABLE,
+] as const
+
+export type RetryableGrpcStatusCode = (typeof RETRYABLE_GRPC_STATUS_CODES)[number]
+
+const RETRYABLE_CODES_SET = new Set<number>(RETRYABLE_GRPC_STATUS_CODES)
+
+/**
+ * Type for errors with a numeric gRPC status code and message.
+ */
+export type GrpcError = {
+  code: number
+  message: string
+}
+
+/**
+ * Checks if an error has gRPC error properties (code and message).
+ * Uses duck typing to avoid fragile instanceof checks.
+ *
+ * @param error - The error to check
+ * @returns true if the error has numeric `code` and string `message` properties
+ */
+export function isGrpcError(error: unknown): error is GrpcError {
+  return (
+    typeof error === 'object' &&
+    error !== null &&
+    'code' in error &&
+    'message' in error &&
+    typeof (error as GrpcError).code === 'number' &&
+    typeof (error as GrpcError).message === 'string'
+  )
+}
+
+/**
+ * Checks if an error is a gRPC error with a retryable status code.
+ *
+ * @param error - The error to check
+ * @returns true if the error has a retryable gRPC status code
+ */
+export function isRetryableGrpcError(error: unknown): error is GrpcError {
+  return isGrpcError(error) && RETRYABLE_CODES_SET.has(error.code)
+}
+
+/**
+ * Gets the gRPC status code from an error if it has one.
+ *
+ * @param error - The error to extract the code from
+ * @returns The gRPC status code, or undefined if not a gRPC error
+ */
+export function getGrpcStatusCode(error: unknown): number | undefined {
+  if (isGrpcError(error)) {
+    return error.code
+  }
+  return undefined
+}
+
+// Re-export for convenience
+export { GrpcStatus }

packages/gcp-pubsub/lib/index.ts

@@ -1,4 +1,6 @@
+export * from './errors/grpcErrors.ts'
 export * from './errors/PubSubConsumerErrorResolver.ts'
+export * from './errors/SubscriptionDoesNotExistError.ts'
 export * from './fakes/FakeConsumerErrorResolver.ts'
 export * from './fakes/TestPubSubPublisher.ts'
 export * from './pubsub/AbstractPubSubConsumer.ts'
@@ -9,4 +11,3 @@ export * from './pubsub/PubSubPublisherManager.ts'
 export * from './schemas/pubSubSchemas.ts'
 export * from './types/MessageTypes.ts'
 export * from './utils/pubSubInitter.ts'
-export * from './utils/pubSubUtils.ts'