Chunk-merge compact concurrency

.changeset/concurrent-initial-chunk-compaction.md

@@ -0,0 +1,6 @@
+---
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-core': patch
+---
+
+Concurrent storage version 4 chunk-merge compaction across buckets during. Configure the shared worker limit with `storage.chunk_compaction_concurrency` (default: 4 with object storage, otherwise 2). Full compactions remain sequential within each job.

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

@@ -22,8 +22,9 @@ import {
   formatIncrementalSyncConfigUpdateLog,
   isCompatible
 } from '@powersync/service-core';
+import { Semaphore } from 'async-mutex';
 import { ObjectId } from 'bson';
-import { DEFAULT_CLEAR_BATCH_THROTTLE_RATE } from '../types/types.js';
+import { DEFAULT_CLEAR_BATCH_THROTTLE_RATE, normalizeChunkCompactionConcurrency } from '../types/types.js';
 import { generateReplicationStreamName } from '../utils/util.js';
 import type { MongoSyncBucketStorage } from './implementation/createMongoSyncBucketStorage.js';
 import { createMongoSyncBucketStorage } from './implementation/createMongoSyncBucketStorage.js';
@@ -42,6 +43,8 @@ export interface MongoBucketStorageOptions {
   checksumOptions?: Omit<MongoChecksumOptions, 'storageConfig'>;
   objectStorage?: ObjectStorage;
   inlineThresholdBytes?: number;
+  /** Shared across chunk-compaction jobs. Default: 4 with object storage, otherwise 2. */
+  chunkCompactionConcurrency?: number;
   /**
    * Prefix for replication stream name and Postgres logical replication slot name.
    */
@@ -69,12 +72,20 @@ export class MongoBucketStorage extends storage.BucketStorageFactory {
   private activeStorageCache: MongoSyncBucketStorage | undefined;
 
   public readonly db: PowerSyncMongo;
+  public readonly chunkCompactionConcurrency: number;
+  public readonly chunkCompactionSlots: Semaphore;
 
   constructor(
     db: PowerSyncMongo,
     private options: MongoBucketStorageOptions
   ) {
     super();
+    this.chunkCompactionConcurrency = normalizeChunkCompactionConcurrency(
+      options.chunkCompactionConcurrency,
+      options.objectStorage != null
+    );
+    // All replication streams created by this factory share the configured limit.
+    this.chunkCompactionSlots = new Semaphore(this.chunkCompactionConcurrency);
     this.client = db.client;
     this.db = db;
     this.replicationStreamNamePrefix = options.replicationStreamNamePrefix;

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

@@ -1,7 +1,11 @@
 import * as lib_mongo from '@powersync/lib-service-mongodb';
 import { ErrorCode, logger, ServiceAssertionError, ServiceError } from '@powersync/lib-services-framework';
 import { POWERSYNC_VERSION, storage } from '@powersync/service-core';
-import { MongoStorageConfig, normalizeClearBatchThrottleRate } from '../../types/types.js';
+import {
+  MongoStorageConfig,
+  normalizeChunkCompactionConcurrency,
+  normalizeClearBatchThrottleRate
+} from '../../types/types.js';
 import { MongoBucketStorage } from '../MongoBucketStorage.js';
 import { MongoReportStorage } from '../MongoReportStorage.js';
 import { PowerSyncMongo } from './db.js';
@@ -24,6 +28,10 @@ export class MongoStorageProvider implements storage.StorageProvider {
     }
 
     const decodedConfig = MongoStorageConfig.decode(storage as any);
+    const chunkCompactionConcurrency = normalizeChunkCompactionConcurrency(
+      decodedConfig.chunk_compaction_concurrency,
+      decodedConfig.object_storage != null
+    );
 
     let objectStorage: ObjectStorage | undefined;
     if (decodedConfig.object_storage?.type === 's3') {
@@ -67,6 +75,7 @@ export class MongoStorageProvider implements storage.StorageProvider {
             maxStalenessSeconds: decodedConfig.bulk_read_preference == 'primary' ? undefined : 90
           });
     const syncStorageFactory = new MongoBucketStorage(database, {
+      chunkCompactionConcurrency,
       replicationStreamNamePrefix: resolvedConfig.slot_name_prefix,
       readPreference,
       clearBatchThrottleRate: normalizeClearBatchThrottleRate(decodedConfig.clear_batch_throttle_rate),