Skip to content
Projects
Groups
Snippets
Help
This project
Loading...
Sign in / Register
Toggle navigation
P
pac
Overview
Overview
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
ai-tools
pac
Commits
7006c97a
Commit
7006c97a
authored
Sep 02, 2026
by
luoqi
Browse files
Options
Browse Files
Download
Plain Diff
merge: 连接池纳入常驻并发 + 回访表名 → main
parents
e19135b4
8d12f2b9
Pipeline
#3653
failed in 0 seconds
Changes
5
Pipelines
1
Hide whitespace changes
Inline
Side-by-side
Showing
5 changed files
with
85 additions
and
18 deletions
+85
-18
apps/pac-service/.env.example
+13
-0
apps/pac-service/src/modules/plan/engine/scenarios/treatment-initiation-recall.scenario.ts
+4
-0
apps/pac-service/src/prisma/prisma.service.ts
+29
-9
apps/pac-service/src/queues/daily-health-report.service.ts
+1
-1
apps/pac-service/tests/persona-recompute-perf.spec.ts
+38
-8
No files found.
apps/pac-service/.env.example
View file @
7006c97a
...
@@ -42,6 +42,16 @@ POSTGRES_DB=pac
...
@@ -42,6 +42,16 @@ 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,自动放大**整个失效**(逃生口,写了就得自己负责够用)。
# ⛔ 但 compose 模式下**在本文件里写 connection_limit 是没用的** —— compose 的 environment:
# 段会整条覆盖 DATABASE_URL。要确认到底生效的是什么,只能看
# `docker inspect <容器> --format '{{range .Config.Env}}{{println .}}{{end}}' | grep DATABASE_URL`。
# 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 +224,9 @@ SENTRY_RELEASE=
...
@@ -214,6 +224,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(集合式)。
...
...
apps/pac-service/src/modules/plan/engine/scenarios/treatment-initiation-recall.scenario.ts
View file @
7006c97a
...
@@ -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
...
...
apps/pac-service/src/prisma/prisma.service.ts
View file @
7006c97a
...
@@ -14,16 +14,28 @@ import { tenantGuardExtension } from './tenant-guard.extension';
...
@@ -14,16 +14,28 @@ 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 路并发 + 日报。
* ⛔ **别再说「测试服 .env 里写了 connection_limit=30 所以不受影响」** —— 这话流传了很久,是错的:
* compose 会在 `environment:` 段覆盖 .env 里的 DATABASE_URL,容器里实际拿到的是
* `postgresql://…@postgres:5432/pac?schema=public`,**没有 connection_limit**
* (2026-09-02 `docker inspect` 实测两台都如此)。测试服同样是默认池,
* 09-01 那天它的健康日报也报了同一句 `connection limit: 5`。
* 判断逃生口生不生效,要看**容器实际环境变量**,不是看 .env 文件。
*
* 生产 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 +47,15 @@ export function withCohortDerivedPool(rawUrl: string | undefined): string | unde
...
@@ -35,7 +47,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
();
...
...
apps/pac-service/src/queues/daily-health-report.service.ts
View file @
7006c97a
...
@@ -253,7 +253,7 @@ export class DailyHealthReportService {
...
@@ -253,7 +253,7 @@ export class DailyHealthReportService {
}
}
if
(
pac
.
kind
===
'return_visit'
)
{
if
(
pac
.
kind
===
'return_visit'
)
{
const
r
=
await
this
.
prisma
.
$queryRaw
<
Array
<
{
c
:
bigint
}
>>
`
const
r
=
await
this
.
prisma
.
$queryRaw
<
Array
<
{
c
:
bigint
}
>>
`
SELECT count(distinct patient_id) AS c FROM patient_return_visit WHERE host_id =
${
hostId
}
::uuid`
;
SELECT count(distinct patient_id) AS c FROM patient_return_visit
s
WHERE host_id =
${
hostId
}
::uuid`
;
return
Number
(
r
[
0
]?.
c
??
0
);
return
Number
(
r
[
0
]?.
c
??
0
);
}
}
if
(
pac
.
kind
===
'image_ai'
)
{
if
(
pac
.
kind
===
'image_ai'
)
{
...
...
apps/pac-service/tests/persona-recompute-perf.spec.ts
View file @
7006c97a
...
@@ -10,9 +10,13 @@ import { withCohortDerivedPool } from '../src/prisma/prisma.service';
...
@@ -10,9 +10,13 @@ import { withCohortDerivedPool } from '../src/prisma/prisma.service';
* ① 批次栅栏 → worker pool:缺口正好是 E[8 次抽样最大值](≈p90)与均值之比;
* ① 批次栅栏 → worker pool:缺口正好是 E[8 次抽样最大值](≈p90)与均值之比;
* 那个 75.7 秒的患者会把同批 7 个 worker 一起冻住。
* 那个 75.7 秒的患者会把同批 7 个 worker 一起冻住。
* ② --concurrency 没放大连接池 —— **潜在**缺陷,咬不咬人看该机 .env:
* ② --concurrency 没放大连接池 —— **潜在**缺陷,咬不咬人看该机 .env:
* 测试服 URL 显式 `connection_limit=30` → 不受影响
* (原文说"测试服 URL 显式 connection_limit=30 所以不受影响" —— 2026-09-02 证伪:
* 生产 URL 没写 → Prisma 默认 核数×2+1,4 核机 = 9 → --concurrency>8 会被卡住
* compose 覆盖了 DATABASE_URL,两台容器里都没有 connection_limit,都是默认池)
* 所以按 --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_
);
});
});
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment