fix(risk): 规则扫描加跨进程锁——手工触发与定时扫描此前可以同时跑
docs/25 P2 最后一项。核实后分清了两层,报告没区分: - **定时扫描是安全的**:risk_scan_scheduler.py 已有 MySQL 连接级咨询锁 (GET_LOCK,锁名 jr_risk_scan_schedule),跨进程互斥。 - **HTTP 端点不安全**:POST /api/v1/risk/alerts/scan → RiskScanService.scan() 只用了 **进程内** asyncio.Lock。多 Web worker、或 Worker 与 API 同时运行时形同虚设。 而扫描的幂等只有应用层的 _exists 查重 —— fin_risk_alert 的 trigger_rule_codes 是 JSON 数组,**无法建唯一索引兜底**(同一交易可命中多条规则,唯一键本应是"交易+规则",而规则 埋在 JSON 里)。所以两条路径并发时会同时查不到、同时插入,产生重复预警。 **改动**: 1. 把 mysql_scan_lock 与锁名移到 pp/infrastructure/db.py —— 端点与调度器**必须共用 同一把锁**,放在基础设施层两个入口才都能引用(service 不该反向依赖 worker)。 调度器改为从那里 import。 2. **端点层加锁**(controllers/risk.py 的 scan 端点):取不到锁就抛 RiskScanBusyError (与 service 内部那把进程内锁用同一错误类型与文案)。 **为什么不加在 RiskScanService.scan() 内部**:GET_LOCK 是**连接级**的,而调度器已经在 它自己的 session 上持锁;被两个入口共用的服务方法若再取同一把锁,取锁的连接并不是持锁的 那一个、必然返回 0 —— 会**把定时扫描自己挡死**。所以锁加在入口层,每个入口只取一次。 **实测**: - 无人持锁时扫描 → **200**「规则扫描完成」 - 本进程先取得跨进程锁后再调端点 → **409**「规则扫描正在执行,请稍后重试」 (同一进程内不同 session 也互斥,说明它是连接级的,正是跨进程所需) - 释放后再调 → **200**,恢复正常 顺带第 4 次遇到 409 复用错误码 RUN_NOT_CANCELLABLE,语义不符;属 P3 待处理项。 ruff / mypy(136 文件) / 639 unit+contract 全绿。
This commit is contained in:
@@ -1,3 +1,7 @@
|
||||
from collections.abc import AsyncIterator
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.engine import make_url
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
@@ -29,3 +33,36 @@ if "init_command" not in _url.query:
|
||||
|
||||
engine = create_async_engine(_url, pool_pre_ping=True)
|
||||
SessionFactory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||||
|
||||
# 规则扫描的**跨进程**咨询锁名。用 MySQL 连接级 `GET_LOCK` 而不是进程内 `asyncio.Lock`:
|
||||
# 后者只在单个进程内互斥,多 Web worker、或 Worker 与 API 同时运行时形同虚设。
|
||||
SCAN_LOCK_NAME = "jr_risk_scan_schedule"
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def mysql_scan_lock() -> AsyncIterator[bool]:
|
||||
"""约束规则扫描的跨进程并发;yield 是否取得锁。
|
||||
|
||||
**为什么两个入口都必须用它**:定时扫描(`worker/risk_scan_scheduler.py`)与手工触发的
|
||||
HTTP 端点(`POST /api/v1/risk/alerts/scan`)最终都调 `RiskScanService.scan()`,
|
||||
而扫描的幂等只有应用层的 `_exists` 查重 —— 两条路径并发时会同时查不到、同时插入,
|
||||
产生重复预警。锁要**共用同一个名字**才能互相排斥。
|
||||
|
||||
**为什么不放在 `scan()` 内部**:`GET_LOCK` 是**连接级**的。调度器已在自己的 session 上
|
||||
持锁,再新建 session 去调 `scan()`;若 `scan()` 又去取同一把锁,取锁的连接并不是持锁的
|
||||
那一个,`GET_LOCK` 返回 0 —— **调度器会把自己挡死**。所以锁加在**入口层**
|
||||
(调度器 / 端点各取一次),而不是被两个入口共用的服务方法里。
|
||||
|
||||
`GET_LOCK(name, 0)` 的 0 表示不等待、立即返回:取不到就跳过本轮,而不是排队。
|
||||
"""
|
||||
async with SessionFactory() as session:
|
||||
acquired = bool(
|
||||
await session.scalar(text("SELECT GET_LOCK(:name, 0)"), {"name": SCAN_LOCK_NAME})
|
||||
)
|
||||
try:
|
||||
yield acquired
|
||||
finally:
|
||||
if acquired:
|
||||
await session.scalar(
|
||||
text("SELECT RELEASE_LOCK(:name)"), {"name": SCAN_LOCK_NAME}
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user