Invalidate changestream if postImage is not available.

Author: rkistner

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

@@ -25,7 +25,15 @@ interface InitResult {
   needsInitialSync: boolean;
 }
 
-export class MissingReplicationSlotError extends Error {
+/**
+ * Thrown when the change stream is not valid anymore, and replication
+ * must be restarted.
+ *
+ * Possible reasons:
+ *  * Some change stream documents do not have postImages.
+ *  * startAfter/resumeToken is not valid anymore.
+ */
+export class ChangeStreamInvalidatedError extends Error {
   constructor(message: string) {
     super(message);
   }
@@ -482,6 +490,21 @@ export class ChangeStream {
   }
 
   async streamChanges() {
+    try {
+      await this.streamChangesInternal();
+    } catch (e) {
+      if (
+        e instanceof mongo.MongoServerError &&
+        e.codeName == 'NoMatchingDocument' &&
+        e.errmsg?.includes('post-image was not found')
+      ) {
+        throw new ChangeStreamInvalidatedError(e.errmsg);
+      }
+      throw e;
+    }
+  }
+
+  async streamChangesInternal() {
     // Auto-activate as soon as initial replication is done
     await this.storage.autoActivate();
 

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

@@ -1,5 +1,5 @@
 import { container } from '@powersync/lib-services-framework';
-import { MissingReplicationSlotError, ChangeStream } from './ChangeStream.js';
+import { ChangeStreamInvalidatedError, ChangeStream } from './ChangeStream.js';
 
 import { replication } from '@powersync/service-core';
 import { ConnectionManagerFactory } from './ConnectionManagerFactory.js';
@@ -40,7 +40,7 @@ export class ChangeStreamReplicationJob extends replication.AbstractReplicationJ
       });
       this.logger.error(`Replication failed`, e);
 
-      if (e instanceof MissingReplicationSlotError) {
+      if (e instanceof ChangeStreamInvalidatedError) {
         // This stops replication on this slot, and creates a new slot
         await this.options.storage.factory.slotRemoved(this.slotName);
       }
@@ -84,8 +84,10 @@ export class ChangeStreamReplicationJob extends replication.AbstractReplicationJ
         // Without this additional log, the cause may not be visible in the logs.
         this.logger.error(`cause`, e.cause);
       }
-      if (e instanceof mongo.MongoError && e.hasErrorLabel('NonResumableChangeStreamError')) {
-        throw new MissingReplicationSlotError(e.message);
+      if (e instanceof ChangeStreamInvalidatedError) {
+        throw e;
+      } else if (e instanceof mongo.MongoError && e.hasErrorLabel('NonResumableChangeStreamError')) {
+        throw new ChangeStreamInvalidatedError(e.message);
       } else {
         // Report the error if relevant, before retrying
         container.reporter.captureException(e, {

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

@@ -402,4 +402,61 @@ bucket_definitions:
 
     expect(data).toMatchObject([]);
   });
+
+  test('postImages - new collection with postImages enabled', async () => {
+    await using context = await ChangeStreamTestContext.open(factory, { postImages: 'autoConfigure' });
+    const { db } = context;
+    await context.updateSyncRules(`
+bucket_definitions:
+  global:
+    data:
+      - SELECT _id as id, description FROM "test_%"`);
+
+    await context.replicateSnapshot();
+
+    await db.createCollection('test_data', {
+      // enabled: true here - everything should work
+      changeStreamPreAndPostImages: { enabled: true }
+    });
+    const collection = db.collection('test_data');
+    const result = await collection.insertOne({ description: 'test1' });
+    const test_id = result.insertedId;
+    await collection.updateOne({ _id: test_id }, { $set: { description: 'test2' } });
+
+    context.startStreaming();
+
+    const data = await context.getBucketData('global[]');
+    expect(data).toMatchObject([
+      putOp('test_data', { id: test_id!.toHexString(), description: 'test1' }),
+      putOp('test_data', { id: test_id!.toHexString(), description: 'test2' })
+    ]);
+  });
+
+  test.only('postImages - new collection with postImages disabled', async () => {
+    await using context = await ChangeStreamTestContext.open(factory, { postImages: 'autoConfigure' });
+    const { db } = context;
+    await context.updateSyncRules(`
+bucket_definitions:
+  global:
+    data:
+      - SELECT _id as id, description FROM "test_data%"`);
+
+    await context.replicateSnapshot();
+
+    await db.createCollection('test_data', {
+      // enabled: false here, but autoConfigure will enable it.
+      // Unfortunately, that is too late, and replication must be restarted.
+      changeStreamPreAndPostImages: { enabled: false }
+    });
+    const collection = db.collection('test_data');
+    const result = await collection.insertOne({ description: 'test1' });
+    const test_id = result.insertedId;
+    await collection.updateOne({ _id: test_id }, { $set: { description: 'test2' } });
+
+    context.startStreaming();
+
+    await expect(() => context.getBucketData('global[]')).rejects.toMatchObject({
+      message: expect.stringContaining('stream was configured to require a post-image for all update events')
+    });
+  });
 }