From b2cf1b3587bbc512ea2f3546875e53dffbd52bf4 Mon Sep 17 00:00:00 2001 From: Ralf Kistner Date: Thu, 10 Sep 2026 12:37:25 +0200 Subject: [PATCH 1/8] Support concurrent chunk-merge compaction. --- .../concurrent-initial-chunk-compaction.md | 5 + .../src/storage/MongoBucketStorage.ts | 10 +- .../implementation/MongoStorageProvider.ts | 8 +- .../implementation/v3/MongoCompactorV3.ts | 213 ++++++++++++++---- .../object-storage/BucketDataObjectStorage.ts | 9 +- .../module-mongodb-storage/src/types/types.ts | 11 + .../src/utils/test-utils.ts | 1 + .../test/src/chunk_compaction_config.test.ts | 23 ++ .../test/src/storage_compacting.test.ts | 118 +++++++++- .../test/src/storage_s3_reading.test.ts | 48 +++- 10 files changed, 389 insertions(+), 57 deletions(-) create mode 100644 .changeset/concurrent-initial-chunk-compaction.md create mode 100644 modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts diff --git a/.changeset/concurrent-initial-chunk-compaction.md b/.changeset/concurrent-initial-chunk-compaction.md new file mode 100644 index 000000000..6639c2ab4 --- /dev/null +++ b/.changeset/concurrent-initial-chunk-compaction.md @@ -0,0 +1,5 @@ +--- +'@powersync/service-module-mongodb-storage': patch +--- + +Overlap V3 initial chunk compaction across concurrent buckets. Configure the shared worker limit with `storage.chunk_compaction_concurrency` (default: 2). diff --git a/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts b/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts index 75632add0..b39cce2ca 100644 --- a/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts +++ b/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; objectStorage?: ObjectStorage; inlineThresholdBytes?: number; + /** Shared across initial chunk-compaction jobs. Default: 2. */ + chunkCompactionConcurrency?: number; /** * Prefix for replication stream name and Postgres logical replication slot name. */ @@ -69,12 +72,17 @@ 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); + // 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; diff --git a/modules/module-mongodb-storage/src/storage/implementation/MongoStorageProvider.ts b/modules/module-mongodb-storage/src/storage/implementation/MongoStorageProvider.ts index 12eff870b..924d0ba36 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/MongoStorageProvider.ts +++ b/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,7 @@ export class MongoStorageProvider implements storage.StorageProvider { } const decodedConfig = MongoStorageConfig.decode(storage as any); + const chunkCompactionConcurrency = normalizeChunkCompactionConcurrency(decodedConfig.chunk_compaction_concurrency); let objectStorage: ObjectStorage | undefined; if (decodedConfig.object_storage?.type === 's3') { @@ -67,6 +72,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), 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 941c5c188..6cc6d472a 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts @@ -1,7 +1,15 @@ import { mongo } from '@powersync/lib-service-mongodb'; import { ReplicationAssertionError, ServiceAssertionError } from '@powersync/lib-services-framework'; -import { addChecksums, formatBytes, InternalOpId, storage, utils } from '@powersync/service-core'; +import { + acquireSemaphoreAbortable, + addChecksums, + formatBytes, + InternalOpId, + storage, + utils +} from '@powersync/service-core'; import { BucketDefinitionId } from '@powersync/service-sync-rules'; +import { setImmediate } from 'node:timers/promises'; import { BucketDataDoc } from '../common/BucketDataDoc.js'; import { BucketDataKey } from '../models.js'; import { ConcurrentCompactionError, MongoCompactor } from '../MongoCompactor.js'; @@ -46,6 +54,7 @@ const DEFAULT_MIN_COMPACT_FULL_INTERVAL_MS = 2 * 60 * 60 * 1000; const DEFAULT_MAX_COMPACT_FULL_INTERVAL_MS = 7 * 24 * 60 * 60 * 1000; const DEFAULT_COMPACT_LEASE_DURATION_MS = 10 * 60 * 1000; const SCHEDULED_COMPACTION_BATCH_SIZE = 100; +const uninterruptibleSignal = new AbortController().signal; interface CompactionGroupResult { documentId: BucketDataKey; @@ -162,7 +171,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC * Batching specifically help to cover cases of many buckets where no compaction is required: * Instead of sequentially claiming and then rescheduling a bucket, this handles it in bulk. * - * Buckets that do need compaction are still claimed and processed sequentially. + * Chunk-only passes overlap a bounded number of buckets. Full compaction stays + * sequential because its working set includes operation deduplication state. * * Any concurrent workers may read the same batch. Rescheduling filters out buckets handled * by a concurrent worker or replication write, while buckets that do need compaction are @@ -182,6 +192,11 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC const dueBefore = new Date(jobStartedAt.getTime() + (options.dueAheadMs ?? 0)); const forceKind = options.forceKind; const rescheduleNotBefore = new Date(dueBefore.getTime() + 1); + // Keep accounting documents bounded by workers, not buckets or scan batches. + const workerUsage = Array.from( + { length: forceKind === CompactionKind.Chunks ? this.storage.factory.chunkCompactionConcurrency : 0 }, + () => new ObjectStorageUsage(this.db, this.group_id, createObjectStorageUsageWriterId()) + ); while (true) { this.signal?.throwIfAborted(); const states = await this.findScheduledBucketBatch(dueBefore); @@ -211,16 +226,19 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC ); await this.rescheduleUnclaimedBuckets(noOpStates, rescheduleNotBefore); - for (const { state, decision, forcedKind } of scheduled) { + const processBucket = async ( + { state, decision, forcedKind }: (typeof scheduled)[number], + objectStorageUsage: ObjectStorageUsage + ) => { const kind = forceKind == null ? decision.kind : forcedKind; if (state.compact_lease == null && kind == null) { - continue; + return; } try { await using lease = await this.claimBucket({ _id: state._id, next_compact_check: { $lte: dueBefore } }); if (lease == null) { - continue; + return; } const claimedDecision = chooseCompactionKind(lease.state, lease.startedAt, this); const claimedKind = @@ -232,7 +250,13 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC // already-published progress. Keep any newer work scheduled. await this.rescheduleClaimedBucket(lease, claimedDecision, rescheduleNotBefore); } else { - await this.compactClaimedBucket(lease, claimedKind, claimedDecision, rescheduleNotBefore); + await this.compactClaimedBucket( + lease, + claimedKind, + claimedDecision, + rescheduleNotBefore, + objectStorageUsage + ); } } catch (error) { if (this.signal?.aborted) { @@ -242,6 +266,63 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC } await this.rescheduleFailedBucket(state, rescheduleNotBefore, error); } + }; + + if (forceKind !== CompactionKind.Chunks) { + for (const entry of scheduled) { + await processBucket(entry, this.objectStorageUsage); + } + } else { + await this.runChunkCompactionWorkers(scheduled, workerUsage, processBucket); + } + } + } + + /** Process one scheduled batch with a fixed pool of workers. */ + private async runChunkCompactionWorkers( + buckets: readonly T[], + workerUsage: readonly ObjectStorageUsage[], + processBucket: (bucket: T, usage: ObjectStorageUsage) => Promise + ): Promise { + const signal = this.signal ?? uninterruptibleSignal; + let nextBucket = 0; + + const runWorker = async (usage: ObjectStorageUsage) => { + while (nextBucket < buckets.length) { + // Taking an entry has no await, so each worker gets a different bucket. + // A worker takes another only after finishing its current bucket. + const bucket = buckets[nextBucket++]; + + // This pool bounds one job; the factory semaphore bounds all jobs together. + // Acquire before claiming the bucket lease, and hold until it is released. + const acquired = await acquireSemaphoreAbortable(this.storage.factory.chunkCompactionSlots, signal); + if (acquired === 'aborted') { + signal.throwIfAborted(); + return; + } + const [, releaseSlot] = acquired; + try { + signal.throwIfAborted(); + await processBucket(bucket, usage); + } finally { + releaseSlot(); + } + // Let replication and other event-loop work run between buckets. + await setImmediate(); + } + }; + + // Concurrent transactions must not all increment the same usage document. + // Reuse one writer for each worker instead of creating one per bucket. + const workers = workerUsage.map(runWorker); + + // Do not release the caller's replication lock or run cleanup while a + // sibling worker still owns a bucket lease or is finishing a replacement. + // Wait for every worker even on failure, then propagate the first error. + const results = await Promise.allSettled(workers); + for (const result of results) { + if (result.status === 'rejected') { + throw result.reason; } } } @@ -315,7 +396,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC lease: CompactionLease, kind: CompactionKind, decision: CompactionDecision, - rescheduleNotBefore?: Date + rescheduleNotBefore?: Date, + objectStorageUsage = this.objectStorageUsage ) { const context = new CompactionContext( lease, @@ -325,7 +407,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC this.compactionTarget(lease.state) ); lease.startRenewal(); - await this.compactSingleBucket(context); + await this.compactSingleBucket(context, objectStorageUsage); } private async rescheduleClaimedBucket(lease: CompactionLease, decision: CompactionDecision, notBefore?: Date) { @@ -361,9 +443,9 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC return new ObjectStorageLifecycle(this.db, this.group_id, this.storage.objectStorage); } - private async compactSingleBucket(context: CompactionContext) { + private async compactSingleBucket(context: CompactionContext, objectStorageUsage: ObjectStorageUsage) { if (context.kind == CompactionKind.Chunks) { - return this.compactSingleBucketChunks(context); + return this.compactSingleBucketChunks(context, objectStorageUsage); } return this.compactSingleBucketFully(context); @@ -375,7 +457,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC * update the persisted checksum state and to decide whether a group can fit * in one chunk. */ - private async compactSingleBucketChunks(context: CompactionContext) { + private async compactSingleBucketChunks(context: CompactionContext, objectStorageUsage: ObjectStorageUsage) { const bucket = context.state._id.b; const resolvedDefinitionId = context.state._id.d; const bucketContext = new BucketDataContextV3(this.db, { @@ -468,7 +550,14 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC const nextSize = pendingSize + doc.size; if (pendingChunks.length > 0 && nextSize > DEFAULT_MAX_DOC_SIZE_BYTES) { - const groupStats = await this.flushChunkMerge(bucket, pendingChunks, collection, dataContext, bucketContext); + const groupStats = await this.flushChunkMerge( + bucket, + pendingChunks, + collection, + dataContext, + bucketContext, + objectStorageUsage + ); compactedTail = combineAdjacentStats(compactedTail, groupStats); pendingChunks = []; pendingSize = 0; @@ -485,7 +574,14 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC } if (pendingChunks.length > 0) { - const groupStats = await this.flushChunkMerge(bucket, pendingChunks, collection, dataContext, bucketContext); + const groupStats = await this.flushChunkMerge( + bucket, + pendingChunks, + collection, + dataContext, + bucketContext, + objectStorageUsage + ); compactedTail = combineAdjacentStats(compactedTail, groupStats); } @@ -519,44 +615,57 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC inputs: BucketDataDocumentV3[], collection: mongo.Collection, context: { replicationStreamId: number; definitionId: string }, - bucketContext: BucketDataContextV3 + bucketContext: BucketDataContextV3, + objectStorageUsage: ObjectStorageUsage ): Promise { if (inputs.length == 1) { return statsForDocument(inputs[0]); } - // The metadata scan deliberately excluded ops. Read inline payloads only - // for this merge group; object-storage payloads are fetched below using - // the same rule. - const inlineInputs = inputs.filter((input) => input.storage_ref == null); - if (inlineInputs.length > 0) { - const inlineDocuments = await collection - .find({ _id: { $in: inlineInputs.map((input) => input._id) } }, { projection: { _id: 1, ops: 1 } }) - .toArray(); - const opsById = new Map(inlineDocuments.map((document) => [document._id.o.toString(), document.ops])); - for (const input of inlineInputs) { - input.ops = opsById.get(input._id.o.toString()); + try { + this.signal?.throwIfAborted(); + + // The metadata scan deliberately excluded ops. Read inline payloads only + // for this merge group; object-storage payloads are fetched below using + // the same rule. + const inlineInputs = inputs.filter((input) => input.storage_ref == null); + if (inlineInputs.length > 0) { + const inlineDocuments = await collection + .find({ _id: { $in: inlineInputs.map((input) => input._id) } }, { projection: { _id: 1, ops: 1 } }) + .toArray(); + const opsById = new Map(inlineDocuments.map((document) => [document._id.o.toString(), document.ops])); + for (const input of inlineInputs) { + input.ops = opsById.get(input._id.o.toString()); + } } - } - await hydrateBucketDataDocuments(inputs, this.storage.objectStorage, { signal: this.signal }); + await hydrateBucketDataDocuments(inputs, this.storage.objectStorage, { signal: this.signal }); - const operations = inputs.flatMap((input) => Array.from(loadBucketDataDocument(context, input))); - const targetOp = inputs.reduce( - (maxTarget, input) => maxOpId(maxTarget, input.target_op), - null - ); - const result = await this.flushCompactionGroup( - bucket, - { - inputs, - ops: operations, - changed: true, - targetOp - }, - bucketContext, - context - ); - return result.stats; + const operations = inputs.flatMap((input) => Array.from(loadBucketDataDocument(context, input))); + const targetOp = inputs.reduce( + (maxTarget, input) => maxOpId(maxTarget, input.target_op), + null + ); + const result = await this.flushCompactionGroup( + bucket, + { + inputs, + ops: operations, + changed: true, + targetOp + }, + bucketContext, + context, + objectStorageUsage + ); + return result.stats; + } finally { + // The scan batch also references these documents. Do not retain hydrated + // operations after finishing this merge group. + for (const input of inputs) { + delete input.ops; + } + await setImmediate(); + } } private async finalizeCompactedBucket({ @@ -935,7 +1044,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC bucket: string, group: PendingCompactionGroup, bucketContext: BucketDataContextV3, - context: { replicationStreamId: number; definitionId: string } + context: { replicationStreamId: number; definitionId: string }, + objectStorageUsage = this.objectStorageUsage ): Promise { if (group.inputs.length == 1 && !group.changed) { return { @@ -993,7 +1103,13 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC writes.deleteMany(bucketContext.collection, { _id: { $in: idsToDelete } }); writes.insertMany(bucketContext.collection, documents); this.finishObjectStorageReplacement(oldStoragePaths, newStoragePaths, uploads, writes); - this.recordObjectStorageReplacement(oldStorageBytes, documents, context.definitionId, writes); + this.recordObjectStorageReplacement( + oldStorageBytes, + documents, + context.definitionId, + writes, + objectStorageUsage + ); await writes.execute(); }, { @@ -1396,7 +1512,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC oldBytes: bigint, newDocuments: Iterable>, definitionId: BucketDefinitionId, - writes: MongoWriteBatch + writes: MongoWriteBatch, + objectStorageUsage = this.objectStorageUsage ): void { if (!this.storage.objectStorage) { return; @@ -1405,7 +1522,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC for (const document of newDocuments) { newBytes += ObjectStorageUsage.bytes(document); } - this.objectStorageUsage.applyDelta(definitionId, newBytes - oldBytes, writes); + objectStorageUsage.applyDelta(definitionId, newBytes - oldBytes, writes); } private async persistBucketData( diff --git a/modules/module-mongodb-storage/src/storage/implementation/v3/object-storage/BucketDataObjectStorage.ts b/modules/module-mongodb-storage/src/storage/implementation/v3/object-storage/BucketDataObjectStorage.ts index 9ae634c39..13ba4f5cd 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/v3/object-storage/BucketDataObjectStorage.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/v3/object-storage/BucketDataObjectStorage.ts @@ -54,9 +54,16 @@ export async function hydrateBucketDataDocuments( return; } using _ = options?.tracer?.span('s3', 'read'); - await Promise.all( + const results = await Promise.allSettled( storedDocuments.map(async (document) => { document.ops = await store.retrieve(document.storage_ref!.path, { signal: options.signal }); }) ); + // Drain every download before propagating failure so callers can release their + // worker slot and payloads without a sibling download hydrating a document later. + for (const result of results) { + if (result.status === 'rejected') { + throw result.reason; + } + } } diff --git a/modules/module-mongodb-storage/src/types/types.ts b/modules/module-mongodb-storage/src/types/types.ts index 32894d7f8..7e5aa2432 100644 --- a/modules/module-mongodb-storage/src/types/types.ts +++ b/modules/module-mongodb-storage/src/types/types.ts @@ -65,6 +65,9 @@ export const MongoStorageConfig = service_types.configFile.BaseStorageConfig.and */ clear_batch_throttle_rate: t.number.optional(), + /** Maximum concurrent buckets across V3 initial chunk-compaction jobs. Default: 2. */ + chunk_compaction_concurrency: t.number.optional(), + object_storage: S3ObjectStorageConfig.optional() }) ); @@ -74,6 +77,14 @@ export type MongoStorageConfigDecoded = t.Decoded; export const DEFAULT_CLEAR_BATCH_THROTTLE_RATE = 0.2; +export function normalizeChunkCompactionConcurrency(value: number | undefined): number { + const concurrency = value ?? 2; + if (!Number.isSafeInteger(concurrency) || concurrency < 1) { + throw new ServiceError(ErrorCode.PSYNC_S3201, 'storage.chunk_compaction_concurrency must be a positive integer'); + } + return concurrency; +} + export function normalizeClearBatchThrottleRate(value: number | undefined): number { const rate = value ?? DEFAULT_CLEAR_BATCH_THROTTLE_RATE; if (!Number.isFinite(rate) || rate < 0 || rate > 20) { diff --git a/modules/module-mongodb-storage/src/utils/test-utils.ts b/modules/module-mongodb-storage/src/utils/test-utils.ts index 922520a54..861bd6cfc 100644 --- a/modules/module-mongodb-storage/src/utils/test-utils.ts +++ b/modules/module-mongodb-storage/src/utils/test-utils.ts @@ -31,6 +31,7 @@ export function mongoTestStorageFactoryGenerator(factoryOptions: MongoTestStorag checksumOptions: factoryOptions.checksumOptions, supportsMultipleSyncConfigs: factoryOptions.supportsMultipleSyncConfigs, objectStorage: factoryOptions.objectStorage, + chunkCompactionConcurrency: factoryOptions.chunkCompactionConcurrency, inlineThresholdBytes: factoryOptions.inlineThresholdBytes }); }, diff --git a/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts b/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts new file mode 100644 index 000000000..67feff130 --- /dev/null +++ b/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts @@ -0,0 +1,23 @@ +import { MongoStorageConfig, normalizeChunkCompactionConcurrency } from '@module/types/types.js'; +import { describe, expect, test } from 'vitest'; + +describe('chunk compaction concurrency configuration', () => { + test('defaults to two workers', () => { + expect(normalizeChunkCompactionConcurrency(undefined)).toBe(2); + }); + + test('decodes a configured worker count', () => { + const config = MongoStorageConfig.decode({ + type: 'mongodb', + uri: 'mongodb://localhost:27017/powersync', + chunk_compaction_concurrency: 4 + }); + expect(normalizeChunkCompactionConcurrency(config.chunk_compaction_concurrency)).toBe(4); + }); + + test.each([0, -1, 1.5, NaN, Infinity, Number.MAX_SAFE_INTEGER + 1])('rejects invalid worker count %s', (value) => { + expect(() => normalizeChunkCompactionConcurrency(value)).toThrow( + 'storage.chunk_compaction_concurrency must be a positive integer' + ); + }); +}); diff --git a/modules/module-mongodb-storage/test/src/storage_compacting.test.ts b/modules/module-mongodb-storage/test/src/storage_compacting.test.ts index 4e05055f8..90f37fa40 100644 --- a/modules/module-mongodb-storage/test/src/storage_compacting.test.ts +++ b/modules/module-mongodb-storage/test/src/storage_compacting.test.ts @@ -7,6 +7,7 @@ import { CompactionLease } from '@module/storage/implementation/v3/CompactionLea import { BucketDataDocumentV3 } from '@module/storage/implementation/v3/models.js'; import { ObjectStorageError } from '@module/storage/implementation/v3/object-storage/ObjectStorage.js'; import { VersionedPowerSyncMongoV3 } from '@module/storage/implementation/v3/VersionedPowerSyncMongoV3.js'; +import { mongoTestStorageFactoryGenerator } from '@module/utils/test-utils.js'; import { replicaIdToSubkey } from '@module/utils/util.js'; import { logger as defaultLogger } from '@powersync/lib-services-framework'; import { @@ -19,6 +20,8 @@ import { import { bucketRequest, compactActive, register, test_utils } from '@powersync/service-core-tests'; import * as bson from 'bson'; import { describe, expect, test, vi } from 'vitest'; +import { env } from './env.js'; +import { MemoryObjectStorage } from './helpers/MemoryObjectStorage.js'; import { INITIALIZED_MONGO_STORAGE_FACTORY, TEST_STORAGE_VERSIONS } from './util.js'; function makeOp( @@ -460,7 +463,7 @@ bucket_definitions: test('aborting scheduled compaction does not reschedule the remaining batch', async () => { const { bucketStorage, collection, bucketStateCollection, ctx, sourceTableId } = await setupV3Storage(); - const buckets = ['first[]', 'second[]', 'third[]']; + const buckets = Array.from({ length: 8 }, (_, index) => `bucket${index}[]`); const documents = buckets.map((bucket) => serializeBucketData(bucket, [makeOp(2, bucket, bucket, { ...ctx, bucket }, sourceTableId)]) ); @@ -499,8 +502,11 @@ bucket_definitions: const states = await bucketStateCollection.find({ '_id.b': { $in: buckets } }).toArray(); const completed = states.filter((state) => state.compacted_state != null); const remaining = states.filter((state) => state.compacted_state == null); - expect(completed).toHaveLength(1); - expect(remaining).toHaveLength(2); + // Already-started buckets may finish after abort, but the queued remainder + // must not be claimed or rescheduled. + expect(completed.length).toBeGreaterThanOrEqual(1); + expect(completed.length).toBeLessThanOrEqual(2); + expect(remaining.length).toBeGreaterThanOrEqual(6); for (const state of remaining) { expect(state.next_compact_check).toEqual(new Date(0)); expect(state.compact_lease).toBeUndefined(); @@ -1637,8 +1643,8 @@ bucket_definitions: describe('Streaming compactor', () => { const BUCKET = 'global[]'; - async function setupV3() { - await using factory = await INITIALIZED_MONGO_STORAGE_FACTORY.factory(); + async function setupV3(factoryGenerator = INITIALIZED_MONGO_STORAGE_FACTORY) { + await using factory = await factoryGenerator.factory(); const syncRules = await factory.updateSyncRules( updateSyncRulesFromYaml( ` @@ -1769,6 +1775,108 @@ bucket_definitions: }); }); + test.each([ + { payloadSize: 100, workers: undefined, abort: false }, + { payloadSize: 450_000, workers: undefined, abort: false }, + { payloadSize: 450_000, workers: 4, abort: false }, + { payloadSize: 100, workers: 1, abort: false }, + { payloadSize: 450_000, workers: 4, abort: true } + ])('chunk compaction shares configured workers across runs: %o', async ({ payloadSize, workers, abort }) => { + const concurrency = workers ?? 2; + const objectStorage = new MemoryObjectStorage(); + const generator = mongoTestStorageFactoryGenerator({ + url: env.MONGO_TEST_URL, + isCI: env.CI, + objectStorage, + chunkCompactionConcurrency: workers, + inlineThresholdBytes: 0 + }); + const { bucketStorage, syncRules, collection, bucketStateCollection, ctx, sourceTableId, db } = + await setupV3(generator); + const bucketCount = 12; + for (let i = 0; i < bucketCount; i++) { + const bucket = `global[${i}]`; + const documents = [1, 2].map((id) => + serializeBucketData(bucket, [makeOp(id, String(id), 'x'.repeat(payloadSize), ctx, sourceTableId)]) + ); + await collection.insertMany(documents); + await bucketStateCollection.insertOne({ + _id: { d: ctx.definitionId, b: bucket }, + last_op: 2n, + next_compact_check: new Date(0), + first_uncompacted_write: new Date(0), + bucket_stats: { + count: 2, + bytes: BigInt(documents.reduce((sum, doc) => sum + doc.size, 0)), + chunks: 2 + } + }); + } + + const gate = Promise.withResolvers(); + const originalPut = objectStorage.put.bind(objectStorage); + let uploads = 0; + const put = vi.spyOn(objectStorage, 'put').mockImplementation(async (...args) => { + uploads++; + await gate.promise; + await originalPut(...args); + }); + const controller = new AbortController(); + let settled = false; + const runs = Promise.allSettled( + Array.from({ length: 2 }, () => + bucketStorage.factory + .getInstance(syncRules) + .compactInitialReplication({ maxOpId: 2n, signal: controller.signal }) + ) + ).finally(() => { + settled = true; + }); + try { + await vi.waitFor(() => expect(uploads).toBe(concurrency)); + // Let other ready workers reach admission; the held uploads must prevent + // any additional payloads from reaching S3, across both compactor instances. + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(uploads).toBe(concurrency); + expect(await bucketStateCollection.countDocuments({ compact_lease: { $exists: true } })).toBeLessThanOrEqual( + concurrency + ); + expect(await collection.countDocuments()).toBe(bucketCount * 2); + if (abort) { + controller.abort(); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(settled).toBe(false); + } + } finally { + gate.resolve(); + await runs; + put.mockRestore(); + } + + const results = await runs; + if (abort) { + expect(results.every((result) => result.status === 'rejected')).toBe(true); + expect(await bucketStateCollection.countDocuments({ compact_lease: { $exists: true } })).toBe(0); + // Cancellation must return all worker slots. + expect(await bucketStorage.compactInitialReplication({ maxOpId: 2n })).toEqual({ buckets: bucketCount }); + } else { + expect(results.every((result) => result.status === 'fulfilled')).toBe(true); + expect(results.reduce((sum, result) => sum + (result.status === 'fulfilled' ? result.value.buckets : 0), 0)).toBe( + bucketCount + ); + } + + const documents = await collection.find().toArray(); + expect(documents).toHaveLength(bucketCount); + expect(documents.every((doc) => doc.count === 2 && doc.checksum === 21n && doc.storage_ref != null)).toBe(true); + expect( + await bucketStateCollection.countDocuments({ 'compacted_state.op_id': 2n, compact_lease: { $exists: false } }) + ).toBe(bucketCount); + const usage = await db.objectStorageUsage.find({ '_id.g': ctx.replicationStreamId }).toArray(); + const activeBytes = usage.reduce((sum, doc) => sum + BigInt(doc.definitions[ctx.definitionId] ?? 0), 0n); + expect(activeBytes).toBe(documents.reduce((sum, doc) => sum + BigInt(doc.storage_ref!.file_size), 0n)); + }); + test('capped chunk compaction repairs stale bucket stats after a committed first merge', async () => { const { bucketStorage, collection, bucketStateCollection, ctx, sourceTableId } = await setupV3(); const operations = [ diff --git a/modules/module-mongodb-storage/test/src/storage_s3_reading.test.ts b/modules/module-mongodb-storage/test/src/storage_s3_reading.test.ts index dc537b479..79d59844b 100644 --- a/modules/module-mongodb-storage/test/src/storage_s3_reading.test.ts +++ b/modules/module-mongodb-storage/test/src/storage_s3_reading.test.ts @@ -2,7 +2,7 @@ import { DeleteObjectsCommand, ListObjectsV2Command } from '@aws-sdk/client-s3'; import { storage, updateSyncRulesFromYaml } from '@powersync/service-core'; import { bucketRequest, test_utils } from '@powersync/service-core-tests'; import * as bson from 'bson'; -import { describe, expect, test } from 'vitest'; +import { describe, expect, test, vi } from 'vitest'; import { MongoSyncBucketStorage } from '../../src/storage/implementation/createMongoSyncBucketStorage.js'; import { VersionedPowerSyncMongoV3 } from '../../src/storage/implementation/v3/VersionedPowerSyncMongoV3.js'; import { hydrateBucketDataDocuments } from '../../src/storage/implementation/v3/object-storage/BucketDataObjectStorage.js'; @@ -576,6 +576,52 @@ describe('S3 object storage reads', () => { expect(maxActiveOperations).toBe(4); }); + test('drains sibling downloads before reporting a hydration failure', async () => { + const objectStorage = new MemoryObjectStorage(); + const gate = Promise.withResolvers(); + const failure = new Error('download failed'); + const documents = ['failed', 'pending'].map((path, index) => ({ + _id: { b: 'bucket', o: BigInt(index + 1) }, + min_op: BigInt(index + 1), + checksum: 0n, + count: 0, + size: 1, + storage_ref: { path, file_size: 1 } + })); + const get = vi.spyOn(objectStorage, 'get').mockImplementation(async (path) => { + if (path === 'failed') { + throw failure; + } + await gate.promise; + return { + data: bson.serialize({ ops: [] }), + metadata: { contentType: 'application/bson', contentEncoding: null } + }; + }); + let settled = false; + const result = hydrateBucketDataDocuments(documents, objectStorage, {}).then( + () => { + settled = true; + return undefined; + }, + (error) => { + settled = true; + return error; + } + ); + try { + await vi.waitFor(() => expect(get).toHaveBeenCalledTimes(2)); + await new Promise((resolve) => setImmediate(resolve)); + expect(settled).toBe(false); + } finally { + gate.resolve(); + await result; + get.mockRestore(); + } + expect(await result).toBe(failure); + expect(documents[1]).toHaveProperty('ops', []); + }); + test('aborts active object downloads', async () => { const objectStorage = new MemoryObjectStorage(); const controller = new AbortController(); From 7f79e7a5476322beee5af417a6ee045be077e7d3 Mon Sep 17 00:00:00 2001 From: Ralf Kistner Date: Thu, 10 Sep 2026 12:50:30 +0200 Subject: [PATCH 2/8] Concurrent chunk-merge compaction also for normal compaction jobs. --- .../concurrent-initial-chunk-compaction.md | 2 +- .../src/storage/MongoBucketStorage.ts | 2 +- .../implementation/v3/MongoCompactorV3.ts | 48 ++++++--- .../module-mongodb-storage/src/types/types.ts | 2 +- .../test/src/storage_compacting.test.ts | 101 ++++++++++++++++++ 5 files changed, 139 insertions(+), 16 deletions(-) diff --git a/.changeset/concurrent-initial-chunk-compaction.md b/.changeset/concurrent-initial-chunk-compaction.md index 6639c2ab4..b2c98c662 100644 --- a/.changeset/concurrent-initial-chunk-compaction.md +++ b/.changeset/concurrent-initial-chunk-compaction.md @@ -2,4 +2,4 @@ '@powersync/service-module-mongodb-storage': patch --- -Overlap V3 initial chunk compaction across concurrent buckets. Configure the shared worker limit with `storage.chunk_compaction_concurrency` (default: 2). +Overlap V3 chunk compaction across concurrent buckets during initial replication and normal scheduled compaction. Configure the shared worker limit with `storage.chunk_compaction_concurrency` (default: 2). Full compactions remain sequential within each job. diff --git a/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts b/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts index b39cce2ca..54d0e39c2 100644 --- a/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts +++ b/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts @@ -43,7 +43,7 @@ export interface MongoBucketStorageOptions { checksumOptions?: Omit; objectStorage?: ObjectStorage; inlineThresholdBytes?: number; - /** Shared across initial chunk-compaction jobs. Default: 2. */ + /** Shared across chunk-compaction jobs. Default: 2. */ chunkCompactionConcurrency?: number; /** * Prefix for replication stream name and Postgres logical replication slot name. 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 6cc6d472a..7f7c432a3 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts @@ -171,7 +171,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC * Batching specifically help to cover cases of many buckets where no compaction is required: * Instead of sequentially claiming and then rescheduling a bucket, this handles it in bulk. * - * Chunk-only passes overlap a bounded number of buckets. Full compaction stays + * Chunk merges overlap a bounded number of buckets. Full compaction stays * sequential because its working set includes operation deduplication state. * * Any concurrent workers may read the same batch. Rescheduling filters out buckets handled @@ -186,15 +186,13 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC // Writers derive next_compact_check from MongoDB's $$NOW. Use the same // clock for the fixed job boundary so clock skew cannot exclude work at // the exact initial-replication interval. - const [{ now: jobStartedAt }] = await this.db.db - .aggregate<{ now: Date }>([{ $documents: [{}] }, { $project: { _id: 0, now: '$$NOW' } }]) - .toArray(); + const jobStartedAt = await this.readCompactionTime(); const dueBefore = new Date(jobStartedAt.getTime() + (options.dueAheadMs ?? 0)); const forceKind = options.forceKind; const rescheduleNotBefore = new Date(dueBefore.getTime() + 1); // Keep accounting documents bounded by workers, not buckets or scan batches. const workerUsage = Array.from( - { length: forceKind === CompactionKind.Chunks ? this.storage.factory.chunkCompactionConcurrency : 0 }, + { length: this.storage.factory.chunkCompactionConcurrency }, () => new ObjectStorageUsage(this.db, this.group_id, createObjectStorageUsageWriterId()) ); while (true) { @@ -203,6 +201,9 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC if (states.length == 0) { break; } + // Keep eligibility bounded by dueBefore, but classify with the current + // server time so buckets that age into full compaction can advance. + const batchStartedAt = await this.readCompactionTime(); const scheduled: { state: BucketStateDocumentV3; @@ -213,7 +214,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC try { scheduled.push({ state, - decision: chooseCompactionKind(state, jobStartedAt, this), + decision: chooseCompactionKind(state, batchStartedAt, this), forcedKind: forcedCompactionKind(state, forceKind, this) }); } catch (error) { @@ -228,7 +229,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC const processBucket = async ( { state, decision, forcedKind }: (typeof scheduled)[number], - objectStorageUsage: ObjectStorageUsage + objectStorageUsage: ObjectStorageUsage, + chunksOnly = false ) => { const kind = forceKind == null ? decision.kind : forcedKind; if (state.compact_lease == null && kind == null) { @@ -243,6 +245,11 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC const claimedDecision = chooseCompactionKind(lease.state, lease.startedAt, this); const claimedKind = forceKind == null ? claimedDecision.kind : forcedCompactionKind(lease.state, forceKind, this); + if (chunksOnly && claimedKind === CompactionKind.Full) { + // The decision changed after scanning. Release the lease without + // rescheduling; the next batch will classify it with a fresh timestamp. + return; + } if (claimedKind == null) { await this.rescheduleClaimedBucket(lease, claimedDecision, rescheduleNotBefore); } else if (this.isCompactionTargetCovered(lease.state, claimedKind)) { @@ -268,16 +275,31 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC } }; - if (forceKind !== CompactionKind.Chunks) { - for (const entry of scheduled) { - await processBucket(entry, this.objectStorageUsage); - } - } else { - await this.runChunkCompactionWorkers(scheduled, workerUsage, processBucket); + const chunkBuckets = scheduled.filter( + ({ decision, forcedKind }) => (forceKind == null ? decision.kind : forcedKind) === CompactionKind.Chunks + ); + const sequentialBuckets = scheduled.filter( + ({ decision, forcedKind }) => (forceKind == null ? decision.kind : forcedKind) !== CompactionKind.Chunks + ); + await this.runChunkCompactionWorkers(chunkBuckets, workerUsage, (entry, usage) => + processBucket(entry, usage, true) + ); + // Full compaction cannot overlap chunk workers from this job, and only + // one full bucket is processed at a time. + for (const entry of sequentialBuckets) { + await processBucket(entry, this.objectStorageUsage); } } } + /** Use MongoDB's clock, matching scheduling and lease timestamps. */ + private async readCompactionTime(): Promise { + const [{ now }] = await this.db.db + .aggregate<{ now: Date }>([{ $documents: [{}] }, { $project: { _id: 0, now: '$$NOW' } }]) + .toArray(); + return now; + } + /** Process one scheduled batch with a fixed pool of workers. */ private async runChunkCompactionWorkers( buckets: readonly T[], diff --git a/modules/module-mongodb-storage/src/types/types.ts b/modules/module-mongodb-storage/src/types/types.ts index 7e5aa2432..1805f4ac4 100644 --- a/modules/module-mongodb-storage/src/types/types.ts +++ b/modules/module-mongodb-storage/src/types/types.ts @@ -65,7 +65,7 @@ export const MongoStorageConfig = service_types.configFile.BaseStorageConfig.and */ clear_batch_throttle_rate: t.number.optional(), - /** Maximum concurrent buckets across V3 initial chunk-compaction jobs. Default: 2. */ + /** Maximum concurrent buckets across V3 chunk-compaction jobs. Default: 2. */ chunk_compaction_concurrency: t.number.optional(), object_storage: S3ObjectStorageConfig.optional() diff --git a/modules/module-mongodb-storage/test/src/storage_compacting.test.ts b/modules/module-mongodb-storage/test/src/storage_compacting.test.ts index 90f37fa40..5504a310e 100644 --- a/modules/module-mongodb-storage/test/src/storage_compacting.test.ts +++ b/modules/module-mongodb-storage/test/src/storage_compacting.test.ts @@ -1877,6 +1877,107 @@ bucket_definitions: expect(activeBytes).toBe(documents.reduce((sum, doc) => sum + BigInt(doc.storage_ref!.file_size), 0n)); }); + test('unforced chunk workers drain before full compaction and reclassify aged buckets on the next scan', async () => { + const { bucketStorage, collection, bucketStateCollection, ctx, sourceTableId } = await setupV3(); + const buckets = ['chunk1[]', 'chunk2[]', 'promoted[]', 'full1[]', 'full2[]']; + for (const bucket of buckets) { + const documents = Array.from({ length: 8 }, (_, index) => + serializeBucketData(bucket, [makeOp(index + 1, String(index), 'data', { ...ctx, bucket }, sourceTableId)]) + ); + await collection.insertMany(documents); + await bucketStateCollection.insertOne({ + _id: { d: ctx.definitionId, b: bucket }, + last_op: 8n, + next_compact_check: new Date(0), + first_uncompacted_write: bucket.startsWith('full') ? new Date(0) : new Date(), + bucket_stats: { count: 8, bytes: BigInt(documents.reduce((sum, doc) => sum + doc.size, 0)), chunks: 8 } + }); + } + + const originalClaim = CompactionLease.claim.bind(CompactionLease); + const jobStartedAt = new Date(); + let serverTime = jobStartedAt; + let promoted = false; + const claim = vi.spyOn(CompactionLease, 'claim').mockImplementation(async (...args) => { + if (!promoted && (args[1]._id as { b: string })?.b === 'promoted[]') { + promoted = true; + // Advance the server clock past the full-compaction interval without + // changing bucket state. A fixed scan timestamp would retry forever. + serverTime = new Date(jobStartedAt.getTime() + 3 * 60 * 60 * 1000); + } + const lease = await originalClaim(...args); + if (lease != null) { + Object.defineProperty(lease, 'startedAt', { value: serverTime }); + } + return lease; + }); + + const compactor = bucketStorage.createMongoCompactor({ maxOpId: 8n }); + const internal = compactor as any; + let timeReads = 0; + const clock = vi.spyOn(internal, 'readCompactionTime').mockImplementation(async () => { + if (++timeReads > 5) { + throw new Error('Compaction did not advance after the bucket aged'); + } + return serverTime; + }); + const scan = vi.spyOn(internal, 'findScheduledBucketBatch'); + const originalChunks = internal.compactSingleBucketChunks.bind(compactor); + const originalFull = internal.compactSingleBucketFully.bind(compactor); + const gate = Promise.withResolvers(); + let activeChunks = 0; + let activeFull = 0; + let maxActiveFull = 0; + const fullBuckets: string[] = []; + const chunks = vi.spyOn(internal, 'compactSingleBucketChunks').mockImplementation(async (...args) => { + activeChunks++; + try { + await gate.promise; + return await originalChunks(...args); + } finally { + activeChunks--; + } + }); + const full = vi.spyOn(internal, 'compactSingleBucketFully').mockImplementation(async (...args) => { + expect(activeChunks).toBe(0); + activeFull++; + maxActiveFull = Math.max(maxActiveFull, activeFull); + fullBuckets.push((args[0] as any).state._id.b); + try { + return await originalFull(...args); + } finally { + activeFull--; + } + }); + const run = compactor.compact(); + try { + await vi.waitFor(() => expect(activeChunks).toBe(2)); + expect(full).not.toHaveBeenCalled(); + } finally { + gate.resolve(); + try { + await run; + expect(timeReads).toBe(3); // Job start, first batch, and promoted bucket's batch. + expect(scan).toHaveBeenCalledTimes(3); // Includes the final empty scan. + for (const [cutoff] of scan.mock.calls) { + expect(cutoff).toEqual(jobStartedAt); + } + } finally { + clock.mockRestore(); + scan.mockRestore(); + claim.mockRestore(); + chunks.mockRestore(); + full.mockRestore(); + } + } + expect(await run).toBe(5); + expect(promoted).toBe(true); + expect(maxActiveFull).toBe(1); + expect(fullBuckets.sort()).toEqual(['full1[]', 'full2[]', 'promoted[]']); + expect(await bucketStateCollection.countDocuments({ compact_lease: { $exists: true } })).toBe(0); + expect(await collection.countDocuments()).toBe(5); + }); + test('capped chunk compaction repairs stale bucket stats after a committed first merge', async () => { const { bucketStorage, collection, bucketStateCollection, ctx, sourceTableId } = await setupV3(); const operations = [ From 5534b3c1f0e4cdb05e90388ebaa08ecc99a9051e Mon Sep 17 00:00:00 2001 From: Ralf Kistner Date: Thu, 10 Sep 2026 13:02:58 +0200 Subject: [PATCH 3/8] Minor refactoring. --- .../implementation/v3/MongoCompactorV3.ts | 7 +++---- .../module-mongodb-storage/src/types/types.ts | 3 ++- packages/service-core/src/sync/util.ts | 5 ++++- .../service-core/test/src/sync/util.test.ts | 19 +++++++++++++++++++ 4 files changed, 28 insertions(+), 6 deletions(-) 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 7f7c432a3..da66bfdf9 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts @@ -54,7 +54,6 @@ const DEFAULT_MIN_COMPACT_FULL_INTERVAL_MS = 2 * 60 * 60 * 1000; const DEFAULT_MAX_COMPACT_FULL_INTERVAL_MS = 7 * 24 * 60 * 60 * 1000; const DEFAULT_COMPACT_LEASE_DURATION_MS = 10 * 60 * 1000; const SCHEDULED_COMPACTION_BATCH_SIZE = 100; -const uninterruptibleSignal = new AbortController().signal; interface CompactionGroupResult { documentId: BucketDataKey; @@ -306,7 +305,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC workerUsage: readonly ObjectStorageUsage[], processBucket: (bucket: T, usage: ObjectStorageUsage) => Promise ): Promise { - const signal = this.signal ?? uninterruptibleSignal; + const signal = this.signal; let nextBucket = 0; const runWorker = async (usage: ObjectStorageUsage) => { @@ -319,12 +318,12 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC // Acquire before claiming the bucket lease, and hold until it is released. const acquired = await acquireSemaphoreAbortable(this.storage.factory.chunkCompactionSlots, signal); if (acquired === 'aborted') { - signal.throwIfAborted(); + signal?.throwIfAborted(); return; } const [, releaseSlot] = acquired; try { - signal.throwIfAborted(); + signal?.throwIfAborted(); await processBucket(bucket, usage); } finally { releaseSlot(); diff --git a/modules/module-mongodb-storage/src/types/types.ts b/modules/module-mongodb-storage/src/types/types.ts index 1805f4ac4..554baa6b1 100644 --- a/modules/module-mongodb-storage/src/types/types.ts +++ b/modules/module-mongodb-storage/src/types/types.ts @@ -76,9 +76,10 @@ export type MongoStorageConfig = t.Encoded; export type MongoStorageConfigDecoded = t.Decoded; export const DEFAULT_CLEAR_BATCH_THROTTLE_RATE = 0.2; +export const DEFAULT_CHUNK_COMPACTION_CONCURRENCY = 2; export function normalizeChunkCompactionConcurrency(value: number | undefined): number { - const concurrency = value ?? 2; + const concurrency = value ?? DEFAULT_CHUNK_COMPACTION_CONCURRENCY; if (!Number.isSafeInteger(concurrency) || concurrency < 1) { throw new ServiceError(ErrorCode.PSYNC_S3201, 'storage.chunk_compaction_concurrency must be a positive integer'); } diff --git a/packages/service-core/src/sync/util.ts b/packages/service-core/src/sync/util.ts index 2a3e6c472..590408357 100644 --- a/packages/service-core/src/sync/util.ts +++ b/packages/service-core/src/sync/util.ts @@ -144,8 +144,11 @@ export async function* transformToBytesTracked( export function acquireSemaphoreAbortable( semaphone: SemaphoreInterface, - abort: AbortSignal + abort?: AbortSignal ): Promise<[number, SemaphoreInterface.Releaser] | 'aborted'> { + if (abort == null) { + return semaphone.acquire(); + } return new Promise((resolve, reject) => { // An already-aborted signal never fires its listener, and we would wait for the semaphore // indefinitely. diff --git a/packages/service-core/test/src/sync/util.test.ts b/packages/service-core/test/src/sync/util.test.ts index 7ff039f94..ee78e2efe 100644 --- a/packages/service-core/test/src/sync/util.test.ts +++ b/packages/service-core/test/src/sync/util.test.ts @@ -15,6 +15,25 @@ describe('isAbortError', () => { }); describe('acquireSemaphoreAbortable', () => { + test('waits for an available slot without a signal', async () => { + const semaphore = new Semaphore(1); + const [, release] = await semaphore.acquire(); + const resolved = vi.fn(); + const waiting = acquireSemaphoreAbortable(semaphore).then((result) => { + resolved(); + return result; + }); + await Promise.resolve(); + expect(resolved).not.toHaveBeenCalled(); + + release(); + const acquired = await waiting; + expect(acquired).not.toBe('aborted'); + expect(semaphore.getValue()).toBe(0); + (acquired as [number, SemaphoreInterface.Releaser])[1](); + expect(semaphore.getValue()).toBe(1); + }); + test('can acquire', async () => { const semaphore = new Semaphore(1); const controller = new AbortController(); From 1f33cf0139d7df23591246b4f6e53b1bf479a844 Mon Sep 17 00:00:00 2001 From: Ralf Kistner Date: Thu, 10 Sep 2026 13:03:42 +0200 Subject: [PATCH 4/8] Max concurrency of 64. --- modules/module-mongodb-storage/src/types/types.ts | 9 ++++++--- .../test/src/chunk_compaction_config.test.ts | 15 +++++++++++---- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/modules/module-mongodb-storage/src/types/types.ts b/modules/module-mongodb-storage/src/types/types.ts index 554baa6b1..df5c571eb 100644 --- a/modules/module-mongodb-storage/src/types/types.ts +++ b/modules/module-mongodb-storage/src/types/types.ts @@ -65,7 +65,7 @@ export const MongoStorageConfig = service_types.configFile.BaseStorageConfig.and */ clear_batch_throttle_rate: t.number.optional(), - /** Maximum concurrent buckets across V3 chunk-compaction jobs. Default: 2. */ + /** Maximum concurrent buckets across V3 chunk-compaction jobs. Range: 1–64. Default: 2. */ chunk_compaction_concurrency: t.number.optional(), object_storage: S3ObjectStorageConfig.optional() @@ -80,8 +80,11 @@ export const DEFAULT_CHUNK_COMPACTION_CONCURRENCY = 2; export function normalizeChunkCompactionConcurrency(value: number | undefined): number { const concurrency = value ?? DEFAULT_CHUNK_COMPACTION_CONCURRENCY; - if (!Number.isSafeInteger(concurrency) || concurrency < 1) { - throw new ServiceError(ErrorCode.PSYNC_S3201, 'storage.chunk_compaction_concurrency must be a positive integer'); + if (!Number.isSafeInteger(concurrency) || concurrency < 1 || concurrency > 64) { + throw new ServiceError( + ErrorCode.PSYNC_S3201, + 'storage.chunk_compaction_concurrency must be an integer between 1 and 64' + ); } return concurrency; } diff --git a/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts b/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts index 67feff130..c306b2198 100644 --- a/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts +++ b/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts @@ -15,9 +15,16 @@ describe('chunk compaction concurrency configuration', () => { expect(normalizeChunkCompactionConcurrency(config.chunk_compaction_concurrency)).toBe(4); }); - test.each([0, -1, 1.5, NaN, Infinity, Number.MAX_SAFE_INTEGER + 1])('rejects invalid worker count %s', (value) => { - expect(() => normalizeChunkCompactionConcurrency(value)).toThrow( - 'storage.chunk_compaction_concurrency must be a positive integer' - ); + test.each([1, 64])('accepts boundary worker count %s', (value) => { + expect(normalizeChunkCompactionConcurrency(value)).toBe(value); }); + + test.each([0, -1, 1.5, 65, NaN, Infinity, Number.MAX_SAFE_INTEGER + 1])( + 'rejects invalid worker count %s', + (value) => { + expect(() => normalizeChunkCompactionConcurrency(value)).toThrow( + 'storage.chunk_compaction_concurrency must be an integer between 1 and 64' + ); + } + ); }); From d101a27d4b4b0e7d9cfb7d251a8dfe38bb3bd4f3 Mon Sep 17 00:00:00 2001 From: Ralf Kistner Date: Thu, 10 Sep 2026 13:05:04 +0200 Subject: [PATCH 5/8] Minor refactor. --- .../implementation/v3/MongoCompactorV3.ts | 58 ++++++++++++++----- 1 file changed, 44 insertions(+), 14 deletions(-) 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 da66bfdf9..ad113489b 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts @@ -469,7 +469,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC return this.compactSingleBucketChunks(context, objectStorageUsage); } - return this.compactSingleBucketFully(context); + return this.compactSingleBucketFully(context, objectStorageUsage); } /** @@ -823,7 +823,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC }; } - private async compactSingleBucketFully(context: CompactionContext) { + private async compactSingleBucketFully(context: CompactionContext, objectStorageUsage: ObjectStorageUsage) { const bucket = context.state._id.b; const resolvedDefinitionId = context.state._id.d; const bucketContext = new BucketDataContextV3(this.db, { @@ -980,7 +980,13 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC }; } else { const flushedGroup = pendingGroup; - const result = await this.flushCompactionGroup(bucket, flushedGroup, bucketContext, dataContext); + const result = await this.flushCompactionGroup( + bucket, + flushedGroup, + bucketContext, + dataContext, + objectStorageUsage + ); compactedStats = combineAdjacentStats(compactedStats, result.stats); if ( lastNotPut != null && @@ -1005,7 +1011,13 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC } if (pendingGroup != null) { - const result = await this.flushCompactionGroup(bucket, pendingGroup, bucketContext, dataContext); + const result = await this.flushCompactionGroup( + bucket, + pendingGroup, + bucketContext, + dataContext, + objectStorageUsage + ); compactedStats = combineAdjacentStats(compactedStats, result.stats); if ( lastNotPut != null && @@ -1031,7 +1043,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC clearBoundary.documentId, bucketContext, collection, - dataContext + dataContext, + objectStorageUsage ); totalOpCount += clearResult.opCountDiff; compactedStats = applyStatsReplacement(compactedStats, clearResult.before, clearResult.after); @@ -1066,7 +1079,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC group: PendingCompactionGroup, bucketContext: BucketDataContextV3, context: { replicationStreamId: number; definitionId: string }, - objectStorageUsage = this.objectStorageUsage + objectStorageUsage: ObjectStorageUsage ): Promise { if (group.inputs.length == 1 && !group.changed) { return { @@ -1161,7 +1174,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC boundaryDocId: BucketDataKey, bucketContext: BucketDataContextV3, collection: mongo.Collection, - context: { replicationStreamId: number; definitionId: string } + context: { replicationStreamId: number; definitionId: string }, + objectStorageUsage: ObjectStorageUsage ): Promise { let opCountDiff = 0; let before = emptyBucketStats(); @@ -1178,7 +1192,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC boundaryDocId, bucketContext, collection, - context + context, + objectStorageUsage ); done = batch.done; opCountDiff += batch.opCountDiff; @@ -1194,7 +1209,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC boundaryDocId, bucketContext, collection, - context + context, + objectStorageUsage ); opCountDiff += boundaryResult.opCountDiff; before = combineAdjacentStats(before, boundaryResult.before); @@ -1212,7 +1228,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC boundaryDocId: BucketDataKey, bucketContext: BucketDataContextV3, collection: mongo.Collection, - context: { replicationStreamId: number; definitionId: string } + context: { replicationStreamId: number; definitionId: string }, + objectStorageUsage: ObjectStorageUsage ): Promise<{ done: boolean; opCountDiff: number } & CompactionStatsReplacement> { const bucket = bucketContext.key.bucket; this.signal?.throwIfAborted(); @@ -1326,7 +1343,13 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC }); writes.insertOne(collection, persisted.documents[0]); this.finishObjectStorageReplacement(oldStoragePaths, persisted.storagePaths, persisted.uploads, writes); - this.recordObjectStorageReplacement(oldStorageBytes, persisted.documents, context.definitionId, writes); + this.recordObjectStorageReplacement( + oldStorageBytes, + persisted.documents, + context.definitionId, + writes, + objectStorageUsage + ); await writes.execute(); opCountDiff = -clearedOpCount + 1; @@ -1348,7 +1371,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC boundaryDocId: BucketDataKey, bucketContext: BucketDataContextV3, collection: mongo.Collection, - context: { replicationStreamId: number; definitionId: string } + context: { replicationStreamId: number; definitionId: string }, + objectStorageUsage: ObjectStorageUsage ): Promise { const bucket = bucketContext.key.bucket; this.signal?.throwIfAborted(); @@ -1473,7 +1497,13 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC }); writes.insertMany(collection, persisted.documents); this.finishObjectStorageReplacement(oldStoragePaths, persisted.storagePaths, persisted.uploads, writes); - this.recordObjectStorageReplacement(oldStorageBytes, persisted.documents, context.definitionId, writes); + this.recordObjectStorageReplacement( + oldStorageBytes, + persisted.documents, + context.definitionId, + writes, + objectStorageUsage + ); await writes.execute(); opCountDiff = -clearedOpCount + 1; @@ -1534,7 +1564,7 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC newDocuments: Iterable>, definitionId: BucketDefinitionId, writes: MongoWriteBatch, - objectStorageUsage = this.objectStorageUsage + objectStorageUsage: ObjectStorageUsage ): void { if (!this.storage.objectStorage) { return; From 6affa10f4a39be01bb7f5448da121ede9bead0f2 Mon Sep 17 00:00:00 2001 From: Ralf Kistner Date: Thu, 10 Sep 2026 13:06:04 +0200 Subject: [PATCH 6/8] Default concurrency to 4 if object-storage is used. --- .changeset/concurrent-initial-chunk-compaction.md | 2 +- .../src/storage/MongoBucketStorage.ts | 7 +++++-- .../src/storage/implementation/MongoStorageProvider.ts | 5 ++++- modules/module-mongodb-storage/src/types/types.ts | 9 ++++++--- .../test/src/chunk_compaction_config.test.ts | 8 ++++++++ .../test/src/storage_compacting.test.ts | 2 +- 6 files changed, 25 insertions(+), 8 deletions(-) diff --git a/.changeset/concurrent-initial-chunk-compaction.md b/.changeset/concurrent-initial-chunk-compaction.md index b2c98c662..1ee5b4994 100644 --- a/.changeset/concurrent-initial-chunk-compaction.md +++ b/.changeset/concurrent-initial-chunk-compaction.md @@ -2,4 +2,4 @@ '@powersync/service-module-mongodb-storage': patch --- -Overlap V3 chunk compaction across concurrent buckets during initial replication and normal scheduled compaction. Configure the shared worker limit with `storage.chunk_compaction_concurrency` (default: 2). Full compactions remain sequential within each job. +Overlap V3 chunk compaction across concurrent buckets during initial replication and normal scheduled compaction. Configure the shared worker limit with `storage.chunk_compaction_concurrency` (default: 4 with object storage, otherwise 2). Full compactions remain sequential within each job. diff --git a/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts b/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts index 54d0e39c2..cf9167044 100644 --- a/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts +++ b/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts @@ -43,7 +43,7 @@ export interface MongoBucketStorageOptions { checksumOptions?: Omit; objectStorage?: ObjectStorage; inlineThresholdBytes?: number; - /** Shared across chunk-compaction jobs. Default: 2. */ + /** 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. @@ -80,7 +80,10 @@ export class MongoBucketStorage extends storage.BucketStorageFactory { private options: MongoBucketStorageOptions ) { super(); - this.chunkCompactionConcurrency = normalizeChunkCompactionConcurrency(options.chunkCompactionConcurrency); + 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; diff --git a/modules/module-mongodb-storage/src/storage/implementation/MongoStorageProvider.ts b/modules/module-mongodb-storage/src/storage/implementation/MongoStorageProvider.ts index 924d0ba36..67d7b39fa 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/MongoStorageProvider.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/MongoStorageProvider.ts @@ -28,7 +28,10 @@ export class MongoStorageProvider implements storage.StorageProvider { } const decodedConfig = MongoStorageConfig.decode(storage as any); - const chunkCompactionConcurrency = normalizeChunkCompactionConcurrency(decodedConfig.chunk_compaction_concurrency); + const chunkCompactionConcurrency = normalizeChunkCompactionConcurrency( + decodedConfig.chunk_compaction_concurrency, + decodedConfig.object_storage != null + ); let objectStorage: ObjectStorage | undefined; if (decodedConfig.object_storage?.type === 's3') { diff --git a/modules/module-mongodb-storage/src/types/types.ts b/modules/module-mongodb-storage/src/types/types.ts index df5c571eb..bfa7cca23 100644 --- a/modules/module-mongodb-storage/src/types/types.ts +++ b/modules/module-mongodb-storage/src/types/types.ts @@ -65,7 +65,7 @@ export const MongoStorageConfig = service_types.configFile.BaseStorageConfig.and */ clear_batch_throttle_rate: t.number.optional(), - /** Maximum concurrent buckets across V3 chunk-compaction jobs. Range: 1–64. Default: 2. */ + /** Maximum concurrent buckets across V3 chunk-compaction jobs. Range: 1–64. Default: 4 with object storage, otherwise 2. */ chunk_compaction_concurrency: t.number.optional(), object_storage: S3ObjectStorageConfig.optional() @@ -77,9 +77,12 @@ export type MongoStorageConfigDecoded = t.Decoded; export const DEFAULT_CLEAR_BATCH_THROTTLE_RATE = 0.2; export const DEFAULT_CHUNK_COMPACTION_CONCURRENCY = 2; +export const DEFAULT_OBJECT_STORAGE_CHUNK_COMPACTION_CONCURRENCY = 4; -export function normalizeChunkCompactionConcurrency(value: number | undefined): number { - const concurrency = value ?? DEFAULT_CHUNK_COMPACTION_CONCURRENCY; +export function normalizeChunkCompactionConcurrency(value: number | undefined, hasObjectStorage = false): number { + const concurrency = + value ?? + (hasObjectStorage ? DEFAULT_OBJECT_STORAGE_CHUNK_COMPACTION_CONCURRENCY : DEFAULT_CHUNK_COMPACTION_CONCURRENCY); if (!Number.isSafeInteger(concurrency) || concurrency < 1 || concurrency > 64) { throw new ServiceError( ErrorCode.PSYNC_S3201, diff --git a/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts b/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts index c306b2198..38f953883 100644 --- a/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts +++ b/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts @@ -6,6 +6,14 @@ describe('chunk compaction concurrency configuration', () => { expect(normalizeChunkCompactionConcurrency(undefined)).toBe(2); }); + test('defaults to four workers with object storage', () => { + expect(normalizeChunkCompactionConcurrency(undefined, true)).toBe(4); + }); + + test.each([false, true])('preserves an explicit override with object storage %s', (hasObjectStorage) => { + expect(normalizeChunkCompactionConcurrency(3, hasObjectStorage)).toBe(3); + }); + test('decodes a configured worker count', () => { const config = MongoStorageConfig.decode({ type: 'mongodb', diff --git a/modules/module-mongodb-storage/test/src/storage_compacting.test.ts b/modules/module-mongodb-storage/test/src/storage_compacting.test.ts index 5504a310e..70c9385f8 100644 --- a/modules/module-mongodb-storage/test/src/storage_compacting.test.ts +++ b/modules/module-mongodb-storage/test/src/storage_compacting.test.ts @@ -1782,7 +1782,7 @@ bucket_definitions: { payloadSize: 100, workers: 1, abort: false }, { payloadSize: 450_000, workers: 4, abort: true } ])('chunk compaction shares configured workers across runs: %o', async ({ payloadSize, workers, abort }) => { - const concurrency = workers ?? 2; + const concurrency = workers ?? 4; const objectStorage = new MemoryObjectStorage(); const generator = mongoTestStorageFactoryGenerator({ url: env.MONGO_TEST_URL, From 3bf3d4e82ba98dc76636e17f47df758e5acb543a Mon Sep 17 00:00:00 2001 From: Ralf Kistner Date: Thu, 10 Sep 2026 13:07:10 +0200 Subject: [PATCH 7/8] Failed workers should stop processing. --- .../implementation/v3/MongoCompactorV3.ts | 49 +++++++++++-------- .../test/src/storage_compacting.test.ts | 43 ++++++++++++++++ 2 files changed, 72 insertions(+), 20 deletions(-) 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 ad113489b..91cb11965 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts @@ -307,29 +307,38 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC ): Promise { const signal = this.signal; let nextBucket = 0; + let failed = false; const runWorker = async (usage: ObjectStorageUsage) => { - while (nextBucket < buckets.length) { - // Taking an entry has no await, so each worker gets a different bucket. - // A worker takes another only after finishing its current bucket. - const bucket = buckets[nextBucket++]; - - // This pool bounds one job; the factory semaphore bounds all jobs together. - // Acquire before claiming the bucket lease, and hold until it is released. - const acquired = await acquireSemaphoreAbortable(this.storage.factory.chunkCompactionSlots, signal); - if (acquired === 'aborted') { - signal?.throwIfAborted(); - return; - } - const [, releaseSlot] = acquired; - try { - signal?.throwIfAborted(); - await processBucket(bucket, usage); - } finally { - releaseSlot(); + try { + while (!failed && nextBucket < buckets.length) { + // Taking an entry has no await, so each worker gets a different bucket. + // A worker takes another only after finishing its current bucket. + const bucket = buckets[nextBucket++]; + + // This pool bounds one job; the factory semaphore bounds all jobs together. + // Acquire before claiming the bucket lease, and hold until it is released. + const acquired = await acquireSemaphoreAbortable(this.storage.factory.chunkCompactionSlots, signal); + if (acquired === 'aborted') { + signal?.throwIfAborted(); + return; + } + const [, releaseSlot] = acquired; + try { + // A sibling may have failed while this worker waited for a slot. + if (failed) return; + signal?.throwIfAborted(); + await processBucket(bucket, usage); + } finally { + releaseSlot(); + } + // Let replication and other event-loop work run between buckets. + await setImmediate(); } - // Let replication and other event-loop work run between buckets. - await setImmediate(); + } catch (error) { + // Drain work already started, but do not let siblings start new buckets. + failed = true; + throw error; } }; diff --git a/modules/module-mongodb-storage/test/src/storage_compacting.test.ts b/modules/module-mongodb-storage/test/src/storage_compacting.test.ts index 70c9385f8..e5e5adc1e 100644 --- a/modules/module-mongodb-storage/test/src/storage_compacting.test.ts +++ b/modules/module-mongodb-storage/test/src/storage_compacting.test.ts @@ -1877,6 +1877,49 @@ bucket_definitions: expect(activeBytes).toBe(documents.reduce((sum, doc) => sum + BigInt(doc.storage_ref!.file_size), 0n)); }); + test.each([2, 3])('failed chunk worker stops queued work and drains siblings (%s workers)', async (workerCount) => { + const { bucketStorage } = await setupV3(); + const compactor = bucketStorage.createMongoCompactor({}) as any; + const failure = new Error('worker failed'); + const failing = Promise.withResolvers(); + const draining = Promise.withResolvers(); + const started: number[] = []; + let settled = false; + // The factory has two slots. With three workers, one also waits for a slot + // when its sibling fails; it must release that slot without starting work. + const run = compactor + .runChunkCompactionWorkers( + [0, 1, 2, 3, 4], + Array.from({ length: workerCount }, () => compactor.objectStorageUsage), + async (bucket: number) => { + started.push(bucket); + if (bucket === 0) await failing.promise; + if (bucket === 1) await draining.promise; + } + ) + .then( + () => undefined, + (error: unknown) => error + ) + .finally(() => { + settled = true; + }); + try { + await vi.waitFor(() => expect(started).toEqual([0, 1])); + failing.reject(failure); + await new Promise((resolve) => setImmediate(resolve)); + expect(started).toEqual([0, 1]); + expect(settled).toBe(false); + } finally { + failing.resolve(); + draining.resolve(); + await run; + } + expect(await run).toBe(failure); + expect(started).toEqual([0, 1]); + expect(bucketStorage.factory.chunkCompactionSlots.getValue()).toBe(2); + }); + test('unforced chunk workers drain before full compaction and reclassify aged buckets on the next scan', async () => { const { bucketStorage, collection, bucketStateCollection, ctx, sourceTableId } = await setupV3(); const buckets = ['chunk1[]', 'chunk2[]', 'promoted[]', 'full1[]', 'full2[]']; From 02c173237f1a05c767592fad1c7c3423fb2b9341 Mon Sep 17 00:00:00 2001 From: Ralf Kistner Date: Thu, 10 Sep 2026 15:35:29 +0200 Subject: [PATCH 8/8] Update changeset. --- .changeset/concurrent-initial-chunk-compaction.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.changeset/concurrent-initial-chunk-compaction.md b/.changeset/concurrent-initial-chunk-compaction.md index 1ee5b4994..2a2142c2a 100644 --- a/.changeset/concurrent-initial-chunk-compaction.md +++ b/.changeset/concurrent-initial-chunk-compaction.md @@ -1,5 +1,6 @@ --- '@powersync/service-module-mongodb-storage': patch +'@powersync/service-core': patch --- -Overlap V3 chunk compaction across concurrent buckets during initial replication and normal scheduled compaction. Configure the shared worker limit with `storage.chunk_compaction_concurrency` (default: 4 with object storage, otherwise 2). Full compactions remain sequential within each job. +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.