Remove mongodb/bson workarounds

.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"
   },

package.json

@@ -25,7 +25,7 @@
     "@changesets/cli": "^2.27.8",
     "@types/node": "^22.13.1",
     "async": "^3.2.4",
-    "bson": "^6.8.0",
+    "bson": "^6.10.3",
     "concurrently": "^8.2.2",
     "inquirer": "^9.2.7",
     "npm-check-updates": "^17.1.2",

packages/rsocket-router/package.json

@@ -27,7 +27,7 @@
   "devDependencies": {
     "@types/uuid": "^9.0.4",
     "@types/ws": "~8.2.0",
-    "bson": "^6.8.0",
+    "bson": "^6.10.3",
     "rsocket-websocket-client": "1.0.0-alpha.3"
   }
 }

packages/service-core/package.json

@@ -29,7 +29,7 @@
     "@powersync/service-types": "workspace:*",
     "async": "^3.2.4",
     "async-mutex": "^0.5.0",
-    "bson": "^6.8.0",
+    "bson": "^6.10.3",
     "commander": "^12.0.0",
     "cors": "^2.8.5",
     "ipaddr.js": "^2.1.0",

packages/service-core/src/storage/bson.ts

@@ -14,11 +14,10 @@ export const BSON_DESERIALIZE_INTERNAL_OPTIONS: bson.DeserializeOptions = {
 };
 
 /**
- * Use for data from external sources.
+ * Use for data from external sources, which could contain arbitrary fields.
  */
 export const BSON_DESERIALIZE_DATA_OPTIONS: bson.DeserializeOptions = {
-  // Temporarily disable due to https://jira.mongodb.org/browse/NODE-6764
-  useBigInt64: false
+  useBigInt64: true
 };
 
 /**
@@ -67,16 +66,11 @@ export const deserializeReplicaId = (id: Buffer): ReplicaId => {
   return deserialized.id;
 };
 
+/**
+ * Deserialize BSON - can be used for BSON containing arbitrary user data.
+ */
 export const deserializeBson = (buffer: Uint8Array): bson.Document => {
-  const doc = bson.deserialize(buffer, BSON_DESERIALIZE_DATA_OPTIONS);
-  // Temporary workaround due to https://jira.mongodb.org/browse/NODE-6764
-  for (let key in doc) {
-    const value = doc[key];
-    if (value instanceof bson.Long) {
-      doc[key] = value.toBigInt();
-    }
-  }
-  return doc;
+  return bson.deserialize(buffer, BSON_DESERIALIZE_DATA_OPTIONS);
 };
 
 export const serializeBson = (document: any): NodeBuffer => {