From 404961690347d4dc6f268e726bc7f4c0f8a00fd0 Mon Sep 17 00:00:00 2001 From: Ralf Kistner Date: Thu, 29 Oct 2026 10:16:19 +0200 Subject: [PATCH] [MongoDB storage v4] Move chunks to object storage if relevant. --- .changeset/public-ideas-stare.md | 5 + .../implementation/v3/MongoCompactorV3.ts | 17 +++- .../storage_s3_compaction_lifecycle.test.ts | 94 ++++++++++++++++++- 3 files changed, 111 insertions(+), 5 deletions(-) create mode 100644 .changeset/public-ideas-stare.md diff --git a/.changeset/public-ideas-stare.md b/.changeset/public-ideas-stare.md new file mode 100644 index 000000000..daa781b35 --- /dev/null +++ b/.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. diff --git a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts index b731b79be..35d948dad 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts +++ b/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 { - 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 { - 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(); 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] : [] ); diff --git a/modules/module-mongodb-storage/test/src/storage_s3_compaction_lifecycle.test.ts b/modules/module-mongodb-storage/test/src/storage_s3_compaction_lifecycle.test.ts index f07e8629d..e8d98cd83 100644 --- a/modules/module-mongodb-storage/test/src/storage_s3_compaction_lifecycle.test.ts +++ b/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();