Avoid change streams on the storage database

.changeset/rich-donuts-cheat.md

@@ -0,0 +1,9 @@
+---
+'@powersync/service-module-postgres-storage': minor
+'@powersync/service-module-mongodb-storage': minor
+'@powersync/service-core-tests': minor
+'@powersync/service-module-postgres': minor
+'@powersync/service-core': minor
+---
+
+[MongoDB Storage] Remove change streams on bucket storage database due to performance overhead.

modules/module-mongodb-storage/src/migrations/db/migrations/1749720702136-checkpoint-events.ts

@@ -0,0 +1,50 @@
+import { migrations } from '@powersync/service-core';
+import * as storage from '../../../storage/storage-index.js';
+import { MongoStorageConfig } from '../../../types/types.js';
+
+export const up: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+  const db = storage.createPowerSyncMongo(configuration.storage as MongoStorageConfig);
+
+  try {
+    await db.createCheckpointEventsCollection();
+
+    await db.write_checkpoints.createIndex(
+      {
+        processed_at_lsn: 1
+      },
+      { name: 'processed_at_lsn' }
+    );
+
+    await db.custom_write_checkpoints.createIndex(
+      {
+        op_id: 1
+      },
+      { name: 'op_id' }
+    );
+  } finally {
+    await db.client.close();
+  }
+};
+
+export const down: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+
+  const db = storage.createPowerSyncMongo(configuration.storage as MongoStorageConfig);
+
+  try {
+    if (await db.write_checkpoints.indexExists('processed_at_lsn')) {
+      await db.write_checkpoints.dropIndex('processed_at_lsn');
+    }
+    if (await db.custom_write_checkpoints.indexExists('op_id')) {
+      await db.custom_write_checkpoints.dropIndex('op_id');
+    }
+    await db.db.dropCollection('checkpoint_events');
+  } finally {
+    await db.client.close();
+  }
+};

modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts

@@ -119,6 +119,7 @@ export class MongoBucketStorage
           }
         }
       );
+      await this.db.notifyCheckpoint();
     } else if (next == null && active?.id == sync_rules_group_id) {
       // Slot removed for "active" sync rules, while there is no "next" one.
       await this.updateSyncRules({
@@ -141,6 +142,7 @@ export class MongoBucketStorage
           }
         }
       );
+      await this.db.notifyCheckpoint();
     } else if (next != null && active?.id == sync_rules_group_id) {
       // Already have next sync rules, but need to stop replicating the active one.
 
@@ -155,6 +157,7 @@ export class MongoBucketStorage
           }
         }
       );
+      await this.db.notifyCheckpoint();
     }
   }
 
@@ -217,6 +220,7 @@ export class MongoBucketStorage
         last_keepalive_ts: null
       };
       await this.db.sync_rules.insertOne(doc);
+      await this.db.notifyCheckpoint();
       rules = new MongoPersistedSyncRulesContent(this.db, doc);
       if (options.lock) {
         const lock = await rules.lock();

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

@@ -139,8 +139,6 @@ export class MongoBucketBatch
         result = r;
       }
     }
-    await batchCreateCustomWriteCheckpoints(this.db, this.write_checkpoint_batch);
-    this.write_checkpoint_batch = [];
     return result;
   }
 
@@ -156,6 +154,11 @@ export class MongoBucketBatch
     await this.withReplicationTransaction(`Flushing ${batch.length} ops`, async (session, opSeq) => {
       resumeBatch = await this.replicateBatch(session, batch, opSeq, options);
 
+      if (this.write_checkpoint_batch.length > 0) {
+        await batchCreateCustomWriteCheckpoints(this.db, session, this.write_checkpoint_batch, opSeq.next());
+        this.write_checkpoint_batch = [];
+      }
+
       last_op = opSeq.last();
     });
 
@@ -627,6 +630,7 @@ export class MongoBucketBatch
         },
         { session }
       );
+      // We don't notify checkpoint here - we don't make any checkpoint updates directly
     });
   }
 
@@ -674,6 +678,7 @@ export class MongoBucketBatch
         },
         { session: this.session }
       );
+      await this.db.notifyCheckpoint();
 
       // Cannot create a checkpoint yet - return false
       return false;
@@ -698,6 +703,23 @@ export class MongoBucketBatch
       update.last_checkpoint = this.persisted_op;
     }
 
+    // Mark relevant write checkpoints as "processed".
+    // This makes it easier to identify write checkpoints that are "valid" in order.
+    await this.db.write_checkpoints.updateMany(
+      {
+        processed_at_lsn: null,
+        'lsns.1': { $lte: lsn }
+      },
+      {
+        $set: {
+          processed_at_lsn: lsn
+        }
+      },
+      {
+        session: this.session
+      }
+    );
+
     await this.db.sync_rules.updateOne(
       {
         _id: this.group_id
@@ -708,6 +730,7 @@ export class MongoBucketBatch
       },
       { session: this.session }
     );
+    await this.db.notifyCheckpoint();
     this.persisted_op = null;
     this.last_checkpoint_lsn = lsn;
     return true;
@@ -730,6 +753,21 @@ export class MongoBucketBatch
       return await this.commit(lsn);
     }
 
+    await this.db.write_checkpoints.updateMany(
+      {
+        processed_at_lsn: null,
+        'lsns.1': { $lte: lsn }
+      },
+      {
+        $set: {
+          processed_at_lsn: lsn
+        }
+      },
+      {
+        session: this.session
+      }
+    );
+
     await this.db.sync_rules.updateOne(
       {
         _id: this.group_id
@@ -745,6 +783,7 @@ export class MongoBucketBatch
       },
       { session: this.session }
     );
+    await this.db.notifyCheckpoint();
     this.last_checkpoint_lsn = lsn;
 
     return true;