Limit slot health check based on time, not iterations.

Author: rkistner

modules/module-postgres/src/replication/WalStream.ts

@@ -2,12 +2,12 @@ import * as lib_postgres from '@powersync/lib-service-postgres';
 import {
   container,
   DatabaseConnectionError,
+  logger as defaultLogger,
   ErrorCode,
   errors,
   Logger,
-  logger as defaultLogger,
-  ReplicationAssertionError,
-  ReplicationAbortedError
+  ReplicationAbortedError,
+  ReplicationAssertionError
 } from '@powersync/lib-services-framework';
 import {
   BucketStorageBatch,
@@ -33,10 +33,10 @@ import {
   toSyncRulesRow
 } from '@powersync/service-sync-rules';
 
+import { ReplicationMetric } from '@powersync/service-types';
 import { PgManager } from './PgManager.js';
 import { getPgOutputRelation, getRelId, referencedColumnTypeIds } from './PgRelation.js';
 import { checkSourceConfiguration, checkTableRls, getReplicationIdentityColumns } from './replication-utils.js';
-import { ReplicationMetric } from '@powersync/service-types';
 import {
   ChunkedSnapshotQuery,
   IdSnapshotQuery,
@@ -335,7 +335,9 @@ export class WalStream {
     wal_status: string;
     invalidation_reason: string | null;
   }): Promise<{ needsNewSlot: boolean }> {
-    let last_error = null;
+    // Start with a placeholder error, should be replaced if there is an actual issue.
+    let last_error = new ReplicationAssertionError(`Slot health check failed to execute`);
+
     const slotName = this.slot_name;
 
     const lost = slot.wal_status == 'lost';
@@ -346,20 +348,11 @@ export class WalStream {
       };
     }
 
-    // Check that replication slot exists
-    for (let i = 120; i >= 0; i--) {
+    // Check that replication slot exists, trying for up to 2 minutes.
+    const startAt = performance.now();
+    while (performance.now() - startAt < 120_000) {
       this.touch();
 
-      if (i == 0) {
-        container.reporter.captureException(last_error, {
-          level: errors.ErrorSeverity.ERROR,
-          metadata: {
-            replication_slot: slotName
-          }
-        });
-
-        throw last_error;
-      }
       try {
         // We peek a large number of changes here, to make it more likely to pick up replication slot errors.
         // For example, "publication does not exist" only occurs here if the peek actually includes changes related
@@ -391,19 +384,18 @@ export class WalStream {
           throw e;
         }
 
-        // Could also be `publication "powersync" does not exist`, although this error may show up much later
-        // in some cases.
-
         if (
           /incorrect prev-link/.test(e.message) ||
           /replication slot.*does not exist/.test(e.message) ||
           /publication.*does not exist/.test(e.message) ||
           /can no longer access replication slot/.test(e.message)
         ) {
+          // Fatal error. In most cases, the `wal_status == 'lost'` check should pick this up, but this
+          // works as a fallback.
+
           container.reporter.captureException(e, {
             level: errors.ErrorSeverity.WARNING,
             metadata: {
-              try_index: i,
               replication_slot: slotName
             }
           });
@@ -422,7 +414,14 @@ export class WalStream {
       }
     }
 
-    throw new ReplicationAssertionError('Unreachable');
+    container.reporter.captureException(last_error, {
+      level: errors.ErrorSeverity.ERROR,
+      metadata: {
+        replication_slot: slotName
+      }
+    });
+
+    throw last_error;
   }
 
   async estimatedCountNumber(db: pgwire.PgConnection, table: storage.SourceTable): Promise<number> {