Merge remote-tracking branch 'origin/main' into optimize-bucket-lookups-2

Author: rkistner

.changeset/cyan-otters-sleep.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-core': patch
+---
+
+Use slot_name_prefix from the replication connection again.

.changeset/fast-flowers-scream.md

@@ -0,0 +1,7 @@
+---
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/lib-service-mongodb': patch
+'@powersync/service-image': patch
+---
+
+Skip large rows, rather than causing hard replication errors

libs/lib-mongodb/package.json

@@ -30,7 +30,7 @@
   "dependencies": {
     "@powersync/lib-services-framework": "workspace:*",
     "bson": "^6.10.3",
-    "mongodb": "^6.13.0",
+    "mongodb": "^6.14.1",
     "ts-codec": "^1.3.0",
     "uri-js": "^4.4.1"
   },

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

@@ -24,7 +24,7 @@ import { idPrefixFilter } from './util.js';
 /**
  * 15MB
  */
-const MAX_ROW_SIZE = 15 * 1024 * 1024;
+export const MAX_ROW_SIZE = 15 * 1024 * 1024;
 
 // Currently, we can only have a single flush() at a time, since it locks the op_id sequence.
 // While the MongoDB transaction retry mechanism handles this okay, using an in-process Mutex

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

@@ -4,8 +4,8 @@ import { EvaluatedParameters, EvaluatedRow } from '@powersync/service-sync-rules
 import * as bson from 'bson';
 
 import { logger } from '@powersync/lib-services-framework';
-import { BucketState, InternalOpId, storage, utils } from '@powersync/service-core';
-import { currentBucketKey } from './MongoBucketBatch.js';
+import { InternalOpId, storage, utils } from '@powersync/service-core';
+import { currentBucketKey, MAX_ROW_SIZE } from './MongoBucketBatch.js';
 import { MongoIdSequence } from './MongoIdSequence.js';
 import { PowerSyncMongo } from './db.js';
 import {
@@ -98,11 +98,20 @@ export class PersistedBatch {
 
     for (const k of options.evaluated) {
       const key = currentBucketKey(k);
-      remaining_buckets.delete(key);
 
       // INSERT
       const recordData = JSONBig.stringify(k.data);
       const checksum = utils.hashData(k.table, k.id, recordData);
+      if (recordData.length > MAX_ROW_SIZE) {
+        // In many cases, the raw data size would have been too large already. But there are cases where
+        // the BSON size is small enough, but the JSON size is too large.
+        // In these cases, we can't store the data, so we skip it, or generate a REMOVE operation if the row
+        // was synced previously.
+        logger.error(`powersync_${this.group_id} Row ${key} too large: ${recordData.length} bytes. Removing.`);
+        continue;
+      }
+
+      remaining_buckets.delete(key);
       this.currentSize += recordData.length + 200;
 
       const op_id = options.op_seq.next();

packages/service-core/src/util/config/compound-config-collector.ts

@@ -170,8 +170,7 @@ export class CompoundConfigCollector {
           baseConfig.api?.parameters?.max_data_fetch_concurrency ?? DEFAULT_MAX_DATA_FETCH_CONCURRENCY
       },
       // TODO maybe move this out of the connection or something
-      // slot_name_prefix: connections[0]?.slot_name_prefix ?? 'powersync_'
-      slot_name_prefix: 'powersync_',
+      slot_name_prefix: baseConfig.replication?.connections?.[0]?.slot_name_prefix ?? 'powersync_',
       parameters: baseConfig.parameters ?? {}
     };
 

pnpm-lock.yaml

@@ -36,7 +36,7 @@
     "ix": "^5.0.0",
     "jose": "^4.15.1",
     "lru-cache": "^10.0.1",
-    "mongodb": "^6.13.0",
+    "mongodb": "^6.14.1",
     "node-fetch": "^3.3.2",
     "pgwire": "github:kagis/pgwire#f1cb95f9a0f42a612bb5a6b67bb2eb793fc5fc87",
     "ts-codec": "^1.3.0",