feat(sync): reconciliation scheduler with digest and bootstrap healing

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 <noreply@anthropic.com>

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<void> =>
-  new Promise((resolve) => setTimeout(resolve, ms));
+/**
+ * 可被停机唤醒的 sleep：对账循环节拍是分钟级，若让 SIGTERM 干等一个完整
+ * interval 才退出不可接受——shutdown 置 stopping 后调用 wake 提前放行，
+ * 循环条件 !stopping 随即退出（在飞一轮的各步均有界，见 shutdown 注释）。
+ */
+let wakeSleeper: (() => void) | null = null;
+const sleepStoppable = (ms: number): Promise<void> =>
+  new Promise((resolve) => {
+    const timer = setTimeout(() => {
+      wakeSleeper = null;
+      resolve();
+    }, ms);
+    wakeSleeper = () => {
+      clearTimeout(timer);
+      wakeSleeper = null;
+      resolve();
+    };
+  });
 
 async function main(): Promise<void> {
   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<void> {
     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<void> | null = null;
-  const loop = async (): Promise<void> => {
+  const senderLoop = async (): Promise<void> => {
     let failureStreak = 0;
     while (!stopping) {
       try {
@@ -94,23 +134,38 @@ async function main(): Promise<void> {
           },
           'outbox push failed（整批未标记，下轮重推）',
         );
-        await sleep(backoffMs);
+        await sleepStoppable(backoffMs);
         continue;
       }
-      await sleep(intervalMs);
+      await sleepStoppable(intervalMs);
     }
   };
-  inFlight = loop();
+  const reconcileLoop = async (): Promise<void> => {
+    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<void>[] = [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) },

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:<type>:<id>:<ver>
+//   合成 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<ReconcileStats> {
+    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<DigestResponse>(
+      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<Map<string, OvAlbumCounts>> {
+    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<string, OvAlbumCounts>();
+    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<T>(
+    base: string,
+    path: string,
+    body: unknown,
+  ): Promise<T> {
+    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;
+  }
+}