perf(plan): 召回批量写也去掉批次栅栏 —— 与画像同一处改法

plan-engine.runAllForHost 的 upsert 段用的是同一种写法:
  for (i += N) await Promise.all(entries.slice(i, i+N).map(...))
每批都要等本批**最慢**那个,快的 worker 干等着。

同一处改法已在 recompute-persona 上实测(测试服 35.7 万患者,concurrency=8):
  并发效率 44% → 94%,吞吐 1,454 → 2,694 人/分
plan 侧单患者耗时的离散度只会更大(upsertPlan 内 reason 数量差异明显),浪费同理。

改:换成 common/run-pool.ts 的连续调度 worker pool(N 个 worker 各自取下一个)。
并发上限、错误处理、计数口径都不变 —— 计数仍靠 JS 单线程 await 恢复后同步自增。

顺带:recompute-plans.cli 在 NestFactory 之前把 PAC_DB_CONCURRENCY 对齐到
PAC_PLAN_BATCH_CONCURRENCY(默认 8)。同 recompute-persona 的理由:PrismaService 在容器
初始化时就定池,不设的话是 Prisma 默认「核数×2+1」(生产 4 核 = 9),把 batch 并发调到
9 以上会被卡住。按默认 8 跑时 9 条刚好够,本项属于**解锁更高并发**,当下不产生收益。

## 验证

- 494 单测通过(36 suites),tsc 干净
- 本地 friday host 全量重算(10,458 plan):15.94s,plansCreated=0,
  跑前跑后 followup_plans 指纹均为 3d5225bf6c1160226739385c6f8e7d8b —— **行为中性**

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

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

@@ -31,6 +31,14 @@ function parseArgs(argv: string[]): Args {
 async function bootstrap() {
   const args = parseArgs(process.argv.slice(2));
   const logger = new Logger('recompute-plans');
+  // 连接池对齐本进程的写并发(PlanEngine 的 upsert 段用 PAC_PLAN_BATCH_CONCURRENCY,默认 8)。
+  // 必须在 NestFactory 之前设 —— PrismaService 在容器初始化时就定池,晚了不生效。
+  // 不设的话池是 Prisma 默认「核数×2+1」(生产 4 核 = 9),把 batch 并发调到 9 以上就会被卡住。
+  // 详见 prisma.service.ts 的 withCohortDerivedPool。
+  if (!process.env.PAC_DB_CONCURRENCY) {
+    const n = Math.max(1, Number(process.env.PAC_PLAN_BATCH_CONCURRENCY) || 8);
+    if (n > 1) process.env.PAC_DB_CONCURRENCY = String(n);
+  }
   const app = await NestFactory.createApplicationContext(AppModule, {
     logger: ['log', 'warn', 'error'],
   });

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

@@ -4,6 +4,7 @@ import { PrismaService } from '../../../prisma/prisma.service';
 import { TreatmentInitiationRecallScenario } from './scenarios/treatment-initiation-recall.scenario';
 import type { PlanScenarioPlugin, ScenarioHit, ScenarioScope } from './scenario.interface';
 import { reasonNeedsRefresh } from './reason-refresh';
+import { runPool } from '../../../common/run-pool';
 
 /**
  * PlanEngineService — 跑所有 scenario plugin,产 FollowupPlan + PlanReason
@@ -197,11 +198,14 @@ export class PlanEngineService {
       await this.prefetchForBatch(scope, patientIds, 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()];
-    for (let i = 0; i < entries.length; i += concurrency) {
-      await Promise.all(
-        entries.slice(i, i + concurrency).map(async ([patientId, hits]) => {
+    await runPool(entries, concurrency, async ([patientId, hits]) => {
       try {
         const result = await this.upsertPlan({
           scope,
@@ -246,9 +250,7 @@ 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) });
     }