Remove mongodb/bson workarounds (#211)

* bson 6.10.3

* Remove bson-related workarounds.

* Add changeset.

* Fix missing await.

Author: rkistner

.changeset/clever-kids-pretend.md

@@ -0,0 +1,11 @@
+---
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-rsocket-router': patch
+'@powersync/service-module-mongodb': patch
+'@powersync/service-core': patch
+'@powersync/lib-services-framework': patch
+'@powersync/lib-service-mongodb': patch
+'@powersync/service-image': patch
+---
+
+Upgrade mongodb and bson packages, removing the need for some workarounds.

libs/lib-mongodb/package.json

@@ -29,8 +29,8 @@
   },
   "dependencies": {
     "@powersync/lib-services-framework": "workspace:*",
-    "bson": "^6.8.0",
-    "mongodb": "^6.11.0",
+    "bson": "^6.10.3",
+    "mongodb": "^6.13.0",
     "ts-codec": "^1.3.0",
     "uri-js": "^4.4.1"
   },

libs/lib-services/package.json

@@ -23,7 +23,7 @@
     "@powersync/service-errors": "workspace:*",
     "ajv": "^8.12.0",
     "better-ajv-errors": "^1.2.0",
-    "bson": "^6.8.0",
+    "bson": "^6.10.3",
     "dotenv": "^16.4.5",
     "ipaddr.js": "^2.1.0",
     "lodash": "^4.17.21",

modules/module-mongodb-storage/package.json

@@ -34,7 +34,7 @@
     "@powersync/service-sync-rules": "workspace:*",
     "@powersync/service-types": "workspace:*",
     "@powersync/lib-service-mongodb": "workspace:*",
-    "bson": "^6.8.0",
+    "bson": "^6.10.3",
     "ts-codec": "^1.3.0",
     "ix": "^5.0.0",
     "lru-cache": "^10.2.2",

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

@@ -5,7 +5,6 @@ import { storage, utils } from '@powersync/service-core';
 import { PowerSyncMongo } from './db.js';
 import { BucketDataDocument, BucketDataKey } from './models.js';
 import { cacheKey } from './OperationBatch.js';
-import { safeBulkWrite } from './util.js';
 
 interface CurrentBucketState {
   /** Bucket name */
@@ -265,7 +264,7 @@ export class MongoCompactor {
   private async flush() {
     if (this.updates.length > 0) {
       logger.info(`Compacting ${this.updates.length} ops`);
-      await safeBulkWrite(this.db.bucket_data, this.updates, {
+      await this.db.bucket_data.bulkWrite(this.updates, {
         // Order is not important.
         // Since checksums are not affected, these operations can happen in any order,
         // and it's fine if the operations are partially applied.

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

@@ -313,13 +313,7 @@ export class MongoSyncBucketStorage
         // 1. We can calculate the document size accurately without serializing again.
         // 2. We can delay parsing the results until it's needed.
         // We manually use bson.deserialize below
-        raw: true,
-
-        // Since we're using raw: true and parsing ourselves later, we don't need bigint
-        // support here.
-        // Disabling due to https://jira.mongodb.org/browse/NODE-6165, and the fact that this
-        // is one of our most common queries.
-        useBigInt64: false
+        raw: true
       }
     ) as unknown as mongo.FindCursor<Buffer>;
 

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

@@ -1,7 +1,6 @@
 import * as framework from '@powersync/lib-services-framework';
 import { storage } from '@powersync/service-core';
 import { PowerSyncMongo } from './db.js';
-import { safeBulkWrite } from './util.js';
 
 export type MongoCheckpointAPIOptions = {
   db: PowerSyncMongo;
@@ -127,8 +126,7 @@ export async function batchCreateCustomWriteCheckpoints(
     return;
   }
 
-  await safeBulkWrite(
-    db.custom_write_checkpoints,
+  await db.custom_write_checkpoints.bulkWrite(
     checkpoints.map((checkpointOptions) => ({
       updateOne: {
         filter: { user_id: checkpointOptions.user_id, sync_rules_id: checkpointOptions.sync_rules_id },

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

@@ -15,7 +15,7 @@ import {
   CurrentDataDocument,
   SourceKey
 } from './models.js';
-import { replicaIdToSubkey, safeBulkWrite } from './util.js';
+import { replicaIdToSubkey } from './util.js';
 
 /**
  * Maximum size of operations we write in a single transaction.
@@ -246,22 +246,21 @@ export class PersistedBatch {
 
   async flush(db: PowerSyncMongo, session: mongo.ClientSession) {
     if (this.bucketData.length > 0) {
-      // calculate total size
-      await safeBulkWrite(db.bucket_data, this.bucketData, {
+      await db.bucket_data.bulkWrite(this.bucketData, {
         session,
         // inserts only - order doesn't matter
         ordered: false
       });
     }
     if (this.bucketParameters.length > 0) {
-      await safeBulkWrite(db.bucket_parameters, this.bucketParameters, {
+      await db.bucket_parameters.bulkWrite(this.bucketParameters, {
         session,
         // inserts only - order doesn't matter
         ordered: false
       });
     }
     if (this.currentData.length > 0) {
-      await safeBulkWrite(db.current_data, this.currentData, {
+      await db.current_data.bulkWrite(this.currentData, {
         session,
         // may update and delete data within the same batch - order matters
         ordered: true

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

@@ -124,48 +124,3 @@ export const connectMongoForTests = (url: string, isCI: boolean) => {
   });
   return new PowerSyncMongo(client);
 };
-
-/**
- * MongoDB bulkWrite internally splits the operations into batches
- * so that no batch exceeds 16MB. However, there are cases where
- * the batch size is very close to 16MB, where additional metadata
- * on the server pushes it over the limit, resulting in this error
- * from the server:
- *
- * > MongoBulkWriteError: BSONObj size: 16814023 (0x1008FC7) is invalid. Size must be between 0 and 16793600(16MB) First element: insert: "bucket_data"
- *
- * We work around the issue by doing our own batching, limiting the
- * batch size to 15MB. This does add additional overhead with
- * BSON.calculateObjectSize.
- */
-export async function safeBulkWrite<T extends mongo.Document>(
-  collection: mongo.Collection<T>,
-  operations: mongo.AnyBulkWriteOperation<T>[],
-  options: mongo.BulkWriteOptions
-) {
-  // Must be below 16MB.
-  // We could probably go a little closer, but 15MB is a safe threshold.
-  const BULK_WRITE_LIMIT = 15 * 1024 * 1024;
-
-  let batch: mongo.AnyBulkWriteOperation<T>[] = [];
-  let currentSize = 0;
-  // Estimated overhead per operation, should be smaller in reality.
-  const keySize = 8;
-  for (let op of operations) {
-    const bsonSize =
-      mongo.BSON.calculateObjectSize(op, {
-        checkKeys: false,
-        ignoreUndefined: true
-      } as any) + keySize;
-    if (batch.length > 0 && currentSize + bsonSize > BULK_WRITE_LIMIT) {
-      await collection.bulkWrite(batch, options);
-      currentSize = 0;
-      batch = [];
-    }
-    batch.push(op);
-    currentSize += bsonSize;
-  }
-  if (batch.length > 0) {
-    await collection.bulkWrite(batch, options);
-  }
-}

modules/module-mongodb/package.json

@@ -34,7 +34,7 @@
     "@powersync/service-sync-rules": "workspace:*",
     "@powersync/service-types": "workspace:*",
     "@powersync/lib-service-mongodb": "workspace:*",
-    "bson": "^6.8.0",
+    "bson": "^6.10.3",
     "ts-codec": "^1.3.0",
     "uuid": "^9.0.1"
   },