Skip to content

P
pac
Commit 887069e5 by luoqi
Options

merge: perf/persona-recompute → main(画像重算吞吐优化 + shm 1g + 版本号并发竞态修复)

parents 0165bc11 2a02d9d1
Showing with 348 additions and 50 deletions
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,14 @@ 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 超过 8 就会被卡住,
// 多出来的 worker 全在排队等连接。详见 prisma.service.ts 里 withCohortDerivedPool 的注释
// (含两台机器的实际 .env 差异;测试服 URL 里显式写了 connection_limit=30,那里不受影响)。
// 只在用户显式给了 --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 +149,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/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/common/run-pool.ts 0 → 100644
/**
* 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: {
// ⭐ 审计日志改「一次终态写」—— 原来是 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,
status: 'running',
},
});
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,12 +274,20 @@ 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) {
// 全部 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 类型
const relations = await this.prisma.patientRelation.findMany({
......@@ -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>();
......@@ -396,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) {
......@@ -420,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() },
});
}
......@@ -461,9 +485,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 +509,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/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) });
}
......
apps/pac-service/src/prisma/prisma.service.ts
......@@ -3,16 +3,29 @@ 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 的旋钮**不会放大池**。
*
* ⚠️ 这是**潜在**缺陷,咬不咬人取决于该机 .env 有没有显式写 connection_limit(核对过两台):
* 测试服 47.251.104.47 URL 里显式 `connection_limit=30` → 本函数提前返回,池=30,不是瓶颈
* (那里观测到的 9 条连接 = 8 个 worker + 1 空闲,不是池上限 —— 别再据此推断)
* 生产 47.99.62.30 URL **没写** → 走 Prisma 默认 `核数×2+1`,4 核机 = **9**
* → `--concurrency` 超过 8 就会被卡在 9,多出来的 worker 排队等连接
* 也就是说:按 --concurrency=8 跑生产,9 条刚好够,本修复不产生收益;它的价值是**解锁更高并发**。
*
* 改成读**通用**的 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 +34,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 0 → 100644
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 没放大连接池 —— **潜在**缺陷,咬不咬人看该机 .env:
* 测试服 URL 显式 `connection_limit=30` → 不受影响
* 生产 URL 没写 → Prisma 默认 核数×2+1,4 核机 = 9 → --concurrency>8 会被卡住
* 所以按 --concurrency=8 跑生产时本修复无收益,价值在于解锁更高并发。
*/
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();
});
});
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:
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment