diff --git a/.changeset/concurrent-initial-chunk-compaction.md b/.changeset/concurrent-initial-chunk-compaction.md new file mode 100644 index 000000000..2a2142c2a --- /dev/null +++ b/.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. diff --git a/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts b/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts index 75632add0..cf9167044 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 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; diff --git a/modules/module-mongodb-storage/src/storage/implementation/MongoStorageProvider.ts b/modules/module-mongodb-storage/src/storage/implementation/MongoStorageProvider.ts index 12eff870b..67d7b39fa 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,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), 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..91cb11965 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'; @@ -162,7 +170,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 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 * by a concurrent worker or replication write, while buckets that do need compaction are @@ -176,18 +185,24 @@ 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: this.storage.factory.chunkCompactionConcurrency }, + () => new ObjectStorageUsage(this.db, this.group_id, createObjectStorageUsageWriterId()) + ); while (true) { this.signal?.throwIfAborted(); const states = await this.findScheduledBucketBatch(dueBefore); 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; @@ -198,7 +213,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) { @@ -211,20 +226,29 @@ 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, + chunksOnly = false + ) => { 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 = 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)) { @@ -232,7 +256,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 +272,87 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC } await this.rescheduleFailedBucket(state, rescheduleNotBefore, error); } + }; + + 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[], + workerUsage: readonly ObjectStorageUsage[], + processBucket: (bucket: T, usage: ObjectStorageUsage) => Promise + ): Promise { + const signal = this.signal; + let nextBucket = 0; + let failed = false; + + const runWorker = async (usage: ObjectStorageUsage) => { + 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(); + } + } catch (error) { + // Drain work already started, but do not let siblings start new buckets. + failed = true; + throw error; + } + }; + + // 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 +426,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 +437,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,12 +473,12 @@ 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); + return this.compactSingleBucketFully(context, objectStorageUsage); } /** @@ -375,7 +487,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 +580,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 +604,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 +645,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({ @@ -693,7 +832,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, { @@ -850,7 +989,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 && @@ -875,7 +1020,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 && @@ -901,7 +1052,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); @@ -935,7 +1087,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: ObjectStorageUsage ): Promise { if (group.inputs.length == 1 && !group.changed) { return { @@ -993,7 +1146,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(); }, { @@ -1024,7 +1183,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(); @@ -1041,7 +1201,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC boundaryDocId, bucketContext, collection, - context + context, + objectStorageUsage ); done = batch.done; opCountDiff += batch.opCountDiff; @@ -1057,7 +1218,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC boundaryDocId, bucketContext, collection, - context + context, + objectStorageUsage ); opCountDiff += boundaryResult.opCountDiff; before = combineAdjacentStats(before, boundaryResult.before); @@ -1075,7 +1237,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(); @@ -1189,7 +1352,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; @@ -1211,7 +1380,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(); @@ -1336,7 +1506,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; @@ -1396,7 +1572,8 @@ export class MongoCompactorV3 extends MongoCompactor implements CompactIntervalC oldBytes: bigint, newDocuments: Iterable>, definitionId: BucketDefinitionId, - writes: MongoWriteBatch + writes: MongoWriteBatch, + objectStorageUsage: ObjectStorageUsage ): void { if (!this.storage.objectStorage) { return; @@ -1405,7 +1582,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..bfa7cca23 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 chunk-compaction jobs. Range: 1–64. Default: 4 with object storage, otherwise 2. */ + chunk_compaction_concurrency: t.number.optional(), + object_storage: S3ObjectStorageConfig.optional() }) ); @@ -73,6 +76,21 @@ 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 const DEFAULT_OBJECT_STORAGE_CHUNK_COMPACTION_CONCURRENCY = 4; + +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, + 'storage.chunk_compaction_concurrency must be an integer between 1 and 64' + ); + } + return concurrency; +} export function normalizeClearBatchThrottleRate(value: number | undefined): number { const rate = value ?? DEFAULT_CLEAR_BATCH_THROTTLE_RATE; 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..38f953883 --- /dev/null +++ b/modules/module-mongodb-storage/test/src/chunk_compaction_config.test.ts @@ -0,0 +1,38 @@ +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('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', + uri: 'mongodb://localhost:27017/powersync', + chunk_compaction_concurrency: 4 + }); + expect(normalizeChunkCompactionConcurrency(config.chunk_compaction_concurrency)).toBe(4); + }); + + 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' + ); + } + ); +}); 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..e5e5adc1e 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,252 @@ 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 ?? 4; + 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.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[]']; + 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 = [ 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(); 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();