perf(persona): 重算吞吐四处优化 —— worker pool + 连接池联动 + 少两次往返

生产/测试服实测基线(测试服 12 万患者样本,concurrency=8):
  单患者 p50 139ms / p90 230ms / p99 386ms / max 75,750ms,均值 165ms
  理论上限 8÷0.165 = 2,909 人/分,实际 ~1,600 → **效率只有 55%**

## ① 批次栅栏 → worker pool(src/common/run-pool.ts)

原写法 `for (i += N) await Promise.all(slice)` 是**栅栏**:每批都等本批最慢那个,
快的 worker 干等。缺口正好是 E[N 次抽样最大值](≈p90)与均值之比。
极端例子:那个 75.7 秒的患者把同批 7 个 worker 一起冻了 75 秒。
改成 N 个 worker 各自从共享游标取下一个,谁空谁取 —— 慢任务只拖住自己那条线。

## ② --concurrency 根本没放大连接池(真 bug)

`withCohortDerivedPool` 只认 `PAC_COHORT_CONCURRENCY`(**摄入**的旋钮),
于是 `recompute-persona --concurrency=N` 的旋钮**静默失效**:实测跑 --concurrency=8 时
进程恰好只有 9 条连接(= 4核×2+1 的 Prisma 默认),超过 9 的 worker 全卡在池上排队。

⚠️ 这个 bug **只在小核机器上咬人**,所以一直没被发现:服务器 4 核 → 默认池 9;
   开发机 16 核 → 默认池 33,本地压根复现不出来。

改:池改读通用的 `PAC_DB_CONCURRENCY`,由各 CLI 在 NestFactory **之前**按自己的
--concurrency 设进来(PrismaService 在容器初始化时就定池,晚了没用);
`PAC_COHORT_CONCURRENCY` 保留为向后兼容别名,两者取大。

## ③ 同一张表查两遍 → 一次取,内存分两份

persona.service 对 patient_facts 每患者查两次(同患者、同索引、两次往返):
  ① status IN (active, fulfilled)                    → factsByType
  ② type=appointment_record AND status != superseded → appointmentsAll
② 是 ① 在 appointment 这类上的放宽,故取 `status != 'superseded'` 一把捞,内存切两份,
语义逐字节等价。

## ⑤ 审计日志 2 写 → 1 写

原 create(status='running') + update(终态)。startedAt 可显式赋值,故耗时统计一字不差。
代价:进程被硬杀(OOM/SIGKILL)时不留痕 —— 评估可接受:try/catch 里的异常仍写 'failed' 行;
唯一丢的是"算到一半进程没了",而那种情况原本留下的是一条**永远卡在 running** 的行(没人清理)。
读侧确认无依赖(daily-health-report 只用 findFirst 最近一条 + 按 status 的 groupBy)。
全量回填还顺带少一半日志表写入(35.6 万 → 17.8 万行/轮)。

往返预算:12.7 → 10.7(-16%)。

## 本地实测(13,268 患者,--force,concurrency=8,同为「全 unchanged」稳态)

  旧代码  93.50s
  优化后  76.89s      → 1.22×,省 17.8%

⚠️ 本地测不出 ① 的真实收益:本机 avg 35.9ms / p90 49ms / **max 436ms**,
   而测试服 avg 165ms / p90 230ms / **max 75,750ms**(174× 均值)。本地没有那种能冻住
   整批的长尾,且延迟低到瓶颈更像是 JS 单线程 CPU 而非 DB 往返。
   所以 1.22× 主要是 ③⑤ 的往返削减(机器无关),① 的收益要到服务器上才显现 ——
   **具体多少不预估,等在测试服实跑对照。**

## 正确性

指纹法:优化后跑完 → 切回旧代码再跑 → 旧代码对全部 13,268 人判 `unchanged`,
persona_features 指纹 4de1dcd0ddc6a2c99e222fd43e0755de 不变。
即**旧算法认为新代码的产出与它自己会算出来的逐字节相同**。

## 测试

494 单测通过(36 suites),service tsc 干净。新增 tests/persona-recompute-perf.spec.ts 12 例:
  runPool — 不丢项 / 并发上限严格且用满 / ⭐慢任务不冻住其他 worker / 边界 / 错误上抛
  连接池 — ⭐PAC_DB_CONCURRENCY 生效 / COHORT 向后兼容 / 取大 / 封顶 40 / =1 不动 URL /
          已有 connection_limit 不覆盖 / 非标准 URL 不抛

## 未做(单独排)

④ gapSelector 每患者平均 2.66 条 SQL(实测分布 0 组 3.4% / 1 组 19.5% / 2 组 25.7% /
  3 组 24.7% / 4+ 26.7%,max 9)。UNION ALL 合一能再省 ~1.7 次往返,但动的是**召回共享**
  的 gap 真理源,风险最高,要配专门的对账测试。

另注:plan-engine.runAllForHost 有一模一样的批次栅栏写法(PAC_PLAN_BATCH_CONCURRENCY),
同样的 5 行改法适用,本次未动。

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

apps/pac-service/src/cli/recompute-persona.cli.ts

@@ -14,6 +14,7 @@ import { Logger } from '@nestjs/common';
 import { AppModule } from '../app.module';
 import { PersonaService } from '../modules/persona/persona.service';
 import { PrismaService } from '../prisma/prisma.service';
+import { runPool } from '../common/run-pool';
 
 interface Args {
   host: string;
@@ -49,6 +50,13 @@ function parseArgs(argv: string[]): Args {
 async function bootstrap() {
   const args = parseArgs(process.argv.slice(2));
   const logger = new Logger('recompute-persona');
+  // ⭐ 必须在 NestFactory 之前设:PrismaService 在容器初始化时就按这个值算 connection_limit。
+  //   不设的话池是 Prisma 默认的「核数×2+1」(4 核机 = 9),--concurrency 超过它就静默失效
+  //   —— 多出来的 worker 全卡在连接池排队,提并发一点不快(2026-07-26 生产实测)。
+  //   只在用户显式给了 --concurrency 时设,不覆盖外部已有的环境变量。
+  if (args.concurrency > 1 && !process.env.PAC_DB_CONCURRENCY) {
+    process.env.PAC_DB_CONCURRENCY = String(args.concurrency);
+  }
   const app = await NestFactory.createApplicationContext(AppModule, {
     logger: ['log', 'warn', 'error'],
   });
@@ -140,11 +148,9 @@ async function bootstrap() {
       if (!verbose && ++done % 2000 === 0) logger.log(`  …进度 ${done}/${total} (failed=${failed})`);
     };
 
-    // 并发分块:每患者 recompute 含 ~15 次串行 DB 往返(I/O 等待为主),
-    // 单进程内并发可重叠这些等待;CPU 占比高的部分再靠多进程 --shard 吃多核。
-    for (let i = 0; i < patients.length; i += args.concurrency) {
-      await Promise.all(patients.slice(i, i + args.concurrency).map(runOne));
-    }
+    // ⭐ worker pool 替代原「每 N 个一批 await Promise.all」的栅栏(实测效率只有 55%,
+    //   缺口=等本批最慢那个)。口径与 runPool 注释同源。
+    await runPool(patients, args.concurrency, (p) => runOne(p));
 
     logger.log('─────────────────────────────────');
     logger.log(

apps/pac-service/src/common/run-pool.ts

+/**
+ * runPool —— 连续调度的 worker pool(替代「每 N 个一批 await Promise.all」的栅栏写法)。
+ *
+ * 为什么不用批次栅栏:每批都要等本批**最慢**那个,快的 worker 白等。
+ * 2026-07-26 在测试服实测 recompute-persona(12 万患者样本):
+ *   单患者 p50 139ms / p90 230ms / p99 386ms / max 75.7s,均值 165ms
+ *   concurrency=8 理论上限 8÷0.165 = 2,909 人/分,实际只有 ~1,600 → **效率 55%**
+ * 缺口正好是 E[8 次抽样最大值](≈p90)与均值之比。极端例子:那个 75.7 秒的患者
+ * 把同批 7 个 worker 一起冻了 75 秒。
+ *
+ * 本实现:N 个 worker 各自从共享游标取下一个,谁空谁取 —— 慢任务只拖住自己那条线。
+ *
+ * 语义与 `for (i+=N) await Promise.all(slice)` 保持一致的地方:
+ *   · 全部处理完才 resolve;
+ *   · 并发上限严格 ≤ concurrency;
+ *   · **不做错误吞咽** —— fn 抛出会让整个 runPool reject(调用方自己 try/catch 每项,
+ *     跟原来 `.map(runOne)` 里 runOne 自带 try/catch 的用法一致)。
+ */
+export async function runPool<T>(
+  items: readonly T[],
+  concurrency: number,
+  fn: (item: T, index: number) => Promise<void>,
+): Promise<void> {
+  const n = Math.max(1, Math.min(Math.floor(concurrency) || 1, items.length));
+  if (items.length === 0) return;
+  let cursor = 0;
+  await Promise.all(
+    Array.from({ length: n }, async () => {
+      for (;;) {
+        const idx = cursor++;
+        if (idx >= items.length) return;
+        await fn(items[idx]!, idx);
+      }
+    }),
+  );
+}

apps/pac-service/src/modules/persona/persona.service.ts

@@ -230,25 +230,36 @@ export class PersonaService {
       };
     }
 
-    const log = await this.prisma.personaRecomputeLog.create({
-      data: {
-        hostId: patient.hostId,
-        tenantId: patient.tenantId,
-        patientId: patient.id,
-        triggeredBy: input.source,
-        status: 'running',
-      },
-    });
+    // ⭐ 审计日志改「一次终态写」—— 原来是 create(running) + update(终态) 两次往返。
+    //   startedAt 可显式赋值,所以耗时统计一字不差(见下方两处 create)。
+    //   代价:进程被硬杀(OOM / SIGKILL)时这一条不会留痕。评估过可接受 ——
+    //     · try/catch 里的异常仍会写 'failed' 行,真正的算法错误照样有记录;
+    //     · 唯一丢的是"算到一半进程没了",而那种情况原本留下的是一条**永远卡在 running**
+    //       的行(没人清理),本身就是噪声;批量回填另有 CLI 进度日志兜底。
+    //   读侧确认无依赖:daily-health-report 只用 findFirst(最近一条 startedAt/status)
+    //     与按 status 的 groupBy 日统计,都不需要 running 行。
+    //   全量回填每患者省 1 次往返(11.7 → 10.7),且少一半日志表写入(35.6 万 → 17.8 万行/轮)。
+    const logBase = {
+      hostId: patient.hostId,
+      tenantId: patient.tenantId,
+      patientId: patient.id,
+      triggeredBy: input.source,
+      startedAt,
+    };
 
     try {
-      const facts = await this.prisma.patientFact.findMany({
+      // ⭐ 一次取,内存分两份 —— 原来对 patient_facts 查两遍(同一患者、同一索引、两次往返):
+      //     ① status IN (active, fulfilled)                    → factsByType
+      //     ② type=appointment_record AND status != superseded → appointmentsAll
+      //   ② 是 ① 的超集在 appointment 这一类上的放宽(多了 cancelled / no_show 等状态),
+      //   所以取 `status != 'superseded'` 一把捞回来,再在内存里切成两份,语义逐字节等价。
+      //   (全量回填时每患者省 1 次往返;12.7 → 11.7)
+      const allFacts = await this.prisma.patientFact.findMany({
         where: {
           hostId: patient.hostId,
           tenantId: patient.tenantId,
           patientId: patient.id,
-          // 跟 chain-composer / scenario SQL 对齐:actual treatment 完成后 status='fulfilled',
-          // 不能只取 'active' 否则已完成治疗在 persona 看不到 → treatment_chain_status 误算 gap
-          status: { in: ['active', 'fulfilled'] },
+          status: { not: 'superseded' },
         },
         select: {
           id: true,
@@ -263,11 +274,19 @@ export class PersonaService {
         },
       });
 
+      // factsByType 口径不变:只收 active / fulfilled。
+      // 跟 chain-composer / scenario SQL 对齐:actual treatment 完成后 status='fulfilled',
+      // 不能只取 'active' 否则已完成治疗在 persona 看不到 → treatment_chain_status 误算 gap
       const factsByType = new Map<string, ActiveFact[]>();
-      for (const f of facts) {
-        const arr = factsByType.get(f.type) ?? [];
-        arr.push(f as ActiveFact);
-        factsByType.set(f.type, arr);
+      // 全部 appointment(各状态,排 superseded)— 特别关注爽约/迟到用(no_show 预约 status≠active)
+      const appointmentsAll: ActiveFact[] = [];
+      for (const f of allFacts) {
+        if (f.status === 'active' || f.status === 'fulfilled') {
+          const arr = factsByType.get(f.type) ?? [];
+          arr.push(f as ActiveFact);
+          factsByType.set(f.type, arr);
+        }
+        if (f.type === 'appointment_record') appointmentsAll.push(f as ActiveFact);
       }
 
       // 关系边(家庭构成等用)— relatedPatientId 是否解析不影响,只看 relationship 类型
@@ -276,21 +295,6 @@ export class PersonaService {
         select: { relationship: true },
       });
 
-      // 全部 appointment(各状态,排 superseded)— 特别关注爽约/迟到用(no_show 预约 status≠active)
-      const appointmentsAll = (await this.prisma.patientFact.findMany({
-        where: {
-          hostId: patient.hostId,
-          tenantId: patient.tenantId,
-          patientId: patient.id,
-          type: 'appointment_record',
-          status: { not: 'superseded' },
-        },
-        select: {
-          id: true, subjectId: true, type: true, kind: true, status: true,
-          occurredAt: true, plannedFor: true, clinicId: true, content: true,
-        },
-      })) as ActiveFact[];
-
       // 潜在治疗 gap(诊断了/建议了但没启动对应治疗)— 复用召回 gap 核心,去时间门(常态属性)。
       //   按患者 active 诊断/建议码剪枝:无相关码 → 不查 SQL。
       const activeCodes = new Set<string>();
@@ -461,9 +465,9 @@ export class PersonaService {
         status = failed === 0 ? 'success' : drafts.length > 0 ? 'partial' : 'failed';
       }
 
-      await this.prisma.personaRecomputeLog.update({
-        where: { id: log.id },
+      await this.prisma.personaRecomputeLog.create({
         data: {
+          ...logBase,
           status,
           producedPersonaId: personaId,
           featureCount: drafts.length,
@@ -485,9 +489,9 @@ export class PersonaService {
         durationMs: Date.now() - startedAt.getTime(),
       };
     } catch (err) {
-      await this.prisma.personaRecomputeLog.update({
-        where: { id: log.id },
+      await this.prisma.personaRecomputeLog.create({
         data: {
+          ...logBase,
           status: 'failed',
           errorMessage: err instanceof Error ? err.message : String(err),
           endedAt: new Date(),

apps/pac-service/src/prisma/prisma.service.ts

@@ -3,16 +3,23 @@ import { PrismaClient } from '@prisma/client';
 import { tenantGuardExtension } from './tenant-guard.extension';
 
 /**
- * 连接池随 cohort 并发自动放大 —— 免去手动给 DATABASE_URL 追加 `?connection_limit=`。
+ * 连接池随**本进程声明的并发**自动放大 —— 免去手动给 DATABASE_URL 追加 `?connection_limit=`。
  *
- * 病根:并发旋钮是 PAC_COHORT_CONCURRENCY(N 个 cohort runner 共享一个池),
- * 但 Prisma 默认池大小是 `核数×2+1`,与 N 无关 → 一调并发就得手动调池。这里把两者联动:
+ * 病根:Prisma 默认池大小是 `核数×2+1`,跟业务并发旋钮完全无关 → 一调并发就得手动调池。
+ * 这里把两者联动:
  *   - 未设 / =1(常驻 API 进程)→ 不改 URL,走 Prisma 默认池;
- *   - >1(摄入是独立 process)→ connection_limit = N×6+5,封顶 40(守住 Postgres max_connections=100);
+ *   - >1(各 CLI 是独立 process)→ connection_limit = N×6+5,封顶 40(守住 Postgres max_connections=100);
  *   - URL 里已显式写了 connection_limit → 尊重,不覆盖(逃生口)。
- * 摄入是独立进程,只有它拿大池,常驻服务仍是小池,天然隔离。
+ * CLI 是独立进程,只有它拿大池,常驻服务仍是小池,天然隔离。
+ *
+ * ⭐ 2026-07-26 修:原来只认 `PAC_COHORT_CONCURRENCY`(摄入的旋钮),于是
+ * `recompute-persona --concurrency=N` 这类 CLI 的旋钮**根本没放大池** —— 生产实测
+ * 跑 `--concurrency=8` 时进程恰好只有 9 条连接(= 4核×2+1 的 Prisma 默认值),
+ * 也就是说 `--concurrency` 一旦超过 9 就是个静默失效的假旋钮,多出来的 worker 全卡在池上排队。
+ * 现在改成读**通用**的 PAC_DB_CONCURRENCY,由各 CLI 在 Nest 启动前按自己的 --concurrency 设进来;
+ * PAC_COHORT_CONCURRENCY 保留为向后兼容的别名,两者取大。
  */
-function withCohortDerivedPool(rawUrl: string | undefined): string | undefined {
+export function withCohortDerivedPool(rawUrl: string | undefined): string | undefined {
   if (!rawUrl) return rawUrl;
   let url: URL;
   try {
@@ -21,7 +28,8 @@ function withCohortDerivedPool(rawUrl: string | undefined): string | undefined {
     return rawUrl; // 非标准 URL(如 unix socket)— 交给 Prisma 原样处理
   }
   if (url.searchParams.has('connection_limit')) return rawUrl;
-  const n = Math.max(1, parseInt(process.env.PAC_COHORT_CONCURRENCY ?? '1', 10) || 1);
+  const num = (v: string | undefined) => Math.max(1, parseInt(v ?? '1', 10) || 1);
+  const n = Math.max(num(process.env.PAC_DB_CONCURRENCY), num(process.env.PAC_COHORT_CONCURRENCY));
   if (n <= 1) return rawUrl;
   url.searchParams.set('connection_limit', String(Math.min(n * 6 + 5, 40)));
   return url.toString();

apps/pac-service/tests/persona-recompute-perf.spec.ts

+import { runPool } from '../src/common/run-pool';
+import { withCohortDerivedPool } from '../src/prisma/prisma.service';
+
+/**
+ * 画像重算的两个吞吐修复(2026-07-26,依据测试服 12 万患者实测):
+ *
+ *   基线  p50 139ms / p90 230ms / p99 386ms / max 75.7s,均值 165ms
+ *         concurrency=8 理论上限 8÷0.165 = 2,909 人/分,实际 ~1,600 → **效率 55%**
+ *
+ *   ① 批次栅栏 → worker pool:缺口正好是 E[8 次抽样最大值](≈p90)与均值之比;
+ *      那个 75.7 秒的患者会把同批 7 个 worker 一起冻住。
+ *   ② --concurrency 没放大连接池:实测跑 --concurrency=8 时进程恰好 9 条连接
+ *      (= 4核×2+1 的 Prisma 默认值),超过 9 的并发全卡在池上排队,旋钮静默失效。
+ */
+
+describe('runPool —— 连续调度,不做批次栅栏', () => {
+  test('全部跑完,且顺序不丢项', async () => {
+    const items = Array.from({ length: 50 }, (_, i) => i);
+    const seen: number[] = [];
+    await runPool(items, 7, async (x) => { seen.push(x); });
+    expect(seen.sort((a, b) => a - b)).toEqual(items);
+  });
+
+  test('并发上限严格 ≤ concurrency', async () => {
+    let inFlight = 0;
+    let peak = 0;
+    await runPool(Array.from({ length: 40 }, (_, i) => i), 5, async () => {
+      peak = Math.max(peak, ++inFlight);
+      await new Promise((r) => setTimeout(r, 5));
+      inFlight--;
+    });
+    expect(peak).toBeLessThanOrEqual(5);
+    expect(peak).toBe(5); // 也要真的用满,不是退化成串行
+  });
+
+  test('⭐ 一个慢任务只拖住自己那条线,不冻住同批其他 worker', async () => {
+    // 4 个任务:第 0 个慢 200ms,其余 3 个各 10ms;并发 2。
+    //   栅栏写法:[0,1] 一批 → 等 200ms;[2,3] 一批 → 10ms;总 ≈210ms
+    //   worker pool:worker A 卡在 0 上,worker B 依次做完 1、2、3;总 ≈200ms 且 1/2/3 早早完成
+    const done: Array<{ id: number; at: number }> = [];
+    const t0 = Date.now();
+    await runPool([0, 1, 2, 3], 2, async (x) => {
+      await new Promise((r) => setTimeout(r, x === 0 ? 200 : 10));
+      done.push({ id: x, at: Date.now() - t0 });
+    });
+    const slow = done.find((d) => d.id === 0)!;
+    const fastest3 = done.filter((d) => d.id !== 0);
+    expect(fastest3).toHaveLength(3);
+    // 三个快任务必须在慢任务之前就都做完了 —— 栅栏写法下 2、3 只能等慢任务结束才开始
+    for (const f of fastest3) expect(f.at).toBeLessThan(slow.at);
+  });
+
+  test('concurrency 大于任务数 / 空列表 都不炸', async () => {
+    const seen: number[] = [];
+    await runPool([1, 2], 99, async (x) => { seen.push(x); });
+    expect(seen.sort()).toEqual([1, 2]);
+    await expect(runPool([], 8, async () => { throw new Error('不该被调用'); })).resolves.toBeUndefined();
+  });
+
+  test('fn 抛错会向上冒(调用方各自 try/catch,跟原 .map(runOne) 用法一致)', async () => {
+    await expect(runPool([1], 2, async () => { throw new Error('boom'); })).rejects.toThrow('boom');
+  });
+});
+
+describe('连接池随并发放大 —— PAC_DB_CONCURRENCY', () => {
+  const URL_ = 'postgresql://u:p@db:5432/pac?schema=public';
+  const saved = { db: process.env.PAC_DB_CONCURRENCY, cohort: process.env.PAC_COHORT_CONCURRENCY };
+  const set = (db?: string, cohort?: string) => {
+    if (db === undefined) delete process.env.PAC_DB_CONCURRENCY; else process.env.PAC_DB_CONCURRENCY = db;
+    if (cohort === undefined) delete process.env.PAC_COHORT_CONCURRENCY; else process.env.PAC_COHORT_CONCURRENCY = cohort;
+  };
+  afterAll(() => set(saved.db, saved.cohort));
+
+  const limitOf = (u: string | undefined) =>
+    u ? new URL(u).searchParams.get('connection_limit') : null;
+  /// 口径与实现同源:n×6+5,封顶 40
+  const expected = (n: number) => String(Math.min(n * 6 + 5, 40));
+
+  test('⭐ 修的正是这条:PAC_DB_CONCURRENCY 要能放大池(原来只认 COHORT,CLI 旋钮静默失效)', () => {
+    set('4', undefined);
+    expect(limitOf(withCohortDerivedPool(URL_))).toBe(expected(4)); // 29
+    set('8', undefined);
+    expect(limitOf(withCohortDerivedPool(URL_))).toBe(expected(8)); // 53 → 封顶 40
+  });
+
+  test('向后兼容:PAC_COHORT_CONCURRENCY 仍然有效(摄入路径不能被改坏)', () => {
+    set(undefined, '4');
+    expect(limitOf(withCohortDerivedPool(URL_))).toBe(expected(4));
+  });
+
+  test('两者都设 → 取大', () => {
+    set('2', '5');
+    expect(limitOf(withCohortDerivedPool(URL_))).toBe(expected(5)); // 35,不是 17
+  });
+
+  test('封顶 40(守住 Postgres max_connections=100)', () => {
+    set('64', undefined);
+    expect(limitOf(withCohortDerivedPool(URL_))).toBe('40');
+  });
+
+  test('都没设 / =1(常驻 API 进程)→ 不动 URL,走 Prisma 默认池', () => {
+    set(undefined, undefined);
+    expect(withCohortDerivedPool(URL_)).toBe(URL_);
+    set('1', '1');
+    expect(withCohortDerivedPool(URL_)).toBe(URL_);
+  });
+
+  test('URL 里已显式写了 connection_limit → 尊重,不覆盖(逃生口)', () => {
+    set('16', undefined);
+    const explicit = 'postgresql://u:p@db:5432/pac?connection_limit=7';
+    expect(withCohortDerivedPool(explicit)).toBe(explicit);
+  });
+
+  test('非标准 URL(unix socket 等)原样返回,不抛', () => {
+    set('8', undefined);
+    expect(withCohortDerivedPool('not-a-url')).toBe('not-a-url');
+    expect(withCohortDerivedPool(undefined)).toBeUndefined();
+  });
+});