merge: cron 防重入 + reparse bind 溢出修复 + 并发否定结论 → test

apps/pac-service/src/modules/plan/engine/scenarios/treatment-initiation-recall.scenario.ts

@@ -220,6 +220,14 @@ export class TreatmentInitiationRecallScenario implements PlanScenarioPlugin {
     // 10 个子场景查询彼此独立(各自 SQL + union-find 合并 + 打分,无共享状态);
     //   合并/去重(下游 hitsByPatient Map + max 分 + Set 比对)与顺序无关 → 并行 = 结果逐字节相同。
     //   PAC_RECALL_SUBSCENARIO_CONCURRENCY=N(>1)开并行(分块 Promise.all);默认串行,行为不变。
+    //
+    // ⛔ **别指望靠这个旋钮提速** —— 2026-08-29 测试服实测(585K 患者 / 87,661 命中,空闲机):
+    //      串行(默认 1):  24.6 / 21.7 / 23.5 分钟(三次)
+    //      并发 3:        24.1 分钟  ← 不但没快,还略慢
+    //    原因:本阶段是**共享磁盘 I/O 受限**,不是查询延迟受限。采样显示全程 DataFileRead,
+    //    并行只是让几条查询抢同一批 page,总读取量一个字节都没少。
+    //    要提速得**减少读取量**(索引 / 收窄扫描范围),不是提高并行度。
+    //    (单条最慢的子场景查询 6.7 分钟,其余 12~35 秒 —— 并行后墙钟被最慢那条兜住。)
     const conc = Math.max(1, Number(process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY) || 1);
     // ⚠️ 不能 hits.push(...subHits):spread 把每个元素当实参压栈,V8 实参上限 ~6.5万;
     //    host 患者到 ~28 万后单子场景命中可超限 → RangeError: Maximum call stack size exceeded

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 → 只重算这些。
-    const changed = await this.prisma.patientFact.findMany({
-      where: {
-        hostId: host.id,
-        supersededAt: { gte: runStart },
-        ...(opts.patientIds?.length ? { patientId: { in: opts.patientIds } } : {}),
-      },
-      select: { patientId: true },
-      distinct: ['patientId'],
-    });
-    const affectedPatientIds = changed.map((a) => a.patientId).filter((x): x is string => !!x);
+    //
+    // ⛔ 必须**分块**查 —— `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 },
+          ...(chunk ? { patientId: { in: chunk } } : {}),
+        },
+        select: { patientId: true },
+        distinct: ['patientId'],
+      });
+      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);
+  });
+});