feat: S3 object storage offloading for V3 bucket data

modules/module-mongodb-storage/package.json

@@ -28,6 +28,8 @@
     }
   },
   "dependencies": {
+    "@aws-sdk/client-s3": "^3.637.0",
+    "@mongodb-js/zstd": "^7.0.0",
     "@powersync/lib-service-mongodb": "workspace:*",
     "@powersync/lib-services-framework": "workspace:*",
     "@powersync/service-core": "workspace:*",

modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts

@@ -20,11 +20,13 @@ import {
 } from './implementation/MongoPersistedSyncRulesContent.js';
 import { syncRuleStateUpdatePipeline } from './implementation/SyncRuleStateUpdate.js';
 import { SyncRuleDocumentV1 } from './implementation/v1/models.js';
+import { ObjectStorage } from './implementation/v3/object-storage/ObjectStorage.js';
 import { VersionedPowerSyncMongoV3 } from './implementation/v3/VersionedPowerSyncMongoV3.js';
 import { ReplicationStreamDocumentV3, SyncConfigDefinition, SyncRuleConfigStateV3 } from './storage-index.js';
 
 export interface MongoBucketStorageOptions {
   checksumOptions?: Omit<MongoChecksumOptions, 'storageConfig' | 'mapping'>;
+  objectStorage?: ObjectStorage;
 }
 
 export class MongoBucketStorage extends storage.BucketStorageFactory {

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

@@ -32,6 +32,7 @@ import { MAX_ROW_SIZE } from './MongoBucketBatchShared.js';
 import { MongoIdSequence } from './MongoIdSequence.js';
 import { batchCreateCustomWriteCheckpoints } from './MongoWriteCheckpointAPI.js';
 import { OperationBatch, RecordOperation } from './OperationBatch.js';
+import { ObjectStorage } from './v3/object-storage/ObjectStorage.js';
 
 // Currently, we can only have a single flush() at a time, since it locks the op_id sequence.
 // While the MongoDB transaction retry mechanism handles this okay, using an in-process Mutex
@@ -63,6 +64,8 @@ export interface MongoBucketBatchOptions {
   tracer?: PerformanceTracer<'storage' | 'evaluate'>;
 
   listSourceRecordCollections?: (groupId: number) => Promise<mongo.Collection<any>[]>;
+
+  objectStorage?: ObjectStorage;
 }
 
 export abstract class MongoBucketBatch

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

@@ -5,6 +5,8 @@ import { MongoStorageConfig } from '../../types/types.js';
 import { MongoBucketStorage } from '../MongoBucketStorage.js';
 import { MongoReportStorage } from '../MongoReportStorage.js';
 import { PowerSyncMongo } from './db.js';
+import { ObjectStorage } from './v3/object-storage/ObjectStorage.js';
+import { S3ObjectStorage } from './v3/object-storage/S3ObjectStorage.js';
 
 export class MongoStorageProvider implements storage.StorageProvider {
   get type() {
@@ -23,6 +25,17 @@ export class MongoStorageProvider implements storage.StorageProvider {
     }
 
     const decodedConfig = MongoStorageConfig.decode(storage as any);
+
+    let objectStorage: ObjectStorage | undefined;
+    if (decodedConfig.object_storage?.type === 's3') {
+      objectStorage = new S3ObjectStorage({
+        bucket: decodedConfig.object_storage.bucket,
+        region: decodedConfig.object_storage.region,
+        prefix: decodedConfig.object_storage.prefix,
+        endpoint: decodedConfig.object_storage.endpoint
+      });
+    }
+
     const client = lib_mongo.db.createMongoClient(decodedConfig, {
       powersyncVersion: POWERSYNC_VERSION,
       maxPoolSize: resolvedConfig.storage.max_pool_size ?? 8
@@ -38,10 +51,16 @@ export class MongoStorageProvider implements storage.StorageProvider {
     await client.connect();
 
     const database = new PowerSyncMongo(client, { database: resolvedConfig.storage.database });
-    const syncStorageFactory = new MongoBucketStorage(database, {
-      // TODO currently need the entire resolved config due to this
-      slot_name_prefix: resolvedConfig.slot_name_prefix
-    });
+    const syncStorageFactory = new MongoBucketStorage(
+      database,
+      {
+        // TODO currently need the entire resolved config due to this
+        slot_name_prefix: resolvedConfig.slot_name_prefix
+      },
+      {
+        objectStorage
+      }
+    );
 
     // Storage factory for reports
     const reportStorageFactory = new MongoReportStorage(database);

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

@@ -36,10 +36,12 @@ import { MongoCompactOptions, MongoCompactor } from './MongoCompactor.js';
 import { MongoParameterCompactor } from './MongoParameterCompactor.js';
 import { MongoPersistedSyncRulesContentV1 } from './MongoPersistedSyncRulesContent.js';
 import { MongoWriteCheckpointAPI } from './MongoWriteCheckpointAPI.js';
+import { ObjectStorage } from './v3/object-storage/ObjectStorage.js';
 
 export interface MongoSyncBucketStorageOptions {
   checksumOptions?: Omit<MongoChecksumOptions, 'storageConfig' | 'mapping'>;
   storageConfig: StorageConfig;
+  objectStorage?: ObjectStorage;
 }
 
 interface InternalCheckpointChanges extends CheckpointChanges {
@@ -75,6 +77,8 @@ export abstract class MongoSyncBucketStorage
 
   readonly checksums: MongoChecksums;
 
+  readonly objectStorage?: ObjectStorage;
+
   private parsedSyncRulesCache: { parsed: HydratedSyncConfig; options: storage.ParseSyncRulesOptions } | undefined;
   private writeCheckpointAPI: MongoWriteCheckpointAPI;
   public readonly logger: Logger;
@@ -91,6 +95,7 @@ export abstract class MongoSyncBucketStorage
   ) {
     super();
     this.storageConfig = options.storageConfig;
+    this.objectStorage = options.objectStorage;
     this.db = factory.db.versioned(this.storageConfig);
     this.checksums = this.createMongoChecksums(options);
     this.writeCheckpointAPI = new MongoWriteCheckpointAPI({
@@ -212,7 +217,8 @@ export abstract class MongoSyncBucketStorage
       markRecordUnavailable: options.markRecordUnavailable,
       hooks: options.hooks,
       syncConfigId: state.syncConfigId,
-      tracer: options.tracer
+      tracer: options.tracer,
+      objectStorage: this.objectStorage
     };
     const writer = this.createWriterImpl(batchOptions);
     this.iterateListeners((cb) => cb.batchStarted?.(writer));

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

@@ -11,6 +11,7 @@ import { currentBucketKey, MAX_ROW_SIZE } from '../MongoBucketBatchShared.js';
 import { MongoIdSequence } from '../MongoIdSequence.js';
 import type { VersionedPowerSyncMongo } from '../db.js';
 import { TaggedBucketParameterDocument } from '../models.js';
+import { ObjectStorage } from '../v3/object-storage/ObjectStorage.js';
 import { BucketDataDoc, BucketKey } from './BucketDataDoc.js';
 import { SourceRecordBucketState, SourceRecordLookupState } from './SourceRecordStore.js';
 
@@ -61,6 +62,7 @@ export interface UpsertCurrentDataOptions {
 
 export interface PersistedBatchOptions {
   logger?: Logger;
+  objectStorage?: ObjectStorage;
 }
 
 /**
@@ -75,6 +77,8 @@ export abstract class PersistedBatch {
   bucketParameters: TaggedBucketParameterDocument[] = [];
   bucketStates: Map<string, BucketStateUpdate> = new Map();
 
+  protected readonly objectStorage?: ObjectStorage;
+
   /**
    * For debug logging only.
    */
@@ -94,6 +98,7 @@ export abstract class PersistedBatch {
   ) {
     this.currentSize = writtenSize;
     this.logger = options?.logger ?? defaultLogger;
+    this.objectStorage = options?.objectStorage;
   }
 
   saveBucketData(options: SaveBucketDataOptions) {

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

@@ -37,7 +37,8 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
 
   protected createPersistedBatch(writtenSize: number): PersistedBatch {
     return new PersistedBatchV3(this.db, this.group_id, this.mapping, writtenSize, {
-      logger: this.logger
+      logger: this.logger,
+      objectStorage: this.options.objectStorage
     });
   }