More tests and fixes.

Author: rkistner

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

@@ -1,5 +1,5 @@
 import { logger } from '@powersync/lib-services-framework';
-import { bson, InternalOpId } from '@powersync/service-core';
+import { bson, CompactOptions, InternalOpId } from '@powersync/service-core';
 import { LRUCache } from 'lru-cache';
 import { PowerSyncMongo } from './db.js';
 import { mongo } from '@powersync/lib-service-mongodb';
@@ -16,7 +16,8 @@ export class MongoParameterCompactor {
   constructor(
     private db: PowerSyncMongo,
     private group_id: number,
-    private checkpoint: InternalOpId
+    private checkpoint: InternalOpId,
+    private options: CompactOptions
   ) {}
 
   async compact() {
@@ -47,7 +48,7 @@ export class MongoParameterCompactor {
 
     // The index doesn't cover sorting by key, so we keep our own cache of the last seen key.
     let lastByKey = new LRUCache<string, InternalOpId>({
-      max: 10_000
+      max: this.options.compactParameterCacheLimit ?? 10_000
     });
     let removeIds: InternalOpId[] = [];
     let removeDeleted: mongo.AnyBulkWriteOperation<BucketParameterDocument>[] = [];
@@ -91,7 +92,7 @@ export class MongoParameterCompactor {
           // there is still an earlier operation with the same key and lookup, that we don't have
           // in the cache due to cache size limits. So we need to explicitly remove all earlier operations.
           removeDeleted.push({
-            deleteOne: {
+            deleteMany: {
               filter: { 'key.g': doc.key.g, lookup: doc.lookup, _id: { $lte: doc._id }, key: doc.key }
             }
           });

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

@@ -704,7 +704,7 @@ export class MongoSyncBucketStorage
     const checkpoint = await this.getCheckpointInternal();
     await new MongoCompactor(this.db, this.group_id, options).compact();
     if (checkpoint != null && options?.compactParameterData) {
-      await new MongoParameterCompactor(this.db, this.group_id, checkpoint.checkpoint).compact();
+      await new MongoParameterCompactor(this.db, this.group_id, checkpoint.checkpoint, options).compact();
     }
   }
 

packages/service-core-tests/src/tests/register-compacting-tests.ts

@@ -1,7 +1,6 @@
 import { storage } from '@powersync/service-core';
 import { expect, test } from 'vitest';
 import * as test_utils from '../test-utils/test-utils-index.js';
-import { ParameterLookup } from '@powersync/service-sync-rules';
 
 const TEST_TABLE = test_utils.makeTestTable('test', ['id']);
 
@@ -433,87 +432,3 @@ bucket_definitions:
     );
   });
 }
-
-export function registerParameterCompactTests(generateStorageFactory: storage.TestStorageFactory) {
-  test('compacting parameters', async () => {
-    await using factory = await generateStorageFactory();
-    const syncRules = await factory.updateSyncRules({
-      content: `
-bucket_definitions:
-  test:
-    parameters: select id from test where id = request.user_id()
-    data: []
-    `
-    });
-    const bucketStorage = factory.getInstance(syncRules);
-
-    await bucketStorage.startBatch(test_utils.BATCH_OPTIONS, async (batch) => {
-      await batch.save({
-        sourceTable: TEST_TABLE,
-        tag: storage.SaveOperationTag.INSERT,
-        after: {
-          id: 't1'
-        },
-        afterReplicaId: 't1'
-      });
-
-      await batch.save({
-        sourceTable: TEST_TABLE,
-        tag: storage.SaveOperationTag.INSERT,
-        after: {
-          id: 't2'
-        },
-        afterReplicaId: 't2'
-      });
-
-      await batch.commit('1/1');
-    });
-
-    const lookup = ParameterLookup.normalized('test', '1', ['t1']);
-
-    const checkpoint1 = await bucketStorage.getCheckpoint();
-    const parameters1 = await checkpoint1.getParameterSets([lookup]);
-    expect(parameters1).toEqual([{ id: 't1' }]);
-
-    await bucketStorage.startBatch(test_utils.BATCH_OPTIONS, async (batch) => {
-      await batch.save({
-        sourceTable: TEST_TABLE,
-        tag: storage.SaveOperationTag.UPDATE,
-        before: {
-          id: 't1'
-        },
-        beforeReplicaId: 't1',
-        after: {
-          id: 't1'
-        },
-        afterReplicaId: 't1'
-      });
-
-      await batch.save({
-        sourceTable: TEST_TABLE,
-        tag: storage.SaveOperationTag.DELETE,
-        before: {
-          id: 't1'
-        },
-        beforeReplicaId: 't1'
-      });
-      await batch.commit('1/2');
-    });
-    const checkpoint2 = await bucketStorage.getCheckpoint();
-    const parameters2 = await checkpoint2.getParameterSets([lookup]);
-    expect(parameters2).toEqual([]);
-
-    const statsBefore = await bucketStorage.factory.getStorageMetrics();
-    await bucketStorage.compact({ compactParameterData: true });
-
-    // Check consistency
-    const parameters1b = await checkpoint1.getParameterSets([lookup]);
-    const parameters2b = await checkpoint2.getParameterSets([lookup]);
-    expect(parameters1b).toEqual([{ id: 't1' }]);
-    expect(parameters2b).toEqual([]);
-
-    // Check storage size
-    const statsAfter = await bucketStorage.factory.getStorageMetrics();
-    expect(statsAfter.parameters_size_bytes).toBeLessThan(statsBefore.parameters_size_bytes);
-  });
-}

packages/service-core-tests/src/tests/tests-index.ts

@@ -1,5 +1,6 @@
 export * from './register-bucket-validation-tests.js';
 export * from './register-compacting-tests.js';
+export * from './register-parameter-compacting-tests.js';
 export * from './register-data-storage-tests.js';
 export * from './register-migration-tests.js';
 export * from './register-sync-tests.js';

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

@@ -203,6 +203,11 @@ export interface CompactOptions {
 
   /** Minimum of 1 */
   moveBatchQueryLimit?: number;
+
+  /**
+   * Internal/testing use: Cache size for compacting parameters.
+   */
+  compactParameterCacheLimit?: number;
 }
 
 export interface ClearStorageOptions {