merge: perf/persona-recompute → test(plan 侧也去掉批次栅栏)

⚠️ 只合代码,**先不部署** —— 测试服正在跑画像回填(优化后代码),部署会重建容器把它杀掉。
等回填 + 链上的 plan 全量重算跑完再部署。

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,58 +198,59 @@ 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]) => {
-          try {
-            const result = await this.upsertPlan({
-              scope,
-              patientId,
-              hits,
-              prefetched: {
-                latest: latestByPatient.get(patientId) ?? null,
-                snoozedAnchors: snoozedByPatient.get(patientId) ?? EMPTY_SNOOZE,
-                personaId: personaByPatient.get(patientId) ?? null,
-                lastVisitClinicId: lastVisitClinicByPatient.get(patientId) ?? null,
-              },
-            });
-            // 计数:JS 单线程,await 恢复后同步自增,chunk 内并发无交错
-            if (result === 'created') plansCreated++;
-            else if (result === 'superseded') plansSuperseded++;
-            else if (result === 'unchanged') plansUnchanged++;
-            else if (result === 'suppressed') plansSuppressed++;
-            logRows.push({
-              hostId: scope.hostId,
-              tenantId: scope.tenantId,
-              patientId,
-              triggeredBy: 'cli:batch',
-              status: 'success',
-              scenarioRunCount: this.scenarios.length,
-              plansCreated: result === 'created' || result === 'superseded' ? 1 : 0,
-              startedAt,
-              endedAt: new Date(),
-            });
-          } catch (err) {
-            logRows.push({
-              hostId: scope.hostId,
-              tenantId: scope.tenantId,
-              patientId,
-              triggeredBy: 'cli:batch',
-              status: 'failed',
-              scenarioRunCount: this.scenarios.length,
-              errorMessage: err instanceof Error ? err.message : String(err),
-              startedAt,
-              endedAt: new Date(),
-            });
-            this.logger.error(
-              `plan upsert failed patient=${patientId}: ${err instanceof Error ? err.message : err}`,
-            );
-          }
-        }),
-      );
-    }
+    await runPool(entries, concurrency, async ([patientId, hits]) => {
+      try {
+        const result = await this.upsertPlan({
+          scope,
+          patientId,
+          hits,
+          prefetched: {
+            latest: latestByPatient.get(patientId) ?? null,
+            snoozedAnchors: snoozedByPatient.get(patientId) ?? EMPTY_SNOOZE,
+            personaId: personaByPatient.get(patientId) ?? null,
+            lastVisitClinicId: lastVisitClinicByPatient.get(patientId) ?? null,
+          },
+        });
+        // 计数:JS 单线程,await 恢复后同步自增,chunk 内并发无交错
+        if (result === 'created') plansCreated++;
+        else if (result === 'superseded') plansSuperseded++;
+        else if (result === 'unchanged') plansUnchanged++;
+        else if (result === 'suppressed') plansSuppressed++;
+        logRows.push({
+          hostId: scope.hostId,
+          tenantId: scope.tenantId,
+          patientId,
+          triggeredBy: 'cli:batch',
+          status: 'success',
+          scenarioRunCount: this.scenarios.length,
+          plansCreated: result === 'created' || result === 'superseded' ? 1 : 0,
+          startedAt,
+          endedAt: new Date(),
+        });
+      } catch (err) {
+        logRows.push({
+          hostId: scope.hostId,
+          tenantId: scope.tenantId,
+          patientId,
+          triggeredBy: 'cli:batch',
+          status: 'failed',
+          scenarioRunCount: this.scenarios.length,
+          errorMessage: err instanceof Error ? err.message : String(err),
+          startedAt,
+          endedAt: new Date(),
+        });
+        this.logger.error(
+          `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) });
     }