From 36082c5a7404c6d5fed178a63c973c836e730a4c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 18 Sep 2026 16:18:31 +0800 Subject: [PATCH] feat(sync): reconciliation scheduler with digest and bootstrap healing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2 Task 4（inbox 对账）： - services/sync 新增 SyncReconciler（runOnce 可测；main 双循环，间隔 SYNC_RECONCILE_INTERVAL_MS env 可调，CN_API_URL 指向 cn 入口）：每轮 先按 album/category/photo/asset 逐类分页拉取 cn reconcile「当前状态 事件」并直接签名推送到既有 content-batches 管道。sinceVersion 取 0 （全量当前态）而非 per-type 已应用最大版本——cn reconcile 以类型级 全局版本阈值过滤，类型内单实体滞后会被 max 掩蔽；ov 的 eventId 幂等 + 版本单调把已追平实体全部判重（recon 确定性 eventId 重扫零成本）， 偏差登记：计划原文对账返回「缺失清单」，契约已定为可重放 SyncEvent 数组（免建第二条拉取通道），openapi 描述同步更新 - cn 新增 POST /internal/sync/digest：每相册权威计数 {categoryCount, publishedPhotoCount, photoCount, assetCount}；ov 比对副本计数（照片 取 [publishedPhotoCount, photoCount] 区间口径——hidden/deleted 照片 若从未以 published 态同步过 ov 合法无行，缺席即撤下），不一致相册 自动触发 cn bootstrap 重建；ov-only 相册（cn 行已删，bootstrap 无法 表达删除）告警登记不自动处置 - 收敛语义两处落位：① sync-inbox 版本比较改「严格更旧才拒绝」—— 同版本新 eventId（bootstrap/reconcile 快照重发）为当前态幂等重断言 应用，修复 ov 缺失副本行；A26 不受影响（更旧版本仍拒绝，cn 侧 entityVersion 与状态一一对应）② album.updated 快照载荷（bootstrap/ reconcile）携带分类全量清单，ov 存在即整体替换 albums_ov.categories ——修复错过的 category 墓碑（Task 3 登记边界：cn 行已删，无现存实体 事件可重放，唯全量替换可收敛） - e2e（reconcile.e2e-spec）：漏事件追赶（ov v1/cn v5 事件投后即失）→ 清扫追平；digest 修复缺失 photos_ov 行（bootstrap 重发 → 同版本 重断言重建）；digest 修复丢失的分类墓碑（全量清单替换收敛）；稳态 空转（applied=0、无 bootstrap）；Task 3 套件保持绿 Co-Authored-By: Claude Code --- .env.example | 6 +- packages/contracts/openapi.yaml | 96 ++++- packages/contracts/src/events.ts | 17 +- packages/contracts/src/generated/openapi.d.ts | 96 ++++- .../api/src/sync-inbox/sync-inbox.service.ts | 50 ++- services/api/src/sync/dto.ts | 12 +- services/api/src/sync/payloads.ts | 34 +- services/api/src/sync/sync.controller.ts | 20 +- services/api/src/sync/sync.service.ts | 176 ++++++++- services/sync/package.json | 4 +- services/sync/src/main.ts | 95 ++++- services/sync/src/reconciler.ts | 293 +++++++++++++++ services/sync/test/reconcile.e2e-spec.ts | 343 ++++++++++++++++++ 13 files changed, 1179 insertions(+), 63 deletions(-) create mode 100644 services/sync/src/reconciler.ts create mode 100644 services/sync/test/reconcile.e2e-spec.ts diff --git a/.env.example b/.env.example index 2d75963..abd7df6 100644 --- a/.env.example +++ b/.env.example @@ -24,10 +24,14 @@ PORT_OV=8081 # P2 Task 3：sync 发送器（services/sync）推送目标——ov 入口基址（不含路径， # 端点路径由发送器拼接为 /v1/internal/sync/content-batches）。 OV_API_URL=http://127.0.0.1:8081 +# P2 Task 4：对账调度器（services/sync reconciler）拉取 reconcile/digest 与 +# 触发 bootstrap 的 cn 入口基址（不含路径）。 +CN_API_URL=http://127.0.0.1:8080 # 以下为 P2 决策初值（D07/D11）的 env 覆盖键：默认值见 # packages/contracts/src/constants.ts（HEARTBEAT_MS 见 src/realtime.ts）， # 此处登记供运维可见与调整；SYNC_BATCH_SIZE/SYNC_PUSH_INTERVAL_MS 自 Task 3 -# 起被 services/sync 发送器读取，其余为前瞻登记（Task 4+ 逐个接线）。 +# 起、SYNC_RECONCILE_INTERVAL_MS 自 Task 4 起被 services/sync 读取，其余为 +# 前瞻登记（Task 7+ 逐个接线）。 SYNC_BATCH_SIZE=100 SYNC_PUSH_INTERVAL_MS=2000 SYNC_RECONCILE_INTERVAL_MS=300000 diff --git a/packages/contracts/openapi.yaml b/packages/contracts/openapi.yaml index ec65e1b..9543c17 100644 --- a/packages/contracts/openapi.yaml +++ b/packages/contracts/openapi.yaml @@ -1369,6 +1369,40 @@ paths: '500': $ref: '#/components/responses/InternalError' + /internal/sync/digest: + post: + operationId: syncDigest + tags: [internal] + summary: 每相册实体计数摘要 + description: | + cn 侧权威计数（P2 Task 4）：每相册 {categoryCount, publishedPhotoCount, + photoCount, assetCount}。ov 对账调度定期拉取并与副本计数比对——不一致 + 的相册触发该相册 bootstrap（当前状态快照重发，含分类全量清单整体替换）， + 收敛错过的墓碑与漏掉的事件。albumId 缺省表示全部相册。鉴权见 [internal] + tag 描述。 + security: [] + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/DigestRequest' + responses: + '200': + description: 每相册计数摘要（albumId 升序）。 + content: + application/json: + schema: + $ref: '#/components/schemas/DigestResponse' + '400': + description: 请求体校验失败（VALIDATION_FAILED）。 + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '500': + $ref: '#/components/responses/InternalError' + components: securitySchemes: bearerAuth: @@ -2719,12 +2753,16 @@ components: ReconcileResponse: type: object - description: 对账结果：按 entityVersion 升序的事件（无缺失为空数组）。 + description: | + 对账结果：按 entityVersion 升序的**当前状态事件**（可直接重放的完整 + SyncEvent，非仅版本清单——P2 Task 4 设计决策：重放走既有 + content-batches 管道，无需第二条拉取通道；sinceVersion 之后无实体时为 + 空数组）。 required: [events] properties: events: type: array - description: 缺失事件（升序）；请求方按序重放。 + description: 当前状态事件（升序）；请求方按序经 content-batches 重放。 items: $ref: '#/components/schemas/SyncEvent' @@ -2750,3 +2788,57 @@ components: format: int64 minimum: 0 description: 快照水位：后续增量对账的 sinceVersion 起点。 + + DigestRequest: + type: object + description: 对账摘要请求（P2 Task 4）。 + properties: + albumId: + type: string + description: 指定相册；缺省为全部相册。 + + AlbumDigest: + type: object + description: | + 单相册权威计数摘要。ov 侧比对口径：albums_ov.categories 元素数应等于 + categoryCount；photos_ov 行数落在 [publishedPhotoCount, photoCount] + 区间内即视为一致（hidden/deleted 照片若从未以 published 态同步过，ov + 合法地无行——缺席即撤下语义）；相册行缺失即不一致。assetCount 供观测 + （ov 无 asset 副本行，不参与比对）。 + required: + - albumId + - categoryCount + - publishedPhotoCount + - photoCount + - assetCount + properties: + albumId: + type: string + description: 相册 ID。 + categoryCount: + type: integer + minimum: 0 + description: 当前分类数（cn 权威行数）。 + publishedPhotoCount: + type: integer + minimum: 0 + description: 可见（published）照片数——ov 必须持有的副本行数下界。 + photoCount: + type: integer + minimum: 0 + description: published+hidden+deleted 照片总数——ov 副本行数上界。 + assetCount: + type: integer + minimum: 0 + description: 生效（effective）资产数（观测用；ov 无 asset 副本行）。 + + DigestResponse: + type: object + description: 每相册计数摘要（albumId 升序）。 + required: [albums] + properties: + albums: + type: array + description: 相册摘要清单；空库为空数组。 + items: + $ref: '#/components/schemas/AlbumDigest' diff --git a/packages/contracts/src/events.ts b/packages/contracts/src/events.ts index 4acb311..f7ccde2 100644 --- a/packages/contracts/src/events.ts +++ b/packages/contracts/src/events.ts @@ -34,8 +34,10 @@ export interface PhotoPublishedPayload { } // album.updated 载荷：ov 副本（albums_ov）展示一张相册所需的标量 + 翻译全集。 -// categories 不在此列——分类是独立事件（category.*，各自携带 entityVersion）； +// categories 默认不在此列——分类是独立事件（category.*，各自携带 entityVersion）； // banner 对象键经 asset.updated 事件同步（albums_ov.bannerAssetKey）。 +// 例外：bootstrap/reconcile 的快照事件携带可选 categories 全量清单（Task 4 +// digest 补漏的收敛原语，见 AlbumCategorySnapshot 注释）。 export interface AlbumUpdatedPayload { albumId: string; publicToken: string; @@ -49,9 +51,22 @@ export interface AlbumUpdatedPayload { startAt: string; // ISO UTC endAt: string | null; // ISO UTC；null 表示未定 translations: AlbumTranslationPayload[]; // locale 升序（确定性输出） + /** 分类全量清单（可选）：存在时 ov **整体替换** albums_ov.categories。 */ + categories?: AlbumCategorySnapshot[]; entityVersion: number; } +// albums_ov.categories JSON 元素的载荷形状。日常增量由 category.updated 逐个 +// upsert / category.deleted 墓碑逐个移除；但**错过的墓碑**（cn 行已删，对账 +// 无法为不存在的行重放事件）只有「全量清单整体替换」才能收敛——bootstrap/ +// reconcile 的 album.updated 快照事件据此携带当前全集（Task 4 digest→bootstrap +// 补漏通道）。缺省（增量事件不携带）时 ov 不触碰 categories，语义不变。 +export interface AlbumCategorySnapshot { + id: string; + sortOrder: number; + translations: CategoryTranslationPayload[]; // locale 升序 +} + export interface AlbumTranslationPayload { locale: string; title: string; diff --git a/packages/contracts/src/generated/openapi.d.ts b/packages/contracts/src/generated/openapi.d.ts index ced2bc2..b2bbd12 100644 --- a/packages/contracts/src/generated/openapi.d.ts +++ b/packages/contracts/src/generated/openapi.d.ts @@ -664,6 +664,30 @@ export interface paths { patch?: never; trace?: never; }; + "/internal/sync/digest": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * 每相册实体计数摘要 + * @description cn 侧权威计数（P2 Task 4）：每相册 {categoryCount, publishedPhotoCount, + * photoCount, assetCount}。ov 对账调度定期拉取并与副本计数比对——不一致 + * 的相册触发该相册 bootstrap（当前状态快照重发，含分类全量清单整体替换）， + * 收敛错过的墓碑与漏掉的事件。albumId 缺省表示全部相册。鉴权见 [internal] + * tag 描述。 + */ + post: operations["syncDigest"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; } export type webhooks = Record; export interface components { @@ -1376,9 +1400,14 @@ export interface components { */ limit: number; }; - /** @description 对账结果：按 entityVersion 升序的事件（无缺失为空数组）。 */ + /** + * @description 对账结果：按 entityVersion 升序的**当前状态事件**（可直接重放的完整 + * SyncEvent，非仅版本清单——P2 Task 4 设计决策：重放走既有 + * content-batches 管道，无需第二条拉取通道；sinceVersion 之后无实体时为 + * 空数组）。 + */ ReconcileResponse: { - /** @description 缺失事件（升序）；请求方按序重放。 */ + /** @description 当前状态事件（升序）；请求方按序经 content-batches 重放。 */ events: components["schemas"]["SyncEvent"][]; }; /** @description 全量快照引导请求。 */ @@ -1396,6 +1425,35 @@ export interface components { */ watermark: number; }; + /** @description 对账摘要请求（P2 Task 4）。 */ + DigestRequest: { + /** @description 指定相册；缺省为全部相册。 */ + albumId?: string; + }; + /** + * @description 单相册权威计数摘要。ov 侧比对口径：albums_ov.categories 元素数应等于 + * categoryCount；photos_ov 行数落在 [publishedPhotoCount, photoCount] + * 区间内即视为一致（hidden/deleted 照片若从未以 published 态同步过，ov + * 合法地无行——缺席即撤下语义）；相册行缺失即不一致。assetCount 供观测 + * （ov 无 asset 副本行，不参与比对）。 + */ + AlbumDigest: { + /** @description 相册 ID。 */ + albumId: string; + /** @description 当前分类数（cn 权威行数）。 */ + categoryCount: number; + /** @description 可见（published）照片数——ov 必须持有的副本行数下界。 */ + publishedPhotoCount: number; + /** @description published+hidden+deleted 照片总数——ov 副本行数上界。 */ + photoCount: number; + /** @description 生效（effective）资产数（观测用；ov 无 asset 副本行）。 */ + assetCount: number; + }; + /** @description 每相册计数摘要（albumId 升序）。 */ + DigestResponse: { + /** @description 相册摘要清单；空库为空数组。 */ + albums: components["schemas"]["AlbumDigest"][]; + }; }; responses: { /** @description 未认证：令牌缺失、无效或已过期（AUTH_TOKEN_EXPIRED）。 */ @@ -2916,4 +2974,38 @@ export interface operations { 500: components["responses"]["InternalError"]; }; }; + syncDigest: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["DigestRequest"]; + }; + }; + responses: { + /** @description 每相册计数摘要（albumId 升序）。 */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["DigestResponse"]; + }; + }; + /** @description 请求体校验失败（VALIDATION_FAILED）。 */ + 400: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["Error"]; + }; + }; + 500: components["responses"]["InternalError"]; + }; + }; } diff --git a/services/api/src/sync-inbox/sync-inbox.service.ts b/services/api/src/sync-inbox/sync-inbox.service.ts index 283a99c..4f01fda 100644 --- a/services/api/src/sync-inbox/sync-inbox.service.ts +++ b/services/api/src/sync-inbox/sync-inbox.service.ts @@ -1,21 +1,27 @@ // services/api/src/sync-inbox/sync-inbox.service.ts —— ov 内容事件收件箱 -// （P2 Task 3：接收 + 持久化 + 简单应用；对账调度属 Task 4）。逐事件单事务： +// （P2 Task 3：接收 + 持久化 + 应用；Task 4 起同版本重断言语义）。逐事件单事务： // // ① eventId 幂等：sync_inbox 主键冲突（含并发窗口）→ duplicate，受理不计应用； -// ② entityVersion 单调：sync_state 已应用版本 ≥ 事件版本 → stale，标记 -// appliedAt（已裁决，非积压）但不投影——旧事件不能复活新状态（A26）； +// ② entityVersion 单调：sync_state 已应用版本 **严格大于** 事件版本 → stale， +// 标记 appliedAt（已裁决，非积压）但不投影——旧事件不能复活新状态（A26）； +// **等于**已应用版本 → 重断言应用（Task 4 语义）：cn 侧 entityVersion 与 +// 状态一一对应（每次变更事务内自增），同版本 = 同状态，重放为幂等重投影 +// ——bootstrap/reconcile 快照重发（新 eventId、同版本）正是靠它修复 ov 侧 +// 缺失/漂移的副本行（digest→bootstrap 补漏通道）；正常增量管线不受影响 +// （同事件重投递走 ① 的 eventId 幂等锚，根本到不了版本比较）； // ③ 依赖缺失暂存：非 album 事件在 albums_ov 无所属相册行时保持 appliedAt // null（暂存不 applied，不投影）；album 事件应用后按 albumId 回扫暂存 // 队列补应用（相册到达即解锁该相册全部暂存事件）； -// ④ 应用投影（版本已更高）：album → albums_ov upsert（不触碰 categories/ -// bannerAssetKey/emergencyDisabled——分别归 category.* / asset.updated / -// 紧急停用通道 Task 5）；category → albums_ov.categories JSON 元素 -// upsert/移除；photo.published → photos_ov upsert（保留 likeTotal/ -// countVersion 权威计数）；photo.hidden/deleted → photos_ov.visibility -// 置位（行不存在 = 从未复制即不可见，缺席即撤下语义）；asset → banner -// 投影 albums_ov.bannerAssetKey（cover/logo 仅推进版本，ov schema 无位）； -// ⑤ sync_state upsert appliedVersion（watermark=false 实体行；水位行归 -// Task 4 对账）。 +// ④ 应用投影（版本不低于已应用）：album → albums_ov upsert（不触碰 +// bannerAssetKey/emergencyDisabled——分别归 asset.updated / 紧急停用通道 +// Task 5；categories 仅当载荷携带全量清单（bootstrap/reconcile 快照事件） +// 时整体替换——错过的 category 墓碑由此收敛，增量事件不携带则不动）； +// category → albums_ov.categories JSON 元素 upsert/移除；photo.published → +// photos_ov upsert（保留 likeTotal/countVersion 权威计数）；photo.hidden/ +// deleted → photos_ov.visibility 置位（行不存在 = 从未复制即不可见，缺席 +// 即撤下语义）；asset → banner 投影 albums_ov.bannerAssetKey（cover/logo +// 仅推进版本，ov schema 无位）； +// ⑤ sync_state upsert appliedVersion（watermark=false 实体行；水位行预留）。 // // 墓碑优先由 ② 保证：tombstone 事件版本必高于其前的任意旧版本，旧 // photo.published/category.updated 重放被版本比较直接拒绝。 @@ -125,7 +131,9 @@ export class SyncInboxService { tx: Prisma.TransactionClient, event: SyncEventWire, ): Promise { - // ② 版本单调：旧版本裁决即终（appliedAt 置位，不入积压）。 + // ② 版本单调：**更旧**版本裁决即终（appliedAt 置位，不入积压）；**等于** + // 已应用版本 → 重断言应用（同版本=同状态，幂等重投影；digest→bootstrap + // 补漏修复缺失副本行的机制，见文件头②）。 const state = await tx.sync_state.findUnique({ where: { entityType_entityId: { @@ -134,7 +142,7 @@ export class SyncInboxService { }, }, }); - if (state !== null && state.appliedVersion >= event.entityVersion) { + if (state !== null && state.appliedVersion > event.entityVersion) { await tx.sync_inbox.update({ where: { eventId: event.eventId }, data: { appliedAt: new Date() }, @@ -186,7 +194,14 @@ export class SyncInboxService { ): Promise { if (event.entityType === 'album') { const p = event.payload as unknown as AlbumUpdatedPayload; - // update 分支不触碰 categories/bannerAssetKey/emergencyDisabled（见文件头④）。 + // categories 全量清单（可选，仅 bootstrap/reconcile 快照事件携带）：整体 + // 替换 albums_ov.categories——错过的 category 墓碑（cn 行已删、无事件可 + // 重放）只有全量替换才能收敛（Task 4 digest→bootstrap 补漏）。增量事件 + // 不携带 → 不触碰（分类元素归 category.* 事件管）。bannerAssetKey/ + // emergencyDisabled 永不触碰（分别归 asset.updated / 紧急停用通道）。 + const categorySnapshot = Array.isArray(p.categories) + ? (p.categories as unknown as Prisma.InputJsonValue) + : undefined; await tx.albums_ov.upsert({ where: { albumId: event.entityId }, create: { @@ -197,7 +212,7 @@ export class SyncInboxService { downloadEnabled: p.downloadEnabled, originalDownloadEnabled: p.originalDownloadEnabled, translations: p.translations as unknown as Prisma.InputJsonValue, - categories: [], + categories: categorySnapshot ?? [], defaultLocale: p.defaultLocale, timezone: p.timezone, startAt: new Date(p.startAt), @@ -211,6 +226,9 @@ export class SyncInboxService { downloadEnabled: p.downloadEnabled, originalDownloadEnabled: p.originalDownloadEnabled, translations: p.translations as unknown as Prisma.InputJsonValue, + ...(categorySnapshot !== undefined + ? { categories: categorySnapshot } + : {}), defaultLocale: p.defaultLocale, timezone: p.timezone, startAt: new Date(p.startAt), diff --git a/services/api/src/sync/dto.ts b/services/api/src/sync/dto.ts index fa8625e..42bb988 100644 --- a/services/api/src/sync/dto.ts +++ b/services/api/src/sync/dto.ts @@ -1,5 +1,6 @@ -// services/api/src/sync/dto.ts —— 内部同步端点请求 DTO（P2 Task 3）。 -// 契约权威：openapi.yaml [internal]（syncBootstrap/syncReconcile）。 +// services/api/src/sync/dto.ts —— 内部同步端点请求 DTO（P2 Task 3 +// bootstrap/reconcile；Task 4 增 digest）。 +// 契约权威：openapi.yaml [internal]（syncBootstrap/syncReconcile/syncDigest）。 // 经全局 ValidationPipe（whitelist）剥离未知字段；校验失败 400 VALIDATION_FAILED。 import { IsIn, @@ -44,3 +45,10 @@ export class ReconcileRequestDto { @Max(1000) limit!: number; } + +/** 契约 DigestRequest（Task 4）：albumId 缺省 = 全部相册。 */ +export class DigestRequestDto { + @IsOptional() + @IsString() + albumId?: string; +} diff --git a/services/api/src/sync/payloads.ts b/services/api/src/sync/payloads.ts index af8338f..c4c98ee 100644 --- a/services/api/src/sync/payloads.ts +++ b/services/api/src/sync/payloads.ts @@ -50,10 +50,17 @@ type PhotoVariantRow = Prisma.photo_variantsGetPayload< Prisma.photo_variantsDefaultArgs >; -/** album.updated 载荷：albums_ov 展示所需标量 + 翻译全集（locale 升序）。 */ +/** + * album.updated 载荷：albums_ov 展示所需标量 + 翻译全集（locale 升序）。 + * categories 可选参数：仅 bootstrap/reconcile 快照事件传入（ov 整体替换 + * albums_ov.categories 的收敛原语，见 contracts events.ts 注释）；业务事务内 + * 增量事件不传（分类元素归 category.* 事件管）。输出按 (sortOrder, id) 升序 + * 确定性排序（与公开列表端点一致），元素翻译 locale 升序。 + */ export function buildAlbumUpdatedPayload( album: AlbumRow, translations: AlbumTranslationRow[], + categories?: AlbumCategoryInput[], ): AlbumUpdatedPayload { return { albumId: album.id, @@ -76,10 +83,35 @@ export function buildAlbumUpdatedPayload( venue: t.venue, address: t.address, })), + ...(categories === undefined + ? {} + : { + categories: [...categories] + .sort((a, b) => + a.category.sortOrder !== b.category.sortOrder + ? a.category.sortOrder - b.category.sortOrder + : a.category.id < b.category.id + ? -1 + : 1, + ) + .map(({ category, translations: categoryTranslations }) => ({ + id: category.id, + sortOrder: category.sortOrder, + translations: [...categoryTranslations] + .sort((a, b) => (a.locale < b.locale ? -1 : 1)) + .map((t) => ({ locale: t.locale, name: t.name })), + })), + }), entityVersion: album.entityVersion, }; } +/** 快照 categories 参数形状：分类行 + 该分类的翻译行（见函数注释）。 */ +export interface AlbumCategoryInput { + category: CategoryRow; + translations: CategoryTranslationRow[]; +} + /** category.updated 载荷：分类标量 + 翻译（locale 升序）。 */ export function buildCategoryUpdatedPayload( category: CategoryRow, diff --git a/services/api/src/sync/sync.controller.ts b/services/api/src/sync/sync.controller.ts index 102f2ec..8138ec7 100644 --- a/services/api/src/sync/sync.controller.ts +++ b/services/api/src/sync/sync.controller.ts @@ -1,11 +1,16 @@ -// services/api/src/sync/sync.controller.ts —— cn 侧内部同步端点（P2 Task 3）。 -// 契约权威：openapi.yaml [internal]（POST /internal/sync/bootstrap、 -// POST /internal/sync/reconcile）。鉴权：InternalSyncGuard（HMAC 原始请求体 +// services/api/src/sync/sync.controller.ts —— cn 侧内部同步端点（P2 Task 3： +// bootstrap/reconcile；Task 4 增 digest）。契约权威：openapi.yaml [internal] +// （POST /internal/sync/bootstrap、POST /internal/sync/reconcile、 +// POST /internal/sync/digest）。鉴权：InternalSyncGuard（HMAC 原始请求体 // 签名 + 时间戳 ±300s，密钥 SYNC_INTERNAL_KEY 仅 env）。仅 cn 入口装配 // （app-cn.module）；ov 入口的 content-batches 在 sync-inbox 模块（同守卫）。 import { Body, Controller, HttpCode, Post, UseGuards } from '@nestjs/common'; import { InternalSyncGuard } from '../shared/internal-sync.guard'; -import { BootstrapRequestDto, ReconcileRequestDto } from './dto'; +import { + BootstrapRequestDto, + DigestRequestDto, + ReconcileRequestDto, +} from './dto'; import { SyncService } from './sync.service'; @Controller('internal/sync') @@ -26,4 +31,11 @@ export class SyncController { reconcile(@Body() dto: ReconcileRequestDto) { return this.sync.reconcile(dto); } + + /** 对账摘要：每相册实体计数（Task 4 ov 副本计数比对 → 不一致触发 bootstrap）。 */ + @Post('digest') + @HttpCode(200) + digest(@Body() dto: DigestRequestDto) { + return this.sync.digest(dto.albumId); + } } diff --git a/services/api/src/sync/sync.service.ts b/services/api/src/sync/sync.service.ts index 13ce636..9008825 100644 --- a/services/api/src/sync/sync.service.ts +++ b/services/api/src/sync/sync.service.ts @@ -1,29 +1,37 @@ -// services/api/src/sync/sync.service.ts —— cn 侧内部同步服务（P2 Task 3）： -// bootstrap（全量快照引导）+ reconcile（缺失事件对账清单）。两端的共同设计 -// 决策（与计划文档 §2A「首次全量快照携带水位」对齐，登记为 Task 3 设计）： +// services/api/src/sync/sync.service.ts —— cn 侧内部同步服务（P2 Task 3： +// bootstrap/reconcile）+ Task 4 digest（每相册实体计数摘要）。共同设计决策 +// （与计划文档 §2A「首次全量快照携带水位」对齐，登记为 Task 3 设计）： // // · bootstrap = 「把当前状态快照事件**写回 outbox**」，而不是另建拉取通道—— // ov 冷启动时经本端点触发 cn 入队，随后由常规发送器（services/sync）批量 // HMAC 推送、ov 按既有 content-batches 幂等/版本逻辑应用。好处：快照与增量 -// 走同一条已被验证的管道（无第二条拉取路径），ov 侧无需新增消费代码； -// 重复 bootstrap 幂等（同实体同版本的新 eventId 在 ov 被 entityVersion -// 比较忽略）。eventCount=入队事件数，watermark=入队事件的最高 entityVersion -// （契约语义「后续增量对账的 sinceVersion 起点」的全局保守值；reconcile 的 -// sinceVersion 按 per-entity 版本过滤，见下）。 +// 走同一条已被验证的管道（无第二条拉取路径），ov 侧无需新增消费代码。 +// 重复 bootstrap 幂等：ov 对**同版本新 eventId** 事件做当前态重断言应用 +// （Task 4 起 sync-inbox 仅拒绝更旧版本；同一版本 = 同一状态，重放无副作用， +// 且正是 digest→bootstrap 补漏通道修复 ov 侧缺失副本行的机制）。 +// eventCount=入队事件数，watermark=入队事件的最高 entityVersion（契约语义 +// 「后续增量对账的 sinceVersion 起点」的全局保守值；reconcile 的 sinceVersion +// 按 per-entity 版本过滤，见下）。 // // · reconcile = 按 entityType 返回「当前状态」事件清单（entityVersion > // sinceVersion 的实体，升序、limit 截断）——数据源是 cn 权威表而非 outbox // 历史流水（快照即最终状态，ov 重放即补齐；outbox 中的中间版本无需重放， // 高版本载荷已包含最终状态）。eventId 为确定性合成 ID（recon::: // ）：与真实 outbox eventId（cuid）永不冲突，同一 (实体,版本) 的 -// 合成事件无论重放多少次在 ov 侧都幂等（eventId 唯一 + 版本比较双保险）。 -// 已知边界（登记）：硬删除实体（category 墓碑后行已删）不再出现在清单中， -// 删除语义依赖事件推送而非对账补漏。 +// 合成事件无论重放多少次在 ov 侧都幂等（eventId 唯一锚）。 +// 已知边界（Task 4 收敛）：硬删除实体（category 墓碑后行已删）不再出现在 +// 清单中——错过的墓碑经 digest 计数比对 → 该相册 bootstrap 重发 album.updated +// （携带分类全量清单，ov 整体替换 categories）收敛。 +// +// · digest（Task 4）= 每相册 {categoryCount, publishedPhotoCount, photoCount, +// assetCount} 权威计数。ov 与副本计数比对（照片取区间语义，见 digest 注释） +// → 不一致相册触发 bootstrap。 import { Injectable } from '@nestjs/common'; import type { components } from '@photo/contracts'; import type { Prisma } from '@photo/db-cn'; import { PrismaService } from '../prisma/prisma.service'; import { + type AlbumCategoryInput, buildAlbumUpdatedPayload, buildAssetUpdatedPayload, buildCategoryUpdatedPayload, @@ -32,6 +40,7 @@ import { export type BootstrapResponse = components['schemas']['BootstrapResponse']; export type ReconcileResponse = components['schemas']['ReconcileResponse']; +export type DigestResponse = components['schemas']['DigestResponse']; export type SyncEventWire = components['schemas']['SyncEvent']; /** 线格式 schema 版本（契约 SyncEvent.schemaVersion 当前为 1）。 */ @@ -128,6 +137,21 @@ export class SyncService { bucket.push(row); } } + // album.updated 快照载荷的分类全量清单（Task 4：ov 整体替换 categories + // 的收敛原语）——分类行与其翻译行按 albumId 配对。 + const categoryInputsByAlbum = new Map(); + for (const category of categories) { + const bucket = categoryInputsByAlbum.get(category.albumId); + const input: AlbumCategoryInput = { + category, + translations: translationsByCategory.get(category.id) ?? [], + }; + if (bucket === undefined) { + categoryInputsByAlbum.set(category.albumId, [input]); + } else { + bucket.push(input); + } + } const rows: Prisma.outbox_eventsCreateManyInput[] = []; let watermark = 0; @@ -159,6 +183,7 @@ export class SyncService { buildAlbumUpdatedPayload( album, translationsByAlbum.get(album.id) ?? [], + categoryInputsByAlbum.get(album.id) ?? [], ), ); } @@ -256,6 +281,10 @@ export class SyncService { bucket.push(row); } } + // album 快照事件携带分类全量清单（与 bootstrap 同形，见文件头说明）。 + const categoryInputs = await this.categoryInputsByAlbumIds( + albums.map((a) => a.id), + ); for (const album of albums) { events.push( this.toWireEvent( @@ -264,7 +293,11 @@ export class SyncService { album.id, album.entityVersion, false, - buildAlbumUpdatedPayload(album, byAlbum.get(album.id) ?? []), + buildAlbumUpdatedPayload( + album, + byAlbum.get(album.id) ?? [], + categoryInputs.get(album.id) ?? [], + ), ), ); } @@ -394,6 +427,125 @@ export class SyncService { return { events }; } + /** + * 对账摘要（Task 4）：cn 侧每相册的实体计数（当前权威值）。ov 侧与自己的 + * 副本计数比对，不一致 → 触发该相册 bootstrap 重建（错过的墓碑/漏掉的 + * 事件由此收敛，见 sync-inbox/Reconciler 设计说明）。photoCount 拆两档： + * publishedPhotoCount = ov **必须**持有的行数（可见照片）；photoCount = + * published+hidden+deleted 总数 = ov 行数上限——hidden/deleted 照片若从未 + * 在 cn 侧以 published 态同步过，ov 侧合法地无行（缺席即撤下语义），故 ov + * 计数落在 [publishedPhotoCount, photoCount] 区间内即视为一致。assetCount + * 供观测（ov 无 asset 副本行，不参与比对）。 + */ + async digest(albumId?: string): Promise { + const albums = await this.prisma.albums.findMany({ + where: albumId === undefined ? {} : { id: albumId }, + select: { id: true }, + orderBy: { id: 'asc' }, + }); + if (albums.length === 0) { + return { albums: [] }; + } + const albumIds = albums.map((a) => a.id); + const [categoryCounts, photoCounts, assetCounts] = await Promise.all([ + this.prisma.categories.groupBy({ + by: ['albumId'], + where: { albumId: { in: albumIds } }, + _count: { _all: true }, + }), + this.prisma.photos.groupBy({ + by: ['albumId', 'visibility'], + where: { + albumId: { in: albumIds }, + visibility: { in: ['published', 'hidden', 'deleted'] }, + }, + _count: { _all: true }, + }), + this.prisma.album_assets.groupBy({ + by: ['albumId'], + where: { albumId: { in: albumIds }, state: 'effective' }, + _count: { _all: true }, + }), + ]); + const categoriesByAlbum = new Map( + categoryCounts.map((row) => [row.albumId, row._count._all]), + ); + const assetsByAlbum = new Map( + assetCounts.map((row) => [row.albumId, row._count._all]), + ); + const photosByAlbum = new Map< + string, + { published: number; total: number } + >(); + for (const row of photoCounts) { + const bucket = photosByAlbum.get(row.albumId) ?? { + published: 0, + total: 0, + }; + if (row.visibility === 'published') { + bucket.published = row._count._all; + } + bucket.total += row._count._all; + photosByAlbum.set(row.albumId, bucket); + } + return { + albums: albums.map(({ id }) => { + const photos = photosByAlbum.get(id); + return { + albumId: id, + categoryCount: categoriesByAlbum.get(id) ?? 0, + publishedPhotoCount: photos?.published ?? 0, + photoCount: photos?.total ?? 0, + assetCount: assetsByAlbum.get(id) ?? 0, + }; + }), + }; + } + + /** 相册 id 集合 → 分类快照输入（分类行 + 翻译行配对；升序确定性由构建器排）。 */ + private async categoryInputsByAlbumIds( + albumIds: string[], + ): Promise> { + const result = new Map(); + if (albumIds.length === 0) { + return result; + } + const categories = await this.prisma.categories.findMany({ + where: { albumId: { in: albumIds } }, + }); + const translations = + categories.length === 0 + ? [] + : await this.prisma.category_translations.findMany({ + where: { categoryId: { in: categories.map((c) => c.id) } }, + }); + const translationsByCategory = new Map< + string, + typeof translations + >(); + for (const row of translations) { + const bucket = translationsByCategory.get(row.categoryId); + if (bucket === undefined) { + translationsByCategory.set(row.categoryId, [row]); + } else { + bucket.push(row); + } + } + for (const category of categories) { + const bucket = result.get(category.albumId); + const input: AlbumCategoryInput = { + category, + translations: translationsByCategory.get(category.id) ?? [], + }; + if (bucket === undefined) { + result.set(category.albumId, [input]); + } else { + bucket.push(input); + } + } + return result; + } + /** 当前状态 → 线格式事件（确定性合成 eventId，见文件头设计说明）。 */ private toWireEvent( entityType: SyncEventWire['entityType'], diff --git a/services/sync/package.json b/services/sync/package.json index de1b91d..8019fe1 100644 --- a/services/sync/package.json +++ b/services/sync/package.json @@ -2,7 +2,7 @@ "name": "@photo/sync", "version": "0.1.0", "private": true, - "description": "CN→OV outbox sender (轮询 outbox_events 推送 content-batches 到 ov 入口；P2 Task 3)", + "description": "CN→OV sync process (outbox sender + ov reconciliation scheduler; P2 Task 3/4)", "scripts": { "build": "tsc -p tsconfig.build.json", "start": "tsc -p tsconfig.build.json && node dist/main.js", @@ -14,13 +14,13 @@ "dependencies": { "@photo/contracts": "workspace:*", "@photo/db-cn": "workspace:*", + "@photo/db-ov": "workspace:*", "pino": "^10.0.0" }, "devDependencies": { "@nestjs/common": "^10.4.6", "@nestjs/core": "^10.4.6", "@nestjs/platform-express": "^10.4.6", - "@photo/db-ov": "workspace:*", "@photo/storage": "workspace:^", "@types/jest": "^29.5.13", "@types/node": "^22.10.0", diff --git a/services/sync/src/main.ts b/services/sync/src/main.ts index 7f659be..6b0181e 100644 --- a/services/sync/src/main.ts +++ b/services/sync/src/main.ts @@ -1,18 +1,28 @@ -// services/sync/src/main.ts —— sync 发送器进程入口（P2 Task 3）。构建产物 -// dist/main.js 由根 dev:sync / 生产启动命令运行；env 一律来自进程环境（根 -// dotenv-cli 注入），此处不读文件。循环：每 SYNC_PUSH_INTERVAL_MS 轮询一次 -// outbox（SYNC_BATCH_SIZE 一批，high 优先——见 sender.ts 排序说明；满批立即 -// 追下一轮，积压排空不被轮询间隔拖慢）。失败退避：连续失败按间隔 ×2^n 指数 -// 递增、60s 封顶，成功即归零。优雅退出：SIGINT/SIGTERM 停轮询（等待在飞 -// 一轮，fetch 超时已封顶 30s）后断开 Prisma。 +// services/sync/src/main.ts —— sync 进程入口（P2 Task 3 发送器 + Task 4 对账 +// 调度）。构建产物 dist/main.js 由根 dev:sync / 生产启动命令运行；env 一律来自 +// 进程环境（根 dotenv-cli 注入），此处不读文件。 +// +// 两个独立循环： +// · 发送循环：每 SYNC_PUSH_INTERVAL_MS 轮询 outbox（SYNC_BATCH_SIZE 一批， +// high 优先——见 sender.ts 排序说明；满批立即追下一轮，积压排空不被轮询 +// 间隔拖慢）。失败退避：连续失败按间隔 ×2^n 指数递增、60s 封顶，成功归零。 +// · 对账循环（Task 4）：每 SYNC_RECONCILE_INTERVAL_MS 一轮 SyncReconciler. +// runOnce（当前态清扫重放 + digest 比对触发 bootstrap，见 reconciler.ts）。 +// 失败仅记日志、下轮再试（分钟级节奏，无需退避——幂等安全）。 +// +// 双库连接：cn 库（outbox 取数）+ ov 库（副本计数直读）。优雅退出： +// SIGINT/SIGTERM 停两循环（等待在飞一轮，fetch 超时已封顶 30s）后断开两连接。 import { pino } from 'pino'; import { PrismaClient } from '@photo/db-cn'; +import { PrismaClient as OvPrismaClient } from '@photo/db-ov'; import { SYNC_BATCH_SIZE, SYNC_INTERNAL_KEY_ENV, SYNC_PUSH_INTERVAL_MS, + SYNC_RECONCILE_INTERVAL_MS, } from '@photo/contracts'; import { OutboxSender } from './sender'; +import { SyncReconciler } from './reconciler'; /** 失败退避封顶（对齐 SYNC_RECONCILE_INTERVAL_MS 量级：分钟级封顶足够）。 */ const MAX_BACKOFF_MS = 60_000; @@ -37,13 +47,35 @@ function readIntEnv( return parsed; } -const sleep = (ms: number): Promise => - new Promise((resolve) => setTimeout(resolve, ms)); +/** + * 可被停机唤醒的 sleep：对账循环节拍是分钟级，若让 SIGTERM 干等一个完整 + * interval 才退出不可接受——shutdown 置 stopping 后调用 wake 提前放行， + * 循环条件 !stopping 随即退出（在飞一轮的各步均有界，见 shutdown 注释）。 + */ +let wakeSleeper: (() => void) | null = null; +const sleepStoppable = (ms: number): Promise => + new Promise((resolve) => { + const timer = setTimeout(() => { + wakeSleeper = null; + resolve(); + }, ms); + wakeSleeper = () => { + clearTimeout(timer); + wakeSleeper = null; + resolve(); + }; + }); async function main(): Promise { const logger = pino({ level: process.env.LOG_LEVEL ?? 'info' }); const intervalMs = readIntEnv(logger, 'SYNC_PUSH_INTERVAL_MS', SYNC_PUSH_INTERVAL_MS); const batchSize = readIntEnv(logger, 'SYNC_BATCH_SIZE', SYNC_BATCH_SIZE); + const reconcileIntervalMs = readIntEnv( + logger, + 'SYNC_RECONCILE_INTERVAL_MS', + SYNC_RECONCILE_INTERVAL_MS, + ); + const cnApiUrl = process.env.CN_API_URL ?? 'http://127.0.0.1:8080'; const ovApiUrl = process.env.OV_API_URL ?? 'http://127.0.0.1:8081'; const internalKey = process.env[SYNC_INTERNAL_KEY_ENV] ?? ''; if (internalKey === '') { @@ -52,23 +84,31 @@ async function main(): Promise { process.exit(1); } - const prisma = new PrismaClient(); + const prismaCn = new PrismaClient(); + const prismaOv = new OvPrismaClient(); const sender = new OutboxSender({ - prisma, + prisma: prismaCn, ovApiUrl, internalKey, batchSize, timeoutMs: REQUEST_TIMEOUT_MS, logger, }); + const reconciler = new SyncReconciler({ + prismaOv, + cnApiUrl, + ovApiUrl, + internalKey, + timeoutMs: REQUEST_TIMEOUT_MS, + logger, + }); logger.info( - { ovApiUrl, intervalMs, batchSize }, - 'sync sender started', + { cnApiUrl, ovApiUrl, intervalMs, batchSize, reconcileIntervalMs }, + 'sync sender + reconciler started', ); let stopping = false; - let inFlight: Promise | null = null; - const loop = async (): Promise => { + const senderLoop = async (): Promise => { let failureStreak = 0; while (!stopping) { try { @@ -94,23 +134,38 @@ async function main(): Promise { }, 'outbox push failed（整批未标记，下轮重推）', ); - await sleep(backoffMs); + await sleepStoppable(backoffMs); continue; } - await sleep(intervalMs); + await sleepStoppable(intervalMs); } }; - inFlight = loop(); + const reconcileLoop = async (): Promise => { + while (!stopping) { + try { + await reconciler.runOnce(); + } catch (err) { + logger.error( + { err: err instanceof Error ? err.message : String(err) }, + 'reconcile cycle failed（幂等安全，下轮再试）', + ); + } + await sleepStoppable(reconcileIntervalMs); + } + }; + const inFlight: Promise[] = [senderLoop(), reconcileLoop()]; const shutdown = (signal: string): void => { if (stopping) return; stopping = true; + wakeSleeper?.(); logger.info({ signal }, 'sync sender shutting down'); void (async () => { try { // 等在飞一轮收尾（runOnce 各步均有界：HTTP 30s 超时 / DB 查询即回）。 - await inFlight; - await prisma.$disconnect(); + await Promise.all(inFlight); + await prismaCn.$disconnect(); + await prismaOv.$disconnect(); } catch (err) { logger.error( { err: err instanceof Error ? err.message : String(err) }, diff --git a/services/sync/src/reconciler.ts b/services/sync/src/reconciler.ts new file mode 100644 index 0000000..ed64bdd --- /dev/null +++ b/services/sync/src/reconciler.ts @@ -0,0 +1,293 @@ +// services/sync/src/reconciler.ts —— ov 补漏对账调度器（P2 Task 4）。 +// runOnce() 两阶段，每 SYNC_RECONCILE_INTERVAL_MS 一轮（main 循环驱动）： +// +// 阶段一「当前态清扫」：对 album/category/photo/asset 四类逐类分页拉取 cn +// reconcile 清单（**当前状态事件**，非版本清单——Task 4 设计决策：重放走 +// 既有 content-batches 管道即幂等，无需第二条拉取通道），直接签名推送到 ov +// content-batches。sinceVersion 取 0（全量当前态）而非「per-type 已应用最大 +// 版本」——cn reconcile 以**类型级全局版本阈值**过滤，类型内单实体滞后 +// （如 A 实体已 v9、B 实体 v3→v4）会被 max 掩蔽；0 则由 ov 的 eventId 幂等 +// + 版本单调把「已追平的实体」全部判重/判旧（确定性 recon::: +// 合成 eventId，重扫即纯重复检查），任何滞后/新实体必然被补上。仿真规模 +// （单册数百照片）下每轮开销可忽略；真实地域部署若需缩量，改点在 cn 侧 +// 提供 per-entity 水位比对（登记为 P5 演进项，不在 P2）。 +// +// 阶段二「digest 比对 → bootstrap 重建」：拉取 cn digest（每相册权威计数： +// categoryCount / publishedPhotoCount..photoCount 区间 / assetCount）与 ov +// 副本计数比对。不一致的相册 → 触发 cn bootstrap {albumId}（当前状态快照 +// 写回 outbox，由常驻发送器投递）。收敛机制依赖两条 Task 4 语义： +// · sync-inbox 对**同版本新 eventId** 事件做当前态重断言应用（cn 的 +// entityVersion 与状态一一对应）→ 修复 ov 缺失的 photos_ov 行； +// · bootstrap/reconcile 的 album.updated 载荷携带分类全量清单，ov 整体 +// 替换 albums_ov.categories → 修复**错过的 category 墓碑**（cn 行已删， +// 任何「重放现存实体」通道都无法表达的删除语义，Task 3 已登记边界）。 +// 照片计数取区间口径：hidden/deleted 照片若从未以 published 态同步过，ov +// 合法地无行（缺席即撤下），故 ov 行数 ∈ [publishedPhotoCount, photoCount] +// 即一致。ov 有而 cn 无的相册（cn 行已删，不可经 bootstrap 重建）仅告警 +// （P4a 管理页消费 sync_state/sync_inbox 的延迟与积压口径）。 +import type { PrismaClient } from '@photo/db-ov'; +import type { components } from '@photo/contracts'; +import { + SYNC_SIGNATURE_HEADER, + SYNC_TIMESTAMP_HEADER, + signSyncBody, +} from '@photo/contracts'; +import type { Logger } from 'pino'; + +type SyncEventWire = components['schemas']['SyncEvent']; +type DigestResponse = components['schemas']['DigestResponse']; + +/** 线格式端点路径（全局前缀 v1 + InternalSyncGuard 域；与 sender 同约定）。 */ +const RECONCILE_PATH = '/v1/internal/sync/reconcile'; +const DIGEST_PATH = '/v1/internal/sync/digest'; +const BOOTSTRAP_PATH = '/v1/internal/sync/bootstrap'; +const CONTENT_BATCHES_PATH = '/v1/internal/sync/content-batches'; + +/** 清扫覆盖的实体类型（slot-state 预留，无当前态可拉）。 */ +const SWEEP_ENTITY_TYPES = ['album', 'category', 'photo', 'asset'] as const; + +const DEFAULT_RECONCILE_LIMIT = 500; // 任务口径单页上限（契约上限 1000） +const DEFAULT_PUSH_BATCH_SIZE = 100; // content-batches 契约单批上限 +const DEFAULT_TIMEOUT_MS = 30_000; +/** 分页保险丝：单类型最多拉 100 页（5 万实体）——防响应形状异常导致的死循环。 */ +const MAX_PAGES_PER_TYPE = 100; + +export interface SyncReconcilerOptions { + /** ov 库客户端（副本计数比对直读；cn 侧数据一律经内部 HTTP，不直连 cn 库）。 */ + prismaOv: PrismaClient; + /** cn 入口基址（如 http://127.0.0.1:8080，不含路径）。 */ + cnApiUrl: string; + /** ov 入口基址（重放事件推送目标）。 */ + ovApiUrl: string; + /** HMAC 密钥（与两端 InternalSyncGuard 同一 SYNC_INTERNAL_KEY 值）。 */ + internalKey: string; + /** reconcile 单页上限（默认 500）。 */ + reconcileLimit?: number; + /** content-batches 单批上限（默认契约 100）。 */ + pushBatchSize?: number; + /** 单次 HTTP 超时毫秒（默认 30s）。 */ + timeoutMs?: number; + logger?: Logger; +} + +/** 单轮 runOnce 的观测口径（main 循环打点；测试用于断言追赶/补漏/稳态）。 */ +export interface ReconcileStats { + /** 阶段一从 cn 拉到的当前态事件数（四类型合计）。 */ + sweptEvents: number; + /** 阶段一重放中 ov 实际应用（新版本/同版本重断言）的条数。 */ + replayApplied: number; + /** 阶段二 cn digest 相册数。 */ + digestAlbums: number; + /** 计数不一致 → 已触发 bootstrap 的相册（ healed by sender 投递）。 */ + mismatchedAlbums: string[]; + /** ov 存在而 cn 无的相册（cn 行已删——不可 bootstrap 重建，仅告警登记）。 */ + ovOrphanAlbums: string[]; +} + +/** ov 侧每相册副本计数（albums_ov.categories 为 JSON，行数小，直接取回计算）。 */ +interface OvAlbumCounts { + categoryCount: number; + photoCount: number; +} + +export class SyncReconciler { + private readonly reconcileLimit: number; + private readonly pushBatchSize: number; + private readonly timeoutMs: number; + private readonly cnBase: string; + private readonly ovBase: string; + + constructor(private readonly options: SyncReconcilerOptions) { + this.reconcileLimit = options.reconcileLimit ?? DEFAULT_RECONCILE_LIMIT; + this.pushBatchSize = options.pushBatchSize ?? DEFAULT_PUSH_BATCH_SIZE; + this.timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS; + this.cnBase = options.cnApiUrl.replace(/\/+$/, ''); + this.ovBase = options.ovApiUrl.replace(/\/+$/, ''); + } + + /** 单轮对账：阶段一清扫重放 + 阶段二 digest 比对触发 bootstrap。 */ + async runOnce(): Promise { + const { sweptEvents, replayApplied } = await this.sweep(); + const digest = await this.digestCompare(); + const stats: ReconcileStats = { + sweptEvents, + replayApplied, + digestAlbums: digest.digestAlbums, + mismatchedAlbums: digest.mismatchedAlbums, + ovOrphanAlbums: digest.ovOrphanAlbums, + }; + this.options.logger?.info( + { + sweptEvents, + replayApplied, + digestAlbums: stats.digestAlbums, + mismatched: stats.mismatchedAlbums.length, + ovOrphans: stats.ovOrphanAlbums.length, + }, + 'reconcile cycle done', + ); + return stats; + } + + /** + * 阶段一：逐类型分页拉取 cn 当前态事件并签名推送到 ov content-batches。 + * 任一类型失败抛错（整轮中止，下轮重来——幂等安全）。 + */ + private async sweep(): Promise<{ + sweptEvents: number; + replayApplied: number; + }> { + let sweptEvents = 0; + let replayApplied = 0; + for (const entityType of SWEEP_ENTITY_TYPES) { + for (let page = 0; page < MAX_PAGES_PER_TYPE; page += 1) { + const response = await this.signedPost<{ + events: SyncEventWire[]; + }>(this.cnBase, RECONCILE_PATH, { + entityType, + sinceVersion: 0, + limit: this.reconcileLimit, + }); + const events = response.events; + sweptEvents += events.length; + for ( + let offset = 0; + offset < events.length; + offset += this.pushBatchSize + ) { + const chunk = events.slice(offset, offset + this.pushBatchSize); + const batch = await this.signedPost<{ + acceptedEventIds: string[]; + applied: number; + }>(this.ovBase, CONTENT_BATCHES_PATH, { batch: chunk }); + replayApplied += batch.applied; + } + if (events.length < this.reconcileLimit) { + break; + } + } + } + return { sweptEvents, replayApplied }; + } + + /** + * 阶段二：cn digest 与 ov 副本计数比对；不一致相册触发 cn bootstrap + * （快照写回 outbox，由常驻发送器投递——本调度器不直接投递 bootstrap 事件， + * 与发送器职责分离）。返回比对口径供打点/测试断言。 + */ + private async digestCompare(): Promise<{ + digestAlbums: number; + mismatchedAlbums: string[]; + ovOrphanAlbums: string[]; + }> { + const digest = await this.signedPost( + this.cnBase, + DIGEST_PATH, + {}, + ); + const cnAlbums = new Map(digest.albums.map((row) => [row.albumId, row])); + const ovCounts = await this.ovAlbumCounts(); + const mismatchedAlbums: string[] = []; + for (const cnRow of digest.albums) { + const ov = ovCounts.get(cnRow.albumId); + if ( + ov === undefined || + ov.categoryCount !== cnRow.categoryCount || + ov.photoCount < cnRow.publishedPhotoCount || + ov.photoCount > cnRow.photoCount + ) { + mismatchedAlbums.push(cnRow.albumId); + } + } + const ovOrphanAlbums = [...ovCounts.keys()] + .filter((albumId) => !cnAlbums.has(albumId)) + .sort(); + if (ovOrphanAlbums.length > 0) { + // cn 行已删（或 ov 指向了错误源）：bootstrap 无法表达「删除相册」， + // 不自动处置——告警登记，人工/P4a 管理通道跟进。 + this.options.logger?.error( + { ovOrphanAlbums }, + 'digest: ov 存在 cn 已不存在的相册副本（不可自动收敛，登记跟进）', + ); + } + for (const albumId of mismatchedAlbums) { + await this.signedPost(this.cnBase, BOOTSTRAP_PATH, { albumId }); + } + if (mismatchedAlbums.length > 0) { + this.options.logger?.warn( + { mismatchedAlbums }, + 'digest: 副本计数不一致，已触发受影响相册 bootstrap 重建', + ); + } + return { + digestAlbums: digest.albums.length, + mismatchedAlbums, + ovOrphanAlbums, + }; + } + + /** ov 每相册副本计数：albums_ov 行（categories JSON 长度）+ photos_ov 分组数。 */ + private async ovAlbumCounts(): Promise> { + const [albumRows, photoRows] = await Promise.all([ + this.options.prismaOv.albums_ov.findMany({ + select: { albumId: true, categories: true }, + orderBy: { albumId: 'asc' }, + }), + this.options.prismaOv.photos_ov.groupBy({ + by: ['albumId'], + _count: { _all: true }, + }), + ]); + const photosByAlbum = new Map( + photoRows.map((row) => [row.albumId, row._count._all]), + ); + const counts = new Map(); + for (const row of albumRows) { + counts.set(row.albumId, { + categoryCount: Array.isArray(row.categories) + ? row.categories.length + : 0, + photoCount: photosByAlbum.get(row.albumId) ?? 0, + }); + } + return counts; + } + + /** + * 签名 POST：同一 JSON.stringify 产物既作签名输入也作请求体（rawBody 验签 + * 字节一致，与 sender.push 同约定）；非 2xx 抛错（调用方整轮重试语义）。 + */ + private async signedPost( + base: string, + path: string, + body: unknown, + ): Promise { + const rawBody = JSON.stringify(body); + const timestamp = Math.floor(Date.now() / 1000).toString(); + let response: Response; + const endpoint = `${base}${path}`; + try { + response = await fetch(endpoint, { + method: 'POST', + headers: { + 'content-type': 'application/json', + [SYNC_TIMESTAMP_HEADER]: timestamp, + [SYNC_SIGNATURE_HEADER]: signSyncBody(this.options.internalKey, rawBody), + }, + body: rawBody, + signal: AbortSignal.timeout(this.timeoutMs), + }); + } catch (err) { + throw new Error( + `${path} 请求失败（${endpoint}）：${err instanceof Error ? err.message : String(err)}`, + ); + } + if (!response.ok) { + const text = await response.text().catch(() => ''); + throw new Error( + `${path} 非 2xx（${endpoint}）：status=${response.status} body=${text.slice(0, 500)}`, + ); + } + return (await response.json()) as T; + } +} diff --git a/services/sync/test/reconcile.e2e-spec.ts b/services/sync/test/reconcile.e2e-spec.ts new file mode 100644 index 0000000..e5bb6c3 --- /dev/null +++ b/services/sync/test/reconcile.e2e-spec.ts @@ -0,0 +1,343 @@ +// services/sync/test/reconcile.e2e-spec.ts —— P2 Task 4 对账调度 E2E。 +// 复用 sync.e2e-spec.ts 的进程内双入口 harness（cn/ov 真实 HTTP + 真实双 +// MySQL + HMAC），被测对象换成生产 SyncReconciler 实例（runOnce 直驱）。 +// +// 覆盖（任务验收场景）： +// ① 漏事件追赶：ov 停在 v1、cn 已到 v5 且 outbox 事件「投后即失」→ +// runOnce 当前态清扫 → ov 副本追到 v5； +// ② digest 修复缺失副本行：直接删 photos_ov 行（不 bump 任何版本）→ +// digest 计数不一致 → 自动触发该相册 bootstrap → 发送器投递 → 行恢复； +// ③ digest 修复错过的分类墓碑（Task 3 登记边界）：cn 删分类且墓碑 outbox +// 行被移除（对账无法为已删行重放）→ digest categoryCount 不一致 → +// bootstrap 的 album.updated 携带分类全量清单 → ov categories 整体替换 +// 收敛为空； +// ④ 稳态空转：一切一致时 runOnce 无 bootstrap 触发、重放全部判重/判旧 +// （applied=0）、outbox 无新增未投递。 +import { NestFactory } from '@nestjs/core'; +import { type NestExpressApplication } from '@nestjs/platform-express'; +import request from 'supertest'; +import sharp from 'sharp'; +import { PrismaClient } from '@photo/db-cn'; +import { PrismaClient as OvPrismaClient } from '@photo/db-ov'; +import { + createStorageFromEnv, + type MinioStorageAdapter, +} from '@photo/storage'; +import { AppCnModule } from '../../api/src/app-cn.module'; +import { AppOvModule } from '../../api/src/app-ov.module'; +import { seedAdmin } from '../../api/src/scripts/seed-admin'; +import { configureApp } from '../../api/src/shared/configure-app'; +import { PublishService } from '../../worker/src/publish.service'; +import { resetDb } from '../../api/test/reset'; +import { resetOvDb } from '../../api/test/reset-ov'; +import { OutboxSender } from '../src/sender'; +import { SyncReconciler } from '../src/reconciler'; + +jest.setTimeout(180_000); + +const ADMIN_ACCOUNT = process.env.ADMIN_ACCOUNT ?? 'admin'; +const ADMIN_PASSWORD = process.env.ADMIN_PASSWORD ?? 'admin-secret-change-me'; +const SYNC_KEY = process.env.SYNC_INTERNAL_KEY ?? 'dev-sync-secret-change-me'; + +/** 测试内生成真实 JPEG（同 sync.e2e-spec；无二进制入库）。 */ +async function makeJpeg( + width: number, + height: number, + rgb: { r: number; g: number; b: number }, +): Promise { + return sharp({ create: { width, height, channels: 3, background: rgb } }) + .jpeg({ quality: 85 }) + .toBuffer(); +} + +describe('sync reconcile + digest healing (e2e)', () => { + let cnApp: NestExpressApplication; + let ovApp: NestExpressApplication; + let cnBase: string; + let ovBase: string; + let prismaCn: PrismaClient; + let prismaOv: OvPrismaClient; + let storage: MinioStorageAdapter; + let sender: OutboxSender; + let reconciler: SyncReconciler; + let adminToken: string; + + let albumId: string; + let categoryId: string; + let photo1Id: string; + let photo2Id: string; + + /** 排空发送器（对账触发的 bootstrap 事件由它投递，职责分离）。 */ + async function drainSender(): Promise { + for (let round = 0; round < 20; round++) { + const stats = await sender.runOnce(); + if (stats.picked === 0) { + return; + } + } + throw new Error('drainSender 超时：20 轮后 outbox 仍有未投递事件'); + } + + /** 上传全链路：init → presign PUT 实写 MinIO → complete → 直驱发布事务。 */ + async function uploadAndPublish( + clientTaskId: string, + jpeg: Buffer, + ): Promise { + const init = await request(cnBase) + .post('/v1/uploads') + .set('Authorization', `Bearer ${adminToken}`) + .send({ + albumId, + clientTaskId, + filename: 'reconcile.jpg', + bytes: jpeg.length, + contentType: 'image/jpeg', + categoryId, + }); + expect(init.status).toBe(201); + const { uploadId, photoId, credential } = init.body as { + uploadId: string; + photoId: string; + credential: { url: string; method: string; headers: Record }; + }; + const put = await fetch(credential.url, { + method: credential.method, + headers: credential.headers, + body: new Uint8Array(jpeg), + }); + expect(put.ok).toBe(true); + const done = await request(cnBase) + .post(`/v1/uploads/${uploadId}/complete`) + .set('Authorization', `Bearer ${adminToken}`); + expect(done.status).toBe(200); + expect(await new PublishService(prismaCn, storage).processPhoto(photoId)).toBe( + 'published', + ); + return photoId; + } + + /** 相册开关翻转一次（publicEnabled 当前值取反；版本 +1、产生 album.updated）。 */ + async function togglePublicEnabled(): Promise { + const current = await prismaCn.albums.findUniqueOrThrow({ + where: { id: albumId }, + select: { publicEnabled: true }, + }); + const res = await request(cnBase) + .patch(`/v1/albums/${albumId}`) + .set('Authorization', `Bearer ${adminToken}`) + .send({ publicEnabled: !current.publicEnabled }); + expect(res.status).toBe(200); + } + + beforeAll(async () => { + expect(SYNC_KEY).not.toBe(''); + await resetDb(); + await resetOvDb(); + + cnApp = await NestFactory.create(AppCnModule, { + rawBody: true, + }); + configureApp(cnApp); + await cnApp.listen(0); + cnBase = await cnApp.getUrl(); + + ovApp = await NestFactory.create(AppOvModule, { + rawBody: true, + }); + configureApp(ovApp); + await ovApp.listen(0); + ovBase = await ovApp.getUrl(); + + prismaCn = new PrismaClient(); + prismaOv = new OvPrismaClient(); + storage = createStorageFromEnv(); + sender = new OutboxSender({ + prisma: prismaCn, + ovApiUrl: ovBase, + internalKey: SYNC_KEY, + }); + reconciler = new SyncReconciler({ + prismaOv, + cnApiUrl: cnBase, + ovApiUrl: ovBase, + internalKey: SYNC_KEY, + }); + + await seedAdmin(prismaCn, ADMIN_ACCOUNT, ADMIN_PASSWORD); + const login = await request(cnBase).post('/v1/auth/password').send({ + account: ADMIN_ACCOUNT, + password: ADMIN_PASSWORD, + }); + expect(login.status).toBe(200); + adminToken = login.body.sessionToken as string; + + // 基线：相册（v1）+ 分类（v1）+ 两张发布照片（各 v2）→ ov 全量同步。 + const album = await request(cnBase) + .post('/v1/albums') + .set('Authorization', `Bearer ${adminToken}`) + .send({ + translations: [ + { locale: 'zh-CN', title: '对账相册' }, + { locale: 'en', title: 'Reconcile Gala' }, + ], + defaultLocale: 'zh-CN', + startAt: '2026-10-01T10:00:00.000Z', + }); + expect(album.status).toBe(201); + albumId = album.body.id as string; + + const category = await request(cnBase) + .post(`/v1/albums/${albumId}/categories`) + .set('Authorization', `Bearer ${adminToken}`) + .send({ + sortOrder: 0, + translations: [ + { locale: 'zh-CN', name: '主舞台' }, + { locale: 'en', name: 'Main Stage' }, + ], + }); + expect(category.status).toBe(201); + categoryId = category.body.id as string; + + photo1Id = await uploadAndPublish( + 'recon-a', + await makeJpeg(900, 600, { r: 170, g: 120, b: 70 }), + ); + photo2Id = await uploadAndPublish( + 'recon-b', + await makeJpeg(800, 600, { r: 70, g: 120, b: 170 }), + ); + await drainSender(); + + expect( + await prismaOv.albums_ov.findUniqueOrThrow({ where: { albumId } }), + ).toMatchObject({ entityVersion: 1, publicEnabled: true }); + expect(await prismaOv.photos_ov.count({ where: { albumId } })).toBe(2); + }); + + afterAll(async () => { + await cnApp.close(); + await ovApp.close(); + await prismaCn.$disconnect(); + await prismaOv.$disconnect(); + }); + + it('① 漏事件追赶：cn 到 v5、事件「投后即失」→ runOnce 清扫 → ov 追到 v5', async () => { + // cn 连续 4 次开关翻转：v1 → v5。 + for (let i = 0; i < 4; i += 1) { + await togglePublicEnabled(); + } + expect( + ( + await prismaCn.albums.findUniqueOrThrow({ + where: { id: albumId }, + select: { entityVersion: true, publicEnabled: true }, + }) + ), + ).toMatchObject({ entityVersion: 5, publicEnabled: true }); + + // 模拟「已确认投递但 ov 从未收到」（发送器停机 + outbox 清扫后的漏账）： + // 未投递事件全部伪造 deliveredAt —— 发送器永不再推，只能靠对账补。 + await prismaCn.outbox_events.updateMany({ + where: { deliveredAt: null }, + data: { deliveredAt: new Date() }, + }); + + const stats = await reconciler.runOnce(); + // 四类型当前态：album v5 + category v1 + photo×2 v2。 + expect(stats.sweptEvents).toBe(4); + // album v5 > ov 已应用 v1 → 应用；其余同版本重断言（Task 4 语义）。 + expect(stats.replayApplied).toBe(4); + expect(stats.mismatchedAlbums).toEqual([]); + + expect( + await prismaOv.albums_ov.findUniqueOrThrow({ where: { albumId } }), + ).toMatchObject({ entityVersion: 5, publicEnabled: true }); + }); + + it('② digest 修复缺失副本行：删 photos_ov 行 → bootstrap 自动触发 → 行恢复', async () => { + // ov 侧漂移：照片行直接丢失（不 bump 任何版本——清扫的确定性 eventId + // 已判重，唯一能发现的是 digest 计数比对）。 + await prismaOv.photos_ov.delete({ where: { photoId: photo1Id } }); + expect(await prismaOv.photos_ov.count({ where: { albumId } })).toBe(1); + + const stats = await reconciler.runOnce(); + expect(stats.mismatchedAlbums).toEqual([albumId]); + // bootstrap 已把该相册当前状态快照写回 outbox（发送器负责投递）。 + expect( + await prismaCn.outbox_events.count({ where: { deliveredAt: null } }), + ).toBeGreaterThan(0); + + await drainSender(); + + // 同版本重断言语义（sync-inbox ②，Task 4）：bootstrap 新 eventId + 同 + // entityVersion → 重应用 → 缺失行按当前态重建（photo2 未受牵连）。 + const restored = await prismaOv.photos_ov.findUniqueOrThrow({ + where: { photoId: photo1Id }, + }); + expect(restored).toMatchObject({ + albumId, + visibility: 'published', + entityVersion: 2, + }); + expect( + await prismaOv.photos_ov.findUniqueOrThrow({ + where: { photoId: photo2Id }, + }), + ).toMatchObject({ visibility: 'published', entityVersion: 2 }); + + // 收敛：再跑一轮 digest 无不一致、无新 bootstrap。 + const steady = await reconciler.runOnce(); + expect(steady.mismatchedAlbums).toEqual([]); + expect( + await prismaCn.outbox_events.count({ where: { deliveredAt: null } }), + ).toBe(0); + }); + + it('③ digest 修复错过的分类墓碑：cn 行已删且墓碑事件丢失 → 分类全量清单整体替换收敛', async () => { + // cn 删除分类（tombstone outbox 事件产生），随后模拟墓碑「投后即失」： + // 直接删除 outbox 行——reconcile 清扫无法为已不存在的行重放任何事件 + // （Task 3 登记边界），ov categories 残留陈旧元素。 + const remove = await request(cnBase) + .delete(`/v1/categories/${categoryId}`) + .set('Authorization', `Bearer ${adminToken}`); + expect(remove.status).toBe(204); + await prismaCn.outbox_events.deleteMany({ + where: { type: 'category.deleted', albumId }, + }); + let replica = await prismaOv.albums_ov.findUniqueOrThrow({ + where: { albumId }, + }); + expect((replica.categories as unknown[]).length).toBe(1); + + const stats = await reconciler.runOnce(); + expect(stats.mismatchedAlbums).toEqual([albumId]); + await drainSender(); + + // bootstrap 的 album.updated 携带分类全量清单（此刻为空）→ ov 整体替换。 + replica = await prismaOv.albums_ov.findUniqueOrThrow({ + where: { albumId }, + }); + expect(replica.categories).toEqual([]); + // cn 侧分类确已删净（对账收敛到权威态，非仅表面计数一致）。 + expect( + await prismaCn.categories.count({ where: { albumId } }), + ).toBe(0); + }); + + it('④ 稳态空转：一致时无 bootstrap、重放全判重（applied=0）', async () => { + const stats = await reconciler.runOnce(); + expect(stats).toMatchObject({ + digestAlbums: 1, + mismatchedAlbums: [], + ovOrphanAlbums: [], + replayApplied: 0, + }); + // 清扫仍会拉当前态（sweptEvents>0）但全部 eventId 判重——无副作用。 + expect(stats.sweptEvents).toBeGreaterThan(0); + expect( + await prismaCn.outbox_events.count({ where: { deliveredAt: null } }), + ).toBe(0); + expect(await prismaOv.photos_ov.count({ where: { albumId } })).toBe(2); + }); +});