Fix error mapping for ChangeStreamInvalidatedError.

Author: rkistner

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

@@ -1,4 +1,4 @@
-import { isMongoNetworkTimeoutError, mongo } from '@powersync/lib-service-mongodb';
+import { isMongoNetworkTimeoutError, isMongoServerError, mongo } from '@powersync/lib-service-mongodb';
 import {
   container,
   DatabaseConnectionError,
@@ -605,13 +605,7 @@ export class ChangeStream {
           }
 
           const originalChangeDocument = await stream.tryNext().catch((e) => {
-            if (isMongoNetworkTimeoutError(e)) {
-              // This typically has an unhelpful message like "connection 2 to 159.41.94.47:27017 timed out".
-              // We wrap the error to make it more useful.
-              throw new DatabaseConnectionError(ErrorCode.PSYNC_S1345, `Timeout while reading MongoDB ChangeStream`, e);
-            } else {
-              throw new DatabaseConnectionError(ErrorCode.PSYNC_S1346, `Error reading MongoDB ChangeStream`, e);
-            }
+            throw mapChangeStreamError(e);
           });
           // The stream was closed, we will only ever receive `null` from it
           if (!originalChangeDocument && stream.closed) {
@@ -793,3 +787,21 @@ async function touch() {
   // or reduce PING_INTERVAL here.
   return container.probes.touch();
 }
+
+function mapChangeStreamError(e: any) {
+  if (isMongoNetworkTimeoutError(e)) {
+    // This typically has an unhelpful message like "connection 2 to 159.41.94.47:27017 timed out".
+    // We wrap the error to make it more useful.
+    throw new DatabaseConnectionError(ErrorCode.PSYNC_S1345, `Timeout while reading MongoDB ChangeStream`, e);
+  } else if (
+    isMongoServerError(e) &&
+    e.codeName == 'NoMatchingDocument' &&
+    e.errmsg?.includes('post-image was not found')
+  ) {
+    throw new ChangeStreamInvalidatedError(e.errmsg, e);
+  } else if (isMongoServerError(e) && e.hasErrorLabel('NonResumableChangeStreamError')) {
+    throw new ChangeStreamInvalidatedError(e.message, e);
+  } else {
+    throw new DatabaseConnectionError(ErrorCode.PSYNC_S1346, `Error reading MongoDB ChangeStream`, e);
+  }
+}

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

@@ -86,8 +86,6 @@ export class ChangeStreamReplicationJob extends replication.AbstractReplicationJ
       }
       if (e instanceof ChangeStreamInvalidatedError) {
         throw e;
-      } else if (isMongoServerError(e) && e.hasErrorLabel('NonResumableChangeStreamError')) {
-        throw new ChangeStreamInvalidatedError(e.message, e);
       } else {
         // Report the error if relevant, before retrying
         container.reporter.captureException(e, {

modules/module-mongodb/test/src/resume.test.ts

@@ -8,6 +8,7 @@ import { describe, expect, test, vi } from 'vitest';
 import { ChangeStreamTestContext } from './change_stream_utils.js';
 import { env } from './env.js';
 import { INITIALIZED_MONGO_STORAGE_FACTORY, INITIALIZED_POSTGRES_STORAGE_FACTORY } from './util.js';
+import { ChangeStreamInvalidatedError } from '@module/replication/ChangeStream.js';
 
 describe('mongo lsn', () => {
   test('LSN with resume tokens should be comparable', () => {
@@ -145,8 +146,7 @@ function defineResumeTest(factoryGenerator: (options?: TestStorageOptions) => Pr
     context2.storage = factory.getInstance(activeContent!);
 
     const error = await context2.startStreaming().catch((ex) => ex);
-    expect(error).exist;
     // The ChangeStreamReplicationJob will detect this and throw a ChangeStreamInvalidatedError
-    expect(isMongoServerError(error) && error.hasErrorLabel('NonResumableChangeStreamError'));
+    expect(error).toBeInstanceOf(ChangeStreamInvalidatedError);
   });
 }