[MongoDB storage v4] Move chunks to object storage if relevant

.changeset/public-ideas-stare.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-module-mongodb-storage': patch
+---
+
+[MongoDB storage] Move chunks to object storage during compacting, if relevant.

modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts

@@ -645,7 +645,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC
     bucketContext: BucketDataContextV3,
     objectStorageUsage: ObjectStorageUsage
   ): Promise<BucketStatsWithChecksum> {
-    if (inputs.length == 1) {
+    if (inputs.length == 1 && !this.shouldMoveToObjectStorage(inputs[0])) {
       return statsForDocument(inputs[0]);
     }
 
@@ -1093,7 +1093,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC
     context: { replicationStreamId: number; definitionId: string },
     objectStorageUsage: ObjectStorageUsage
   ): Promise<CompactionGroupResult> {
-    if (group.inputs.length == 1 && !group.changed) {
+    if (group.inputs.length == 1 && !group.changed && !this.shouldMoveToObjectStorage(group.inputs[0])) {
       return {
         documentId: group.inputs[0]._id,
         stats: statsForDocument(group.inputs[0])
@@ -1588,6 +1588,14 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC
     objectStorageUsage.applyDelta(definitionId, newBytes - oldBytes, writes);
   }
 
+  private shouldMoveToObjectStorage(document: BucketDataDocumentV3): boolean {
+    return (
+      this.storage.objectStorage != null &&
+      document.storage_ref == null &&
+      document.size > this.storage.inlineThresholdBytes
+    );
+  }
+
   private async persistBucketData(
     bucket: string,
     chunks: BucketDataDoc[][],
@@ -1607,8 +1615,9 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC
     const store = new BucketDataObjectStorage(this.storage.objectStorage);
     const storagePaths = new Set<string>();
     const lifecycle = this.objectStorageLifecycle;
-    // Base placement on the final compacted size. Unchanged documents are not
-    // rewritten, while small MOVE/merge results and CLEAR ops stay inline.
+    // Base placement on the final compacted size. Unchanged documents are only
+    // rewritten to move them to object storage, while small MOVE/merge results
+    // and CLEAR ops stay inline.
     const storedIndexes = serializedChunks.flatMap((document, index) =>
       document.size > this.storage.inlineThresholdBytes ? [index] : []
     );

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

@@ -1,7 +1,7 @@
 import * as lib_mongo from '@powersync/lib-service-mongodb';
 import { storage, updateSyncRulesFromYaml } from '@powersync/service-core';
 import { bucketRequest, compactActive, getTestStorage, test_utils } from '@powersync/service-core-tests';
-import { describe, expect, test } from 'vitest';
+import { describe, expect, test, vi } from 'vitest';
 import { MongoSyncBucketStorage } from '../../src/storage/implementation/createMongoSyncBucketStorage.js';
 import { MongoSyncBucketStorageV3 } from '../../src/storage/implementation/v3/MongoSyncBucketStorageV3.js';
 import { VersionedPowerSyncMongoV3 } from '../../src/storage/implementation/v3/VersionedPowerSyncMongoV3.js';
@@ -53,6 +53,98 @@ async function expectUsageMatchesBucketData(bucketStorage: MongoSyncBucketStorag
 }
 
 describe('S3 compaction storage lifecycle', () => {
+  test.each(
+    [false, true].flatMap((compactChunksOnly) =>
+      [
+        ...[-1, 0, 1].map((thresholdOffset) => ({ thresholdOffset, initiallyInline: true })),
+        { thresholdOffset: -1, initiallyInline: false }
+      ].map((placement) => ({ compactChunksOnly, ...placement }))
+    )
+  )(
+    'unchanged chunk placement: V4, chunks only=$compactChunksOnly, inline=$initiallyInline, threshold offset=$thresholdOffset',
+    async ({ compactChunksOnly, thresholdOffset, initiallyInline }) => {
+      const { memoryStorage, factory: factoryGen } = memoryS3Factory({
+        inlineThresholdBytes: initiallyInline ? 10_000 : 0
+      });
+      await using factory = await factoryGen.factory();
+      const syncRules = await factory.updateSyncRules(
+        updateSyncRulesFromYaml(
+          `
+config:
+  edition: 3
+streams:
+  global:
+    query: SELECT id, description FROM items
+`,
+          { storageVersion: 4 }
+        )
+      );
+      const bucketStorage = (await getTestStorage(factory, syncRules)) as MongoSyncBucketStorageV3;
+      await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS);
+      const sourceTable = await test_utils.resolveTestTable(writer, 'items', ['id'], factoryGen, 1);
+      await writer.markAllSnapshotDone('1/1');
+      await writer.save({
+        sourceTable,
+        tag: storage.SaveOperationTag.INSERT,
+        after: { id: 'A', description: 'unchanged' },
+        afterReplicaId: test_utils.rid('A')
+      });
+      await writer.commit('1/1');
+
+      const definitionId = syncRules.syncConfigContent[0].mapping.allBucketDefinitionIds()[0];
+      const db = bucketStorage.db;
+      const collection = db.bucketData(bucketStorage.replicationStreamId, definitionId);
+      const [before] = await collection.find({}).toArray();
+      expect(before.ops != null).toBe(initiallyInline);
+      expect(before.storage_ref != null).toBe(!initiallyInline);
+      const checkpoint = await bucketStorage.getCheckpoint();
+      const request = bucketRequest(syncRules.syncConfigContent[0], 'global|0[]', 0n);
+      const dataBefore = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request]));
+
+      // Simulate changing the threshold after this chunk was written.
+      const reconfiguredStorage = new MongoSyncBucketStorageV3(
+        factory,
+        bucketStorage.replicationStreamId,
+        bucketStorage.replicationStream,
+        bucketStorage.replicationStreamName,
+        undefined,
+        {
+          storageConfig: bucketStorage.storageConfig,
+          objectStorage: memoryStorage,
+          inlineThresholdBytes: before.size + thresholdOffset
+        }
+      );
+      const put = vi.spyOn(memoryStorage, 'put');
+      expect(
+        await reconfiguredStorage
+          .createMongoCompactor({
+            maxOpId: checkpoint.checkpoint,
+            ...(compactChunksOnly
+              ? { compactChunksOnly: true, forceChunkCompaction: true }
+              : { compactBuckets: [request.bucket] })
+          })
+          .compact()
+      ).toBe(1);
+
+      const documents = await collection.find({}).toArray();
+      expect(documents).toHaveLength(1);
+      const [after] = documents;
+      if (initiallyInline && thresholdOffset < 0) {
+        expect(after.storage_ref).toBeDefined();
+        expect(after.ops).toBeUndefined();
+        expect(put).toHaveBeenCalledTimes(1);
+        const { ops, ...metadata } = before;
+        expect(after).toMatchObject(metadata);
+      } else {
+        expect(after).toEqual(before);
+        expect(put).not.toHaveBeenCalled();
+      }
+      await expectUsageMatchesBucketData(reconfiguredStorage, definitionId);
+      const dataAfter = await test_utils.getBatchArray(reconfiguredStorage.getBucketDataBatch(checkpoint, [request]));
+      expect(test_utils.getBatchData(dataAfter)).toEqual(test_utils.getBatchData(dataBefore));
+    }
+  );
+
   test('a later compaction recovers from a transient object storage failure', async () => {
     const { memoryStorage, factory: factoryGen } = memoryS3Factory();
     await using factory = await factoryGen.factory();