From 55b673093c452a5243b075cdc86e0aea488b0f1b Mon Sep 17 00:00:00 2001 From: enol5423 Date: Fri, 14 Aug 2026 18:48:30 +0600 Subject: [PATCH] fix(server): respect backpressure in the sync stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit send() called response.write() but discarded its boolean return value. Every sync handler produces items via for-await loops over async DB cursors with no gating, so a client reading slower than the server can produce (a large AssetOcrV1 backfill over a slow mobile connection, per #29925) lets the response's internal write buffer grow unbounded — observed as either the V8 heap OOM'ing or the kernel OOM-killer taking down the api worker, and on constrained hosts taking other services with it. write() returns false once the stream's internal buffer exceeds its highWaterMark, signaling the producer should pause until 'drain' fires. send() now awaits that signal, and every call site awaits send() in turn, so a slow reader naturally throttles the DB cursor instead of the server continuing to buffer data the client hasn't caught up to yet. Refs #29925 --- server/src/services/sync.service.spec.ts | 78 +++++++++++++ server/src/services/sync.service.ts | 143 ++++++++++++----------- 2 files changed, 155 insertions(+), 66 deletions(-) create mode 100644 server/src/services/sync.service.spec.ts diff --git a/server/src/services/sync.service.spec.ts b/server/src/services/sync.service.spec.ts new file mode 100644 index 0000000000..ff505c1e67 --- /dev/null +++ b/server/src/services/sync.service.spec.ts @@ -0,0 +1,78 @@ +import { Writable } from 'node:stream'; +import { SyncEntityType } from 'src/enum'; +import { send } from 'src/services/sync.service'; +import { serialize } from 'src/utils/sync'; + +type TestStream = { + stream: Writable; + chunks: string[]; + flushNext: () => void; + pendingCount: () => number; +}; + +const createTestStream = (highWaterMark: number): TestStream => { + const chunks: string[] = []; + const pendingCallbacks: Array<() => void> = []; + + const stream = new Writable({ + highWaterMark, + write(chunk, _encoding, callback) { + chunks.push(chunk.toString()); + pendingCallbacks.push(callback); + }, + }); + + return { + stream, + chunks, + flushNext: () => pendingCallbacks.shift()?.(), + pendingCount: () => pendingCallbacks.length, + }; +}; + +describe('send', () => { + const item = { + type: SyncEntityType.SyncCompleteV1 as const, + data: {}, + ids: ['now-id'] as [string], + }; + + it('resolves immediately when the stream has capacity', async () => { + // A large highWaterMark means write() never signals backpressure for a + // single small item. + const { stream, chunks, flushNext } = createTestStream(1024 * 1024); + + const sendPromise = send(stream, item); + flushNext(); + await sendPromise; + + expect(chunks).toEqual([serialize(item)]); + }); + + it('waits for the drain event before resolving when the stream signals backpressure', async () => { + // A tiny highWaterMark means the very first write already exceeds + // capacity, so write() returns false and send() must wait for 'drain'. + const { stream, chunks, flushNext, pendingCount } = createTestStream(1); + + let resolved = false; + const sendPromise = send(stream, item).then(() => { + resolved = true; + }); + + // Let any pending microtasks run; send() should still be waiting on the + // underlying write to complete and 'drain' to fire — it must not resolve + // just because write() was called. + await Promise.resolve(); + await Promise.resolve(); + expect(resolved).toBe(false); + expect(pendingCount()).toBe(1); + + // Completing the write lets the stream's internal buffer drop back below + // highWaterMark, which is what triggers the 'drain' event. + flushNext(); + await sendPromise; + + expect(resolved).toBe(true); + expect(chunks).toEqual([serialize(item)]); + }); +}); diff --git a/server/src/services/sync.service.ts b/server/src/services/sync.service.ts index e3842b1503..74693cf29d 100644 --- a/server/src/services/sync.service.ts +++ b/server/src/services/sync.service.ts @@ -1,6 +1,7 @@ import { BadRequestException, ForbiddenException, Injectable } from '@nestjs/common'; import { Insertable } from 'kysely'; import { DateTime, Duration } from 'luxon'; +import { once } from 'node:events'; import { Writable } from 'node:stream'; import { OnJob } from 'src/decorators'; import { AuthDto } from 'src/dtos/auth.dto'; @@ -42,12 +43,22 @@ const isEntityBackfillComplete = (createId: string, checkpoint: SyncAck | undefi const getStartId = (createId: string, checkpoint: SyncAck | undefined): string | undefined => createId === checkpoint?.updateId ? checkpoint?.extraId : undefined; -const send = (response: Writable, item: SerializeOptions) => { - response.write(serialize(item)); +export const send = async ( + response: Writable, + item: SerializeOptions, +) => { + // response.write() returns false once the stream's internal buffer exceeds + // its highWaterMark. Every sync handler below produces items faster than a + // slow/mobile client can consume them (see #29925), so without waiting for + // 'drain' here the buffer grows unbounded regardless of read speed. + const canWriteMore = response.write(serialize(item)); + if (!canWriteMore) { + await once(response, 'drain'); + } }; -const sendEntityBackfillCompleteAck = (response: Writable, ackType: SyncEntityType, id: string) => { - send(response, { type: SyncEntityType.SyncAckV1, data: {}, ackType, ids: [id, COMPLETE_ID] }); +const sendEntityBackfillCompleteAck = async (response: Writable, ackType: SyncEntityType, id: string) => { + await send(response, { type: SyncEntityType.SyncAckV1, data: {}, ackType, ids: [id, COMPLETE_ID] }); }; export const SYNC_TYPES_ORDER = [ @@ -141,7 +152,7 @@ export class SyncService extends BaseService { const isPendingSyncReset = await this.sessionRepository.isPendingSyncReset(session.id); if (isPendingSyncReset) { - send(response, { type: SyncEntityType.SyncResetV1, ids: ['reset'], data: {} }); + await send(response, { type: SyncEntityType.SyncResetV1, ids: ['reset'], data: {} }); response.end(); return; } @@ -150,7 +161,7 @@ export class SyncService extends BaseService { const checkpointMap: CheckpointMap = Object.fromEntries(checkpoints.map(({ type, ack }) => [type, fromAck(ack)])); if (this.needsFullSync(checkpointMap)) { - send(response, { type: SyncEntityType.SyncResetV1, ids: ['reset'], data: {} }); + await send(response, { type: SyncEntityType.SyncResetV1, ids: ['reset'], data: {} }); response.end(); return; } @@ -201,7 +212,7 @@ export class SyncService extends BaseService { await handler(); } - send(response, { type: SyncEntityType.SyncCompleteV1, ids: [nowId], data: {} }); + await send(response, { type: SyncEntityType.SyncCompleteV1, ids: [nowId], data: {} }); response.end(); } @@ -242,7 +253,7 @@ export class SyncService extends BaseService { const upsertType = SyncEntityType.AuthUserV1; const upserts = this.syncRepository.authUser.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, profileImagePath, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data: { ...data, hasProfileImage: !!profileImagePath } }); + await send(response, { type: upsertType, ids: [updateId], data: { ...data, hasProfileImage: !!profileImagePath } }); } } @@ -250,13 +261,13 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.UserDeleteV1; const deletes = this.syncRepository.user.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.UserV1; const upserts = this.syncRepository.user.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, profileImagePath, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data: { ...data, hasProfileImage: !!profileImagePath } }); + await send(response, { type: upsertType, ids: [updateId], data: { ...data, hasProfileImage: !!profileImagePath } }); } } @@ -264,13 +275,13 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.PartnerDeleteV1; const deletes = this.syncRepository.partner.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.PartnerV1; const upserts = this.syncRepository.partner.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -282,13 +293,13 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.AssetDeleteV1; const deletes = this.syncRepository.asset.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.AssetV2; const upserts = this.syncRepository.asset.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data: mapSyncAssetV2(data) }); + await send(response, { type: upsertType, ids: [updateId], data: mapSyncAssetV2(data) }); } } @@ -307,7 +318,7 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.PartnerAssetDeleteV1; const deletes = this.syncRepository.partnerAsset.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const backfillType = SyncEntityType.PartnerAssetBackfillV2; @@ -334,14 +345,14 @@ export class SyncService extends BaseService { ); for await (const { updateId, ...data } of backfill) { - send(response, { + await send(response, { type: backfillType, ids: [createId, updateId], data: mapSyncAssetV2(data), }); } - sendEntityBackfillCompleteAck(response, backfillType, createId); + await sendEntityBackfillCompleteAck(response, backfillType, createId); } } else if (partners.length > 0) { await this.upsertBackfillCheckpoint({ @@ -353,7 +364,7 @@ export class SyncService extends BaseService { const upserts = this.syncRepository.partnerAsset.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data: mapSyncAssetV2(data) }); + await send(response, { type: upsertType, ids: [updateId], data: mapSyncAssetV2(data) }); } } @@ -361,7 +372,7 @@ export class SyncService extends BaseService { const upsertType = SyncEntityType.AssetExifV1; const upserts = this.syncRepository.assetExif.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -370,13 +381,13 @@ export class SyncService extends BaseService { const deletes = this.syncRepository.assetEdit.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.AssetEditV1; const upserts = this.syncRepository.assetEdit.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -411,10 +422,10 @@ export class SyncService extends BaseService { ); for await (const { updateId, ...data } of backfill) { - send(response, { type: backfillType, ids: [partner.createId, updateId], data }); + await send(response, { type: backfillType, ids: [partner.createId, updateId], data }); } - sendEntityBackfillCompleteAck(response, backfillType, partner.createId); + await sendEntityBackfillCompleteAck(response, backfillType, partner.createId); } } else if (partners.length > 0) { await this.upsertBackfillCheckpoint({ @@ -426,7 +437,7 @@ export class SyncService extends BaseService { const upserts = this.syncRepository.partnerAssetExif.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -434,14 +445,14 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.AlbumDeleteV1; const deletes = this.syncRepository.album.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.AlbumV1; const upserts = this.syncRepository.album.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { const albumUsers = await this.syncRepository.album.getAlbumUsers(data.id); - send(response, { + await send(response, { type: upsertType, ids: [updateId], // TODO: return null instead of '' in v4 @@ -454,14 +465,14 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.AlbumDeleteV1; const deletes = this.syncRepository.album.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.AlbumV2; const upserts = this.syncRepository.album.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { // TODO: return null instead of '' in v4 - send(response, { type: upsertType, ids: [updateId], data: { ...data, description: data.description ?? '' } }); + await send(response, { type: upsertType, ids: [updateId], data: { ...data, description: data.description ?? '' } }); } } @@ -474,7 +485,7 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.AlbumUserDeleteV1; const deletes = this.syncRepository.albumUser.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const backfillType = SyncEntityType.AlbumUserBackfillV1; @@ -501,10 +512,10 @@ export class SyncService extends BaseService { ); for await (const { updateId, ...data } of backfill) { - send(response, { type: backfillType, ids: [createId, updateId], data }); + await send(response, { type: backfillType, ids: [createId, updateId], data }); } - sendEntityBackfillCompleteAck(response, backfillType, createId); + await sendEntityBackfillCompleteAck(response, backfillType, createId); } } else if (albums.length > 0) { await this.upsertBackfillCheckpoint({ @@ -516,7 +527,7 @@ export class SyncService extends BaseService { const upserts = this.syncRepository.albumUser.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -559,10 +570,10 @@ export class SyncService extends BaseService { ); for await (const { updateId, ...data } of backfill) { - send(response, { type: backfillType, ids: [createId, updateId], data: mapSyncAssetV2(data) }); + await send(response, { type: backfillType, ids: [createId, updateId], data: mapSyncAssetV2(data) }); } - sendEntityBackfillCompleteAck(response, backfillType, createId); + await sendEntityBackfillCompleteAck(response, backfillType, createId); } } else if (albums.length > 0) { await this.upsertBackfillCheckpoint({ @@ -578,7 +589,7 @@ export class SyncService extends BaseService { createCheckpoint, ); for await (const { updateId, ...data } of updates) { - send(response, { type: updateType, ids: [updateId], data: mapSyncAssetV2(data) }); + await send(response, { type: updateType, ids: [updateId], data: mapSyncAssetV2(data) }); } } @@ -586,7 +597,7 @@ export class SyncService extends BaseService { let isFirst = true; for await (const { updateId, ...data } of creates) { if (isFirst) { - send(response, { + await send(response, { type: SyncEntityType.SyncAckV1, data: {}, ackType: SyncEntityType.AlbumAssetUpdateV2, @@ -594,7 +605,7 @@ export class SyncService extends BaseService { }); isFirst = false; } - send(response, { type: createType, ids: [updateId], data: mapSyncAssetV2(data) }); + await send(response, { type: createType, ids: [updateId], data: mapSyncAssetV2(data) }); } } @@ -630,10 +641,10 @@ export class SyncService extends BaseService { ); for await (const { updateId, ...data } of backfill) { - send(response, { type: backfillType, ids: [createId, updateId], data }); + await send(response, { type: backfillType, ids: [createId, updateId], data }); } - sendEntityBackfillCompleteAck(response, backfillType, createId); + await sendEntityBackfillCompleteAck(response, backfillType, createId); } } else if (albums.length > 0) { await this.upsertBackfillCheckpoint({ @@ -649,7 +660,7 @@ export class SyncService extends BaseService { createCheckpoint, ); for await (const { updateId, ...data } of updates) { - send(response, { type: updateType, ids: [updateId], data }); + await send(response, { type: updateType, ids: [updateId], data }); } } @@ -657,7 +668,7 @@ export class SyncService extends BaseService { let isFirst = true; for await (const { updateId, ...data } of creates) { if (isFirst) { - send(response, { + await send(response, { type: SyncEntityType.SyncAckV1, data: {}, ackType: SyncEntityType.AlbumAssetExifUpdateV1, @@ -665,7 +676,7 @@ export class SyncService extends BaseService { }); isFirst = false; } - send(response, { type: createType, ids: [updateId], data }); + await send(response, { type: createType, ids: [updateId], data }); } } @@ -678,7 +689,7 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.AlbumToAssetDeleteV1; const deletes = this.syncRepository.albumToAsset.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const backfillType = SyncEntityType.AlbumToAssetBackfillV1; @@ -705,10 +716,10 @@ export class SyncService extends BaseService { ); for await (const { updateId, ...data } of backfill) { - send(response, { type: backfillType, ids: [createId, updateId], data }); + await send(response, { type: backfillType, ids: [createId, updateId], data }); } - sendEntityBackfillCompleteAck(response, backfillType, createId); + await sendEntityBackfillCompleteAck(response, backfillType, createId); } } else if (albums.length > 0) { await this.upsertBackfillCheckpoint({ @@ -720,7 +731,7 @@ export class SyncService extends BaseService { const upserts = this.syncRepository.albumToAsset.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -728,13 +739,13 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.MemoryDeleteV1; const deletes = this.syncRepository.memory.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.MemoryV1; const upserts = this.syncRepository.memory.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -742,13 +753,13 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.MemoryToAssetDeleteV1; const deletes = this.syncRepository.memoryToAsset.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.MemoryToAssetV1; const upserts = this.syncRepository.memoryToAsset.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -756,13 +767,13 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.StackDeleteV1; const deletes = this.syncRepository.stack.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.StackV1; const upserts = this.syncRepository.stack.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -775,7 +786,7 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.PartnerStackDeleteV1; const deletes = this.syncRepository.partnerStack.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const backfillType = SyncEntityType.PartnerStackBackfillV1; @@ -802,14 +813,14 @@ export class SyncService extends BaseService { ); for await (const { updateId, ...data } of backfill) { - send(response, { + await send(response, { type: backfillType, ids: [createId, updateId], data, }); } - sendEntityBackfillCompleteAck(response, backfillType, createId); + await sendEntityBackfillCompleteAck(response, backfillType, createId); } } else if (partners.length > 0) { await this.upsertBackfillCheckpoint({ @@ -821,7 +832,7 @@ export class SyncService extends BaseService { const upserts = this.syncRepository.partnerStack.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -829,13 +840,13 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.PersonDeleteV1; const deletes = this.syncRepository.person.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.PersonV1; const upserts = this.syncRepository.person.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -849,13 +860,13 @@ export class SyncService extends BaseService { const deleteType = SyncEntityType.AssetFaceDeleteV1; const deletes = this.syncRepository.assetFace.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.AssetFaceV2; const upserts = this.syncRepository.assetFace.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -864,14 +875,14 @@ export class SyncService extends BaseService { const deletes = this.syncRepository.userMetadata.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.UserMetadataV1; const upserts = this.syncRepository.userMetadata.getUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -888,7 +899,7 @@ export class SyncService extends BaseService { ); for await (const { id, ...data } of deletes) { - send(response, { type: deleteType, ids: [id], data }); + await send(response, { type: deleteType, ids: [id], data }); } const upsertType = SyncEntityType.AssetMetadataV1; @@ -898,7 +909,7 @@ export class SyncService extends BaseService { ); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } } @@ -915,7 +926,7 @@ export class SyncService extends BaseService { ); for await (const row of deletes) { - send(response, { type: deleteType, ids: [row.id], data: row }); + await send(response, { type: deleteType, ids: [row.id], data: row }); } const upsertType = SyncEntityType.AssetOcrV1; @@ -925,7 +936,7 @@ export class SyncService extends BaseService { ); for await (const { updateId, ...data } of upserts) { - send(response, { type: upsertType, ids: [updateId], data }); + await send(response, { type: upsertType, ids: [updateId], data }); } }