merge: perf/persona-recompute → test(shm 1g + 画像版本号并发竞态修复)

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

@@ -400,6 +400,9 @@ export class PersonaService {
         status = verdict === 'unchanged' ? 'unchanged' : 'refreshed';
         await this.prisma.$transaction(
           async (tx) => {
+            // 同下面建版本那支拿同一把患者级 advisory 锁:否则并发写者可能正在 supersede
+            // 这一行 / 建新版本,我们的就地刷新就写到了一个刚变成历史的版本上。
+            await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtext(${patient.id}))`;
             if (verdict === 'volatile') {
               // key 集合与语义指纹都已确认相同 → 按 (personaId,key) 逐条就地改,保住行 id。
               for (const d of drafts) {
@@ -424,12 +427,29 @@ export class PersonaService {
           { maxWait: 30000, timeout: 60000 },
         );
       } else {
-        const nextVersion = (latest?.version ?? 0) + 1;
+        // ⭐ 版本号必须在**锁内重读**,不能用事务外那次 findFirst 的结果。
+        //   2026-07-27 测试服实测:CLI 全量回填跨过了 02:00 的 stale-scan cron,两个进程
+        //   同时给同一患者建新版本 —— 各自在事务外读到同一个 latest.version、算出同一个
+        //   nextVersion → `Unique constraint failed on (patient_id, version)`。
+        //   那一轮 CLI 挂 688 个、定时任务自己挂 1,604 个。
+        //   BullMQ 内部按 patient_id 串行挡得住自己人,挡不住外面另起的 CLI 进程。
+        //   修法:进事务先拿**按患者的 advisory 事务锁**(随事务自动释放,跨进程有效),
+        //   再重读当前最高版本 —— 迟到的那个会看到已经变大的版本号,顺序叠上去,不再冲突。
+        let nextVersion = (latest?.version ?? 0) + 1;
         const created = await this.prisma.$transaction(
           async (tx) => {
-            if (active) {
+            await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtext(${patient.id}))`;
+            // 锁内重读:并发写者可能已经把版本推高,也可能已经把我们看到的 active 换掉了
+            const fresh = await tx.persona.findFirst({
+              where: { patientId: patient.id },
+              orderBy: { version: 'desc' },
+              select: { id: true, version: true, supersededAt: true },
+            });
+            nextVersion = (fresh?.version ?? 0) + 1;
+            const freshActive = fresh && !fresh.supersededAt ? fresh : null;
+            if (freshActive) {
               await tx.persona.update({
-                where: { id: active.id },
+                where: { id: freshActive.id },
                 data: { supersededAt: new Date() },
               });
             }

apps/pac-service/tests/persona-watermark.spec.ts

@@ -76,9 +76,25 @@ function makeHarness(o: Opts) {
     featureUpdate: [] as Record<string, unknown>[],
     watermarkQueries: 0,
     txnFindFirst: 0,
+    /// 事务内拿的患者级 advisory 锁(防 CLI 与 stale-scan cron 抢版本号,见 persona.service)
+    advisoryLocks: [] as string[],
+    /// 事务内重读最高版本的次数 —— 版本号必须锁内重读,不能用事务外那次
+    versionRereads: 0,
   };
   const tx = {
+    $executeRaw: jest.fn().mockImplementation((strings: TemplateStringsArray) => {
+      calls.advisoryLocks.push(strings.join('?'));
+      return Promise.resolve(1);
+    }),
     persona: {
+      findFirst: jest.fn().mockImplementation(() => {
+        calls.versionRereads++;
+        return Promise.resolve(
+          o.persona
+            ? { id: 'persona-old', version: 3, supersededAt: null, ...o.persona }
+            : null,
+        );
+      }),
       update: jest.fn().mockImplementation(({ data }) => {
         calls.personaUpdate.push(data);
         return Promise.resolve({});
@@ -293,3 +309,66 @@ describe('查询瘦身', () => {
     expect(calls.txnFindFirst).toBe(0); // 旧实现这里会再查一次 latest event_seq
   });
 });
+
+/**
+ * 并发写版本号 —— 2026-07-27 测试服实测事故的回归。
+ *
+ * CLI 全量回填(concurrency=8)跨过了 02:00 的 stale-scan cron,两个**独立进程**同时给
+ * 同一批患者建新版本:各自在事务外读到同一个 latest.version、算出同一个 nextVersion →
+ *   `Unique constraint failed on the fields: (patient_id, version)`
+ * 那一轮 CLI 挂 688 个、定时任务自己挂 1,604 个(它伤得更重)。
+ *
+ * BullMQ processor 内部按 patient_id 串行,挡得住自己人,挡不住外面另起的 CLI 进程。
+ * 修法:两条写路径进事务先拿**按患者的 advisory 事务锁**(pg_advisory_xact_lock,
+ * 随事务自动释放、跨进程有效),建版本那支再**在锁内重读**最高版本 —— 迟到者看到
+ * 已经变大的版本号,顺序叠上去,不再撞唯一约束。
+ */
+describe('并发安全 —— CLI 与 cron 抢版本号', () => {
+  test('⭐ 建新版本前必须拿患者级 advisory 锁', async () => {
+    const { svc, calls } = await build({
+      persona: { eventWatermark: 100n, factWatermark: T0 },
+      storedFeatures: [{ key: 'gender', description: '男性', score: null, data: null, evidence: {} }],
+      now: { eventSeq: 200n, factAt: T1 }, // 水位前进 → 走建版本那支
+    });
+    await svc.recompute({ patientId: 'pat-1', source: 'test' });
+    expect(calls.personaCreate).toBe(1);
+    expect(calls.advisoryLocks).toHaveLength(1);
+    expect(calls.advisoryLocks[0]).toContain('pg_advisory_xact_lock');
+  });
+
+  test('⭐ 版本号在**锁内重读**,不能用事务外那次查询的结果', async () => {
+    const { svc, calls } = await build({
+      persona: { eventWatermark: 100n, factWatermark: T0 },
+      storedFeatures: [{ key: 'gender', description: '男性', score: null, data: null, evidence: {} }],
+      now: { eventSeq: 200n, factAt: T1 },
+    });
+    await svc.recompute({ patientId: 'pat-1', source: 'test' });
+    expect(calls.versionRereads).toBe(1); // 锁拿到之后又查了一次最高版本
+  });
+
+  test('⭐ 就地刷新那支也要拿同一把锁(否则会写到刚被 supersede 的版本上)', async () => {
+    const { svc, calls } = await build({
+      persona: { eventWatermark: 100n, factWatermark: T0 },
+      now: { eventSeq: 100n, factAt: T1 },
+      // 只有天数漂移 → volatile → 就地刷新,不升版本(口径同上面 refreshed 那条)
+      storedFeatures: storedFrom(RFM()),
+      draft: RFM({
+        description: '重要价值 · 距上次 151 天 · 累计¥154,350',
+        data: { segment: 'important_value', recencyDays: 151, monetaryCents: 15435000 },
+      }),
+    });
+    await svc.recompute({ patientId: 'pat-1', source: 'test' });
+    expect(calls.personaCreate).toBe(0); // 确认走的是就地刷新
+    expect(calls.advisoryLocks).toHaveLength(1);
+    expect(calls.advisoryLocks[0]).toContain('pg_advisory_xact_lock');
+  });
+
+  test('noop(闸拦下)不该白拿锁 —— 压根没进事务', async () => {
+    const { svc, calls } = await build({
+      persona: { eventWatermark: 200n, factWatermark: T1 },
+      now: { eventSeq: 200n, factAt: T1 }, // 水位没动 → noop
+    });
+    await svc.recompute({ patientId: 'pat-1', source: 'test' });
+    expect(calls.advisoryLocks).toHaveLength(0);
+  });
+});

docker-compose.prod.yml

@@ -22,6 +22,13 @@ services:
     image: postgres:16-alpine
     restart: always
     env_file: ./apps/pac-service/.env   # 读 POSTGRES_USER/PASSWORD/DB
+    # ⭐ Docker 给 /dev/shm 的默认只有 64MB,PG 的并行查询 worker 用共享内存段交换中间结果,
+    #   一超就是 `could not resize shared memory segment ... No space left on device`。
+    #   2026-07-27 测试服实测踩到:召回 selectHits 把 10 个子场景 Promise.all 并行跑,
+    #   每个都是 2500 万 patient_facts 上带 LATERAL 的重查询、各自还起 parallel worker,
+    #   并发的段远超 64MB → plan 全量重算当场崩。
+    #   1g 是 PG 容器的常规配法(官方镜像文档亦如此建议);托管 DB 形态下本服务不启用,无影响。
+    shm_size: 1g
     volumes:
       - postgres_data:/var/lib/postgresql/data
     ports: