More abort error improvements.

Author: rkistner

modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts

@@ -561,7 +561,7 @@ export class MongoSyncBucketStorage
   async clear(options?: storage.ClearStorageOptions): Promise<void> {
     while (true) {
       if (options?.signal?.aborted) {
-        throw new ReplicationAbortedError('Aborted clearing data');
+        throw new ReplicationAbortedError('Aborted clearing data', options.signal.reason);
       }
       try {
         await this.clearIteration();

modules/module-mongodb/src/replication/ChangeStream.ts

@@ -497,7 +497,7 @@ export class ChangeStream {
       }
 
       if (this.abort_signal.aborted) {
-        throw new ReplicationAbortedError(`Aborted initial replication`);
+        throw new ReplicationAbortedError(`Aborted initial replication`, this.abort_signal.reason);
       }
 
       // Pre-fetch next batch, so that we can read and write concurrently

modules/module-mysql/src/replication/BinLogStream.ts

@@ -324,7 +324,10 @@ export class BinLogStream {
 
     for await (let row of stream) {
       if (this.stopped) {
-        throw new ReplicationAbortedError('Abort signal received - initial replication interrupted.');
+        throw new ReplicationAbortedError(
+          'Abort signal received - initial replication interrupted.',
+          this.abortSignal.reason
+        );
       }
 
       if (columns == null) {

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

@@ -139,10 +139,10 @@ export class WalStream {
 
     // We wrap in our own abort controller so we can trigger abort internally.
     options.abort_signal.addEventListener('abort', () => {
-      this.abortController.abort();
+      this.abortController.abort(options.abort_signal.reason);
     });
     if (options.abort_signal.aborted) {
-      this.abortController.abort();
+      this.abortController.abort(options.abort_signal.reason);
     }
 
     this.snapshotter = new PostgresSnapshotter({ ...options, abort_signal: this.abortSignal });
@@ -620,6 +620,8 @@ export class WalStream {
         }
       }
     );
+
+    throw new ReplicationAbortedError(`Replication stream aborted`, this.abortSignal.reason);
   }
 
   async ack(lsn: string, replicationStream: pgwire.ReplicationStream) {