mirror of
https://github.com/immich-app/immich
synced 2026-08-15 13:03:57 +00:00
fix(server): respect backpressure in the sync stream
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
This commit is contained in:
enol5423
parent
447cc40a50
commit
55b673093c
2 changed files with 155 additions and 66 deletions
78
server/src/services/sync.service.spec.ts
Normal file
78
server/src/services/sync.service.spec.ts
Normal file
|
|
@ -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)]);
|
||||
});
|
||||
});
|
||||
|
|
@ -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 });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue