merge: plan 写入段端到端分批 → main

apps/pac-service/src/modules/plan/engine/plan-engine.service.ts

@@ -26,6 +26,34 @@ import { PlanLabelService } from '../plan-label.service';
  *   - 有 active plan + hits 不同 → supersede 旧 + 创建新版本
  *   - 已有 assigned plan → 不动(避免抢正在跟进的客服)— v1 用 simple skip
  */
+/**
+ * plan 写入段的**端到端批大小**(按患者切)。
+ *
+ * 🔴 2026-09-04 生产 OOM 宕机 2.5 小时的直接对策。内核日志:
+ *   `Killed process (node) anon-rss:6,321,668kB(6.03GB)`,机器 14GB,之后整机 swap 抖死、
+ *   sshd 与 Web 全部无响应。docker-compose.prod.yml 那段注释 2026-08-26 就预言过
+ *   「这是止血不是根治:池子还在涨,8G 迟早也会到顶」—— 那天到了。
+ *
+ * 堆账(550,049 命中患者 / 约 96 万条 hit,按 V8 对象布局建模):
+ *   latestByPatient 1.93GB + hitsByPatient 1.06GB + persona/visit 0.14GB
+ *   + activePlans 0.09GB + logRows 0.07GB ≈ **活跃保留 3.3GB**,实测 RSS 6.03GB(≈1.8×)。
+ * 病根不是"取数没分块"(取数早就是 2000 一块),是**产物全程不释放**:
+ *   runPool 的闭包(见下方批循环)把 latest/persona/visit/logRows 全部 context-allocate,
+ *   V8 要到 runAllForHost 整个 frame 结束才可能回收 → 第 3 步扫全池时它们全还在,
+ *   那一下就是峰值。所以要把「取数分块」升格成「端到端分批」:每批取数→写库→释放。
+ *
+ * ⚠️ 上限 20000:latest/persona 两条走 Prisma `patientId: { in: ids }`,受 PG 32767 bind 上限约束。
+ * ⚠️ 显式传 0 = single-shot(不分批),等价于改造前的行为 —— **这是回滚开关**,
+ *   照 cold-import 的 resolveCohortBatchSize 同一形状(传 0 回退 single-shot)。
+ * ⛔ 别把 snooze 预取塞进批循环(见 prefetchSnoozeAnchors 的注释):那是已经被消掉的
+ *   O(患者数) 项,塞回去到 200 万患者就是 1 次 vs 1000 次。
+ */
+export function resolvePlanBatchSize(env: NodeJS.ProcessEnv = process.env): number {
+  const raw = env.PAC_PLAN_BATCH_SIZE;
+  if (raw !== undefined && raw.trim() !== '' && Number(raw) === 0) return 0; // 0 = single-shot 回退
+  return Math.min(20_000, Math.max(1, Number(raw) || 2000));
+}
+
 @Injectable()
 export class PlanEngineService {
   private readonly logger = new Logger(PlanEngineService.name);
@@ -272,21 +300,54 @@ export class PlanEngineService {
     //   注:selectHits 的全表扫不在本优化内(时间规则随时可改,每轮必须全量重选才正确)。
     const selectMs = Date.now() - tSelect;
     const patientIds = [...hitsByPatient.keys()];
+    /**
+     * ⭐ 第 3 步(stale-close)是整条链**唯一**的硬全局依赖:它拿本轮命中集对整个召回池取补集
+     *   (`!hitsByPatient.has(pid)` → supersede)。分批之后 hitsByPatient 会被逐批清空,
+     *   所以这里先把「本轮命中过谁」单独留成一个 Set —— 只存 id,不存 hit 载荷。
+     * 🔴 别把第 3 步搬进批循环:每批都会把**其他批**的患者判成"信号消失",
+     *   结果是整池 supersede + 每条认领单一条 auto_release,**而且不报错** ——
+     *   下面那行「本轮关闭 N 条无信号 plan」的日志反而会把它解释成「这不是 bug」。
+     *   第 3 步必须等所有批跑完再跑一次。
+     */
+    const hitPatientIds = new Set(patientIds);
     const tPrefetch = Date.now();
-    const { latestByPatient, snoozedByPatient, personaByPatient, lastVisitClinicByPatient, touchedPlanIds } =
-      await this.prefetchForBatch(scope, patientIds, now);
-    const prefetchMs = Date.now() - tPrefetch;
+    // snooze 整轮一次(见 prefetchSnoozeAnchors);其余三条按批取。
+    const snoozedByPatient = await this.prefetchSnoozeAnchors(scope, now);
+    let prefetchMs = Date.now() - tPrefetch;
     const tWrite = Date.now();
     const EMPTY_SNOOZE = new Map<string, Date>();
-    const logRows: Prisma.PlanGenerationLogCreateManyInput[] = [];
     //   ⭐ 2026-07-26:写操作从「每 N 个一批 await Promise.all」的**栅栏**改成连续调度的
     //   worker pool(见 common/run-pool.ts)。栅栏每批都要等本批最慢那个,快的 worker 干等。
     //   同一处改法在 recompute-persona 上实测过(测试服 35.7 万患者,concurrency=8):
     //     并发效率 44% → 94%,吞吐 1,454 → 2,694 人/分。
     //   plan 侧单患者耗时的离散度只会更大(upsertPlan 里 reason 数量差异明显),栅栏的浪费同理。
     const concurrency = Math.max(1, Number(process.env.PAC_PLAN_BATCH_CONCURRENCY) || 8);
-    const entries = [...hitsByPatient.entries()];
-    await runPool(entries, concurrency, async ([patientId, hits]) => {
+    /**
+     * ⭐ 端到端批循环(2026-09-04)。每批:预取 → 写库 → **释放**。
+     *
+     * ⛔ **循环体里不要加 try/catch**。今天的语义是「selectHits 抛错发生在任何写之前 → 整轮中止,
+     *   第 3 步不跑」;分批后前 N 批已经落库了,若在这里吞掉异常"把剩下的批跑完",
+     *   第 3 步就会拿着**残缺**的命中集去关池子 —— 后果同上面 hitPatientIds 那段注释。
+     *   逐患者的 try/catch(下面 runPool 回调里那个)原样保留,那个是有意的。
+     */
+    const batchSize = resolvePlanBatchSize();
+    const step = batchSize > 0 ? batchSize : Math.max(1, patientIds.length);
+    let batches = 0;
+    for (let off = 0; off < patientIds.length; off += step) {
+      const ids = patientIds.slice(off, off + step);
+      batches += 1;
+      const tb = Date.now();
+      const { latestByPatient, personaByPatient, lastVisitClinicByPatient, touchedPlanIds } =
+        await this.prefetchOneBatch(scope, ids);
+      prefetchMs += Date.now() - tb;
+      const logRows: Prisma.PlanGenerationLogCreateManyInput[] = [];
+      // hit 的所有权交给本批的局部数组,随即从 Map 里删掉 —— 批末连同 latest/persona/visit/logRows
+      // 一起出作用域被回收。这是本次改造真正省内存的那一下(hitsByPatient 满载约 1.06GB)。
+      const entries = ids.map(
+        (pid) => [pid, hitsByPatient.get(pid) ?? []] as [string, ScenarioHitWithKey[]],
+      );
+      for (const pid of ids) hitsByPatient.delete(pid);
+      await runPool(entries, concurrency, async ([patientId, hits]) => {
       try {
         const result = await this.upsertPlan({
           scope,
@@ -332,9 +393,21 @@ export class PlanEngineService {
           `plan upsert failed patient=${patientId}: ${err instanceof Error ? err.message : err}`,
         );
       }
-    });
-    for (let i = 0; i < logRows.length; i += 5000) {
-      await this.prisma.planGenerationLog.createMany({ data: logRows.slice(i, i + 5000) });
+      });
+      for (let i = 0; i < logRows.length; i += 5000) {
+        await this.prisma.planGenerationLog.createMany({ data: logRows.slice(i, i + 5000) });
+      }
+      // 内存水位:沿用 cold-import 的 `rss=` 字样便于跟历史并排 grep;额外带 heapUsed ——
+      // 判 8G 堆上限还剩多少余量只能看 heapUsed,rss 含 Prisma Rust 引擎与碎片,不够精确。
+      // 每批一行太吵:每 5 批 + 末批各打一次(同 cold-import 的做法)。
+      if (batches % 5 === 0 || off + step >= patientIds.length) {
+        const mu = process.memoryUsage();
+        const mb = (n: number) => Math.round(n / 1024 / 1024);
+        this.logger.log(
+          `[plan] 批 ${batches} 患者=${ids.length} 累计=${off + ids.length}/${patientIds.length} ` +
+            `rss=${mb(mu.rss)}MB heapUsed=${mb(mu.heapUsed)}MB heapTotal=${mb(mu.heapTotal)}MB`,
+        );
+      }
     }
 
     // 3. ⭐ 缺口1 修复:关闭"有 active plan 但本轮 0 命中"的患者的遗留 plan。
@@ -359,8 +432,11 @@ export class PlanEngineService {
     //   ⭐ 子集模式(--clinics)必须**同步收窄**:本轮只评估了子集,子集外的患者天然 0 命中,
     //     不收窄就会把整个 host 的召回池当成"信号全消失"清空,认领中的单还各记一条 auto_release。
     //     用内存 Set 过滤而非 SQL `patientId: { in: ids }` —— 子集动辄数万,会撞 PG 32767 bind 上限。
+    //   ⚠️ 2026-09-04:判据从 `hitsByPatient.has` 换成 `hitPatientIds.has` —— 分批之后
+    //     hitsByPatient 已被逐批清空(hit 载荷交给各批局部数组以便回收),只有那个 id Set 还在。
+    //     两者在改造前后是同一个集合(Set 由 hitsByPatient.keys() 构造且此后只读)。
     const staleRows = activePlans.filter(
-      (pl) => !hitsByPatient.has(pl.patientId) && (!scopedPatientIds || scopedPatientIds.has(pl.patientId)),
+      (pl) => !hitPatientIds.has(pl.patientId) && (!scopedPatientIds || scopedPatientIds.has(pl.patientId)),
     );
     if (staleRows.length > 0) {
       // ⚠️ 必须分片:原实现是一条 `id: { in: staleIds }`,PG bind 变量上限 32767 —— 池子上了
@@ -422,13 +498,17 @@ export class PlanEngineService {
       this.logger.warn(`[标签] 补齐失败（不阻断生成，夜间刷新会兜住）：${e instanceof Error ? e.message : e}`);
     }
 
+    // ⚠️ 「预取=」现在是 N 批之和(含整轮一次的 snooze),「写入=」含批循环全部开销;
+    //   加 `批=` 便于跟历史对照。判据仍只看整轮墙钟,别拿分段和历史逐项比。
     this.logger.log(
       `[plan] 阶段耗时 场景=${selectMs}ms 预取=${prefetchMs}ms 写入=${Date.now() - tWrite}ms ` +
-        `命中患者=${patientIds.length} 总计=${Date.now() - startedAt.getTime()}ms`,
+        `批=${batches}(size=${batchSize === 0 ? 'single-shot' : batchSize}) ` +
+        `命中患者=${hitPatientIds.size} 总计=${Date.now() - startedAt.getTime()}ms`,
     );
     return {
       scenariosRun: this.scenarios.length,
-      patientsHit: hitsByPatient.size,
+      // ⚠️ 不能用 hitsByPatient.size —— 分批后它已被清空,恒为 0。
+      patientsHit: hitPatientIds.size,
       plansCreated,
       plansSuperseded,
       plansUnchanged,
@@ -913,23 +993,17 @@ export class PlanEngineService {
    * 取数口径与 upsertPlan(findFirst orderBy version desc)/ fetchSnoozedSignalKeys(终态+未到期)
    * 严格对齐;IN 列表分块防参数超限。
    */
-  private async prefetchForBatch(
+  /**
+   * snooze 抑制集 —— **整轮查一次,绝不能进批循环**。
+   * 它的代价跟「终态且冷静期未到期的计划数」走,跟患者数无关(生产实测全库 106 行)。
+   * 2026-09-04 把 prefetchForBatch 拆成「整轮一次」+「每批一次」两半时单独抽出来,
+   * 就是为了让这条纪律在类型上也成立:它压根拿不到 patientIds,想塞进批里也塞不了。
+   */
+  private async prefetchSnoozeAnchors(
     scope: ScenarioScope,
-    patientIds: string[],
     now: Date,
-  ): Promise<{
-    latestByPatient: Map<string, PlanWithReasons>;
-    snoozedByPatient: Map<string, Map<string, Date>>;
-    personaByPatient: Map<string, string>;
-    lastVisitClinicByPatient: Map<string, string>;
-    /// 「客服碰过」的 planId 集合 —— ⚠️ **只覆盖带 assignment_id 的单**:
-    /// 没进过批次的单本来就没有归因可继承,为它们查 view 事件是白扫全表。
-    touchedPlanIds: Set<string>;
-  }> {
-    const latestByPatient = new Map<string, PlanWithReasons>();
+  ): Promise<Map<string, Map<string, Date>>> {
     const snoozedByPatient = new Map<string, Map<string, Date>>();
-    const personaByPatient = new Map<string, string>();
-    const lastVisitClinicByPatient = new Map<string, string>();
 
     // ⭐ snooze 抑制集**提到循环外一次查完** —— 它的代价跟「终态且冷静期未到期的计划数」走,
     //   跟患者数无关。2026-08-30 生产实测:全库符合条件的只有 **106 行**,
@@ -965,10 +1039,31 @@ export class PlanEngineService {
         snoozedByPatient.set(pid, buildSnoozeAnchors(plans));
       }
     }
+    return snoozedByPatient;
+  }
 
-    const CHUNK = 2000;
-    for (let i = 0; i < patientIds.length; i += CHUNK) {
-      const ids = patientIds.slice(i, i + CHUNK);
+  /**
+   * **一批**患者的预取(不含 snooze —— 那个整轮一次,见 prefetchSnoozeAnchors)。
+   *
+   * 2026-09-04 从 prefetchForBatch 里拆出来:原先这里有个内部 `CHUNK = 2000` 循环,
+   * 结果**累积**在跨全量存活的几个 Map 里 —— 取数是分块的,内存不是。
+   * 现在把切批权交给调用方,本方法只管一批,返回的 Map 生命周期 = 该批,批末即可回收。
+   */
+  private async prefetchOneBatch(
+    scope: ScenarioScope,
+    ids: string[],
+  ): Promise<{
+    latestByPatient: Map<string, PlanWithReasons>;
+    personaByPatient: Map<string, string>;
+    lastVisitClinicByPatient: Map<string, string>;
+    /// 「客服碰过」的 planId 集合 —— ⚠️ **只覆盖带 assignment_id 的单**:
+    /// 没进过批次的单本来就没有归因可继承,为它们查 view 事件是白扫全表。
+    touchedPlanIds: Set<string>;
+  }> {
+    const latestByPatient = new Map<string, PlanWithReasons>();
+    const personaByPatient = new Map<string, string>();
+    const lastVisitClinicByPatient = new Map<string, string>();
+    {
       // ⭐ 三条彼此独立,**并行发** —— 原来是串行,每 chunk 的墙钟 = 三条之和;
       //   并行后 = 最慢那条。三条都只读、无共享状态,并行不改任何口径。
       //   并发度就是 3(不随患者数涨),不会挤爆连接池(Prisma 默认池 = 核数×2+1)。
@@ -1029,9 +1124,7 @@ export class PlanEngineService {
       for (const r of rows) touchedPlanIds.add(r.planId);
     }
 
-    return {
-      latestByPatient, snoozedByPatient, personaByPatient, lastVisitClinicByPatient, touchedPlanIds,
-    };
+    return { latestByPatient, personaByPatient, lastVisitClinicByPatient, touchedPlanIds };
   }
 
   /// 单刷路径:解析该患者最后一次到诊(encounter/emr)所在诊所(批量路径走 prefetchForBatch)。

apps/pac-service/tests/plan-engine-batch.spec.ts

-import { PlanEngineService } from '../src/modules/plan/engine/plan-engine.service';
+import { PlanEngineService, resolvePlanBatchSize } from '../src/modules/plan/engine/plan-engine.service';
 import type { ScenarioHit } from '../src/modules/plan/engine/scenario.interface';
 
 /**
@@ -1156,3 +1156,109 @@ describe('归因继承的边界 — 判据是「客服碰过没」', () => {
     for (const [args] of called) expect(args.where.planId.in).not.toContain('p-g');
   });
 });
+
+/**
+ * 🔴 2026-09-04 端到端分批改造的**核心判据**:结果与批大小无关。
+ *
+ * 事故背景:生产 pac-service 涨到 6.03GB 常驻被内核 OOM killer 打掉,整机 swap 抖死 2.5 小时。
+ *   堆账里 latestByPatient(1.93GB)+ hitsByPatient(1.06GB)是大头,而它们原先**全程不释放**
+ *   —— 取数早就是 2000 一块,但产物累积在跨全量存活的 Map 里,runPool 的闭包又把它们
+ *   context-allocate 到整个 runAllForHost frame 结束。改造把取数分块升格成端到端分批。
+ *
+ * 为什么这组用例是必须的:上面那 20+ 条既有用例**一条都跑不到分批路径的差异** ——
+ *   它们每个只有 1~4 个患者,默认批大小 2000,永远只有一批。分批写错了它们全绿。
+ *   这里用 describe.each 把同一份 fixture 在不同批大小下各跑一遍,直接编码
+ *   「结果与批大小无关」这条不变式。
+ *
+ * ⚠️ batchSize=1 是最凶的一档:每批一个患者,同时压测三件事 ——
+ *   ① 跨批不误关(stale-close 拿的是跨批累加的 hitPatientIds,不是被逐批清空的 hitsByPatient)
+ *   ② touchedPlanIds 按批算与全量算等价
+ *   ③ snooze 提在循环外之后仍能被每一批查到
+ */
+describe('端到端分批 — 结果必须与 PAC_PLAN_BATCH_SIZE 无关', () => {
+  const prev = process.env.PAC_PLAN_BATCH_SIZE;
+  afterEach(() => {
+    if (prev === undefined) delete process.env.PAC_PLAN_BATCH_SIZE;
+    else process.env.PAC_PLAN_BATCH_SIZE = prev;
+  });
+
+  /// 与上面「混合一批」同一份 fixture:5 种结局各一,外加一个 0 命中的患者压 stale-close
+  const runMixed = async () => {
+    const { prisma, plans, logs } = makeStore({
+      plans: [
+        { id: 'u-old', patientId: 'p2', status: 'active', priorityScore: 50, reasons: [{ scenario: SCEN, subKey: 'k@1' }] },
+        { id: 's-old', patientId: 'p3', status: 'active', reasons: [{ scenario: SCEN, subKey: 'k@old' }] },
+        {
+          id: 't-term',
+          patientId: 'p4',
+          status: 'abandoned',
+          snoozedUntil: new Date('2026-12-01T00:00:00Z'),
+          reasons: [{ scenario: SCEN, subKey: 'k@4' }],
+        },
+        // ⭐ p9 本轮 0 命中且有 active plan → 必须被 stale-close 关掉。
+        //   分批写错(第 3 步进了循环 / 用了被清空的 hitsByPatient)时,
+        //   这里会连 p1~p4 一起关掉,plansClosed 从 1 变 5 —— 这条就是照妖镜。
+        { id: 'stale-old', patientId: 'p9', status: 'active', reasons: [{ scenario: SCEN, subKey: 'k@gone' }] },
+      ],
+    });
+    const res = await engine(
+      prisma,
+      makeScenario([
+        hit('p1', 'k@1'),
+        hit('p2', 'k@1', 50),
+        hit('p3', 'k@new'),
+        hit('p4', 'k@4'),
+      ]),
+    ).runAllForHost({ hostId: HOST, tenantId: TENANT, now: NOW });
+    return { res, plans, logs };
+  };
+
+  /// 把最终状态压成一个可比对的快照(引擎返回值 + 落库结果),批大小不该改变它任何一位
+  const snapshot = (r: Awaited<ReturnType<typeof runMixed>>) => ({
+    result: {
+      scenariosRun: r.res.scenariosRun,
+      patientsHit: r.res.patientsHit,
+      plansCreated: r.res.plansCreated,
+      plansSuperseded: r.res.plansSuperseded,
+      plansUnchanged: r.res.plansUnchanged,
+      plansSuppressed: r.res.plansSuppressed,
+      plansClosed: r.res.plansClosed,
+      plansSkippedAssigned: r.res.plansSkippedAssigned,
+    },
+    // plan 终态:按 (patientId, version) 排序,避免并发写入顺序带来的假差异
+    plans: r.plans
+      .map((p) => `${p.patientId}|v${p.version}|${p.status}|${p.priorityScore}`)
+      .sort(),
+    logs: r.logs.map((l) => `${l.patientId}|${l.status}`).sort(),
+  });
+
+  const SIZES = [0, 1, 2, 3, 2000];
+  let baseline: ReturnType<typeof snapshot> | null = null;
+
+  test.each(SIZES)('批大小 %s → 结果与基线逐字段一致', async (size) => {
+    process.env.PAC_PLAN_BATCH_SIZE = String(size);
+    const snap = snapshot(await runMixed());
+    // 先自证这份 fixture 真的走到了 5 种结局,否则"全相等"可能只是都没跑
+    expect(snap.result).toMatchObject({
+      patientsHit: 4,
+      plansCreated: 1,
+      plansUnchanged: 1,
+      plansSuperseded: 1,
+      plansSuppressed: 1,
+      plansClosed: 1, // ⭐ 只关 p9;若变成 5 说明分批把其他批的患者误判成"信号消失"
+    });
+    if (baseline === null) baseline = snap;
+    else expect(snap).toEqual(baseline);
+  });
+
+  test('⛔ 0 = single-shot 回退开关(等价于改造前的不分批行为)', () => {
+    expect(resolvePlanBatchSize({ PAC_PLAN_BATCH_SIZE: '0' } as NodeJS.ProcessEnv)).toBe(0);
+  });
+
+  test('⛔ 批大小夹在 [1, 20000] —— 上限守 PG 32767 bind(latest/persona 走 patientId in ids)', () => {
+    expect(resolvePlanBatchSize({ PAC_PLAN_BATCH_SIZE: '999999' } as NodeJS.ProcessEnv)).toBe(20_000);
+    expect(resolvePlanBatchSize({ PAC_PLAN_BATCH_SIZE: '-5' } as NodeJS.ProcessEnv)).toBe(1);
+    expect(resolvePlanBatchSize({} as NodeJS.ProcessEnv)).toBe(2000);
+    expect(resolvePlanBatchSize({ PAC_PLAN_BATCH_SIZE: 'abc' } as NodeJS.ProcessEnv)).toBe(2000);
+  });
+});