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 = <T extends keyof SyncItem, D extends SyncItem[T]>(response: Writable, item: SerializeOptions<T, D>) => {
-  response.write(serialize(item));
+export const send = async <T extends keyof SyncItem, D extends SyncItem[T]>(
+  response: Writable,
+  item: SerializeOptions<T, D>,
+) => {
+  // 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 });
     }
   }