diff --git a/.changeset/hip-views-do.md b/.changeset/hip-views-do.md new file mode 100644 index 000000000..342d5962e --- /dev/null +++ b/.changeset/hip-views-do.md @@ -0,0 +1,8 @@ +--- +'@powersync/service-module-postgres-storage': patch +'@powersync/service-module-mongodb-storage': patch +'@powersync/service-core-tests': patch +'@powersync/service-module-mongodb': patch +--- + +Stability and performance fixes for MongoDB storage V3. diff --git a/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts b/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts index 5765f8b21..9183b7814 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts @@ -276,13 +276,13 @@ export abstract class MongoSyncBucketStorage checkpoint: MongoReplicationCheckpoint, dataBuckets: storage.BucketDataRequest[], options?: storage.BucketDataBatchOptions - ): AsyncIterable; + ): AsyncIterable; async *getBucketDataBatch( checkpoint: storage.ReplicationCheckpoint, dataBuckets: storage.BucketDataRequest[], options?: storage.BucketDataBatchOptions - ): AsyncIterable { + ): AsyncIterable { yield* this.getBucketDataBatchImpl(checkpoint as MongoReplicationCheckpoint, dataBuckets, options); } diff --git a/modules/module-mongodb-storage/src/storage/implementation/common/PersistedBatch.ts b/modules/module-mongodb-storage/src/storage/implementation/common/PersistedBatch.ts index 020c9987e..8a1df88c4 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/common/PersistedBatch.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/common/PersistedBatch.ts @@ -35,7 +35,7 @@ const MAX_TRANSACTION_BATCH_SIZE = 30_000_000; */ const MAX_TRANSACTION_DOC_COUNT = 2_000; -export const DEFAULT_INLINE_THRESHOLD_BYTES = 1024; +export const DEFAULT_INLINE_THRESHOLD_BYTES = 16 * 1024; export interface SaveBucketDataOptions { op_seq: MongoIdSequence; diff --git a/modules/module-mongodb-storage/src/storage/implementation/models.ts b/modules/module-mongodb-storage/src/storage/implementation/models.ts index fb9593d75..dfcb9117a 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/models.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/models.ts @@ -78,6 +78,11 @@ export interface BucketDataProperties { row_id?: string; checksum: bigint; data: string | null; + /** + * V1-only. + * + * V3 stores this on the BucketDataDocumentV3 instead of the individual ops. + */ target_op?: bigint | null; } 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 a6d0ea838..cd4434968 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoCompactorV3.ts @@ -1,12 +1,12 @@ import { mongo } from '@powersync/lib-service-mongodb'; import { logger, ReplicationAssertionError, ServiceAssertionError } from '@powersync/lib-services-framework'; -import { addChecksums, storage, utils } from '@powersync/service-core'; +import { addChecksums, InternalOpId, storage, utils } from '@powersync/service-core'; import { BucketDefinitionId } from '@powersync/service-sync-rules'; import { BucketDataDoc } from '../common/BucketDataDoc.js'; import { BucketDataKey, BucketStateDocumentBase } from '../models.js'; import { ConcurrentCompactionError, DirtyBucket, MongoCompactor } from '../MongoCompactor.js'; import { cacheKey } from '../OperationBatch.js'; -import { loadBucketDataDocument, serializeBucketData } from './bucket-format.js'; +import { loadBucketDataDocument, maxOpId, serializeBucketData } from './bucket-format.js'; import { BucketDataContextV3 } from './BucketDataContextV3.js'; import { DEFAULT_MAX_DOC_SIZE_BYTES } from './chunking.js'; import { BucketDataDocumentV3, BucketStateDocumentV3 } from './models.js'; @@ -24,6 +24,7 @@ interface PendingCompactionGroup { inputs: BucketDataDocumentV3[]; ops: BucketDataDoc[]; changed: boolean; + targetOp: InternalOpId | null; } /** @@ -283,16 +284,17 @@ export class MongoCompactorV3 extends MongoCompactor { let changed = false; const compactedOps: BucketDataDoc[] = []; + let maxTargetOp: InternalOpId | null = doc.target_op ?? null; for (let index = originalOps.length - 1; index >= 0; index--) { const op = originalOps[index]; if (op.op == 'PUT' || op.op == 'REMOVE') { const key = `${op.table}/${op.row_id}/${cacheKey(op.source_table!, op.source_key!)}`; const targetOp = seen.get(key); if (targetOp != null) { + maxTargetOp = maxOpId(maxTargetOp, targetOp); compactedOps.push({ ...op, op: 'MOVE', - target_op: targetOp, table: undefined, row_id: undefined, source_table: undefined, @@ -341,19 +343,21 @@ export class MongoCompactorV3 extends MongoCompactor { const candidate: PendingCompactionGroup = { inputs: [doc], ops: compactedOps, - changed + changed, + targetOp: maxTargetOp }; if (pendingGroup == null) { pendingGroup = candidate; } else { const mergedOps: BucketDataDoc[] = [...candidate.ops, ...pendingGroup.ops]; - const mergedSize = serializeBucketData(bucket, mergedOps).size; + const mergedSize = serializeBucketData(bucket, mergedOps, { targetOp: maxTargetOp }).size; if (mergedSize <= DEFAULT_MAX_DOC_SIZE_BYTES) { pendingGroup = { inputs: [...candidate.inputs, ...pendingGroup.inputs], ops: mergedOps, - changed: candidate.changed || pendingGroup.changed + changed: candidate.changed || pendingGroup.changed, + targetOp: maxOpId(maxTargetOp, pendingGroup.targetOp) }; } else { const flushedGroup = pendingGroup; @@ -432,6 +436,12 @@ export class MongoCompactorV3 extends MongoCompactor { logger.info(`Compacted bucket ${bucket}: ${totalOpCount} surviving ops`); } + /** + * Persist replacement objects before starting the transaction, then atomically + * publish their lifecycle markers alongside the MongoDB document replacement. + * If verification or the transaction fails, the prepared markers retain enough + * information for the uploaded objects to be cleaned up later. + */ private async flushCompactionGroup( bucket: string, group: PendingCompactionGroup, @@ -442,29 +452,17 @@ export class MongoCompactorV3 extends MongoCompactor { return group.inputs[0]._id; } - const [newDoc] = await this.replaceCompactionDocuments(bucket, group.inputs, [group.ops], bucketContext, context); - return newDoc._id; - } - - /** - * Persist replacement objects before starting the transaction, then atomically - * publish their lifecycle markers alongside the MongoDB document replacement. - * If verification or the transaction fails, the prepared markers retain enough - * information for the uploaded objects to be cleaned up later. - */ - private async replaceCompactionDocuments( - bucket: string, - inputs: BucketDataDocumentV3[], - chunks: BucketDataDoc[][], - bucketContext: BucketDataContextV3, - context: { replicationStreamId: number; definitionId: string } - ): Promise { + const inputs = group.inputs; const idsToDelete = inputs.map((doc) => doc._id); const expectedDocCount = inputs.length; const expectedChecksum = inputs.reduce((sum, doc) => sum + doc.checksum, 0n); const expectedOpCount = inputs.reduce((sum, doc) => sum + doc.count, 0); const oldStoragePaths = inputs.flatMap((doc) => (doc.storage_ref ? [doc.storage_ref.path] : [])); - const { documents, storagePaths: newStoragePaths, uploads } = await this.persistBucketData(bucket, chunks, context); + const { + documents, + storagePaths: newStoragePaths, + uploads + } = await this.persistBucketData(bucket, [group.ops], context, undefined, { targetOp: group.targetOp }); const session = this.db.client.startSession(); try { await session.withTransaction( @@ -509,7 +507,7 @@ export class MongoCompactorV3 extends MongoCompactor { } finally { await session.endSession(); } - return documents; + return documents[0]._id; } /** @@ -664,10 +662,11 @@ export class MongoCompactorV3 extends MongoCompactor { o: lastDocId!.o, op: 'CLEAR' as const, checksum: BigInt(combinedChecksum), - data: null, - target_op: maxTargetOp + data: null } satisfies BucketDataDoc; - const persisted = await this.persistBucketData(bucket, [[clearOp]], context, prepared); + const persisted = await this.persistBucketData(bucket, [[clearOp]], context, prepared, { + targetOp: maxTargetOp + }); await collection.insertOne(persisted.documents[0], { session }); await this.finishObjectStorageReplacement(oldStoragePaths, persisted.storagePaths, persisted.uploads, session); @@ -742,6 +741,7 @@ export class MongoCompactorV3 extends MongoCompactor { oldStoragePaths.push(doc.storage_ref.path); } await hydrateBucketDataDocuments([doc], this.storage.objectStorage, { signal: this.signal }); + maxTargetOp = maxOpId(maxTargetOp, doc.target_op); for (const op of loadBucketDataDocument(context, doc)) { if (!isBoundaryDoc && op.op != 'CLEAR') { throw new ReplicationAssertionError( @@ -757,9 +757,6 @@ export class MongoCompactorV3 extends MongoCompactor { } combinedChecksum = addChecksums(combinedChecksum, Number(op.checksum)); clearedOpCount++; - if (op.target_op != null && (maxTargetOp == null || op.target_op > maxTargetOp)) { - maxTargetOp = op.target_op; - } } else if (isBoundaryDoc) { boundarySurvivors.push(op); } else { @@ -790,8 +787,7 @@ export class MongoCompactorV3 extends MongoCompactor { o: lastNotPut, op: 'CLEAR' as const, checksum: BigInt(combinedChecksum), - data: null, - target_op: maxTargetOp + data: null } satisfies BucketDataDoc; const chunks: BucketDataDoc[][] = [[clearOp]]; if (boundarySurvivors.length > 0) { @@ -799,7 +795,9 @@ export class MongoCompactorV3 extends MongoCompactor { // them together cannot increase its stored ops payload. chunks.push(boundarySurvivors); } - const persisted = await this.persistBucketData(bucket, chunks, context, prepared); + const persisted = await this.persistBucketData(bucket, chunks, context, prepared, { + targetOp: maxTargetOp ?? undefined + }); await collection.insertMany(persisted.documents, { session }); await this.finishObjectStorageReplacement(oldStoragePaths, persisted.storagePaths, persisted.uploads, session); @@ -858,9 +856,10 @@ export class MongoCompactorV3 extends MongoCompactor { bucket: string, chunks: BucketDataDoc[][], context: { replicationStreamId: number; definitionId: string }, - preparedUploads?: PreparedObjectStorageUpload[] + preparedUploads?: PreparedObjectStorageUpload[], + options?: { targetOp?: InternalOpId | null } ): Promise<{ documents: BucketDataDocumentV3[]; storagePaths: Set; uploads: PreparedObjectStorageUpload[] }> { - const serializedChunks = chunks.map((chunk) => serializeBucketData(bucket, chunk)); + const serializedChunks = chunks.map((chunk) => serializeBucketData(bucket, chunk, options)); if (!this.storage.objectStorage) { return { documents: serializedChunks, diff --git a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts index a4b313003..f67dff166 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts @@ -24,7 +24,7 @@ import { MongoCompactOptions, MongoCompactor } from '../MongoCompactor.js'; import { MongoParameterCompactor } from '../MongoParameterCompactor.js'; import { MongoPersistedReplicationStream } from '../MongoPersistedReplicationStream.js'; import { MongoSyncBucketStorage, MongoSyncBucketStorageOptions } from '../MongoSyncBucketStorage.js'; -import { loadBucketDataDocument } from './bucket-format.js'; +import { loadBucketDataDocument, maxOpId } from './bucket-format.js'; import { BucketDataDocumentV3, BucketParameterDocumentV3, @@ -59,67 +59,50 @@ export interface MongoSyncBucketStorageContextV3 { const BUCKET_DATA_FETCH_BATCH_LIMIT_BYTES = 16 * 1024 * 1024; /** - * Keep the documents hydrated for one sync response within a bounded payload. - * The first document is always included so an oversized operation cannot prevent - * forward progress. + * Keep the documents hydrated for one sync response within a bounded payload or slightly higher. + * + * This deserializes on-demand, so no deserialization is performed for discarded data. */ -function cutBucketDataBatch(documents: BucketDataDocumentV3[]): { +function cutBucketDataBatch(rawDocuments: Buffer[]): { documents: BucketDataDocumentV3[]; wasCut: boolean; } { let cumulativeBytes = 0; - for (let index = 0; index < documents.length; index++) { - cumulativeBytes += documents[index].size; - if (cumulativeBytes > BUCKET_DATA_FETCH_BATCH_LIMIT_BYTES && index > 0) { + let documents: BucketDataDocumentV3[] = []; + for (const raw of rawDocuments) { + const doc = bson.deserialize(raw, storage.BSON_DESERIALIZE_INTERNAL_OPTIONS) as BucketDataDocumentV3; + documents.push(doc); + cumulativeBytes += doc.size; + if (cumulativeBytes > BUCKET_DATA_FETCH_BATCH_LIMIT_BYTES) { return { - documents: documents.slice(0, index), - wasCut: true + documents, + wasCut: documents.length < rawDocuments.length }; } } return { documents, wasCut: false }; } -function* walkDocumentOps( - data: BucketDataDoc[], - documentOpCounts: number[] -): Generator<{ row: BucketDataDoc; docIndex: number; isLastOpInDocument: boolean }> { - let opIndex = 0; - for (const [docIndex, opCount] of documentOpCounts.entries()) { - for (let i = 0; i < opCount; i++) { - yield { row: data[opIndex++], docIndex, isLastOpInDocument: i === opCount - 1 }; - } - } -} - function extractRowsFromDocument( doc: BucketDataDocumentV3, context: { replicationStreamId: number; definitionId: string }, - bucketMap: Map, - endOpId: InternalOpId, - remainingLimit: number -): { rows: BucketDataDoc[]; remainingLimit: number; limitReached: boolean } { + bucketStart: InternalOpId, + endOpId: InternalOpId +): BucketDataDoc[] { const rows: BucketDataDoc[] = []; for (const row of loadBucketDataDocument(context, doc)) { - const bucket = row.bucketKey.bucket; - const bucketStart = bucketMap.get(bucket); - if (bucketStart == null) { - throw new Error(`data for unexpected bucket: ${bucket}`); - } + // In theory a binary search could be faster than a linear scan to find the start. + // In practice, most cases should not filter out anything here. if (row.o <= bucketStart) { continue; } if (row.o > endOpId) { - continue; + break; } rows.push(row); - remainingLimit--; - if (remainingLimit <= 0) { - return { rows, remainingLimit, limitReached: true }; - } } - return { rows, remainingLimit, limitReached: false }; + return rows; } export class MongoSyncBucketStorageV3 extends MongoSyncBucketStorage { @@ -385,7 +368,7 @@ export class MongoSyncBucketStorageV3 extends MongoSyncBucketStorage { checkpoint: MongoSyncBucketStorageCheckpoint, dataBuckets: storage.BucketDataRequest[], options?: storage.BucketDataBatchOptions - ): AsyncIterable { + ): AsyncIterable { return getBucketDataBatchV3(this.versionContext, checkpoint, dataBuckets, options); } @@ -577,7 +560,7 @@ export async function* getBucketDataBatchV3( checkpoint: MongoSyncBucketStorageCheckpoint, dataBuckets: storage.BucketDataRequest[], options?: storage.BucketDataBatchOptions -): AsyncIterable { +): AsyncIterable { if (dataBuckets.length == 0) { return; } @@ -602,8 +585,9 @@ export async function* getBucketDataBatchV3( const batchLimit = options?.limit ?? storage.DEFAULT_DOCUMENT_BATCH_LIMIT; const chunkSizeLimitBytes = options?.chunkLimitBytes ?? storage.DEFAULT_DOCUMENT_CHUNK_LIMIT_BYTES; const end = checkpoint.checkpoint; - let remainingLimit = batchLimit; + // Group requests by definition, so that we can query each definition's bucket data in a single query. + // We only return the results of a single query per batch. const requestsByDefinition = new Map(); for (const request of dataBuckets) { const definitionId = ctx.mapping.bucketSourceId(request.source); @@ -614,9 +598,6 @@ export async function* getBucketDataBatchV3( const definitionGroups = Array.from(requestsByDefinition.entries()); for (const [groupIndex, [definitionId, requests]] of definitionGroups.entries()) { - if (remainingLimit <= 0) { - break; - } const hasLaterDefinitionGroups = groupIndex < definitionGroups.length - 1; const bucketMap = new Map(requests.map((request) => [request.bucket, request.start])); const filters = Array.from(bucketMap.entries()).map(([bucket, start]) => ({ @@ -632,9 +613,8 @@ export async function* getBucketDataBatchV3( // MongoDB Filter doesn't accept the $or operator in its type. const filter = { $or: filters } as unknown as mongo.Filter; const context = { replicationStreamId: ctx.replicationStreamId, definitionId }; - const limit = remainingLimit; - const cursorOptions = { limit: remainingLimit, batchSize: remainingLimit + 1 }; + const cursorOptions = { limit: batchLimit, batchSize: batchLimit + 1 }; // raw: true returns Buffers, but the driver typing doesn't reflect that // without an explicit cast to FindCursor. @@ -652,97 +632,44 @@ export async function* getBucketDataBatchV3( throw lib_mongo.mapQueryError(e, 'while reading bucket data'); }); - if (cursorOptions.limit != null && rawData.length >= cursorOptions.limit) { + if (rawData.length >= cursorOptions.limit) { hasMore = true; } - const data: BucketDataDoc[] = []; - const documentOpCounts: number[] = []; - let sharedRemainingLimit = limit; - let limitReached = false; - // Buckets whose matched document contributed no rows after filtering. - const completeEmptyBuckets = new Set(); - - // Deserialize all docs once - const deserializedDocs: BucketDataDocumentV3[] = rawData.map( - (raw) => bson.deserialize(raw, storage.BSON_DESERIALIZE_INTERNAL_OPTIONS) as BucketDataDocumentV3 - ); - const cutBatch = cutBucketDataBatch(deserializedDocs); + // Deserialize the raw documents and cut the batch to a bounded size. Any data not + // making the cut will be read in the next round, using a fresh query. + const cutBatch = cutBucketDataBatch(rawData); const docs = cutBatch.documents; if (cutBatch.wasCut) { hasMore = true; } + // Hydrate any operations from object storage. + // In the future we can do this in a more pipelined fashion, but for now we hydrate + // the entire batch at once. await hydrateBucketDataDocuments(docs, ctx.objectStorage, { signal: options?.signal }); - for (const doc of docs) { - const { - rows, - remainingLimit, - limitReached: docLimitReached - } = extractRowsFromDocument(doc, context, bucketMap, end, sharedRemainingLimit); - if (rows.length == 0) { - // The document straddles the requested (start, end] window: it matched the - // query, but none of its ops are in range. Since its _id.o (max op) must be - // > end (any op <= end would have been > start, and thus in range), and - // document ranges per bucket are disjoint, no later document for this bucket - // can match either. The bucket is complete through the checkpoint. - completeEmptyBuckets.add(doc._id.b); - } - data.push(...rows); - documentOpCounts.push(rows.length); - sharedRemainingLimit = remainingLimit; - if (docLimitReached) { - limitReached = true; - break; - } - } - - const batchHasMore = hasMore || limitReached; - - // Empty chunks are not forwarded to clients, but report progress to the caller: - // the bucket's position advances to the checkpoint, so it is not re-requested. - // If the batch produced no data at all, the last empty chunk also carries the - // has_more signal, so the caller re-requests the remaining buckets instead of - // treating an all-filtered batch as the end of the stream. - const emptyBuckets = Array.from(completeEmptyBuckets); - for (const [index, bucket] of emptyBuckets.entries()) { - const startOpId = bucketMap.get(bucket); - if (startOpId == null) { - throw new ServiceAssertionError(`data for unexpected bucket: ${bucket}`); - } - const isLastChunkOfBatch = data.length == 0 && index == emptyBuckets.length - 1; - yield { - chunkData: { - bucket, - after: internalToExternalOpId(startOpId), - has_more: isLastChunkOfBatch && batchHasMore, - data: [], - next_after: internalToExternalOpId(end) - }, - targetOp: null - }; - } - - if (data.length == 0) { - if (batchHasMore) { - // The remaining documents are read in the next round, after the caller has - // advanced the positions of the empty buckets above. - return; - } - continue; - } - - remainingLimit -= data.length; - let currentChunkSizeBytes = 0; let currentChunk: utils.SyncBucketData | null = null; let targetOp: InternalOpId | null = null; + let seenBuckets = new Set(); + const batchHasMore = hasMore; - for (const { row, docIndex, isLastOpInDocument } of walkDocumentOps(data, documentOpCounts)) { - const bucket = row.bucketKey.bucket; + for (const doc of docs) { + const bucket = doc._id.b; + seenBuckets.add(bucket); + const bucketStart = bucketMap.get(bucket); + if (bucketStart == null) { + throw new ServiceAssertionError(`data for unexpected bucket: ${bucket}`); + } - if (currentChunk == null || currentChunk.bucket != bucket || currentChunkSizeBytes >= chunkSizeLimitBytes) { + // Reached a new bucket or size limit: yield the current chunk and start a new one. + if ( + currentChunk == null || + currentChunk.bucket != bucket || + currentChunkSizeBytes >= chunkSizeLimitBytes || + currentChunk.data.length >= batchLimit + ) { let start: ProtocolOpId | undefined = undefined; if (currentChunk != null) { if (currentChunk.bucket == bucket) { @@ -758,12 +685,9 @@ export async function* getBucketDataBatchV3( } if (start == null) { - const startOpId = bucketMap.get(bucket); - if (startOpId == null) { - throw new Error(`data for unexpected bucket: ${bucket}`); - } - start = internalToExternalOpId(startOpId); + start = internalToExternalOpId(bucketStart); } + currentChunk = { bucket, after: start, @@ -773,27 +697,47 @@ export async function* getBucketDataBatchV3( }; } - const entry = mapOpEntry(row); - if (row.target_op != null && (targetOp == null || row.target_op > targetOp)) { - targetOp = row.target_op; - } - - currentChunk.data.push(entry); - currentChunk.next_after = entry.op_id; - - if (isLastOpInDocument) { - currentChunkSizeBytes += docs[docIndex].size; - } + const rows = extractRowsFromDocument(doc, context, bucketStart, end); + currentChunk.data.push(...rows.map(mapOpEntry)); + currentChunk.next_after = currentChunk.data.at(-1)?.op_id ?? internalToExternalOpId(end); + targetOp = maxOpId(targetOp, doc.target_op); + currentChunkSizeBytes += doc.size; } if (currentChunk != null) { const yieldChunk = currentChunk; - yieldChunk.has_more = batchHasMore || (remainingLimit <= 0 && hasLaterDefinitionGroups); + // The last chunk may contain more data that was cut in this batch. + yieldChunk.has_more = batchHasMore; yield { chunkData: yieldChunk, targetOp }; } - if (batchHasMore || remainingLimit <= 0) { - return; + if (!batchHasMore) { + for (const bucket of bucketMap.keys()) { + if (!seenBuckets.has(bucket)) { + // We processed everything for this definition group, but this bucket had no data in the batch. + // Yield an empty chunk to indicate that it is complete. + // This prevents re-querying the same bucket in the next batch. + yield { + chunkData: { + bucket, + after: internalToExternalOpId(bucketMap.get(bucket)!), + has_more: false, + data: [], + next_after: internalToExternalOpId(end) + }, + targetOp + }; + } + } + } + + if (currentChunk != null) { + // We yielded data in this group (aside from empty buckets). + // Return to the caller to allow them to process it before continuing to the next group. + yield { hasMore: batchHasMore || hasLaterDefinitionGroups }; + break; + } else { + // No data in this definition group - continue in the next group. } } } diff --git a/modules/module-mongodb-storage/src/storage/implementation/v3/PersistedBatchV3.ts b/modules/module-mongodb-storage/src/storage/implementation/v3/PersistedBatchV3.ts index f381ababc..563b65033 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/v3/PersistedBatchV3.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/v3/PersistedBatchV3.ts @@ -1,6 +1,6 @@ import { mongo } from '@powersync/lib-service-mongodb'; import { ReplicationAssertionError } from '@powersync/lib-services-framework'; -import { InternalOpId, storage } from '@powersync/service-core'; +import { BucketDefinitionMapping, InternalOpId, storage } from '@powersync/service-core'; import { BucketDataSource, BucketDefinitionId } from '@powersync/service-sync-rules'; import * as bson from 'bson'; import { mongoTableId } from '../../../utils/util.js'; @@ -8,6 +8,7 @@ import { BucketDataDoc } from '../common/BucketDataDoc.js'; import { BucketStateUpdate, PersistedBatch, + PersistedBatchOptions, SaveParameterDataOptions, UpsertCurrentDataOptions } from '../common/PersistedBatch.js'; @@ -28,9 +29,23 @@ import { VersionedPowerSyncMongoV3 } from './VersionedPowerSyncMongoV3.js'; export class PersistedBatchV3 extends PersistedBatch { currentData: { sourceTableId: bson.ObjectId; operation: mongo.AnyBulkWriteOperation }[] = []; sourceTablePendingDeletes = new Map(); + protected readonly objectStorageLifecycle?: ObjectStorageLifecycle; declare protected readonly db: VersionedPowerSyncMongoV3; + constructor( + db: VersionedPowerSyncMongoV3, + group_id: number, + mapping: BucketDefinitionMapping, + writtenSize: number, + options?: PersistedBatchOptions + ) { + super(db, group_id, mapping, writtenSize, options); + if (this.objectStorage) { + this.objectStorageLifecycle = new ObjectStorageLifecycle(this.db, this.group_id, this.objectStorage); + } + } + // Abstract override from PersistedBatch (V3-specific error message) protected override checkDefinitionId(definitionId: BucketDefinitionId | null): BucketDefinitionId { @@ -210,65 +225,63 @@ export class PersistedBatchV3 extends PersistedBatch { operationsByDefinition.set(document.bucketKey.definitionId, existing); } - for (const [definitionId, documents] of operationsByDefinition.entries()) { - const operationsByBucket = new Map(); - for (const document of documents) { - const existing = operationsByBucket.get(document.bucketKey.bucket) ?? []; - existing.push(document); - operationsByBucket.set(document.bucketKey.bucket, existing); - } - - const inserts: mongo.AnyBulkWriteOperation[] = []; - - if (!this.objectStorage) { - for (const [bucket, ops] of operationsByBucket.entries()) { - const chunks = chunkBucketData(ops); - for (const chunk of chunks) { - inserts.push({ + let uploadCount = 0; + const plans = Array.from(operationsByDefinition, ([definitionId, documents]) => { + const operationsByBucket = Map.groupBy(documents, (document) => document.bucketKey.bucket); + const lifecycle = this.objectStorageLifecycle; + const createInserts: (() => Promise>)[] = []; + + for (const [bucket, ops] of operationsByBucket) { + for (const chunk of chunkBucketData(ops)) { + const serialized = serializeBucketData(bucket, chunk); + if (lifecycle == null || serialized.size <= this.inlineThresholdBytes) { + createInserts.push(async () => ({ insertOne: { - document: serializeBucketData(bucket, chunk) + document: serialized } - }); + })); + continue; } - } - } else { - const lifecycle = new ObjectStorageLifecycle(this.db, this.group_id, this.objectStorage); - for (const [bucket, ops] of operationsByBucket.entries()) { - const chunks = chunkBucketData(ops); - for (const chunk of chunks) { + uploadCount += 1; + createInserts.push(async () => { const minOp = chunk[0].o; const maxOp = chunk[chunk.length - 1].o; - const serialized = serializeBucketData(bucket, chunk); const { ops: bucketOps, ...metadata } = serialized; - - if (serialized.size <= this.inlineThresholdBytes) { - // Small enough to store inline - inserts.push({ - insertOne: { - document: serialized - } - }); - } else { - const path = lifecycle.allocatePath(definitionId, bucket, minOp, maxOp); - const { fileSize } = await lifecycle.bucketData.store(path, bucketOps!); - - inserts.push({ - insertOne: { - document: { - ...metadata, - storage_ref: { - path, - file_size: fileSize - } + const path = lifecycle.allocatePath(definitionId, bucket, minOp, maxOp); + const { fileSize } = await lifecycle.bucketData.store(path, bucketOps!); + return { + insertOne: { + document: { + ...metadata, + storage_ref: { + path, + file_size: fileSize } } - }); - } - } + } + }; + }); } } + return { definitionId, createInserts }; + }); + + const createAllInserts = () => + Promise.all( + plans.map(async ({ definitionId, createInserts }) => ({ + definitionId, + inserts: await Promise.all(createInserts.map((createInsert) => createInsert())) + })) + ); + + // S3ObjectStorage applies one shared concurrency limit across all callers, + // so replication can schedule its uploads together without creating a + // separate limiter here. + const writes = await createAllInserts(); + + for (const { definitionId, inserts } of writes) { if (inserts.length > 0) { await this.db.bucketData(this.group_id, definitionId).bulkWrite(inserts, { session, diff --git a/modules/module-mongodb-storage/src/storage/implementation/v3/bucket-format.ts b/modules/module-mongodb-storage/src/storage/implementation/v3/bucket-format.ts index 97c984d72..2014d4d71 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/v3/bucket-format.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/v3/bucket-format.ts @@ -1,4 +1,5 @@ -import { bson } from '@powersync/service-core'; +import { ServiceAssertionError } from '@powersync/lib-services-framework'; +import { bson, InternalOpId } from '@powersync/service-core'; import { BucketDataDoc, BucketKey } from '../common/BucketDataDoc.js'; import { BucketDataDocumentV3, BucketOperation } from './models.js'; @@ -10,21 +11,18 @@ import { BucketDataDocumentV3, BucketOperation } from './models.js'; export function serializeBucketData( bucket: string, operations: BucketDataDoc[], - options?: { compactionTargetOp?: bigint } + options?: { targetOp?: InternalOpId | null } ): BucketDataDocumentV3 { const minOp = operations[0].o; const maxOp = operations[operations.length - 1].o; let totalChecksum = 0n; - let maxTargetOp: bigint | null = options?.compactionTargetOp ?? null; + let maxTargetOp: InternalOpId | null = options?.targetOp ?? null; let hasClearOp = false; const ops: BucketOperation[] = operations.map((op) => { totalChecksum += op.checksum; - if (op.target_op != null && (maxTargetOp == null || op.target_op > maxTargetOp)) { - maxTargetOp = op.target_op; - } if (op.op == 'CLEAR') { hasClearOp = true; } @@ -64,7 +62,7 @@ export function* loadBucketDataDocument( ): Generator { const { _id, ops } = doc; if (!ops) { - throw new Error( + throw new ServiceAssertionError( `Missing ops array on BucketDataDocumentV3 at _id.o=${_id.o}. Callers must patch doc.ops from S3 before calling this function.` ); } @@ -83,8 +81,17 @@ export function* loadBucketDataDocument( table: op.table, row_id: op.row_id, checksum: op.checksum, - data: op.data, - target_op: doc.target_op ?? null + data: op.data }; } } + +export function maxOpId(a: InternalOpId | null | undefined, b: InternalOpId | null | undefined): InternalOpId | null { + if (a == null) { + return b ?? null; + } + if (b == null) { + return a ?? null; + } + return a > b ? a : b; +} diff --git a/modules/module-mongodb-storage/test/src/__snapshots__/storage_sync.test.ts.snap b/modules/module-mongodb-storage/test/src/__snapshots__/storage_sync.test.ts.snap index a42306be5..c55031dd1 100644 --- a/modules/module-mongodb-storage/test/src/__snapshots__/storage_sync.test.ts.snap +++ b/modules/module-mongodb-storage/test/src/__snapshots__/storage_sync.test.ts.snap @@ -589,200 +589,6 @@ exports[`sync - mongodb > storage v1 > sync global data 1`] = ` ] `; -exports[`sync - mongodb > storage v1 > sync interrupts low-priority buckets on new checkpoints (2) 1`] = ` -[ - { - "checkpoint": { - "buckets": [ - { - "bucket": "b0a[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "b0b[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "b1[]", - "checksum": -1096116670, - "count": 1, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "last_op_id": "4001", - "streams": [ - { - "errors": [], - "is_default": true, - "name": "b0a", - }, - { - "errors": [], - "is_default": true, - "name": "b0b", - }, - { - "errors": [], - "is_default": true, - "name": "b1", - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "0", - "bucket": "b1[]", - "data": undefined, - "has_more": false, - "next_after": "1", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4001", - "priority": 1, - }, - }, - { - "data": { - "after": "0", - "bucket": "b0a[]", - "data": undefined, - "has_more": true, - "next_after": "2000", - }, - }, - { - "data": { - "after": "2000", - "bucket": "b0a[]", - "data": undefined, - "has_more": true, - "next_after": "4000", - }, - }, - { - "checkpoint_diff": { - "last_op_id": "4004", - "removed_buckets": [], - "updated_buckets": [ - { - "bucket": "b0a[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "b0b[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "b1[]", - "checksum": 1841937527, - "count": 2, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "1", - "bucket": "b1[]", - "data": undefined, - "has_more": false, - "next_after": "4002", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4004", - "priority": 1, - }, - }, - { - "data": { - "after": "4000", - "bucket": "b0a[]", - "data": undefined, - "has_more": false, - "next_after": "4003", - }, - }, - { - "data": { - "after": "0", - "bucket": "b0b[]", - "data": undefined, - "has_more": true, - "next_after": "1999", - }, - }, - { - "data": { - "after": "1999", - "bucket": "b0b[]", - "data": undefined, - "has_more": true, - "next_after": "3999", - }, - }, - { - "data": { - "after": "3999", - "bucket": "b0b[]", - "data": undefined, - "has_more": false, - "next_after": "4004", - }, - }, - { - "checkpoint_complete": { - "last_op_id": "4004", - }, - }, -] -`; - exports[`sync - mongodb > storage v1 > sync legacy non-raw data 1`] = ` [ { @@ -1778,200 +1584,6 @@ exports[`sync - mongodb > storage v2 > sync global data 1`] = ` ] `; -exports[`sync - mongodb > storage v2 > sync interrupts low-priority buckets on new checkpoints (2) 1`] = ` -[ - { - "checkpoint": { - "buckets": [ - { - "bucket": "1#b0a[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "1#b0b[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "1#b1[]", - "checksum": -1096116670, - "count": 1, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "last_op_id": "4001", - "streams": [ - { - "errors": [], - "is_default": true, - "name": "b0a", - }, - { - "errors": [], - "is_default": true, - "name": "b0b", - }, - { - "errors": [], - "is_default": true, - "name": "b1", - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "0", - "bucket": "1#b1[]", - "data": undefined, - "has_more": false, - "next_after": "1", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4001", - "priority": 1, - }, - }, - { - "data": { - "after": "0", - "bucket": "1#b0a[]", - "data": undefined, - "has_more": true, - "next_after": "2000", - }, - }, - { - "data": { - "after": "2000", - "bucket": "1#b0a[]", - "data": undefined, - "has_more": true, - "next_after": "4000", - }, - }, - { - "checkpoint_diff": { - "last_op_id": "4004", - "removed_buckets": [], - "updated_buckets": [ - { - "bucket": "1#b0a[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "1#b0b[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "1#b1[]", - "checksum": 1841937527, - "count": 2, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "1", - "bucket": "1#b1[]", - "data": undefined, - "has_more": false, - "next_after": "4002", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4004", - "priority": 1, - }, - }, - { - "data": { - "after": "4000", - "bucket": "1#b0a[]", - "data": undefined, - "has_more": false, - "next_after": "4003", - }, - }, - { - "data": { - "after": "0", - "bucket": "1#b0b[]", - "data": undefined, - "has_more": true, - "next_after": "1999", - }, - }, - { - "data": { - "after": "1999", - "bucket": "1#b0b[]", - "data": undefined, - "has_more": true, - "next_after": "3999", - }, - }, - { - "data": { - "after": "3999", - "bucket": "1#b0b[]", - "data": undefined, - "has_more": false, - "next_after": "4004", - }, - }, - { - "checkpoint_complete": { - "last_op_id": "4004", - }, - }, -] -`; - exports[`sync - mongodb > storage v2 > sync legacy non-raw data 1`] = ` [ { @@ -2967,200 +2579,6 @@ exports[`sync - mongodb > storage v3 > sync global data 1`] = ` ] `; -exports[`sync - mongodb > storage v3 > sync interrupts low-priority buckets on new checkpoints (2) 1`] = ` -[ - { - "checkpoint": { - "buckets": [ - { - "bucket": "b0a.1.1[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "b0b.1.2[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "b1.1.3[]", - "checksum": -1096116670, - "count": 1, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "last_op_id": "4001", - "streams": [ - { - "errors": [], - "is_default": true, - "name": "b0a", - }, - { - "errors": [], - "is_default": true, - "name": "b0b", - }, - { - "errors": [], - "is_default": true, - "name": "b1", - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "0", - "bucket": "b1.1.3[]", - "data": undefined, - "has_more": false, - "next_after": "1", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4001", - "priority": 1, - }, - }, - { - "data": { - "after": "0", - "bucket": "b0a.1.1[]", - "data": undefined, - "has_more": true, - "next_after": "2000", - }, - }, - { - "data": { - "after": "2000", - "bucket": "b0a.1.1[]", - "data": undefined, - "has_more": true, - "next_after": "4000", - }, - }, - { - "checkpoint_diff": { - "last_op_id": "4004", - "removed_buckets": [], - "updated_buckets": [ - { - "bucket": "b0a.1.1[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "b0b.1.2[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "b1.1.3[]", - "checksum": 1841937527, - "count": 2, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "1", - "bucket": "b1.1.3[]", - "data": undefined, - "has_more": false, - "next_after": "4002", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4004", - "priority": 1, - }, - }, - { - "data": { - "after": "4000", - "bucket": "b0a.1.1[]", - "data": undefined, - "has_more": false, - "next_after": "4003", - }, - }, - { - "data": { - "after": "0", - "bucket": "b0b.1.2[]", - "data": undefined, - "has_more": true, - "next_after": "1999", - }, - }, - { - "data": { - "after": "1999", - "bucket": "b0b.1.2[]", - "data": undefined, - "has_more": true, - "next_after": "3999", - }, - }, - { - "data": { - "after": "3999", - "bucket": "b0b.1.2[]", - "data": undefined, - "has_more": false, - "next_after": "4004", - }, - }, - { - "checkpoint_complete": { - "last_op_id": "4004", - }, - }, -] -`; - exports[`sync - mongodb > storage v3 > sync legacy non-raw data 1`] = ` [ { @@ -3281,7 +2699,7 @@ exports[`sync - mongodb > storage v3 > sync updates to data query only 2`] = ` }, { "data": { - "after": "0", + "after": "1", "bucket": "by_user.1.1["user1"]", "data": [ { 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 544c10a37..a91edc32f 100644 --- a/modules/module-mongodb-storage/test/src/storage_compacting.test.ts +++ b/modules/module-mongodb-storage/test/src/storage_compacting.test.ts @@ -296,7 +296,6 @@ describe('V3 invariant verification', () => { row_id: 'row1', checksum: 1n, data: '{"id":"row1"}', - target_op: null, ...overrides }; } @@ -351,8 +350,7 @@ bucket_definitions: table: TABLE, row_id: rowId, checksum: BigInt(opId * 7), - data: overrides?.op === 'REMOVE' ? null : JSON.stringify({ id: rowId, description: data }), - target_op: null + data: overrides?.op === 'REMOVE' ? null : JSON.stringify({ id: rowId, description: data }) }; } @@ -455,25 +453,6 @@ bucket_definitions: } }); - test('3. target_op correctness - max of non-null target_ops', () => { - const ops = [ - makeBucketDataDoc({ o: 1n, target_op: null }), - makeBucketDataDoc({ o: 2n, target_op: 10n }), - makeBucketDataDoc({ o: 3n, target_op: 5n }), - makeBucketDataDoc({ o: 4n, target_op: null }) - ]; - - const doc = serializeBucketData('test[]', ops); - expect(doc.target_op).toBe(10n); - }); - - test('3. target_op correctness - all null yields null', () => { - const ops = [makeBucketDataDoc({ o: 1n, target_op: null }), makeBucketDataDoc({ o: 2n, target_op: null })]; - - const doc = serializeBucketData('test[]', ops); - expect(doc.target_op).toBeNull(); - }); - test('4. no overlapping ranges - multiple documents', () => { const opsA = [makeBucketDataDoc({ o: 1n }), makeBucketDataDoc({ o: 3n })]; const opsB = [makeBucketDataDoc({ o: 5n }), makeBucketDataDoc({ o: 8n })]; @@ -817,8 +796,7 @@ bucket_definitions: table: TABLE, row_id: rowId, checksum: BigInt(opId * 7), - data: JSON.stringify({ id: rowId, description: data }), - target_op: null + data: JSON.stringify({ id: rowId, description: data }) }; } @@ -868,7 +846,7 @@ bucket_definitions: // Compaction rechunks the bucket into one document spanning the cached // checkpoint. The requested endpoint is the end of the new document. await collection.deleteMany({}); - await collection.insertOne(serializeBucketData(BUCKET, ops, { compactionTargetOp: 60n })); + await collection.insertOne(serializeBucketData(BUCKET, ops, { targetOp: 60n })); const result = await bucketStorage.getChecksums(test_utils.testCheckpoint(60n), [request]); const checksumResult = result.get(BUCKET)!; @@ -888,7 +866,7 @@ bucket_definitions: makeOp(50, 'E', 'e1', ctx, sourceTableId), makeOp(60, 'F', 'f1', ctx, sourceTableId) ]; - const doc = serializeBucketData(BUCKET, ops, { compactionTargetOp: 60n }); + const doc = serializeBucketData(BUCKET, ops, { targetOp: 60n }); await collection.insertMany([doc]); const checksumAllOps = ops.reduce((sum, op) => addChecksums(sum, Number(op.checksum)), 0); @@ -1014,8 +992,7 @@ bucket_definitions: table: TABLE, row_id: rowId, checksum: BigInt(opId * 7), - data: overrides?.op === 'REMOVE' ? null : JSON.stringify({ id: rowId, description: data }), - target_op: null + data: overrides?.op === 'REMOVE' ? null : JSON.stringify({ id: rowId, description: data }) }; } @@ -1282,8 +1259,7 @@ bucket_definitions: table: TABLE, row_id: rowId, checksum: BigInt(opId * 7), - data: overrides?.op === 'REMOVE' ? null : JSON.stringify({ id: rowId, description: data }), - target_op: null + data: overrides?.op === 'REMOVE' ? null : JSON.stringify({ id: rowId, description: data }) }; } @@ -1495,8 +1471,7 @@ bucket_definitions: table: TABLE, row_id: rowId, checksum: BigInt(opId * 7), - data: overrides?.op === 'REMOVE' ? null : JSON.stringify({ id: rowId, description: data }), - target_op: null + data: overrides?.op === 'REMOVE' ? null : JSON.stringify({ id: rowId, description: data }) }; } diff --git a/modules/module-mongodb-storage/test/src/storage_s3_checksums.test.ts b/modules/module-mongodb-storage/test/src/storage_s3_checksums.test.ts index fff7ecb52..d34d687f2 100644 --- a/modules/module-mongodb-storage/test/src/storage_s3_checksums.test.ts +++ b/modules/module-mongodb-storage/test/src/storage_s3_checksums.test.ts @@ -184,7 +184,7 @@ describe('V3 checksums with S3 object storage', () => { expect(checksum.checksum).toBe(groundTruth); // The two superseded A operations collapse into CLEAR. - const batchAfter = await test_utils.fromAsync(bucketStorage.getBucketDataBatch(checkpoint, [request])); + const batchAfter = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request])); const dataAfter = batchAfter.flatMap((chunk) => chunk.chunkData.data); expect(dataAfter).toMatchObject([{ op: 'CLEAR' }, { object_id: 'B', op: 'PUT' }, { object_id: 'A', op: 'PUT' }]); }); diff --git a/modules/module-mongodb-storage/test/src/storage_s3_compaction_lifecycle.test.ts b/modules/module-mongodb-storage/test/src/storage_s3_compaction_lifecycle.test.ts index 65f47a2f4..7c4a20934 100644 --- a/modules/module-mongodb-storage/test/src/storage_s3_compaction_lifecycle.test.ts +++ b/modules/module-mongodb-storage/test/src/storage_s3_compaction_lifecycle.test.ts @@ -71,7 +71,7 @@ describe('S3 compaction storage lifecycle', () => { }); expect(injectedFailure).toBe(true); - const batch = await test_utils.fromAsync(bucketStorage.getBucketDataBatch(checkpoint, [request])); + const batch = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request])); const data = batch.flatMap((chunk) => chunk.chunkData.data); expect(data).toHaveLength(2); }); @@ -258,7 +258,7 @@ describe('S3 compaction storage lifecycle', () => { // Retired objects remain readable during the reference grace period. expect([...lowerPaths].every((path) => memoryStorage.store.has(path))).toBe(true); - const batch = await test_utils.fromAsync(bucketStorage.getBucketDataBatch(readCheckpoint, [request])); + const batch = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(readCheckpoint, [request])); const data = batch.flatMap((chunk) => chunk.chunkData.data); expect(data).toHaveLength(12); for (let i = 7; i <= 12; i++) { @@ -347,7 +347,7 @@ describe('S3 compaction storage lifecycle', () => { expect(bucketStateBefore!.estimate_since_compact!.count).toBeGreaterThan(0); // Record the input operations and object path. - const batchBefore = await test_utils.fromAsync(bucketStorage.getBucketDataBatch(checkpoint, [request])); + const batchBefore = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request])); const dataBefore = test_utils.getBatchData(batchBefore); expect(dataBefore).toHaveLength(4); const oldS3Paths = new Set(docsBefore.map((doc) => doc.storage_ref!.path)); @@ -405,7 +405,7 @@ describe('S3 compaction storage lifecycle', () => { // The replacement object is readable and contains the expected compacted // operation sequence. - const batchAfter = await test_utils.fromAsync(bucketStorage.getBucketDataBatch(checkpoint, [request])); + const batchAfter = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request])); const dataAfter = batchAfter.flatMap((chunk) => chunk.chunkData.data); expect(dataAfter).toMatchObject([ { op: 'MOVE' }, 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 1d0c7bc06..13633469a 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 @@ -511,7 +511,7 @@ describe('S3 object storage reads', () => { expect(new Set(memoryStorage.store.keys())).toEqual(new Set([documents[1].storage_ref!.path])); // Read back. Both S3-backed and inline ops should be returned. - const batch = await test_utils.fromAsync( + const batch = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(documents[1]._id.o), [ bucketRequest(syncRules.syncConfigContent[0], 'global[]', 0n) ]) diff --git a/modules/module-mongodb-storage/test/src/storage_sync.test.ts b/modules/module-mongodb-storage/test/src/storage_sync.test.ts index dc8988b30..d595bff5b 100644 --- a/modules/module-mongodb-storage/test/src/storage_sync.test.ts +++ b/modules/module-mongodb-storage/test/src/storage_sync.test.ts @@ -191,7 +191,7 @@ function registerSyncStorageTests(storageConfig: storage.TestStorageConfig, stor const checkpoint = flushResult!.flushed_op; const options: storage.BucketDataBatchOptions = {}; - const batch1 = await test_utils.fromAsync( + const batch1 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( test_utils.testCheckpoint(checkpoint), [bucketRequest(syncRulesContent, 'global[]', 0n)], @@ -208,7 +208,7 @@ function registerSyncStorageTests(storageConfig: storage.TestStorageConfig, stor next_after: '2' }); - const batch2 = await test_utils.fromAsync( + const batch2 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( test_utils.testCheckpoint(checkpoint), [bucketRequest(syncRulesContent, 'global[]', batch1[0].chunkData.next_after)], @@ -224,7 +224,7 @@ function registerSyncStorageTests(storageConfig: storage.TestStorageConfig, stor next_after: '3' }); - const batch3 = await test_utils.fromAsync( + const batch3 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( test_utils.testCheckpoint(checkpoint), [bucketRequest(syncRulesContent, 'global[]', batch2[0].chunkData.next_after)], @@ -1837,7 +1837,7 @@ describe('sync - mongodb', () => { async function getFilteredOps(start: number, checkpoint: number): Promise { const { syncRules, bucketStorage } = await setupFilteringTest(); const request = bucketRequest(syncRules.syncConfigContent[0], 'global[]', BigInt(start)); - const batch = await test_utils.fromAsync( + const batch = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(BigInt(checkpoint)), [request]) ); const ops = batch.flatMap((b) => b.chunkData.data.map((d) => BigInt(d.op_id))); @@ -1984,7 +1984,7 @@ describe('sync - mongodb', () => { const roundRequests = requests .filter((request) => pending.has(request.bucket)) .map((request) => ({ ...request, start: positions.get(request.bucket)! })); - const batch = await test_utils.fromAsync( + const batch = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(end), roundRequests) ); let anyHasMore = false; diff --git a/modules/module-mongodb/src/replication/MongoSnapshotter.ts b/modules/module-mongodb/src/replication/MongoSnapshotter.ts index 37cf45242..ffe4fc248 100644 --- a/modules/module-mongodb/src/replication/MongoSnapshotter.ts +++ b/modules/module-mongodb/src/replication/MongoSnapshotter.ts @@ -99,6 +99,9 @@ export class MongoSnapshotter { this.abortSignal.addEventListener('abort', () => { this.nextItemQueued?.resolve(); }); + + // Errors here should not result in uncaught rejection - calling waitForInitialSnapshot is optional. + void this.initialSnapshotDone.promise.catch(() => {}); } private get usePostImages() { diff --git a/modules/module-mongodb/test/src/change_stream_utils.ts b/modules/module-mongodb/test/src/change_stream_utils.ts index 6188aaf00..ce15586e0 100644 --- a/modules/module-mongodb/test/src/change_stream_utils.ts +++ b/modules/module-mongodb/test/src/change_stream_utils.ts @@ -5,6 +5,7 @@ import { createCoreReplicationMetrics, initializeCoreReplicationMetrics, InternalOpId, + isBatchEnd, LEGACY_STORAGE_VERSION, OplogEntry, ProtocolOpId, @@ -273,12 +274,23 @@ export class ChangeStreamTestContext { while (true) { const batch = this.storage!.getBucketDataBatch(checkpoint, map); - const batches = await test_utils.fromAsync(batch); - data = data.concat(batches[0]?.chunkData.data ?? []); - if (batches.length == 0 || !batches[0]!.chunkData.has_more) { + const chunks = await test_utils.fromAsync(batch); + if (chunks.length == 0) { break; } - map = [bucketRequest(syncConfigContent, bucket, BigInt(batches[0]!.chunkData.next_after))]; + for (let chunk of chunks) { + if (isBatchEnd(chunk)) { + if (!chunk.hasMore) { + return data; + } + } else { + data = data.concat(chunk.chunkData.data ?? []); + map = [bucketRequest(syncConfigContent, bucket, BigInt(chunk.chunkData.next_after))]; + if (!chunk.chunkData.has_more) { + return data; + } + } + } } return data; } diff --git a/modules/module-mssql/test/src/CDCStreamTestContext.ts b/modules/module-mssql/test/src/CDCStreamTestContext.ts index 1bb2c49c5..1b063d518 100644 --- a/modules/module-mssql/test/src/CDCStreamTestContext.ts +++ b/modules/module-mssql/test/src/CDCStreamTestContext.ts @@ -203,7 +203,7 @@ export class CDCStreamTestContext implements AsyncDisposable { while (true) { const batch = this.storage!.getBucketDataBatch(checkpoint, map); - const batches = await test_utils.fromAsync(batch); + const batches = await test_utils.getBatchArray(batch); data = data.concat(batches[0]?.chunkData.data ?? []); if (batches.length == 0 || !batches[0]!.chunkData.has_more) { break; @@ -230,7 +230,7 @@ export class CDCStreamTestContext implements AsyncDisposable { const syncConfigContent = this.getSyncConfigContent(); const map = [bucketRequest(syncConfigContent, bucket, start)]; const batch = this.storage!.getBucketDataBatch(checkpoint, map); - const batches = await test_utils.fromAsync(batch); + const batches = await test_utils.getBatchArray(batch); return batches[0]?.chunkData.data ?? []; } } diff --git a/modules/module-mysql/test/src/BinlogStreamUtils.ts b/modules/module-mysql/test/src/BinlogStreamUtils.ts index dcbc5ad34..c9f1a0723 100644 --- a/modules/module-mysql/test/src/BinlogStreamUtils.ts +++ b/modules/module-mysql/test/src/BinlogStreamUtils.ts @@ -167,7 +167,7 @@ export class BinlogStreamTestContext { const checkpoint = await this.getCheckpoint(options); const syncConfigContent = this.getSyncConfigContent(); const map = Object.entries(buckets).map(([bucket, start]) => bucketRequest(syncConfigContent, bucket, start)); - return test_utils.fromAsync(this.storage!.getBucketDataBatch(checkpoint, map)); + return test_utils.getBatchArray(this.storage!.getBucketDataBatch(checkpoint, map)); } async getBucketData( @@ -183,7 +183,7 @@ export class BinlogStreamTestContext { const checkpoint = await this.getCheckpoint(options); const map = [bucketRequest(syncConfigContent, bucket, start)]; const batch = this.storage!.getBucketDataBatch(checkpoint, map); - const batches = await test_utils.fromAsync(batch); + const batches = await test_utils.getBatchArray(batch); return batches[0]?.chunkData.data ?? []; } } diff --git a/modules/module-postgres-storage/test/src/__snapshots__/storage_sync.test.ts.snap b/modules/module-postgres-storage/test/src/__snapshots__/storage_sync.test.ts.snap index 39488d4f1..e474292fe 100644 --- a/modules/module-postgres-storage/test/src/__snapshots__/storage_sync.test.ts.snap +++ b/modules/module-postgres-storage/test/src/__snapshots__/storage_sync.test.ts.snap @@ -589,200 +589,6 @@ exports[`sync - postgres > storage v1 > sync global data 1`] = ` ] `; -exports[`sync - postgres > storage v1 > sync interrupts low-priority buckets on new checkpoints (2) 1`] = ` -[ - { - "checkpoint": { - "buckets": [ - { - "bucket": "b0a[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "b0b[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "b1[]", - "checksum": -1096116670, - "count": 1, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "last_op_id": "4001", - "streams": [ - { - "errors": [], - "is_default": true, - "name": "b0a", - }, - { - "errors": [], - "is_default": true, - "name": "b0b", - }, - { - "errors": [], - "is_default": true, - "name": "b1", - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "0", - "bucket": "b1[]", - "data": undefined, - "has_more": false, - "next_after": "1", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4001", - "priority": 1, - }, - }, - { - "data": { - "after": "0", - "bucket": "b0a[]", - "data": undefined, - "has_more": true, - "next_after": "2000", - }, - }, - { - "data": { - "after": "2000", - "bucket": "b0a[]", - "data": undefined, - "has_more": true, - "next_after": "4000", - }, - }, - { - "checkpoint_diff": { - "last_op_id": "4004", - "removed_buckets": [], - "updated_buckets": [ - { - "bucket": "b0a[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "b0b[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "b1[]", - "checksum": 1841937527, - "count": 2, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "1", - "bucket": "b1[]", - "data": undefined, - "has_more": false, - "next_after": "4002", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4004", - "priority": 1, - }, - }, - { - "data": { - "after": "4000", - "bucket": "b0a[]", - "data": undefined, - "has_more": false, - "next_after": "4003", - }, - }, - { - "data": { - "after": "0", - "bucket": "b0b[]", - "data": undefined, - "has_more": true, - "next_after": "1999", - }, - }, - { - "data": { - "after": "1999", - "bucket": "b0b[]", - "data": undefined, - "has_more": true, - "next_after": "3999", - }, - }, - { - "data": { - "after": "3999", - "bucket": "b0b[]", - "data": undefined, - "has_more": false, - "next_after": "4004", - }, - }, - { - "checkpoint_complete": { - "last_op_id": "4004", - }, - }, -] -`; - exports[`sync - postgres > storage v1 > sync legacy non-raw data 1`] = ` [ { @@ -1778,200 +1584,6 @@ exports[`sync - postgres > storage v2 > sync global data 1`] = ` ] `; -exports[`sync - postgres > storage v2 > sync interrupts low-priority buckets on new checkpoints (2) 1`] = ` -[ - { - "checkpoint": { - "buckets": [ - { - "bucket": "1#b0a[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "1#b0b[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "1#b1[]", - "checksum": -1096116670, - "count": 1, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "last_op_id": "4001", - "streams": [ - { - "errors": [], - "is_default": true, - "name": "b0a", - }, - { - "errors": [], - "is_default": true, - "name": "b0b", - }, - { - "errors": [], - "is_default": true, - "name": "b1", - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "0", - "bucket": "1#b1[]", - "data": undefined, - "has_more": false, - "next_after": "1", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4001", - "priority": 1, - }, - }, - { - "data": { - "after": "0", - "bucket": "1#b0a[]", - "data": undefined, - "has_more": true, - "next_after": "2000", - }, - }, - { - "data": { - "after": "2000", - "bucket": "1#b0a[]", - "data": undefined, - "has_more": true, - "next_after": "4000", - }, - }, - { - "checkpoint_diff": { - "last_op_id": "4004", - "removed_buckets": [], - "updated_buckets": [ - { - "bucket": "1#b0a[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "1#b0b[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "1#b1[]", - "checksum": 1841937527, - "count": 2, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "1", - "bucket": "1#b1[]", - "data": undefined, - "has_more": false, - "next_after": "4002", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4004", - "priority": 1, - }, - }, - { - "data": { - "after": "4000", - "bucket": "1#b0a[]", - "data": undefined, - "has_more": false, - "next_after": "4003", - }, - }, - { - "data": { - "after": "0", - "bucket": "1#b0b[]", - "data": undefined, - "has_more": true, - "next_after": "1999", - }, - }, - { - "data": { - "after": "1999", - "bucket": "1#b0b[]", - "data": undefined, - "has_more": true, - "next_after": "3999", - }, - }, - { - "data": { - "after": "3999", - "bucket": "1#b0b[]", - "data": undefined, - "has_more": false, - "next_after": "4004", - }, - }, - { - "checkpoint_complete": { - "last_op_id": "4004", - }, - }, -] -`; - exports[`sync - postgres > storage v2 > sync legacy non-raw data 1`] = ` [ { @@ -2967,200 +2579,6 @@ exports[`sync - postgres > storage v3 > sync global data 1`] = ` ] `; -exports[`sync - postgres > storage v3 > sync interrupts low-priority buckets on new checkpoints (2) 1`] = ` -[ - { - "checkpoint": { - "buckets": [ - { - "bucket": "1#b0a[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "1#b0b[]", - "checksum": -659831575, - "count": 2000, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "1#b1[]", - "checksum": -1096116670, - "count": 1, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "last_op_id": "4001", - "streams": [ - { - "errors": [], - "is_default": true, - "name": "b0a", - }, - { - "errors": [], - "is_default": true, - "name": "b0b", - }, - { - "errors": [], - "is_default": true, - "name": "b1", - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "0", - "bucket": "1#b1[]", - "data": undefined, - "has_more": false, - "next_after": "1", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4001", - "priority": 1, - }, - }, - { - "data": { - "after": "0", - "bucket": "1#b0a[]", - "data": undefined, - "has_more": true, - "next_after": "2000", - }, - }, - { - "data": { - "after": "2000", - "bucket": "1#b0a[]", - "data": undefined, - "has_more": true, - "next_after": "4000", - }, - }, - { - "checkpoint_diff": { - "last_op_id": "4004", - "removed_buckets": [], - "updated_buckets": [ - { - "bucket": "1#b0a[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 0, - }, - ], - }, - { - "bucket": "1#b0b[]", - "checksum": 883076828, - "count": 2001, - "priority": 2, - "subscriptions": [ - { - "default": 1, - }, - ], - }, - { - "bucket": "1#b1[]", - "checksum": 1841937527, - "count": 2, - "priority": 1, - "subscriptions": [ - { - "default": 2, - }, - ], - }, - ], - "write_checkpoint": undefined, - }, - }, - { - "data": { - "after": "1", - "bucket": "1#b1[]", - "data": undefined, - "has_more": false, - "next_after": "4002", - }, - }, - { - "partial_checkpoint_complete": { - "last_op_id": "4004", - "priority": 1, - }, - }, - { - "data": { - "after": "4000", - "bucket": "1#b0a[]", - "data": undefined, - "has_more": false, - "next_after": "4003", - }, - }, - { - "data": { - "after": "0", - "bucket": "1#b0b[]", - "data": undefined, - "has_more": true, - "next_after": "1999", - }, - }, - { - "data": { - "after": "1999", - "bucket": "1#b0b[]", - "data": undefined, - "has_more": true, - "next_after": "3999", - }, - }, - { - "data": { - "after": "3999", - "bucket": "1#b0b[]", - "data": undefined, - "has_more": false, - "next_after": "4004", - }, - }, - { - "checkpoint_complete": { - "last_op_id": "4004", - }, - }, -] -`; - exports[`sync - postgres > storage v3 > sync legacy non-raw data 1`] = ` [ { diff --git a/modules/module-postgres-storage/test/src/storage.test.ts b/modules/module-postgres-storage/test/src/storage.test.ts index 382d9d4c9..0620a2b4a 100644 --- a/modules/module-postgres-storage/test/src/storage.test.ts +++ b/modules/module-postgres-storage/test/src/storage.test.ts @@ -308,7 +308,7 @@ bucket_definitions: const options: storage.BucketDataBatchOptions = {}; - const batch1 = await test_utils.fromAsync( + const batch1 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [globalBucket], options) ); expect(test_utils.getBatchData(batch1)).toEqual([ @@ -320,7 +320,7 @@ bucket_definitions: next_after: '1' }); - const batch2 = await test_utils.fromAsync( + const batch2 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( test_utils.testCheckpoint(checkpoint), [{ ...globalBucket, start: BigInt(batch1[0].chunkData.next_after) }], @@ -336,7 +336,7 @@ bucket_definitions: next_after: '2' }); - const batch3 = await test_utils.fromAsync( + const batch3 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( test_utils.testCheckpoint(checkpoint), [{ ...globalBucket, start: BigInt(batch2[0].chunkData.next_after) }], @@ -352,7 +352,7 @@ bucket_definitions: next_after: '3' }); - const batch4 = await test_utils.fromAsync( + const batch4 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( test_utils.testCheckpoint(checkpoint), [{ ...globalBucket, start: BigInt(batch3[0].chunkData.next_after) }], diff --git a/modules/module-postgres-storage/test/src/storage_compacting.test.ts b/modules/module-postgres-storage/test/src/storage_compacting.test.ts index 711e67bdd..43d4132ae 100644 --- a/modules/module-postgres-storage/test/src/storage_compacting.test.ts +++ b/modules/module-postgres-storage/test/src/storage_compacting.test.ts @@ -49,7 +49,7 @@ bucket_definitions: minBucketChanges: 1 }); - const batch = await test_utils.oneFromAsync( + const batch = await test_utils.getSingleBatchItem( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [ bucketRequest(syncRulesContent, 'global[]', 0n) ]) @@ -110,7 +110,7 @@ bucket_definitions: })(); const checkpoint = result!.flushed_op; - const rowsBefore = await test_utils.oneFromAsync( + const rowsBefore = await test_utils.getSingleBatchItem( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) ); const dataBefore = test_utils.getBatchData(rowsBefore); @@ -123,7 +123,7 @@ bucket_definitions: ); // The method wraps in a transaction; on assertion error the bucket must remain unchanged. - const rowsAfter = await test_utils.oneFromAsync( + const rowsAfter = await test_utils.getSingleBatchItem( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) ); expect(test_utils.getBatchData(rowsAfter)).toEqual(dataBefore); diff --git a/modules/module-postgres-storage/test/src/storage_sync.test.ts b/modules/module-postgres-storage/test/src/storage_sync.test.ts index fa88aedcc..1a5ffef25 100644 --- a/modules/module-postgres-storage/test/src/storage_sync.test.ts +++ b/modules/module-postgres-storage/test/src/storage_sync.test.ts @@ -90,7 +90,7 @@ function registerStorageVersionTests(storageVersion: number) { const options: storage.BucketDataBatchOptions = {}; - const batch1 = await test_utils.fromAsync( + const batch1 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [globalBucket], options) ); expect(test_utils.getBatchData(batch1)).toEqual([ @@ -102,7 +102,7 @@ function registerStorageVersionTests(storageVersion: number) { next_after: '1' }); - const batch2 = await test_utils.fromAsync( + const batch2 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( test_utils.testCheckpoint(checkpoint), [{ ...globalBucket, start: BigInt(batch1[0].chunkData.next_after) }], @@ -118,7 +118,7 @@ function registerStorageVersionTests(storageVersion: number) { next_after: '2' }); - const batch3 = await test_utils.fromAsync( + const batch3 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( test_utils.testCheckpoint(checkpoint), [{ ...globalBucket, start: BigInt(batch2[0].chunkData.next_after) }], @@ -134,7 +134,7 @@ function registerStorageVersionTests(storageVersion: number) { next_after: '3' }); - const batch4 = await test_utils.fromAsync( + const batch4 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( test_utils.testCheckpoint(checkpoint), [{ ...globalBucket, start: BigInt(batch3[0].chunkData.next_after) }], diff --git a/modules/module-postgres/test/src/wal_stream_utils.ts b/modules/module-postgres/test/src/wal_stream_utils.ts index e1ecf814a..f533d8d61 100644 --- a/modules/module-postgres/test/src/wal_stream_utils.ts +++ b/modules/module-postgres/test/src/wal_stream_utils.ts @@ -231,7 +231,7 @@ export class WalStreamTestContext implements AsyncDisposable { const checkpoint = await this.storage!.getCheckpoint(); const map = [bucketRequest(syncConfigContent, bucket, start)]; const batch = this.storage!.getBucketDataBatch(checkpoint, map); - const batches = await test_utils.fromAsync(batch); + const batches = await test_utils.getBatchArray(batch); return batches[0]?.chunkData.data ?? []; } diff --git a/packages/service-core-tests/src/test-utils/AbstractStreamTestContext.ts b/packages/service-core-tests/src/test-utils/AbstractStreamTestContext.ts index 85817c0d6..f038e6b9e 100644 --- a/packages/service-core-tests/src/test-utils/AbstractStreamTestContext.ts +++ b/packages/service-core-tests/src/test-utils/AbstractStreamTestContext.ts @@ -10,8 +10,7 @@ import { updateSyncRulesFromYaml } from '@powersync/service-core'; import { StorageDataHelpers } from './StorageDataHelpers.js'; -import { bucketRequest } from './general-utils.js'; -import { fromAsync } from './stream_utils.js'; +import { bucketRequest, getBatchArray } from './general-utils.js'; export abstract class AbstractStreamTestContext implements AsyncDisposable { protected abortController = new AbortController(); @@ -178,7 +177,7 @@ export abstract class AbstractStreamTestContext implements AsyncDisposable { const checkpoint = await this.storage!.getCheckpoint(); const map = [bucketRequest(syncConfigContent, bucket, start)]; const batch = this.storage!.getBucketDataBatch(checkpoint, map); - const batches = await fromAsync(batch); + const batches = await getBatchArray(batch); return batches[0]?.chunkData.data ?? []; } } diff --git a/packages/service-core-tests/src/test-utils/StorageDataHelpers.ts b/packages/service-core-tests/src/test-utils/StorageDataHelpers.ts index 057265a51..3bb7cba18 100644 --- a/packages/service-core-tests/src/test-utils/StorageDataHelpers.ts +++ b/packages/service-core-tests/src/test-utils/StorageDataHelpers.ts @@ -1,12 +1,16 @@ import { + BucketDataBatchOptions, + BucketDataRequest, InternalOpId, + isBatchEnd, OplogEntry, ParsedSyncConfigSet, PersistedSyncConfigContent, ReplicationCheckpoint, + SyncBucketDataChunk, SyncRulesBucketStorage } from '@powersync/service-core'; -import { bucketRequest } from './general-utils.js'; +import { bucketRequest, getBatchArray } from './general-utils.js'; import { fromAsync } from './stream_utils.js'; export class StorageDataHelpers { @@ -28,7 +32,7 @@ export class StorageDataHelpers { while (true) { const batch = this.storage!.getBucketDataBatch(checkpoint, map); - const batches = await fromAsync(batch); + const batches = await getBatchArray(batch); data = data.concat(batches[0]?.chunkData.data ?? []); if (batches.length == 0 || !batches[0]!.chunkData.has_more) { break; @@ -42,4 +46,43 @@ export class StorageDataHelpers { const map = Object.entries(buckets).map(([bucket, start]) => bucketRequest(this.syncRules, bucket, start)); return fromAsync(this.storage!.getBucketDataBatch(checkpoint, map)); } + + async getAllBucketData( + requests: BucketDataRequest[], + checkpoint: ReplicationCheckpoint, + options?: BucketDataBatchOptions + ): Promise { + let remainingBuckets = new Map(requests.map((r) => [r.bucket, r])); + let chunks: SyncBucketDataChunk[] = []; + while (true) { + let hasMore = false; + for await (let chunk of this.storage!.getBucketDataBatch(checkpoint, [...remainingBuckets.values()], options)) { + if (isBatchEnd(chunk)) { + if (chunk.hasMore) { + hasMore = true; + break; + } else { + return chunks; + } + } else { + chunks.push(chunk); + if (chunk.chunkData.has_more) { + hasMore = true; + const r = remainingBuckets.get(chunk.chunkData.bucket)!; + remainingBuckets.set(chunk.chunkData.bucket, { + start: BigInt(chunk.chunkData.next_after), + bucket: chunk.chunkData.bucket, + source: r.source + }); + } else { + remainingBuckets.delete(chunk.chunkData.bucket); + } + } + } + if (!hasMore) { + break; + } + } + return chunks; + } } diff --git a/packages/service-core-tests/src/test-utils/general-utils.ts b/packages/service-core-tests/src/test-utils/general-utils.ts index 0b2a89225..c29e57f1d 100644 --- a/packages/service-core-tests/src/test-utils/general-utils.ts +++ b/packages/service-core-tests/src/test-utils/general-utils.ts @@ -1,6 +1,7 @@ -import { BucketDataRequest, InternalOpId, JwtPayload, storage, utils } from '@powersync/service-core'; +import { BucketDataRequest, InternalOpId, isBatchEnd, JwtPayload, storage, utils } from '@powersync/service-core'; import { GetQuerierOptions, RequestParameters } from '@powersync/service-sync-rules'; import * as bson from 'bson'; +import { fromAsync } from './stream_utils.js'; export const ZERO_LSN = '0/0'; @@ -86,7 +87,10 @@ export async function resolveTestTable( } export function getBatchData( - batch: utils.SyncBucketData[] | storage.SyncBucketDataChunk[] | storage.SyncBucketDataChunk + batch: + | utils.SyncBucketData[] + | (storage.SyncBucketDataChunk | storage.SyncBucketDataBatchEnd)[] + | storage.SyncBucketDataChunk ) { const first = getFirst(batch); if (first == null) { @@ -152,8 +156,28 @@ export function getBatchMeta( }; } +export async function getBatchArray( + data: AsyncIterable +): Promise { + const array = await fromAsync(data); + return array.filter((c) => !isBatchEnd(c) && c.chunkData.data.length > 0) as storage.SyncBucketDataChunk[]; +} + +export async function getSingleBatchItem( + data: AsyncIterable +): Promise { + const array = await getBatchArray(data); + if (array.length != 1) { + throw new Error(`Expected a single batch item, got ${array.length}`); + } + return array[0]; +} + function getFirst( - batch: utils.SyncBucketData[] | storage.SyncBucketDataChunk[] | storage.SyncBucketDataChunk + batch: + | utils.SyncBucketData[] + | (storage.SyncBucketDataChunk | storage.SyncBucketDataBatchEnd)[] + | storage.SyncBucketDataChunk ): utils.SyncBucketData | null { if (!Array.isArray(batch)) { return batch.chunkData; diff --git a/packages/service-core-tests/src/test-utils/stream_utils.ts b/packages/service-core-tests/src/test-utils/stream_utils.ts index d29586997..451ed11b0 100644 --- a/packages/service-core-tests/src/test-utils/stream_utils.ts +++ b/packages/service-core-tests/src/test-utils/stream_utils.ts @@ -22,17 +22,6 @@ export function compareIds(a: utils.OplogEntry, b: utils.OplogEntry) { return a.object_id!.localeCompare(b.object_id!); } -export async function oneFromAsync(source: Iterable | AsyncIterable): Promise { - const items: T[] = []; - for await (const item of source) { - items.push(item); - } - if (items.length != 1) { - throw new Error(`One item expected, got: ${items.length}`); - } - return items[0]; -} - export async function fromAsync(source: Iterable | AsyncIterable): Promise { const items: T[] = []; for await (const item of source) { diff --git a/packages/service-core-tests/src/tests/register-compacting-tests.ts b/packages/service-core-tests/src/tests/register-compacting-tests.ts index 2f7a385cb..a30a6cca9 100644 --- a/packages/service-core-tests/src/tests/register-compacting-tests.ts +++ b/packages/service-core-tests/src/tests/register-compacting-tests.ts @@ -56,7 +56,7 @@ bucket_definitions: const request = bucketRequest(syncRulesContent, 'global[]'); - const batchBefore = await test_utils.oneFromAsync( + const batchBefore = await test_utils.getSingleBatchItem( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) ); const dataBefore = batchBefore.chunkData.data; @@ -89,7 +89,7 @@ bucket_definitions: minChangeRatio: 0 }); - const batchAfter = await test_utils.oneFromAsync( + const batchAfter = await test_utils.getSingleBatchItem( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) ); const dataAfter = batchAfter.chunkData.data; @@ -176,7 +176,7 @@ bucket_definitions: const checkpoint = writer.last_flushed_op!; const request = bucketRequest(syncRulesContent, 'global[]'); - const batchBefore = await test_utils.oneFromAsync( + const batchBefore = await test_utils.getSingleBatchItem( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) ); const dataBefore = batchBefore.chunkData.data; @@ -210,7 +210,7 @@ bucket_definitions: minChangeRatio: 0 }); - const batchAfter = await test_utils.oneFromAsync( + const batchAfter = await test_utils.getSingleBatchItem( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) ); const dataAfter = batchAfter.chunkData.data; @@ -309,7 +309,7 @@ bucket_definitions: minChangeRatio: 0 }); - const batchAfter = await test_utils.oneFromAsync( + const batchAfter = await test_utils.getSingleBatchItem( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint2), [request]) ); const dataAfter = batchAfter.chunkData.data; @@ -424,7 +424,7 @@ bucket_definitions: minChangeRatio: 0 }); - const batchAfter = await test_utils.fromAsync( + const batchAfter = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( test_utils.testCheckpoint(checkpoint), bucketRequestMap(syncRulesContent, [ @@ -665,7 +665,7 @@ bucket_definitions: minChangeRatio: 0 }); - const batchAfterDefaultCompact = await test_utils.oneFromAsync( + const batchAfterDefaultCompact = await test_utils.getSingleBatchItem( bucketStorage.getBucketDataBatch( test_utils.testCheckpoint(checkpoint2), bucketRequestMap(syncRulesContent, [['global[]', 0n]]) diff --git a/packages/service-core-tests/src/tests/register-data-storage-data-tests.ts b/packages/service-core-tests/src/tests/register-data-storage-data-tests.ts index 4401c7b80..6aa7973d4 100644 --- a/packages/service-core-tests/src/tests/register-data-storage-data-tests.ts +++ b/packages/service-core-tests/src/tests/register-data-storage-data-tests.ts @@ -72,12 +72,10 @@ bucket_definitions: }); await writer.commit('1/1'); - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); const request = bucketRequest(syncRules, 'global[]'); - const batch = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) - ); + const batch = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request])); const data = batch[0].chunkData.data.map((d) => { return { op: d.op, @@ -94,9 +92,7 @@ bucket_definitions: { op: 'REMOVE', object_id: 'test1', checksum: c2 } ]); - const checksums = [ - ...(await bucketStorage.getChecksums(test_utils.testCheckpoint(checkpoint), [request])).values() - ]; + const checksums = [...(await bucketStorage.getChecksums(checkpoint, [request])).values()]; expect(checksums).toEqual([ { bucket: request.bucket, @@ -144,12 +140,10 @@ bucket_definitions: }); await writer.commit('2/1'); - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); const request = bucketRequest(syncRules, 'global[]'); - const batch = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) - ); + const batch = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request])); const data = batch[0].chunkData.data.map((d) => { return { op: d.op, @@ -162,9 +156,7 @@ bucket_definitions: expect(data).toEqual([{ op: 'PUT', object_id: 'test1', checksum: c1 }]); - const checksums = [ - ...(await bucketStorage.getChecksums(test_utils.testCheckpoint(checkpoint), [request])).values() - ]; + const checksums = [...(await bucketStorage.getChecksums(checkpoint, [request])).values()]; expect(checksums).toEqual([ { bucket: request.bucket, @@ -217,12 +209,10 @@ bucket_definitions: }); await writer.commit('2/1'); - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); const request = bucketRequest(syncRules, 'global[]'); - const batch = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) - ); + const batch = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request])); const data = batch[0].chunkData.data.map((d) => { return { op: d.op, @@ -235,9 +225,7 @@ bucket_definitions: expect(data).toEqual([{ op: 'PUT', object_id: 'test1', checksum: c1 }]); - const checksums = [ - ...(await bucketStorage.getChecksums(test_utils.testCheckpoint(checkpoint), [request])).values() - ]; + const checksums = [...(await bucketStorage.getChecksums(checkpoint, [request])).values()]; expect(checksums).toEqual([ { bucket: request.bucket, @@ -284,12 +272,10 @@ bucket_definitions: }); await writer.commit('1/1'); - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); const request = bucketRequest(syncRules, 'global[]'); - const batch = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) - ); + const batch = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request])); const data = batch[0].chunkData.data.map((d) => { return { op: d.op, @@ -302,9 +288,7 @@ bucket_definitions: expect(data).toEqual([{ op: 'PUT', object_id: 'test1', checksum: c1 }]); - const checksums = [ - ...(await bucketStorage.getChecksums(test_utils.testCheckpoint(checkpoint), [request])).values() - ]; + const checksums = [...(await bucketStorage.getChecksums(checkpoint, [request])).values()]; expect(checksums).toEqual([ { bucket: request.bucket, @@ -375,12 +359,10 @@ bucket_definitions: await writer.commit('2/1'); } - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); const request = bucketRequest(syncRules, 'global[]'); - const batch = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) - ); + const batch = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request])); expect(reduceBucket(batch[0].chunkData.data).slice(1)).toEqual([]); @@ -448,9 +430,9 @@ bucket_definitions: }); await writer.commit('1/1'); - const { checkpoint } = await bucketStorage.getCheckpoint(); - const batch = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [bucketRequest(syncRules, 'global[]')]) + const checkpoint = await bucketStorage.getCheckpoint(); + const batch = await test_utils.getBatchArray( + bucketStorage.getBucketDataBatch(checkpoint, [bucketRequest(syncRules, 'global[]')]) ); const data = batch[0].chunkData.data.map((d) => { return { @@ -514,12 +496,10 @@ bucket_definitions: }); await writer.flush(); - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); const request = bucketRequest(syncRules, 'global[]'); - const batch = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) - ); + const batch = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request])); const data = batch[0].chunkData.data.map((d) => { return { op: d.op, @@ -536,9 +516,7 @@ bucket_definitions: { op: 'REMOVE', object_id: 'test1', checksum: c2 } ]); - const checksums = [ - ...(await bucketStorage.getChecksums(test_utils.testCheckpoint(checkpoint), [request])).values() - ]; + const checksums = [...(await bucketStorage.getChecksums(checkpoint, [request])).values()]; expect(checksums).toEqual([ { bucket: bucketRequest(syncRules, 'global[]').bucket, @@ -638,12 +616,10 @@ bucket_definitions: await writer.commit('2/1'); - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); const request = bucketRequest(syncRules, 'global[]'); - const batch = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request]) - ); + const batch = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request])); const data = batch[0].chunkData.data.map((d) => { return { @@ -663,9 +639,7 @@ bucket_definitions: { op: 'REMOVE', object_id: 'test1', checksum: c2 } ]); - const checksums = [ - ...(await bucketStorage.getChecksums(test_utils.testCheckpoint(checkpoint), [request])).values() - ]; + const checksums = [...(await bucketStorage.getChecksums(checkpoint, [request])).values()]; expect(checksums).toEqual([ { bucket: bucketRequest(syncRules, 'global[]').bucket, @@ -808,7 +782,7 @@ bucket_definitions: const checkpoint2 = result2!.flushed_op; const request = bucketRequest(syncRules, 'global[]', checkpoint1); - const batch = await test_utils.fromAsync( + const batch = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint2), [request]) ); @@ -914,7 +888,7 @@ bucket_definitions: const checkpoint3 = result3!.flushed_op; const request = bucketRequest(syncRules, 'global[]'); - const batch = await test_utils.fromAsync( + const batch = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint3), [{ ...request, start: checkpoint1 }]) ); const data = batch[0].chunkData.data.map((d) => { @@ -1028,7 +1002,7 @@ bucket_definitions: const checkpoint3 = result3!.flushed_op; const request = bucketRequest(syncRules, 'global[]'); - const batch = await test_utils.fromAsync( + const batch = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint3), [{ ...request, start: checkpoint1 }]) ); const data = batch[0].chunkData.data.map((d) => { @@ -1129,16 +1103,14 @@ bucket_definitions: await writer.commit('1/1'); - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); const options: storage.BucketDataBatchOptions = { chunkLimitBytes: 16 * 1024 * 1024 }; const request = bucketRequest(syncRules, 'global[]'); - const batch1 = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request], options) - ); + const batch1 = await test_utils.getBatchArray(bucketStorage.getBucketDataBatch(checkpoint, [request], options)); expect(test_utils.getBatchData(batch1)).toEqual([ { op_id: '1', op: 'PUT', object_id: 'test1', checksum: 2871785649 }, { op_id: '2', op: 'PUT', object_id: 'large1', checksum: 454746904 } @@ -1149,9 +1121,9 @@ bucket_definitions: next_after: '2' }); - const batch2 = await test_utils.fromAsync( + const batch2 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( - test_utils.testCheckpoint(checkpoint), + checkpoint, [{ ...request, start: BigInt(batch1[0].chunkData.next_after) }], options ) @@ -1166,9 +1138,9 @@ bucket_definitions: next_after: '4' }); - const batch3 = await test_utils.fromAsync( + const batch3 = await test_utils.getBatchArray( bucketStorage.getBucketDataBatch( - test_utils.testCheckpoint(checkpoint), + checkpoint, [{ ...request, start: BigInt(batch2[0].chunkData.next_after) }], options ) @@ -1178,7 +1150,8 @@ bucket_definitions: }); test('long batch', async () => { - // Test syncing a batch of data that is limited by count. + // Test syncing a batch of data that is limited by count. Compressed storage limits persisted + // documents and finishes yielding all operations from a document once it has been hydrated. await using factory = await generateStorageFactory(); const { stream: replicationStream, content: syncRules } = await test_utils.deploySyncRules( factory, @@ -1213,62 +1186,71 @@ bucket_definitions: await writer.commit('1/1'); - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); const request = bucketRequest(syncRules, 'global[]'); - const batch1 = await test_utils.oneFromAsync( - bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(checkpoint), [request], { limit: 4 }) + const batch1 = await test_utils.getSingleBatchItem( + bucketStorage.getBucketDataBatch(checkpoint, [request], { limit: 4 }) ); - expect(test_utils.getBatchData(batch1)).toEqual([ + const allOperations = [ { op_id: '1', op: 'PUT', object_id: 'test1', checksum: 2871785649 }, { op_id: '2', op: 'PUT', object_id: 'test2', checksum: 730027011 }, { op_id: '3', op: 'PUT', object_id: 'test3', checksum: 1359888332 }, - { op_id: '4', op: 'PUT', object_id: 'test4', checksum: 2049153252 } - ]); + { op_id: '4', op: 'PUT', object_id: 'test4', checksum: 2049153252 }, + { op_id: '5', op: 'PUT', object_id: 'test5', checksum: 3686902721 }, + { op_id: '6', op: 'PUT', object_id: 'test6', checksum: 1974820016 } + ]; - expect(test_utils.getBatchMeta(batch1)).toEqual({ - after: '0', - has_more: true, - next_after: '4' - }); + if (config.compressedBucketStorage) { + expect(test_utils.getBatchData(batch1)).toEqual(allOperations); + expect(test_utils.getBatchMeta(batch1)).toEqual({ + after: '0', + has_more: false, + next_after: '6' + }); - const batch2 = await test_utils.oneFromAsync( - bucketStorage.getBucketDataBatch( - test_utils.testCheckpoint(checkpoint), - [{ ...request, start: BigInt(batch1.chunkData.next_after) }], - { + const batch2 = await test_utils.getBatchArray( + bucketStorage.getBucketDataBatch(checkpoint, [{ ...request, start: BigInt(batch1.chunkData.next_after) }], { limit: 4 - } - ) - ); - expect(test_utils.getBatchData(batch2)).toEqual([ - { op_id: '5', op: 'PUT', object_id: 'test5', checksum: 3686902721 }, - { op_id: '6', op: 'PUT', object_id: 'test6', checksum: 1974820016 } - ]); + }) + ); + expect(test_utils.getBatchData(batch2)).toEqual([]); + expect(test_utils.getBatchMeta(batch2)).toEqual(null); + } else { + expect(test_utils.getBatchData(batch1)).toEqual(allOperations.slice(0, 4)); - expect(test_utils.getBatchMeta(batch2)).toEqual({ - after: '4', - has_more: false, - next_after: '6' - }); + expect(test_utils.getBatchMeta(batch1)).toEqual({ + after: '0', + has_more: true, + next_after: '4' + }); - const batch3 = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch( - test_utils.testCheckpoint(checkpoint), - [{ ...request, start: BigInt(batch2.chunkData.next_after) }], - { + const batch2 = await test_utils.getSingleBatchItem( + bucketStorage.getBucketDataBatch(checkpoint, [{ ...request, start: BigInt(batch1.chunkData.next_after) }], { limit: 4 - } - ) - ); - expect(test_utils.getBatchData(batch3)).toEqual([]); + }) + ); + expect(test_utils.getBatchData(batch2)).toEqual(allOperations.slice(4)); - expect(test_utils.getBatchMeta(batch3)).toEqual(null); + expect(test_utils.getBatchMeta(batch2)).toEqual({ + after: '4', + has_more: false, + next_after: '6' + }); + + const batch3 = await test_utils.getBatchArray( + bucketStorage.getBucketDataBatch(checkpoint, [{ ...request, start: BigInt(batch2.chunkData.next_after) }], { + limit: 4 + }) + ); + expect(test_utils.getBatchData(batch3)).toEqual([]); + expect(test_utils.getBatchMeta(batch3)).toEqual(null); + } }); describe('batch has_more', () => { - const setup = async (options: BucketDataBatchOptions) => { + const setup = async (options: BucketDataBatchOptions, commitAfter?: number) => { await using factory = await generateStorageFactory(); const { stream: replicationStream, content: syncRules } = await test_utils.deploySyncRules( factory, @@ -1301,27 +1283,30 @@ bucket_definitions: }, afterReplicaId: `test${i}` }); + if (i == commitAfter) { + await writer.commit('1/1'); + } } - await writer.commit('1/1'); + await writer.commit(commitAfter == null ? '1/1' : '1/2'); - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); const global1Request = bucketRequest(syncRules, 'global1[]', 0n); const global2Request = bucketRequest(syncRules, 'global2[]', 0n); - const batch = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch( - test_utils.testCheckpoint(checkpoint), - [global1Request, global2Request], - options - ) + const batch = await new test_utils.StorageDataHelpers(bucketStorage, syncRules).getAllBucketData( + [global1Request, global2Request], + checkpoint, + options ); return { batch, global1Request, global2Request }; }; test('batch has_more (1)', async () => { - const { batch, global1Request, global2Request } = await setup({ limit: 5 }); - expect(batch.length).toEqual(2); + // The commit boundary creates two compressed documents for global2. The limit selects the global1 + // document and the first global2 document, leaving the second global2 document for the next batch. + const { batch, global1Request, global2Request } = await setup({ limit: 2 }, 5); + expect(batch.length).toBeGreaterThanOrEqual(2); expect(batch[0].chunkData.bucket).toEqual(global1Request.bucket); expect(batch[1].chunkData.bucket).toEqual(global2Request.bucket); @@ -1330,12 +1315,20 @@ bucket_definitions: { op_id: '1', op: 'PUT', object_id: 'test1', checksum: 2871785649 } ]); - expect(test_utils.getBatchData(batch[1])).toEqual([ + const global2Operations = [ { op_id: '2', op: 'PUT', object_id: 'test2', checksum: 730027011 }, { op_id: '3', op: 'PUT', object_id: 'test3', checksum: 1359888332 }, { op_id: '4', op: 'PUT', object_id: 'test4', checksum: 2049153252 }, - { op_id: '5', op: 'PUT', object_id: 'test5', checksum: 3686902721 } - ]); + { op_id: '5', op: 'PUT', object_id: 'test5', checksum: 3686902721 }, + { op_id: '6', op: 'PUT', object_id: 'test6', checksum: 1974820016 }, + { op_id: '7', op: 'PUT', object_id: 'test7', checksum: 2477637855 }, + { op_id: '8', op: 'PUT', object_id: 'test8', checksum: 3644033632 }, + { op_id: '9', op: 'PUT', object_id: 'test9', checksum: 1011055869 }, + { op_id: '10', op: 'PUT', object_id: 'test10', checksum: 1331456365 } + ]; + expect(test_utils.getBatchData(batch[1])).toEqual( + config.compressedBucketStorage ? global2Operations.slice(0, 4) : global2Operations.slice(0, 1) + ); expect(test_utils.getBatchMeta(batch[0])).toEqual({ after: '0', @@ -1346,7 +1339,7 @@ bucket_definitions: expect(test_utils.getBatchMeta(batch[1])).toEqual({ after: '0', has_more: true, - next_after: '5' + next_after: config.compressedBucketStorage ? '5' : '2' }); }); @@ -1391,7 +1384,8 @@ bucket_definitions: const { batch, global1Request, global2Request } = await setup({ limit: 3, chunkLimitBytes: 50 }); if (config.compressedBucketStorage) { - // In v3+, ops in the same bucket share a document, so ops 2 and 3 (global2) are batched together + // In v3+, ops in the same bucket share a document. Once hydrated, the entire document is yielded even + // when it exceeds the byte limit; the byte limit controls whether another document is fetched. expect(batch.length).toEqual(2); expect(batch[0].chunkData.bucket).toEqual(global1Request.bucket); expect(batch[1].chunkData.bucket).toEqual(global2Request.bucket); @@ -1402,7 +1396,14 @@ bucket_definitions: expect(test_utils.getBatchData(batch[1])).toEqual([ { op_id: '2', op: 'PUT', object_id: 'test2', checksum: 730027011 }, - { op_id: '3', op: 'PUT', object_id: 'test3', checksum: 1359888332 } + { op_id: '3', op: 'PUT', object_id: 'test3', checksum: 1359888332 }, + { op_id: '4', op: 'PUT', object_id: 'test4', checksum: 2049153252 }, + { op_id: '5', op: 'PUT', object_id: 'test5', checksum: 3686902721 }, + { op_id: '6', op: 'PUT', object_id: 'test6', checksum: 1974820016 }, + { op_id: '7', op: 'PUT', object_id: 'test7', checksum: 2477637855 }, + { op_id: '8', op: 'PUT', object_id: 'test8', checksum: 3644033632 }, + { op_id: '9', op: 'PUT', object_id: 'test9', checksum: 1011055869 }, + { op_id: '10', op: 'PUT', object_id: 'test10', checksum: 1331456365 } ]); expect(test_utils.getBatchMeta(batch[0])).toEqual({ @@ -1413,11 +1414,11 @@ bucket_definitions: expect(test_utils.getBatchMeta(batch[1])).toEqual({ after: '0', - has_more: true, - next_after: '3' + has_more: false, + next_after: '10' }); } else { - expect(batch.length).toEqual(3); + expect(batch.length).toBeGreaterThanOrEqual(3); expect(batch[0].chunkData.bucket).toEqual(global1Request.bucket); expect(batch[1].chunkData.bucket).toEqual(global2Request.bucket); expect(batch[2].chunkData.bucket).toEqual(global2Request.bucket); @@ -1560,15 +1561,13 @@ bucket_definitions: afterReplicaId: test_utils.rid('test1') }); await writer.commit('1/1'); - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); const request = bucketRequest(syncRules, 'global[]'); - const checksums = [ - ...(await bucketStorage.getChecksums(test_utils.testCheckpoint(checkpoint), [request])).values() - ]; + const checksums = [...(await bucketStorage.getChecksums(checkpoint, [request])).values()]; expect(checksums).toEqual([{ bucket: request.bucket, checksum: 1917136889, count: 1 }]); const checksums2 = [ - ...(await bucketStorage.getChecksums(test_utils.testCheckpoint(checkpoint + 1n), [request])).values() + ...(await bucketStorage.getChecksums(test_utils.testCheckpoint(checkpoint.checkpoint + 1n), [request])).values() ]; expect(checksums2).toEqual([{ bucket: request.bucket, checksum: 1917136889, count: 1 }]); }); @@ -1577,7 +1576,7 @@ bucket_definitions: test('empty checkpoints (1)', async () => { await using factory = await generateStorageFactory(); - const { stream: replicationStream, content: syncRules } = await test_utils.deploySyncRules( + const { stream: replicationStream } = await test_utils.deploySyncRules( factory, updateSyncRulesFromYaml( ` @@ -1614,7 +1613,7 @@ bucket_definitions: test('empty checkpoints (2)', async () => { await using factory = await generateStorageFactory(); - const { stream: replicationStream, content: syncRules } = await test_utils.deploySyncRules( + const { stream: replicationStream } = await test_utils.deploySyncRules( factory, updateSyncRulesFromYaml( ` @@ -1771,8 +1770,8 @@ bucket_definitions: const cp = await bucketStorage.getCheckpoint(); expect(cp.lsn).toEqual('3/1'); - const data = await test_utils.fromAsync( - bucketStorage.getBucketDataBatch(test_utils.testCheckpoint(cp.checkpoint), [bucketRequest(syncRules, 'global[]')]) + const data = await test_utils.getBatchArray( + bucketStorage.getBucketDataBatch(cp, [bucketRequest(syncRules, 'global[]')]) ); expect(data).toEqual([]); @@ -1823,15 +1822,13 @@ bucket_definitions: } } await writer.commit('1/1'); - const { checkpoint } = await bucketStorage.getCheckpoint(); + const checkpoint = await bucketStorage.getCheckpoint(); bucketStorage.clearChecksumCache(); const users = ['u1', 'u2', 'u3', 'u4']; const expectedChecksums = [346204588, 5261081, 134760718, -302639724]; const bucketRequests = users.map((user) => bucketRequest(syncRules, `user["${user}"]`)); - const checksums = [ - ...(await bucketStorage.getChecksums(test_utils.testCheckpoint(checkpoint), bucketRequests)).values() - ]; + const checksums = [...(await bucketStorage.getChecksums(checkpoint, bucketRequests)).values()]; checksums.sort((a, b) => a.bucket.localeCompare(b.bucket)); const expected = bucketRequests.map((request, index) => ({ bucket: request.bucket, diff --git a/packages/service-core-tests/src/tests/register-sync-tests.ts b/packages/service-core-tests/src/tests/register-sync-tests.ts index 6ab3c0882..2bb4c6c75 100644 --- a/packages/service-core-tests/src/tests/register-sync-tests.ts +++ b/packages/service-core-tests/src/tests/register-sync-tests.ts @@ -251,29 +251,36 @@ streams: expect(lines).toMatchSnapshot(); }); - test('sync interrupts low-priority buckets on new checkpoints', async () => { + test('carries pending low-priority buckets into a new checkpoint', async () => { await using f = await factory(); const syncRules = await updateSyncRules(f, { content: ` bucket_definitions: - b0: + b0a: priority: 2 data: - - SELECT * FROM test WHERE LENGTH(id) <= 5; + - SELECT * FROM test WHERE substring(id, 1, 6) = 'first-'; + b0b: + priority: 3 + data: + - SELECT * FROM test WHERE substring(id, 1, 4) = 'low-'; b1: priority: 1 data: - - SELECT * FROM test WHERE LENGTH(id) > 5; + - SELECT * FROM test WHERE substring(id, 1, 8) = 'highprio'; ` }); const bucketStorage = f.getInstance(syncRules); await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS); const testTable = await test_utils.resolveTestTable(writer, 'test', ['id'], config); + const syncRulesContent = syncRules.syncConfigContent[0]; + const b0aBucket = test_utils.bucketRequest(syncRulesContent, 'b0a[]').bucket; + const b0bBucket = test_utils.bucketRequest(syncRulesContent, 'b0b[]').bucket; + const b1Bucket = test_utils.bucketRequest(syncRulesContent, 'b1[]').bucket; await writer.markAllSnapshotDone('0/1'); - // Initial data: Add one priority row and 10k low-priority rows. await writer.save({ sourceTable: testTable, tag: storage.SaveOperationTag.INSERT, @@ -283,15 +290,26 @@ bucket_definitions: }, afterReplicaId: 'highprio' }); - for (let i = 0; i < 10_000; i++) { + for (let i = 0; i < 999; i++) { await writer.save({ sourceTable: testTable, tag: storage.SaveOperationTag.INSERT, after: { - id: `${i}`, - description: 'low prio' + id: `first-${i}`, + description: 'first low-priority bucket' }, - afterReplicaId: `${i}` + afterReplicaId: `first-${i}` + }); + } + for (let i = 0; i < 9_001; i++) { + await writer.save({ + sourceTable: testTable, + tag: storage.SaveOperationTag.INSERT, + after: { + id: `low-${i}`, + description: 'last low-priority bucket' + }, + afterReplicaId: `low-${i}` }); } @@ -311,8 +329,12 @@ bucket_definitions: isEncodingAsBson: false }); - let sentCheckpoints = 0; - let sentRows = 0; + const dataLines: Array<{ bucket: string; objectIds: string[]; afterCheckpointDiff: boolean }> = []; + const partialCheckpoints: Array<{ priority: number; afterCheckpointDiff: boolean }> = []; + const checkpoints: any[] = []; + const checkpointCompletes: any[] = []; + let afterCheckpointDiff = false; + let interruptionTriggered = false; for await (let next of stream) { if (typeof next == 'string') { @@ -320,10 +342,13 @@ bucket_definitions: } if (typeof next === 'object' && next !== null) { if ('partial_checkpoint_complete' in next) { - if (sentCheckpoints == 1) { - // Save new data to interrupt the low-priority sync. + partialCheckpoints.push({ + priority: next.partial_checkpoint_complete.priority, + afterCheckpointDiff + }); - // Add another high-priority row. This should interrupt the long-running low-priority sync. + if (next.partial_checkpoint_complete.priority == 2 && !interruptionTriggered) { + interruptionTriggered = true; await writer.save({ sourceTable: testTable, tag: storage.SaveOperationTag.INSERT, @@ -335,28 +360,76 @@ bucket_definitions: }); await writer.commit('0/2'); - } else { - // Low-priority sync from the first checkpoint was interrupted. This should not happen before - // 1000 low-priority items were synchronized. - expect(sentCheckpoints).toBe(2); - expect(sentRows).toBeGreaterThan(1000); + // Let the checkpoint watcher observe the update before requesting the next priority. + await timers.setTimeout(50); } } if ('checkpoint' in next || 'checkpoint_diff' in next) { - sentCheckpoints += 1; + checkpoints.push(next); + afterCheckpointDiff = 'checkpoint_diff' in next; } if ('data' in next) { - sentRows += next.data.data.length; + dataLines.push({ + bucket: next.data.bucket, + objectIds: next.data.data.flatMap((entry: any) => (entry.object_id == null ? [] : [entry.object_id])), + afterCheckpointDiff + }); } if ('checkpoint_complete' in next) { + checkpointCompletes.push(next); break; } } } - expect(sentCheckpoints).toBe(2); - expect(sentRows).toBe(10002); + // Expected flow (data may be split into any number of chunks): + // + // checkpoint + // data: b1 contains highprio + // partial_checkpoint_complete: priority 1 + // data: b0a contains all 999 first-* rows (1,000 operations including highprio) + // partial_checkpoint_complete: priority 2 + // ## add highprio2, interrupting before b0b starts + // checkpoint_diff + // data: b1 contains highprio2 + // partial_checkpoint_complete: priority 1 + // data: b0b contains all 9,001 low-* rows carried over from the interrupted checkpoint + // checkpoint_complete: only for the new checkpoint + expect(interruptionTriggered).toBe(true); + expect(checkpoints).toHaveLength(2); + expect(checkpoints[0]).toHaveProperty('checkpoint'); + expect(checkpoints[1]).toHaveProperty('checkpoint_diff'); + expect(partialCheckpoints).toEqual([ + { priority: 1, afterCheckpointDiff: false }, + { priority: 2, afterCheckpointDiff: false }, + { priority: 1, afterCheckpointDiff: true } + ]); + expect(checkpointCompletes).toEqual([ + { + checkpoint_complete: { + last_op_id: checkpoints[1].checkpoint_diff.last_op_id + } + } + ]); + + const objectIdsFor = (bucket: string, afterDiff: boolean) => + dataLines + .filter((line) => line.bucket == bucket && line.afterCheckpointDiff == afterDiff) + .flatMap((line) => line.objectIds); + + expect(objectIdsFor(b1Bucket, false)).toEqual(['highprio']); + const b0aObjectIds = objectIdsFor(b0aBucket, false); + expect(new Set(b0aObjectIds)).toEqual(new Set(Array.from({ length: 999 }, (_, i) => `first-${i}`))); + expect(b0aObjectIds).toHaveLength(999); + expect(objectIdsFor(b0bBucket, false)).toEqual([]); + + expect(objectIdsFor(b1Bucket, true)).toEqual(['highprio2']); + expect(objectIdsFor(b0aBucket, true)).toEqual([]); + + const b0bObjectIds = objectIdsFor(b0bBucket, true); + expect(new Set(b0bObjectIds)).toEqual(new Set(Array.from({ length: 9_001 }, (_, i) => `low-${i}`))); + expect(b0bObjectIds).toHaveLength(9_001); }); test('sync interruptions with unrelated data', async () => { @@ -490,37 +563,36 @@ bucket_definitions: expect(sentRows).toBe(10002); }); - test('sync interrupts low-priority buckets on new checkpoints (2)', async () => { + test('restarts updated low-priority buckets at a new checkpoint', async () => { await using f = await factory(); - // bucket0a -> send all data - // then interrupt checkpoint with new data for all buckets - // -> data for all buckets should be sent in the new checkpoint - const syncRules = await updateSyncRules(f, { content: ` bucket_definitions: b0a: priority: 2 data: - - SELECT * FROM test WHERE LENGTH(id) <= 5; + - SELECT * FROM test WHERE substring(id, 1, 6) = 'first-'; b0b: - priority: 2 + priority: 3 data: - - SELECT * FROM test WHERE LENGTH(id) <= 5; + - SELECT * FROM test WHERE substring(id, 1, 4) = 'low-'; b1: priority: 1 data: - - SELECT * FROM test WHERE LENGTH(id) > 5; + - SELECT * FROM test WHERE substring(id, 1, 8) = 'highprio'; ` }); const bucketStorage = f.getInstance(syncRules); await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS); const testTable = await test_utils.resolveTestTable(writer, 'test', ['id'], config); + const syncRulesContent = syncRules.syncConfigContent[0]; + const b0aBucket = test_utils.bucketRequest(syncRulesContent, 'b0a[]').bucket; + const b0bBucket = test_utils.bucketRequest(syncRulesContent, 'b0b[]').bucket; + const b1Bucket = test_utils.bucketRequest(syncRulesContent, 'b1[]').bucket; await writer.markAllSnapshotDone('0/1'); - // Initial data: Add one priority row and 10k low-priority rows. await writer.save({ sourceTable: testTable, tag: storage.SaveOperationTag.INSERT, @@ -530,15 +602,26 @@ bucket_definitions: }, afterReplicaId: 'highprio' }); + for (let i = 0; i < 999; i++) { + await writer.save({ + sourceTable: testTable, + tag: storage.SaveOperationTag.INSERT, + after: { + id: `first-${i}`, + description: 'first low-priority bucket' + }, + afterReplicaId: `first-${i}` + }); + } for (let i = 0; i < 2_000; i++) { await writer.save({ sourceTable: testTable, tag: storage.SaveOperationTag.INSERT, after: { - id: `${i}`, + id: `low-${i}`, description: 'low prio' }, - afterReplicaId: `${i}` + afterReplicaId: `low-${i}` }); } @@ -558,8 +641,12 @@ bucket_definitions: isEncodingAsBson: false }); - let sentRows = 0; - let lines: any[] = []; + const dataLines: Array<{ bucket: string; objectIds: string[]; afterCheckpointDiff: boolean }> = []; + const partialCheckpoints: Array<{ priority: number; afterCheckpointDiff: boolean }> = []; + const checkpoints: any[] = []; + const checkpointCompletes: any[] = []; + let afterCheckpointDiff = false; + let interruptionTriggered = false; for await (let next of stream) { if (typeof next == 'string') { @@ -567,19 +654,13 @@ bucket_definitions: } if (typeof next === 'object' && next !== null) { if ('partial_checkpoint_complete' in next) { - lines.push(next); - } - if ('checkpoint' in next || 'checkpoint_diff' in next) { - lines.push(next); - } + partialCheckpoints.push({ + priority: next.partial_checkpoint_complete.priority, + afterCheckpointDiff + }); - if ('data' in next) { - lines.push({ data: { ...next.data, data: undefined } }); - sentRows += next.data.data.length; - - if (sentRows == 1001) { - // Save new data to interrupt the low-priority sync. - // Add another high-priority row. This should interrupt the long-running low-priority sync. + if (next.partial_checkpoint_complete.priority == 2 && !interruptionTriggered) { + interruptionTriggered = true; await writer.save({ sourceTable: testTable, tag: storage.SaveOperationTag.INSERT, @@ -590,51 +671,87 @@ bucket_definitions: afterReplicaId: 'highprio2' }); - // Also add a low-priority row await writer.save({ sourceTable: testTable, tag: storage.SaveOperationTag.INSERT, after: { - id: '2001', + id: 'low-2000', description: 'Another low-priority row' }, - afterReplicaId: '2001' + afterReplicaId: 'low-2000' }); await writer.commit('0/2'); - } - - if (sentRows >= 1000 && sentRows <= 2001) { - // pause for a bit to give the stream time to process interruptions. - // This covers the data batch above and the next one. + // Let the checkpoint watcher observe the update before requesting the next priority. await timers.setTimeout(50); } } + if ('checkpoint' in next || 'checkpoint_diff' in next) { + checkpoints.push(next); + afterCheckpointDiff = 'checkpoint_diff' in next; + } + + if ('data' in next) { + dataLines.push({ + bucket: next.data.bucket, + objectIds: next.data.data.flatMap((entry: any) => (entry.object_id == null ? [] : [entry.object_id])), + afterCheckpointDiff + }); + } if ('checkpoint_complete' in next) { - lines.push(next); + checkpointCompletes.push(next); break; } } } - // Expected lines (full details in snapshot): + // Expected flow (data may be split into any number of chunks): // - // checkpoint (4001) - // data (b1[] 0 -> 1) - // partial_checkpoint_complete (4001, priority 1) - // data (b0a[], 0 -> 2000) - // ## adds new data, interrupting the checkpoint - // data (b0a[], 2000 -> 4000) # expected - stream is already busy with this by the time it receives the interruption - // checkpoint_diff (4004) - // data (b1[], 1 -> 4002) - // partial_checkpoint_complete (4004, priority 1) - // data (b0a[], 4000 -> 4003) - // data (b0b[], 0 -> 1999) - // data (b0b[], 1999 -> 3999) - // data (b0b[], 3999 -> 4004) - // checkpoint_complete (4004) - expect(lines).toMatchSnapshot(); - expect(sentRows).toBe(4004); + // checkpoint + // data: b1 contains highprio + // partial_checkpoint_complete: priority 1 + // data: b0a contains all 999 first-* rows (1,000 operations including highprio) + // partial_checkpoint_complete: priority 2 + // ## add highprio2 and low-2000, interrupting before b0b starts + // checkpoint_diff + // data: b1 contains highprio2 + // partial_checkpoint_complete: priority 1 + // data: b0b contains all 2,001 low-* rows + // checkpoint_complete: only for the new checkpoint + expect(interruptionTriggered).toBe(true); + expect(checkpoints).toHaveLength(2); + expect(checkpoints[0]).toHaveProperty('checkpoint'); + expect(checkpoints[1]).toHaveProperty('checkpoint_diff'); + expect(partialCheckpoints).toEqual([ + { priority: 1, afterCheckpointDiff: false }, + { priority: 2, afterCheckpointDiff: false }, + { priority: 1, afterCheckpointDiff: true } + ]); + expect(checkpointCompletes).toEqual([ + { + checkpoint_complete: { + last_op_id: checkpoints[1].checkpoint_diff.last_op_id + } + } + ]); + + const objectIdsFor = (bucket: string, afterDiff: boolean) => + dataLines + .filter((line) => line.bucket == bucket && line.afterCheckpointDiff == afterDiff) + .flatMap((line) => line.objectIds); + + expect(objectIdsFor(b1Bucket, false)).toEqual(['highprio']); + const b0aObjectIds = objectIdsFor(b0aBucket, false); + expect(new Set(b0aObjectIds)).toEqual(new Set(Array.from({ length: 999 }, (_, i) => `first-${i}`))); + expect(b0aObjectIds).toHaveLength(999); + expect(objectIdsFor(b0bBucket, false)).toEqual([]); + + expect(objectIdsFor(b1Bucket, true)).toEqual(['highprio2']); + expect(objectIdsFor(b0aBucket, true)).toEqual([]); + + const b0bObjectIds = objectIdsFor(b0bBucket, true); + expect(new Set(b0bObjectIds)).toEqual(new Set(Array.from({ length: 2_001 }, (_, i) => `low-${i}`))); + expect(b0bObjectIds).toHaveLength(2_001); }); test('sends checkpoint complete line for empty checkpoint', async () => { diff --git a/packages/service-core/src/storage/SyncRulesBucketStorage.ts b/packages/service-core/src/storage/SyncRulesBucketStorage.ts index c626ccaef..49648dbe7 100644 --- a/packages/service-core/src/storage/SyncRulesBucketStorage.ts +++ b/packages/service-core/src/storage/SyncRulesBucketStorage.ts @@ -133,6 +133,13 @@ export interface SyncRulesBucketStorage * 1. Separate buckets. * 2. Limit the size of each individual chunk according to options.batchSizeLimitBytes. * + * The batch may not contain all data for the checkpoint, if the checkpoint is large. The caller must + * continue querying if either: + * 1. The last chunk for any bucket has has_more = true. + * 2. A SyncBucketDataBatchEnd is returned with hasMore = true. + * + * The first check can be skipped if a SyncBucketDataBatchEnd is returned with hasMore = false. + * * @param checkpoint the checkpoint * @param dataBuckets current bucket states * @param options batch size options @@ -141,7 +148,7 @@ export interface SyncRulesBucketStorage checkpoint: ReplicationCheckpoint, dataBuckets: BucketDataRequest[], options?: BucketDataBatchOptions - ): AsyncIterable; + ): AsyncIterable; /** * Compute checksums for a given list of buckets. @@ -426,6 +433,20 @@ export interface SyncBucketDataChunk { targetOp: util.InternalOpId | null; } +export interface SyncBucketDataBatchEnd { + /** + * True if there may be more data for this checkpoint, and the caller should continue querying. + * + * This is different from `SyncBucketDataChunk.has_more`, which is per-bucket. This is a global signal for the + * entire request, and may be true even if there is no returned chunk with has_more: true. + */ + hasMore: boolean; +} + +export function isBatchEnd(chunk: SyncBucketDataChunk | SyncBucketDataBatchEnd): chunk is SyncBucketDataBatchEnd { + return (chunk as SyncBucketDataBatchEnd).hasMore !== undefined; +} + export interface ReplicationCheckpoint { readonly checkpoint: util.InternalOpId; readonly lsn: string | null; diff --git a/packages/service-core/src/sync/sync.ts b/packages/service-core/src/sync/sync.ts index 47db7eebf..09f1485fd 100644 --- a/packages/service-core/src/sync/sync.ts +++ b/packages/service-core/src/sync/sync.ts @@ -6,6 +6,7 @@ import * as storage from '../storage/storage-index.js'; import * as util from '../util/util-index.js'; import { Logger, logger as defaultLogger } from '@powersync/lib-services-framework'; +import { isBatchEnd } from '../storage/storage-index.js'; import { mergeAsyncIterables } from '../streams/streams-index.js'; import { PerformanceTracer, type Span } from '../tracing/PerformanceTracer.js'; import { BucketChecksumState, CheckpointLine, type SyncCheckpointTraceCategory } from './BucketChecksumState.js'; @@ -459,11 +460,17 @@ async function* bucketDataBatch( // in-flight storage work when the connection itself is closed. signal: abort_connection }); - for await (let { chunkData: r, targetOp } of dataBatches) { + for await (let chunk of dataBatches) { // Abort in current batch if the connection is closed if (abort_connection.aborted) { return null; } + if (isBatchEnd(chunk)) { + // This replaces any other has_more value, since the batch end is the last chunk. + has_more = chunk.hasMore; + break; + } + const { chunkData: r, targetOp } = chunk; if (r.has_more) { has_more = true; }