fix(调度): cron 回调防重入 + reparse 两处 bind 变量溢出

## 一、cron 防重入(sync-incremental.scheduler.ts)

🔴 2026-08-29 生产事故:plan 段耗时涨过 2 小时后,cron(每 2 小时)照常触发下一轮,
两轮 plan 段并发抢同一批 I/O → 都更慢 → 更易被再下一轮套圈 → 雪崩。
实测 08-28 20:15 起连续多轮 plan 段一次都没跑完;到 08-29 11:19(即触发原因
reparse 结束 1.4 小时后)仍有两轮在并行互拖 —— 雪崩是**自持**的,源头消失也不会自愈。

现有的锁都挡不住,这是本修复存在的理由:
  ① NestJS CronJob **默认不防重入** —— 上一次回调还在 await,下一次照样进;
  ② sync_logs 的 partial UNIQUE(host_id) WHERE status='running' 只覆盖**摄入段**,
     摄入一结束锁就放了,而最慢的 persona / plan 段还在跑。
所以必须在回调入口用进程内 Set 挡。跳过而非排队:摄入是游标增量,下轮自然 catchup;
plan 是时间驱动全量,跳一轮只是晚 2 小时评估,远好过雪崩。
释放放在 finally —— 抛异常时不释放会把该 host 锁死到进程重启。

tests/scheduler-reentrancy-guard.spec.ts 锁四条:并发跳过 / 结束后放行 /
按 host 而非全局 / 抛异常也释放。

## 二、reparse 两处 bind 变量溢出(cold-import.service.ts)

`patientId: { in: [...] }` 直接塞完整患者清单会撞 PG 的 32767 上限。

实跑路径(第 3 步统计受影响患者)—— 2026-08-29 生产实测:18 万患者的 reparse
跑满 61/61 批、写完全部事实(superseded=9,062)之后**倒在最后一步**:
  Assertion violation: too many bind variables ... received 32769
6.6 小时的活全干完,只因收尾统计炸掉而 exit 1。

dry-run 路径同病:>3.2 万患者直接崩,而 --patients-file 的文档恰恰说
「按受影响患者收窄是最有效的提速手段(可达 250 倍)」—— 最需要先 dry-run 探路的
大清单场景,正好是它唯一不工作的场景。

两处都按 3000 分块(与 PAC_REPARSE_BATCH 同款),每块 3001 个变量,离上限很远。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

apps/pac-service/src/modules/sync/cold-import/cold-import.service.ts

@@ -185,10 +185,22 @@ export class ColdImportService {
 
     // 2a. dryRun:报 scope(患者数 + 各资源 txn 数),不写、不分批装配(便宜)。
     if (opts.dryRun) {
+      // ⛔ 同样分块 —— 早先这里也是整份清单塞 in,>3.2 万患者时 dry-run 直接崩,
+      //   而 --patients-file 的文档恰恰说「按受影响患者收窄是最有效的提速手段(可达 250 倍)」:
+      //   最需要先 dry-run 探一探的大清单场景,正好是它唯一不工作的场景。
+      const DRY_CHUNK = 3000;
       for (const cfg of reparseableCfgs) {
-        const n = await this.prisma.patientTransaction.count({
-          where: { hostId: host.id, subjectType: cfg.emits!.subjectType, ...(opts.patientIds?.length ? { patientId: { in: opts.patientIds } } : {}) },
+        let n = 0;
+        const chunks: Array<string[] | null> = opts.patientIds?.length
+          ? Array.from({ length: Math.ceil(opts.patientIds.length / DRY_CHUNK) }, (_, i) =>
+              opts.patientIds!.slice(i * DRY_CHUNK, (i + 1) * DRY_CHUNK),
+            )
+          : [null];
+        for (const chunk of chunks) {
+          n += await this.prisma.patientTransaction.count({
+            where: { hostId: host.id, subjectType: cfg.emits!.subjectType, ...(chunk ? { patientId: { in: chunk } } : {}) },
           });
+        }
         this.logger.log(`reparse[dry]: ${cfg.canonical}(${cfg.emits!.subjectType}) txns=${n} → 实跑按版本流 supersede 变更的、跳过不变的`);
       }
       this.logger.log(`reparse[dry]: 范围 ${scopePatientIds.length} 患者;去掉 --dry-run 实跑(非破坏)。`);
@@ -276,16 +288,34 @@ export class ColdImportService {
     }
 
     // 3. 受影响 patientId = 本次真正被 supersede(内容变了)的 fact 的 distinct patient → 只重算这些。
+    //
+    // ⛔ 必须**分块**查 —— `patientId: { in: [...] }` 直接塞完整清单会撞 PG 的
+    //   32767 bind 变量上限。2026-08-29 生产实测:18 万患者的 reparse 跑满 61/61 批、
+    //   写完全部事实之后,**倒在这最后一步**:
+    //     `Assertion violation: too many bind variables ... received 32769`
+    //   6.6 小时的活全干完了,只因收尾统计炸掉而 exit 1 —— 最难受的一种失败。
+    //   (同族的另一处在 dryRun 分支的 count,见下方注释。)
+    //   分块大小取 BATCH 同款 3000:每块 3001 个变量,离上限很远。
+    const CHANGED_CHUNK = 3000;
+    const affectedSet = new Set<string>();
+    const scanChunks: Array<string[] | null> = opts.patientIds?.length
+      ? Array.from({ length: Math.ceil(opts.patientIds.length / CHANGED_CHUNK) }, (_, i) =>
+          opts.patientIds!.slice(i * CHANGED_CHUNK, (i + 1) * CHANGED_CHUNK),
+        )
+      : [null]; // 不限定患者 → 一次全查(where 里没有 in,无变量上限问题)
+    for (const chunk of scanChunks) {
       const changed = await this.prisma.patientFact.findMany({
         where: {
           hostId: host.id,
           supersededAt: { gte: runStart },
-        ...(opts.patientIds?.length ? { patientId: { in: opts.patientIds } } : {}),
+          ...(chunk ? { patientId: { in: chunk } } : {}),
         },
         select: { patientId: true },
         distinct: ['patientId'],
       });
-    const affectedPatientIds = changed.map((a) => a.patientId).filter((x): x is string => !!x);
+      for (const c of changed) if (c.patientId) affectedSet.add(c.patientId);
+    }
+    const affectedPatientIds = [...affectedSet];
     return { perResource, affectedPatientIds, dryRunDiffs };
   }
 

apps/pac-service/src/queues/sync-incremental.scheduler.ts

@@ -41,6 +41,24 @@ import { schedulerDisabled } from './scheduler-switch';
 export class SyncIncrementalSchedulerService implements OnModuleInit {
   private readonly logger = new Logger(SyncIncrementalSchedulerService.name);
 
+  /**
+   * 本进程内「该 host 正在跑」的闸 —— **防套圈**。
+   *
+   * 🔴 2026-08-29 生产事故:plan 段耗时涨到 2 小时以上后,cron(每 2 小时)照常触发下一轮,
+   *   两轮的 plan 段并发抢同一批 I/O → 两轮都更慢 → 更容易被再下一轮套圈 → 雪崩。
+   *   实测 08-28 20:15 起连续多轮 plan 段一次都没跑完,直到 08-29 上午仍有两轮在并行。
+   *
+   * ⚠️ 为什么现有的锁挡不住:
+   *   ① NestJS 的 CronJob **默认不防重入** —— 上一次回调还在 await,下一次照样进;
+   *   ② `sync_logs` 的 partial UNIQUE(host_id) WHERE status='running' 只覆盖**摄入段**,
+   *      摄入一结束锁就放了,而 persona / plan 段还在跑,恰恰是最慢的部分。
+   *   所以必须在**回调入口**挡,不能靠库里的锁。
+   *
+   * 跳过而不是排队:摄入是游标增量,跳过这轮的数据下轮自然 catchup;
+   * plan 是时间驱动的全量,跳一轮只是晚 2 小时评估,远好过雪崩。
+   */
+  private readonly runningHosts = new Set<string>();
+
   constructor(
     private readonly prisma: PrismaService,
     private readonly coldImport: ColdImportService,
@@ -168,6 +186,15 @@ export class SyncIncrementalSchedulerService implements OnModuleInit {
 
   /// 单 host 跑一轮(cron 回调用,吞异常不影响该 host 下次 / 别的 host)
   private async runHostSafe(host: string): Promise<void> {
+    // ⛔ 上一轮还没跑完就跳过本轮 —— 见 runningHosts 的注释(防套圈雪崩)
+    if (this.runningHosts.has(host)) {
+      this.logger.warn(
+        `sync-incremental: host=${host} **跳过本轮** —— 上一轮仍在运行(防套圈)。` +
+          `连续出现说明单轮已撑不下 cron 间隔,需要查 plan 段耗时。`,
+      );
+      return;
+    }
+    this.runningHosts.add(host);
     try {
       await this.runOne(path.join(this.dataDir(), host));
     } catch (err) {
@@ -176,6 +203,9 @@ export class SyncIncrementalSchedulerService implements OnModuleInit {
       } else {
         this.logger.error(`sync-incremental: host=${host} failed: ${(err as Error).message}`);
       }
+    } finally {
+      // ⚠️ 必须在 finally —— 抛异常时不释放会把该 host 永久锁死到进程重启
+      this.runningHosts.delete(host);
     }
   }
 

apps/pac-service/tests/scheduler-reentrancy-guard.spec.ts

+import { SyncIncrementalSchedulerService } from '../src/queues/sync-incremental.scheduler';
+
+/**
+ * cron 回调防重入(runHostSafe 的 runningHosts 闸)回归。
+ *
+ * 🔴 2026-08-29 生产事故:plan 段耗时涨过 2 小时后,cron(每 2 小时)照常触发下一轮,
+ *   两轮的 plan 段并发抢同一批 I/O → 都更慢 → 更易被再下一轮套圈 → 雪崩。
+ *   实测 08-28 20:15 起连续多轮 plan 段一次都没跑完,到 08-29 上午仍有两轮在并行。
+ *
+ * 为什么现有的锁挡不住(这两条是本用例存在的理由):
+ *   ① NestJS CronJob **默认不防重入** —— 上一次回调还在 await,下一次照样进;
+ *   ② `sync_logs` 的 partial UNIQUE(host_id) WHERE status='running' 只覆盖**摄入段**,
+ *      摄入一结束锁就放了,而最慢的 persona / plan 段还在跑。
+ *
+ * 跑:
+ *   pnpm test -- scheduler-reentrancy-guard
+ */
+
+/** 造一个只关心 runOne 编排的实例;runOne 用可控的 promise 替换 */
+function makeService() {
+  const svc = new SyncIncrementalSchedulerService(
+    {} as never, {} as never, {} as never, {} as never, {} as never, {} as never,
+  );
+  const calls: string[] = [];
+  let release!: () => void;
+  const gate = new Promise<void>((r) => { release = r; });
+  (svc as unknown as { runOne: (dir: string) => Promise<void> }).runOne = async (dir: string) => {
+    calls.push(dir);
+    await gate; // 卡住不返回 = 模拟"上一轮还在跑"
+  };
+  const run = (host: string) =>
+    (svc as unknown as { runHostSafe: (h: string) => Promise<void> }).runHostSafe(host);
+  return { svc, calls, run, release };
+}
+
+describe('runHostSafe 防重入', () => {
+  test('🔴 上一轮未结束时,同 host 的下一轮被跳过(不并发进 runOne)', async () => {
+    const { calls, run, release } = makeService();
+    const first = run('jvs-dw'); // 卡在 gate 上
+    await run('jvs-dw'); // 第二次应立即返回
+    await run('jvs-dw'); // 第三次同样
+    expect(calls).toHaveLength(1); // ⛔ 只有第一轮真正进了 runOne
+    release();
+    await first;
+  });
+
+  test('上一轮结束后,下一轮正常放行', async () => {
+    const { calls, run, release } = makeService();
+    const first = run('jvs-dw');
+    release();
+    await first;
+    await run('jvs-dw');
+    expect(calls).toHaveLength(2);
+  });
+
+  test('不同 host 互不阻塞(闸是按 host 的,不是全局)', async () => {
+    const { calls, run, release } = makeService();
+    const a = run('jvs-dw');
+    const b = run('friday');
+    // runOne 收到的是 path.join(dataDir, host) 的完整路径,按后缀断言
+    expect(calls.map((d) => d.split('/').pop())).toEqual(['jvs-dw', 'friday']);
+    release();
+    await Promise.all([a, b]);
+  });
+
+  test('⛔ runOne 抛异常也必须释放闸 —— 否则该 host 被永久锁死到进程重启', async () => {
+    const svc = new SyncIncrementalSchedulerService(
+      {} as never, {} as never, {} as never, {} as never, {} as never, {} as never,
+    );
+    let n = 0;
+    (svc as unknown as { runOne: (dir: string) => Promise<void> }).runOne = async () => {
+      n++;
+      throw new Error('boom');
+    };
+    const run = (h: string) =>
+      (svc as unknown as { runHostSafe: (h: string) => Promise<void> }).runHostSafe(h);
+    await run('jvs-dw'); // runHostSafe 吞异常
+    await run('jvs-dw'); // 闸已释放 → 应能再进
+    expect(n).toBe(2);
+  });
+});