Fix checksum calculations in large buckets with > 4m rows (#282)

* Convert checksums to long when aggregating.

* Fix the compactor.

* Changeset.

Author: rkistner

.changeset/popular-keys-switch.md

@@ -0,0 +1,7 @@
+---
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-core': patch
+'@powersync/service-image': patch
+---
+
+[MongoDB Storage] Fix checksum calculations in buckets with more than 4 million operations

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

@@ -317,7 +317,7 @@ export class MongoCompactor {
             let numberOfOpsToClear = 0;
             for await (let op of query.stream()) {
               if (op.op == 'MOVE' || op.op == 'REMOVE' || op.op == 'CLEAR') {
-                checksum = utils.addChecksums(checksum, op.checksum);
+                checksum = utils.addChecksums(checksum, Number(op.checksum));
                 lastOpId = op._id;
                 numberOfOpsToClear += 1;
                 if (op.op != 'CLEAR') {
@@ -358,7 +358,7 @@ export class MongoCompactor {
               {
                 _id: lastOpId!,
                 op: 'CLEAR',
-                checksum: checksum,
+                checksum: BigInt(checksum),
                 data: null,
                 target_op: targetOp
               },

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

@@ -473,7 +473,10 @@ export class MongoSyncBucketStorage
           {
             $group: {
               _id: '$_id.b',
-              checksum_total: { $sum: '$checksum' },
+              // Historically, checksum may be stored as 'int' or 'double'.
+              // More recently, this should be a 'long'.
+              // $toLong ensures that we always sum it as a long, avoiding inaccuracies in the calculations.
+              checksum_total: { $sum: { $toLong: '$checksum' } },
               count: { $sum: 1 },
               has_clear_op: {
                 $max: {

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

@@ -97,7 +97,7 @@ export class PersistedBatch {
       remaining_buckets.set(key, b);
     }
 
-    const dchecksum = utils.hashDelete(replicaIdToSubkey(options.table.id, options.sourceKey));
+    const dchecksum = BigInt(utils.hashDelete(replicaIdToSubkey(options.table.id, options.sourceKey)));
 
     for (const k of options.evaluated) {
       const key = currentBucketKey(k);
@@ -133,7 +133,7 @@ export class PersistedBatch {
             source_key: options.sourceKey,
             table: k.table,
             row_id: k.id,
-            checksum: checksum,
+            checksum: BigInt(checksum),
             data: recordData
           }
         }

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

@@ -56,7 +56,7 @@ export interface BucketDataDocument {
   source_key?: ReplicaId;
   table?: string;
   row_id?: string;
-  checksum: number;
+  checksum: bigint;
   data: string | null;
   target_op?: bigint | null;
 }

modules/module-mongodb-storage/test/src/storage_sync.test.ts

@@ -117,5 +117,12 @@ describe('sync - mongodb', () => {
       has_more: false,
       next_after: '4'
     });
+
+    // Test that the checksum type is correct.
+    // Specifically, test that it never persisted as double.
+    const checksumTypes = await factory.db.bucket_data
+      .aggregate([{ $group: { _id: { $type: '$checksum' }, count: { $sum: 1 } } }])
+      .toArray();
+    expect(checksumTypes).toEqual([{ _id: 'long', count: 4 }]);
   });
 });