Merge branch 'main' into track-source

Author: rkistner

.changeset/slow-plants-deliver.md

@@ -0,0 +1,7 @@
+---
+'@powersync/service-module-postgres-storage': patch
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-core-tests': patch
+---
+
+[Postgres Storage] Fix concurrency issue in compacting, leading to "[PSYNC_S1101] Unexpected PUT operation".

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

@@ -86,18 +86,18 @@ export class MongoCompactor {
   constructor(
     private storage: MongoSyncBucketStorage,
     private db: VersionedPowerSyncMongo,
-    options?: MongoCompactOptions
+    options: MongoCompactOptions
   ) {
     this.group_id = storage.group_id;
-    this.idLimitBytes = (options?.memoryLimitMB ?? DEFAULT_MEMORY_LIMIT_MB) * 1024 * 1024;
-    this.moveBatchLimit = options?.moveBatchLimit ?? DEFAULT_MOVE_BATCH_LIMIT;
-    this.moveBatchQueryLimit = options?.moveBatchQueryLimit ?? DEFAULT_MOVE_BATCH_QUERY_LIMIT;
-    this.clearBatchLimit = options?.clearBatchLimit ?? DEFAULT_CLEAR_BATCH_LIMIT;
-    this.minBucketChanges = options?.minBucketChanges ?? DEFAULT_MIN_BUCKET_CHANGES;
-    this.minChangeRatio = options?.minChangeRatio ?? DEFAULT_MIN_CHANGE_RATIO;
-    this.maxOpId = options?.maxOpId ?? 0n;
-    this.buckets = options?.compactBuckets;
-    this.signal = options?.signal;
+    this.idLimitBytes = (options.memoryLimitMB ?? DEFAULT_MEMORY_LIMIT_MB) * 1024 * 1024;
+    this.moveBatchLimit = options.moveBatchLimit ?? DEFAULT_MOVE_BATCH_LIMIT;
+    this.moveBatchQueryLimit = options.moveBatchQueryLimit ?? DEFAULT_MOVE_BATCH_QUERY_LIMIT;
+    this.clearBatchLimit = options.clearBatchLimit ?? DEFAULT_CLEAR_BATCH_LIMIT;
+    this.minBucketChanges = options.minBucketChanges ?? DEFAULT_MIN_BUCKET_CHANGES;
+    this.minChangeRatio = options.minChangeRatio ?? DEFAULT_MIN_CHANGE_RATIO;
+    this.maxOpId = options.maxOpId ?? 0n;
+    this.buckets = options.compactBuckets;
+    this.signal = options.signal;
   }
 
   /**

modules/module-postgres-storage/src/storage/PostgresCompactor.ts

@@ -51,20 +51,20 @@ export class PostgresCompactor {
   private moveBatchLimit: number;
   private moveBatchQueryLimit: number;
   private clearBatchLimit: number;
-  private maxOpId: InternalOpId | undefined;
+  private maxOpId: InternalOpId;
   private buckets: string[] | undefined;
 
   constructor(
     private db: lib_postgres.DatabaseClient,
     private group_id: number,
-    options?: PostgresCompactOptions
+    options: PostgresCompactOptions
   ) {
-    this.idLimitBytes = (options?.memoryLimitMB ?? DEFAULT_MEMORY_LIMIT_MB) * 1024 * 1024;
-    this.moveBatchLimit = options?.moveBatchLimit ?? DEFAULT_MOVE_BATCH_LIMIT;
-    this.moveBatchQueryLimit = options?.moveBatchQueryLimit ?? DEFAULT_MOVE_BATCH_QUERY_LIMIT;
-    this.clearBatchLimit = options?.clearBatchLimit ?? DEFAULT_CLEAR_BATCH_LIMIT;
-    this.maxOpId = options?.maxOpId;
-    this.buckets = options?.compactBuckets;
+    this.idLimitBytes = (options.memoryLimitMB ?? DEFAULT_MEMORY_LIMIT_MB) * 1024 * 1024;
+    this.moveBatchLimit = options.moveBatchLimit ?? DEFAULT_MOVE_BATCH_LIMIT;
+    this.moveBatchQueryLimit = options.moveBatchQueryLimit ?? DEFAULT_MOVE_BATCH_QUERY_LIMIT;
+    this.clearBatchLimit = options.clearBatchLimit ?? DEFAULT_CLEAR_BATCH_LIMIT;
+    this.maxOpId = options.maxOpId ?? 0n;
+    this.buckets = options.compactBuckets;
   }
 
   /**
@@ -240,6 +240,15 @@ export class PostgresCompactor {
     }
   }
 
+  /**
+   * Expose the internal clearBucket() method to tests.
+   *
+   * @deprecated Only for tests
+   */
+  clearBucketForTests(bucket: string, op: InternalOpId) {
+    return this.clearBucket(bucket, op);
+  }
+
   /**
    * Perform a CLEAR compact for a bucket.
    *

modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts

@@ -127,8 +127,18 @@ export class PostgresSyncRulesStorage
     `.execute();
   }
 
-  compact(options?: storage.CompactOptions): Promise<void> {
-    return new PostgresCompactor(this.db, this.group_id, options).compact();
+  async compact(options?: storage.CompactOptions): Promise<void> {
+    let maxOpId = options?.maxOpId;
+    if (maxOpId == null) {
+      const checkpoint = await this.getCheckpoint();
+      // Note: If there is no active checkpoint, this will be 0, in which case no compacting is performed
+      maxOpId = checkpoint.checkpoint;
+    }
+
+    return new PostgresCompactor(this.db, this.group_id, {
+      ...options,
+      maxOpId
+    }).compact();
   }
 
   async populatePersistentChecksumCache(options: PopulateChecksumCacheOptions): Promise<PopulateChecksumCacheResults> {

modules/module-postgres-storage/test/src/storage_compacting.test.ts

@@ -1,6 +1,7 @@
 import { storage, updateSyncRulesFromYaml } from '@powersync/service-core';
 import { bucketRequest, bucketRequestMap, register, test_utils } from '@powersync/service-core-tests';
 import { describe, expect, test } from 'vitest';
+import { PostgresCompactor } from '../../src/storage/PostgresCompactor.js';
 import { POSTGRES_STORAGE_FACTORY } from './util.js';
 
 describe('Postgres Sync Bucket Storage Compact', () => register.registerCompactTests(POSTGRES_STORAGE_FACTORY));
@@ -31,6 +32,7 @@ bucket_definitions:
         after: { id: 't1' },
         afterReplicaId: test_utils.rid('t1')
       });
+      await batch.markAllSnapshotDone('1/1');
       await batch.commit('1/1');
     });
 
@@ -52,4 +54,65 @@ bucket_definitions:
       { op_id: '2', op: 'PUT', object_id: 't1' }
     ]);
   });
+
+  test('clearBucket fails fast when prefix includes PUT', async () => {
+    // This tests the specific implementation, to check that our operation type guard is working
+    // for CLEAR compacting.
+    await using factory = await POSTGRES_STORAGE_FACTORY.factory();
+    const syncRules = await factory.updateSyncRules(
+      updateSyncRulesFromYaml(`
+bucket_definitions:
+  global:
+    data: [select * from test]
+      `)
+    );
+    const bucketStorage = factory.getInstance(syncRules);
+    const bucket = bucketRequest(syncRules, 'global[]');
+
+    const result = await bucketStorage.startBatch(test_utils.BATCH_OPTIONS, async (batch) => {
+      await batch.markAllSnapshotDone('1/1');
+      await batch.save({
+        sourceTable: TEST_TABLE,
+        tag: storage.SaveOperationTag.INSERT,
+        after: { id: 't1' },
+        afterReplicaId: test_utils.rid('t1')
+      });
+      await batch.save({
+        sourceTable: TEST_TABLE,
+        tag: storage.SaveOperationTag.DELETE,
+        before: { id: 't1' },
+        beforeReplicaId: test_utils.rid('t1')
+      });
+      await batch.save({
+        sourceTable: TEST_TABLE,
+        tag: storage.SaveOperationTag.INSERT,
+        after: { id: 't2' },
+        afterReplicaId: test_utils.rid('t2')
+      });
+      await batch.save({
+        sourceTable: TEST_TABLE,
+        tag: storage.SaveOperationTag.DELETE,
+        before: { id: 't2' },
+        beforeReplicaId: test_utils.rid('t2')
+      });
+      await batch.commit('1/1');
+    });
+
+    const checkpoint = result!.flushed_op;
+    const rowsBefore = await test_utils.oneFromAsync(
+      bucketStorage.getBucketDataBatch(checkpoint, bucketRequestMap(syncRules, [['global[]', 0n]]))
+    );
+    const dataBefore = test_utils.getBatchData(rowsBefore);
+    const clearToOpId = BigInt(dataBefore[2].op_id);
+
+    const compactor = new PostgresCompactor(factory.db, bucketStorage.group_id, {});
+    // Trigger the private method directly
+    await expect(compactor.clearBucketForTests(bucket, clearToOpId)).rejects.toThrow(/Unexpected PUT operation/);
+
+    // The method wraps in a transaction; on assertion error the bucket must remain unchanged.
+    const rowsAfter = await test_utils.oneFromAsync(
+      bucketStorage.getBucketDataBatch(checkpoint, bucketRequestMap(syncRules, [['global[]', 0n]]))
+    );
+    expect(test_utils.getBatchData(rowsAfter)).toEqual(dataBefore);
+  });
 });

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

@@ -596,4 +596,61 @@ bucket_definitions:
     // storage-specific checksum - just check that it does not change
     expect(globalChecksum).toMatchSnapshot();
   });
+
+  test('defaults maxOpId to current checkpoint', async () => {
+    await using factory = await generateStorageFactory();
+    const syncRules = await factory.updateSyncRules(
+      updateSyncRulesFromYaml(`
+bucket_definitions:
+  global:
+    data: [select * from test]
+      `)
+    );
+    const bucketStorage = factory.getInstance(syncRules);
+
+    const result1 = await bucketStorage.startBatch(test_utils.BATCH_OPTIONS, async (batch) => {
+      await batch.markAllSnapshotDone('1/1');
+      await batch.save({
+        sourceTable: TEST_TABLE,
+        tag: storage.SaveOperationTag.INSERT,
+        after: { id: 't1' },
+        afterReplicaId: test_utils.rid('t1')
+      });
+      await batch.commit('1/1');
+    });
+
+    const checkpoint1 = result1!.flushed_op;
+
+    const result2 = await bucketStorage.startBatch(test_utils.BATCH_OPTIONS, async (batch) => {
+      // This is flushed but not committed (does not advance the checkpoint)
+      await batch.save({
+        sourceTable: TEST_TABLE,
+        tag: storage.SaveOperationTag.UPDATE,
+        after: { id: 't1' },
+        afterReplicaId: test_utils.rid('t1')
+      });
+    });
+    const checkpoint2 = result2!.flushed_op;
+
+    const checkpointBeforeCompact = await bucketStorage.getCheckpoint();
+    expect(checkpointBeforeCompact.checkpoint).toEqual(checkpoint1);
+
+    // With default options, Postgres compaction should use the active checkpoint.
+    await bucketStorage.compact({
+      moveBatchLimit: 1,
+      moveBatchQueryLimit: 1,
+      minBucketChanges: 1,
+      minChangeRatio: 0
+    });
+
+    const batchAfterDefaultCompact = await test_utils.oneFromAsync(
+      bucketStorage.getBucketDataBatch(checkpoint2, bucketRequestMap(syncRules, [['global[]', 0n]]))
+    );
+
+    // Operation 1 should remain a PUT because op_id=2 is above the default maxOpId checkpoint.
+    expect(batchAfterDefaultCompact.chunkData.data).toMatchObject([
+      { op_id: '1', op: 'PUT', object_id: 't1' },
+      { op_id: '2', op: 'PUT', object_id: 't1' }
+    ]);
+  });
 }