Improve handling of abort errors.

Author: rkistner

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

@@ -363,7 +363,7 @@ export class PostgresSnapshotter {
           await this.markSnapshotDone();
         }
       }
-      throw new ReplicationAbortedError();
+      throw new ReplicationAbortedError(`Replication loop aborted`, this.abortSignal.reason);
     } catch (e) {
       // If initial snapshot already completed, this has no effect
       this.initialSnapshotDone.reject(e);
@@ -622,7 +622,7 @@ export class PostgresSnapshotter {
 
       if (this.abortSignal.aborted) {
         // We only abort after flushing
-        throw new ReplicationAbortedError(`Table snapshot interrupted`);
+        throw new ReplicationAbortedError(`Table snapshot interrupted`, this.abortSignal.reason);
       }
     }
   }

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

@@ -347,12 +347,23 @@ export class WalStream {
       this.initPromise = this.initReplication();
       await this.initPromise;
       // These Promises are both expected to run until aborted or error.
-      streamPromise = this.streamChanges().finally(() => {
-        this.abortController.abort();
-      });
-      loopPromise = this.snapshotter.replicationLoop().finally(() => {
-        this.abortController.abort();
-      });
+      streamPromise = this.streamChanges()
+        .then(() => {
+          throw new ReplicationAssertionError(`Replication stream exited unexpectedly`);
+        })
+        .catch((e) => {
+          this.abortController.abort(e);
+          throw e;
+        });
+      loopPromise = this.snapshotter
+        .replicationLoop()
+        .then(() => {
+          throw new ReplicationAssertionError(`Replication snapshotter exited unexpectedly`);
+        })
+        .catch((e) => {
+          this.abortController.abort(e);
+          throw e;
+        });
       const results = await Promise.allSettled([loopPromise, streamPromise]);
       // First, prioritize non-aborted errors
       for (let result of results) {
@@ -366,16 +377,21 @@ export class WalStream {
           throw result.reason;
         }
       }
+
       // If we get here, both Promises completed successfully, which is unexpected.
       throw new ReplicationAssertionError(`Replication loop exited unexpectedly`);
     } catch (e) {
       await this.storage.reportError(e);
       throw e;
     } finally {
+      // Just to make sure
       this.abortController.abort();
     }
   }
 
+  /**
+   * For tests: Wait until the initial snapshot is complete.
+   */
   public async waitForInitialSnapshot() {
     if (this.initPromise == null) {
       throw new ReplicationAssertionError('replicate() must be called before waitForInitialSnapshot()');

packages/service-errors/src/errors.ts

@@ -151,11 +151,13 @@ export class ServiceAssertionError extends ServiceError {
 export class ReplicationAbortedError extends ServiceError {
   static readonly CODE = ErrorCode.PSYNC_S1103;
 
-  constructor(description?: string) {
+  constructor(description?: string, cause?: any) {
     super({
       code: ReplicationAbortedError.CODE,
       description: description ?? 'Replication aborted'
     });
+
+    this.cause = cause;
   }
 }