perf(persona): recompute CLI 加 --concurrency(默认1)—— I/O-bound,并发 ~4.5x

实测推翻旧注记"persona 重算 CPU-bound、异步并发不提速":每患者 recompute 含 ~15 次
串行 DB 往返(2 次全量 fact 读 + 多次 findFirst),实为 **I/O-bound**。单进程分块
Promise.all 并发可重叠这些 DB 等待 —— 本地 500 患者 17s(串行)→ 4s(并发16),~4.5x,
concurrency 8-16 后 plateau(DB 成瓶颈)。正确性验证:500/500 各 1 active persona、
0 重复版本、0 空特征(不同患者独立写 + 同患者不并发,无竞争)。

- --concurrency=N:分块并发(默认 1 向后兼容)
- --shard=i/k:多进程分片(按患者序号取模,吃多核;DB 先饱和时用)
- 大批(>500)只打进度行不逐条刷屏;orderBy id 稳定序供分片

全量重算 131689 患者:7h(串行)→ ~1.6h(并发12)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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

@@ -19,16 +19,24 @@ interface Args {
   pid?: string;
   pids?: string[]; // 多个 externalId(逗号分隔),定向重算受影响子集用
   force?: boolean; // 跳过水位幂等闸:算法/特征变更后(数据没变)也强制重算
+  concurrency: number; // 并发度:每患者 recompute 含 ~15 次 DB 往返,并发可重叠 I/O 等待
+  shard?: { i: number; k: number }; // 分片 i/k:多进程并行(吃多核),按患者序号取模
 }
 
 function parseArgs(argv: string[]): Args {
-  const args: Args = { host: 'demo' };
+  const args: Args = { host: 'demo', concurrency: 1 };
   for (const a of argv) {
     if (a.startsWith('--host=')) args.host = a.slice('--host='.length);
     else if (a.startsWith('--pid=')) args.pid = a.slice('--pid='.length);
     else if (a === '--force') args.force = true;
     else if (a.startsWith('--pids=')) {
       args.pids = a.slice('--pids='.length).split(',').map((s) => s.trim()).filter(Boolean);
+    } else if (a.startsWith('--concurrency=')) {
+      const n = parseInt(a.slice('--concurrency='.length), 10);
+      if (Number.isFinite(n) && n >= 1) args.concurrency = n;
+    } else if (a.startsWith('--shard=')) {
+      const m = a.slice('--shard='.length).match(/^(\d+)\/(\d+)$/);
+      if (m) args.shard = { i: Number(m[1]), k: Number(m[2]) };
     }
   }
   return args;
@@ -48,31 +56,37 @@ async function bootstrap() {
     const host = await prisma.host.findUnique({ where: { name: args.host } });
     if (!host) throw new Error(`Host '${args.host}' not found`);
 
-    const patients = await prisma.patient.findMany({
+    let patients = await prisma.patient.findMany({
       where: {
         hostId: host.id,
         active: true,
         ...(args.pids?.length ? { externalId: { in: args.pids } } : args.pid ? { externalId: args.pid } : {}),
       },
       select: { id: true, externalId: true, name: true },
+      orderBy: { id: 'asc' }, // 稳定序,供分片按下标取模(各进程不重不漏)
     });
+    if (args.shard) {
+      const { i, k } = args.shard;
+      patients = patients.filter((_, idx) => idx % k === i);
+    }
 
-    logger.log(`Recomputing ${patients.length} patients for host=${args.host} ...`);
+    const total = patients.length;
+    logger.log(
+      `Recomputing ${total} patients for host=${args.host} ` +
+        `(concurrency=${args.concurrency}${args.shard ? `, shard=${args.shard.i}/${args.shard.k}` : ''}) ...`,
+    );
 
-    // noop = 水位闸命中(自上次重算无新事件,沿用旧 persona),是正常结果,
-    // 不能并入 failed —— 否则全量重跑(无新增事实)会误报"全失败"。
-    let ok = 0, partial = 0, noop = 0, failed = 0;
-    // 串行:persona 重算 Node-CPU-bound(特征计算占主),异步并发不提速(见 scheduler 注释)。
-    for (const p of patients) {
+    // noop = 水位闸命中(自上次重算无新事件,沿用旧 persona),是正常结果,不并入 failed。
+    let ok = 0, partial = 0, noop = 0, failed = 0, done = 0;
+    const verbose = total <= 500; // 小批逐条日志;大批只打进度,免刷屏
+    const runOne = async (p: { id: string; externalId: string; name: string | null }) => {
       try {
-        const r = await svc.recompute({
-          patientId: p.id,
-          source: `manual:cli`,
-          force: args.force,
-        });
+        const r = await svc.recompute({ patientId: p.id, source: `manual:cli`, force: args.force });
+        if (verbose) {
           logger.log(
             `  ${p.externalId.padEnd(6)} ${(p.name ?? '').padEnd(12)} v${r.version} features=${r.featureCount} status=${r.status} (${r.durationMs}ms)`,
           );
+        }
         if (r.status === 'success') ok++;
         else if (r.status === 'partial') partial++;
         else if (r.status === 'noop') noop++;
@@ -81,6 +95,13 @@ async function bootstrap() {
         failed++;
         logger.error(`  ${p.externalId} FAILED: ${err instanceof Error ? err.message : err}`);
       }
+      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));
     }
 
     logger.log('─────────────────────────────────');