Commit db15fe0e by luoqi

fix(db): 连接池把常驻服务自己的并发旋钮也算进来 —— 每日健康日报连续三天超时的根因

现象:企微收不到每日健康日报。09-01 / 09-02 两天都在 **09:00:10** 失败
(= cron 09:00:00 + 10s pool_timeout),报错是拿不到连接而不是推送失败:

  Invalid `prisma.followupPlan.count()` invocation:
  Timed out fetching a new connection from the connection pool
  (Current connection pool timeout: 10, connection limit: 5)

根因两条叠加:

① **常驻服务的并发旋钮没被算进池。** withCohortDerivedPool 只认 PAC_DB_CONCURRENCY /
   PAC_COHORT_CONCURRENCY(都是 CLI 的旋钮),而 PAC_RECALL_SUBSCENARIO_CONCURRENCY=4
   (2026-08-30 在生产开启,场景段 ×2.31)是**常驻进程**的旋钮 —— 批次因此长期占住 4 条连接,
   池却还是 Prisma 默认。函数开头那句「一调并发就得手动调池,这里把两者联动」说的就是这个病,
   只是当时只联动了 CLI 那半边。

② **生产的默认池只有 5,不是注释里写的 9。** Prisma 默认池按**物理核**×2+1 算:
   生产机 nproc=4 但 `Core(s) per socket`=2 / `Thread(s) per core`=2 → 物理核 2 → 池 = 5。
   原注释按逻辑核算成 9,低估了一倍,实现和测试文件里都跟着错。

5 条连接里批次占 4 条,日报要并发发 6 个 count → 排队 → 10 秒超时。
同一根因还打掉过一条 plan upsert(`Unable to start a transaction in the given time`,
09-01 21:47,549,106 个命中患者里 1 个)。

改动:把 PAC_RECALL_SUBSCENARIO_CONCURRENCY 并入取最大 → 生产池 4×6+5 = 29。
生产 RDS max_connections=820(实测,当时全库仅 23 条在用),29 条毫无压力。

️ 09:00 撞在 08:15 那轮的召回场景段(08:44~09:35)中间是**天天必撞**,不是偶发;
   这里选择扩池而不是挪 cron —— 挪 cron 只躲开这一个碰撞,扩池同时修掉 API/plan 侧的抢连接。

顺带把三处写错或缺失的口径补上:实现注释、.env.example(DATABASE_URL 与并发旋钮的耦合、
显式 connection_limit 会让自动放大整个失效)、以及 conc 读取处的反向指引。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
parent e19135b4
...@@ -42,6 +42,13 @@ POSTGRES_DB=pac ...@@ -42,6 +42,13 @@ POSTGRES_DB=pac
# #
# staging: postgresql://pac:<staging-pwd>@<staging-pg-host>:5532/pac?schema=public # staging: postgresql://pac:<staging-pwd>@<staging-pg-host>:5532/pac?schema=public
# production: postgresql://pac:<prod-pwd>@<prod-pg-host>:5532/pac?schema=public # production: postgresql://pac:<prod-pwd>@<prod-pg-host>:5532/pac?schema=public
#
# ⚠️ 连接池:URL 里**不写** connection_limit 时,PrismaService 会按并发旋钮自动放大
# (PAC_DB_CONCURRENCY / PAC_COHORT_CONCURRENCY / PAC_RECALL_SUBSCENARIO_CONCURRENCY 取最大,
# 池 = N×6+5 封顶 40);全都 ≤1 才落回 Prisma 默认池。
# Prisma 默认池是「**物理核**×2+1」,不是逻辑核 —— 生产 nproc=4 但物理核只有 2,默认只有 **5** 条。
# URL 里一旦显式写了 connection_limit,自动放大**整个失效**(逃生口,写了就得自己负责够用)。
# 2026-09-02 生产事故就出在这:开了召回并发却没算进池,批次占满 5 条,每日健康日报连续三天超时。
DATABASE_URL=postgresql://pac:pac@localhost:5532/pac?schema=public DATABASE_URL=postgresql://pac:pac@localhost:5532/pac?schema=public
# staging+prod: redis://<host>:6479 (BullMQ 队列用,丢了会丢未消费的 plan-asset-generate 任务) # staging+prod: redis://<host>:6479 (BullMQ 队列用,丢了会丢未消费的 plan-asset-generate 任务)
...@@ -214,6 +221,9 @@ SENTRY_RELEASE= ...@@ -214,6 +221,9 @@ SENTRY_RELEASE=
# 瓶颈是**延迟**(逐次索引探查在等 page 返回)不是磁盘吞吐,所以并发 2 就超线性(×1.50)。 # 瓶颈是**延迟**(逐次索引探查在等 page 返回)不是磁盘吞吐,所以并发 2 就超线性(×1.50)。
# ⚠️ 判据只能看 `[plan] 阶段耗时` 的**墙钟** —— 并发下单条查询的 sql= 会被争抢拉长, # ⚠️ 判据只能看 `[plan] 阶段耗时` 的**墙钟** —— 并发下单条查询的 sql= 会被争抢拉长,
# 看单条会误判成"变慢了"。详见 treatment-initiation-recall.scenario.ts 里 conc 处注释。 # 看单条会误判成"变慢了"。详见 treatment-initiation-recall.scenario.ts 里 conc 处注释。
# ⚠️ **这个值会放大数据库连接池**(见上面 DATABASE_URL 处):批次会长时间占住这么多条连接,
# 不放大池的话常驻服务的其他活(每日健康日报、API)就抢不到 —— 2026-09-02 实际发生过。
# URL 里显式写了 connection_limit 的机器不吃这套自动放大,得自己保证池够大。
PAC_RECALL_SUBSCENARIO_CONCURRENCY=4 PAC_RECALL_SUBSCENARIO_CONCURRENCY=4
# gap 计算形态:legacy(默认,逐行相关子查询)/ setbased(集合式)。 # gap 计算形态:legacy(默认,逐行相关子查询)/ setbased(集合式)。
......
...@@ -251,6 +251,10 @@ export class TreatmentInitiationRecallScenario implements PlanScenarioPlugin { ...@@ -251,6 +251,10 @@ export class TreatmentInitiationRecallScenario implements PlanScenarioPlugin {
// ⚠️ **读数注意**:并发下**单条**查询的 sql= 会被争抢拉长(生产 caries 23.8→27.7 分, // ⚠️ **读数注意**:并发下**单条**查询的 sql= 会被争抢拉长(生产 caries 23.8→27.7 分,
// perio 12.6→16.9 分),看单条会误以为变慢了。**判据只能是阶段墙钟**, // perio 12.6→16.9 分),看单条会误以为变慢了。**判据只能是阶段墙钟**,
// 不能看各条耗时之和 —— 上面那次误判有一半就栽在这。 // 不能看各条耗时之和 —— 上面那次误判有一半就栽在这。
// ⚠️ 这个旋钮**同时是连接池的输入** —— 批次会长时间占住 conc 条连接。
// withCohortDerivedPool() 已把它算进池(N×6+5);2026-09-02 之前没算,生产池只有 5,
// 批次占 4 条 → 每日 09:00 健康日报的并发 count 抢不到连接,连续三天超时。
// ⛔ 调大它之前先确认该机 DATABASE_URL 没有显式 connection_limit(写了就绕过自动放大)。
const conc = Math.max(1, Number(process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY) || 1); const conc = Math.max(1, Number(process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY) || 1);
// ⚠️ 不能 hits.push(...subHits):spread 把每个元素当实参压栈,V8 实参上限 ~6.5万; // ⚠️ 不能 hits.push(...subHits):spread 把每个元素当实参压栈,V8 实参上限 ~6.5万;
// host 患者到 ~28 万后单子场景命中可超限 → RangeError: Maximum call stack size exceeded // host 患者到 ~28 万后单子场景命中可超限 → RangeError: Maximum call stack size exceeded
......
...@@ -14,16 +14,24 @@ import { tenantGuardExtension } from './tenant-guard.extension'; ...@@ -14,16 +14,24 @@ import { tenantGuardExtension } from './tenant-guard.extension';
* *
* ⭐ 2026-07-26 修:原来只认 `PAC_COHORT_CONCURRENCY`(**摄入**的旋钮),于是 * ⭐ 2026-07-26 修:原来只认 `PAC_COHORT_CONCURRENCY`(**摄入**的旋钮),于是
* `recompute-persona --concurrency=N` 这类 CLI 的旋钮**不会放大池**。 * `recompute-persona --concurrency=N` 这类 CLI 的旋钮**不会放大池**。
* 改成读**通用**的 PAC_DB_CONCURRENCY,由各 CLI 在 Nest 启动前按自己的 --concurrency 设进来;
* PAC_COHORT_CONCURRENCY 保留为向后兼容的别名,取最大者。
* *
* ⚠️ 这是**潜在**缺陷,咬不咬人取决于该机 .env 有没有显式写 connection_limit(核对过两台): * 🔴 2026-09-02 再修:**常驻服务自己也有并发旋钮,一样没被算进来** —— 生产事故。
* 测试服 47.251.104.47 URL 里显式 `connection_limit=30` → 本函数提前返回,池=30,不是瓶颈 * `PAC_RECALL_SUBSCENARIO_CONCURRENCY=4`(召回子场景并行,2026-08-30 在生产开启)让批次
* (那里观测到的 9 条连接 = 8 个 worker + 1 空闲,不是池上限 —— 别再据此推断) * 长时间占住 4 条连接,而常驻进程走的是 Prisma 默认池。结果:每日 09:00 的健康日报
* 生产 47.99.62.30 URL **没写** → 走 Prisma 默认 `核数×2+1`,4 核机 = **9** * (6 个并发 count)抢不到连接,**连续三天在 09:00:10 整点超时**,企微收不到日报;
* → `--concurrency` 超过 8 就会被卡在 9,多出来的 worker 排队等连接 * 同一根因还打掉过一条 plan upsert(`Unable to start a transaction in the given time`)。
* 也就是说:按 --concurrency=8 跑生产,9 条刚好够,本修复不产生收益;它的价值是**解锁更高并发** * 本函数开头那句"一调并发就得手动调池"说的就是这个病,只是当时没把常驻服务的旋钮算进来
* *
* 改成读**通用**的 PAC_DB_CONCURRENCY,由各 CLI 在 Nest 启动前按自己的 --concurrency 设进来; * ⚠️ **Prisma 默认池是「物理核」×2+1,不是逻辑核** —— 这里以前写错过,导致低估了生产的池:
* PAC_COHORT_CONCURRENCY 保留为向后兼容的别名,两者取大。 * 生产 47.99.62.30:`nproc`=4 但 `Core(s) per socket`=2 / `Thread(s) per core`=2,
* 物理核只有 2 → 默认池 = 2×2+1 = **5**(不是按 4 核算的 9)。5 条连接扛不住 4 路并发 + 日报。
* 测试服 47.251.104.47:URL 里显式 `connection_limit=30` → 本函数提前返回,不受影响
* (那里观测到的 9 条连接 = 8 个 worker + 1 空闲,不是池上限 —— 别再据此推断)。
*
* 生产 RDS `max_connections=820`(2026-09-02 实测,当时全库仅 23 条在用),
* 封顶 40 对它绰绰有余;封顶注释里那个 100 是早年自建 PG 的值,保守留着。
*/ */
export function withCohortDerivedPool(rawUrl: string | undefined): string | undefined { export function withCohortDerivedPool(rawUrl: string | undefined): string | undefined {
if (!rawUrl) return rawUrl; if (!rawUrl) return rawUrl;
...@@ -35,7 +43,15 @@ export function withCohortDerivedPool(rawUrl: string | undefined): string | unde ...@@ -35,7 +43,15 @@ export function withCohortDerivedPool(rawUrl: string | undefined): string | unde
} }
if (url.searchParams.has('connection_limit')) return rawUrl; if (url.searchParams.has('connection_limit')) return rawUrl;
const num = (v: string | undefined) => Math.max(1, parseInt(v ?? '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)); // 三个旋钮取最大:
// PAC_DB_CONCURRENCY 通用(各 CLI 按自己的 --concurrency 在 Nest 启动前设)
// PAC_COHORT_CONCURRENCY 摄入的历史别名,向后兼容
// PAC_RECALL_SUBSCENARIO_CONCURRENCY **常驻服务**的召回并行度(批次会长时间占住这么多条)
const n = Math.max(
num(process.env.PAC_DB_CONCURRENCY),
num(process.env.PAC_COHORT_CONCURRENCY),
num(process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY),
);
if (n <= 1) return rawUrl; if (n <= 1) return rawUrl;
url.searchParams.set('connection_limit', String(Math.min(n * 6 + 5, 40))); url.searchParams.set('connection_limit', String(Math.min(n * 6 + 5, 40)));
return url.toString(); return url.toString();
......
...@@ -11,8 +11,12 @@ import { withCohortDerivedPool } from '../src/prisma/prisma.service'; ...@@ -11,8 +11,12 @@ import { withCohortDerivedPool } from '../src/prisma/prisma.service';
* 那个 75.7 秒的患者会把同批 7 个 worker 一起冻住。 * 那个 75.7 秒的患者会把同批 7 个 worker 一起冻住。
* ② --concurrency 没放大连接池 —— **潜在**缺陷,咬不咬人看该机 .env: * ② --concurrency 没放大连接池 —— **潜在**缺陷,咬不咬人看该机 .env:
* 测试服 URL 显式 `connection_limit=30` → 不受影响 * 测试服 URL 显式 `connection_limit=30` → 不受影响
* 生产 URL 没写 → Prisma 默认 核数×2+1,4 核机 = 9 → --concurrency>8 会被卡住 * 生产 URL 没写 → 走 Prisma 默认池
* 所以按 --concurrency=8 跑生产时本修复无收益,价值在于解锁更高并发。 *
* 🔴 2026-09-02 补:上面那句"4 核机 = 9"是**错的**,已在实现里更正 ——
* Prisma 默认池按**物理核**算(×2+1),生产 nproc=4 但物理核只有 2 → 池 = **5**。
* 而常驻服务的 `PAC_RECALL_SUBSCENARIO_CONCURRENCY=4` 当时压根没被算进池 →
* 批次占住 4 条,每日健康日报连续三天在 09:00:10 超时。下面 describe 里锁的就是这条。
*/ */
describe('runPool —— 连续调度,不做批次栅栏', () => { describe('runPool —— 连续调度,不做批次栅栏', () => {
...@@ -66,12 +70,17 @@ describe('runPool —— 连续调度,不做批次栅栏', () => { ...@@ -66,12 +70,17 @@ describe('runPool —— 连续调度,不做批次栅栏', () => {
describe('连接池随并发放大 —— PAC_DB_CONCURRENCY', () => { describe('连接池随并发放大 —— PAC_DB_CONCURRENCY', () => {
const URL_ = 'postgresql://u:p@db:5432/pac?schema=public'; 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 saved = {
const set = (db?: string, cohort?: string) => { db: process.env.PAC_DB_CONCURRENCY,
cohort: process.env.PAC_COHORT_CONCURRENCY,
recall: process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY,
};
const set = (db?: string, cohort?: string, recall?: string) => {
if (db === undefined) delete process.env.PAC_DB_CONCURRENCY; else process.env.PAC_DB_CONCURRENCY = db; 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; if (cohort === undefined) delete process.env.PAC_COHORT_CONCURRENCY; else process.env.PAC_COHORT_CONCURRENCY = cohort;
if (recall === undefined) delete process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY; else process.env.PAC_RECALL_SUBSCENARIO_CONCURRENCY = recall;
}; };
afterAll(() => set(saved.db, saved.cohort)); afterAll(() => set(saved.db, saved.cohort, saved.recall));
const limitOf = (u: string | undefined) => const limitOf = (u: string | undefined) =>
u ? new URL(u).searchParams.get('connection_limit') : null; u ? new URL(u).searchParams.get('connection_limit') : null;
...@@ -101,9 +110,30 @@ describe('连接池随并发放大 —— PAC_DB_CONCURRENCY', () => { ...@@ -101,9 +110,30 @@ describe('连接池随并发放大 —— PAC_DB_CONCURRENCY', () => {
}); });
test('都没设 / =1(常驻 API 进程)→ 不动 URL,走 Prisma 默认池', () => { test('都没设 / =1(常驻 API 进程)→ 不动 URL,走 Prisma 默认池', () => {
set(undefined, undefined); set(undefined, undefined, undefined);
expect(withCohortDerivedPool(URL_)).toBe(URL_);
set('1', '1', '1');
expect(withCohortDerivedPool(URL_)).toBe(URL_); expect(withCohortDerivedPool(URL_)).toBe(URL_);
set('1', '1'); });
// ── 🔴 2026-09-02 生产事故:常驻服务的并发旋钮没被算进池 ──
// PAC_RECALL_SUBSCENARIO_CONCURRENCY=4 让批次长期占住 4 条连接,而常驻进程用的是
// Prisma 默认池(生产 2 物理核 → 5)。每日 09:00 健康日报的 6 个并发 count 抢不到连接,
// 连续三天在 09:00:10(= 10s pool_timeout)整点失败,企微收不到日报。
test('⭐ 常驻服务的召回并发旋钮也要放大池(否则批次占满,日报/API 抢不到连接)', () => {
set(undefined, undefined, '4');
expect(limitOf(withCohortDerivedPool(URL_))).toBe(expected(4)); // 29,不再是默认的 5
});
test('召回旋钮与另外两个一起取最大,不会互相盖掉', () => {
set('2', '3', '6');
expect(limitOf(withCohortDerivedPool(URL_))).toBe(expected(6)); // 41 → 封顶 40
set('9', '3', '4');
expect(limitOf(withCohortDerivedPool(URL_))).toBe(expected(9)); // CLI 旋钮更大时以它为准
});
test('召回旋钮 =1(默认串行)→ 行为不变,仍走 Prisma 默认池', () => {
set(undefined, undefined, '1');
expect(withCohortDerivedPool(URL_)).toBe(URL_); expect(withCohortDerivedPool(URL_)).toBe(URL_);
}); });
......
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