fix(#213): 治理 async 路由内同步阻塞事件循环(R-P1-10)

AST 盘点 app/api:9 个「假 async(函数体无 await)」+ 22 个真 async。

1) 纯同步阻塞且不依赖事件循环 → 改同步 def(FastAPI 自动 run_in_threadpool):
   - deps.get_current_user / require_role._check:全站每请求最高频,db.get() 同步 ORM
   - uploads.confirm_import / preview_diff:pandas + ingest_table 批量 upsert
   - staff_assignments.import_snapshot:pandas xlsx 导入

2) 虽无 await 但内部 asyncio.get_running_loop().create_task 挂后台任务,必须留在
   事件循环线程 → 保留 async def,仅把唯一同步阻塞点包 run_in_threadpool:
   - ai_onboarding.create_draft(占位行 DB 插入)
   - governance.create_scan / create_judge(占位行 DB 插入,await 提交后再入队保证可见)
   - fetch.start_schedule(仅内存态判断+调度,无阻塞 IO)

3) uploads.parse_upload 保留 async(需 await file.read),save_upload_file /
   parse_file(pandas) 改为 await run_in_threadpool。

验证:ruff app+tests 全绿;onboarding/governance 定向 44 项通过;全量 pytest exit 0。
不做全仓 async 化(守工单不做项)。台账 R-P1-10 回填。
This commit is contained in:
2026-10-07 22:21:32 +08:00
parent ea9f936df5
commit 5fef31e90c
7 changed files with 42 additions and 12 deletions

View File

@@ -14,11 +14,14 @@ from app.models.user import MANAGER_ROLES, RoleEnum, User
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_PREFIX if hasattr(settings, 'API_PREFIX') else '/api/v1'}/auth/login", auto_error=False)
async def get_current_user(
# 注意:这里刻意用同步 def 而非 async —— 本依赖是全站最高频依赖(每个鉴权请求都执行),
# 体内 db.get() 是同步 SQLAlchemy 调用;若声明成 async,它会在事件循环线程里阻塞,
# 拖垮同进程所有并发请求。FastAPI 对普通 def 依赖自动放入线程池(run_in_threadpool)执行(#213 R-P1-10)。
def get_current_user(
token: Annotated[str | None, Depends(oauth2_scheme)],
db: Annotated[Session, Depends(get_db)],
) -> User:
"""从 JWT 解析当前用户"""
"""从 JWT 解析当前用户(同步 def:FastAPI 自动线程池调度,避免阻塞事件循环)"""
if not token:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
@@ -61,7 +64,8 @@ def require_role(*roles: RoleEnum):
allowed = [r.value for r in roles]
async def _check(current_user: Annotated[User, Depends(get_current_user)]) -> User:
def _check(current_user: Annotated[User, Depends(get_current_user)]) -> User:
# 同步 def:仅做内存角色比对、无 IO;与 get_current_user 一致由线程池调度(#213)。
if current_user.role not in allowed:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
return current_user

View File

@@ -19,6 +19,7 @@ import logging
from fastapi import APIRouter, HTTPException, Query, status
from sqlalchemy import func, select
from starlette.concurrency import run_in_threadpool
from app.api.deps import AdminUser, CurrentUser, DBSession
from app.core.database import SessionLocal
@@ -109,6 +110,9 @@ def _load_visible(db, draft_id: int, user) -> DataSourceSpec:
@router.post("/drafts", response_model=R, status_code=status.HTTP_202_ACCEPTED,
summary="提交样例,异步起草(202 + draft_id,轮询 GET /drafts/{id})")
# #213: 保留 async def —— 本端点末尾 start_draft_task() 依赖 asyncio.get_running_loop()
# 在线程上 create_task(fire-and-forget LLM 起草),故不能改成纯同步 def;
# 仅把其中唯一的阻塞同步 DB 调用 create_draft_placeholder 放入线程池,避免阻塞事件循环。
async def create_draft(payload: dict, db: DBSession, user: CurrentUser):
"""上传样例 → 预建占位行(drafting)→ 立即返回 draft_id;后台经 L1 门闸跑 LLM。
@@ -126,7 +130,9 @@ async def create_draft(payload: dict, db: DBSession, user: CurrentUser):
raise HTTPException(400, "samples 每行必须是 JSON 对象(列名→值)")
try:
draft = create_draft_placeholder(
# #213: 同步 DB 插入放入线程池;start_draft_task 必须留在事件循环线程 create_task。
draft = await run_in_threadpool(
create_draft_placeholder,
db, table_hint=payload.get("table_hint"), samples=samples, created_by=user.id,
)
except DraftError as e:

View File

@@ -576,6 +576,9 @@ def update_schedule_config(payload: ScheduleConfigIn, db: DBSession, me: AdminUs
@router.post("/schedule/start", response_model=R[ScheduleStatus], summary="启动定时同步")
# #213: 必须保留 async def —— 体内无 await,但 bms_scheduler.start() 内部用
# asyncio.create_task 挂载到「当前运行的事件循环」;改成同步 def 会在 anyio 线程池线程执行,
# 那里没有运行中的 loop(get_running_loop 抛 RuntimeError)。本端点只做内存态判断 + 调度,无阻塞 IO。
async def start_schedule(current_user: CurrentUser):
"""启动 BMS + SCRM 定时同步(8-23/0 每小时 + 5:00 补跑 + 8:00 全量兜底)

View File

@@ -30,6 +30,7 @@ import logging
from fastapi import APIRouter, HTTPException, Query, status
from sqlalchemy import func, select
from starlette.concurrency import run_in_threadpool
from app.api.deps import AdminUser, DBSession
from app.models.governance_finding import (
@@ -170,12 +171,15 @@ def _start_placeholder_event(db, action: str, title: str) -> SystemEvent:
# ═══════════════════════════════════════════════════════════════════════
@router.post("/scans", response_model=R, status_code=status.HTTP_202_ACCEPTED,
summary="手动触发一次数据体检(202 + scan_id,异步)")
# #213: 保留 async def —— start_scan_task() 内部 asyncio.get_running_loop().create_task,
# 必须在事件循环线程调度;仅把同步 DB 占位行插入放入线程池(commit=True,await 完成后再入队,
# 保证后台任务可见该占位事件)。
async def create_scan(db: DBSession, user: AdminUser):
"""提交体检 → 立即返回 scan_id;后台扫 D1/D2/D3 并落 findings。
轮询 GET /governance/scans/{scan_id} 直到 finished=true。
"""
ev = _start_placeholder_event(db, EVENT_SCAN, "数据体检: 已提交")
ev = await run_in_threadpool(_start_placeholder_event, db, EVENT_SCAN, "数据体检: 已提交")
start_scan_task(ev.id)
logger.info("governance 手动体检已提交 scan_id=%s user=%s", ev.id, user.username)
return R(data={"scan_id": ev.id})
@@ -191,12 +195,14 @@ def get_scan(scan_id: int, db: DBSession, _: AdminUser):
@router.post("/judge", response_model=R, status_code=status.HTTP_202_ACCEPTED,
summary="手动触发一次研判(202 + judge_id,异步,经 L1 门闸)")
# #213: 保留 async def —— start_judge_task() 同样在事件循环上 create_task(研判等 LLM);
# 仅把同步 DB 占位行插入放入线程池(await 提交完成后再入队,保证可见性)。
async def create_judge(db: DBSession, user: AdminUser):
"""提交研判 → 返回 judge_id;后台对全部 open finding 走「规则短路 → 批量 AI」。
规则命中的 finding **不调 LLM**(零成本),故可放心对全量 open 提交。
"""
ev = _start_placeholder_event(db, EVENT_JUDGE, "治理研判: 已提交")
ev = await run_in_threadpool(_start_placeholder_event, db, EVENT_JUDGE, "治理研判: 已提交")
start_judge_task(ev.id)
logger.info("governance 手动研判已提交 judge_id=%s user=%s", ev.id, user.username)
return R(data={"judge_id": ev.id})

View File

@@ -146,7 +146,9 @@ def import_snapshot_preview(
@router.post("/import-snapshot", response_model=R[ImportResponse], summary="HR 人力快照 xlsx 导入")
async def import_snapshot(
# #213: 改同步 def —— 体内 load_dataframe(pandas xlsx) 为阻塞重活且无 await;普通 def 由
# FastAPI 自动线程池调度,避免 HR 快照导入卡住事件循环。
def import_snapshot(
payload: dict,
db: DBSession,
current_user: AdminUser,

View File

@@ -15,6 +15,7 @@ from fastapi import (
)
from fastapi.responses import FileResponse, Response
from sqlalchemy import select
from starlette.concurrency import run_in_threadpool
from app.api.deps import CurrentUser, DBSession
from app.core.config import settings
@@ -73,7 +74,10 @@ async def parse_upload(
raise HTTPException(status_code=413, detail=f"文件超过 {settings.MAX_UPLOAD_SIZE_MB}MB 限制")
try:
# P0-3: save_upload_file 落盘前校验扩展名白名单 + magic bytes
path, actual_stored_path = save_upload_file(content, file.filename or "upload")
# #213: 同步落盘/校验放入线程池,避免阻塞事件循环(本函数因需 await file.read 保留 async)。
path, actual_stored_path = await run_in_threadpool(
save_upload_file, content, file.filename or "upload"
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
filename = file.filename
@@ -81,7 +85,8 @@ async def parse_upload(
raise HTTPException(status_code=400, detail="必须提供 file 或 stored_path 参数")
try:
preview = parse_file(path, data_type, sheet_name=sheet_name)
# #213: pandas 解析是 CPU/IO 重活,放入线程池,避免大文件解析阻塞事件循环。
preview = await run_in_threadpool(parse_file, path, data_type, sheet_name=sheet_name)
except Exception as e:
raise HTTPException(status_code=400, detail=f"文件解析失败: {e}") from e
preview["stored_path"] = actual_stored_path
@@ -90,7 +95,9 @@ async def parse_upload(
@router.post("/preview-diff", response_model=R, summary="diff 预览(导入前查看 existing/new/abnormal)")
async def preview_diff(
# #213: 改同步 def —— 体内 load_dataframe(pandas) / compute_diff(同步 DB) 全是阻塞调用且无
# await;声明 async 反而让重活跑在事件循环线程。普通 def 由 FastAPI 自动放入线程池执行。
def preview_diff(
payload: DiffPreviewRequest,
db: DBSession,
current_user: CurrentUser,
@@ -121,7 +128,9 @@ async def preview_diff(
@router.post("/import", response_model=R[ImportResponse], summary="确认导入(upsert+关联匹配)")
async def confirm_import(
# #213: 改同步 def —— load_dataframe + ingest_table(批量 upsert) 是最重的同步调用且无 await,
# 必须跑在线程池而非事件循环线程;关联匹配等后处理已在 BackgroundTasks 中异步执行。
def confirm_import(
payload: dict,
db: DBSession,
current_user: CurrentUser,

View File

@@ -372,7 +372,7 @@ sudo rm -f /opt/csctalon/backups/pre-org-20260914.dump # 108M (09
| R-P1-7 | P1 | 菜单漂移 vs A-20260910-004(LOCKED) | 待处理 | 报告 §3.1 W1;须发起 LOCKED 文档人工审核修订 |
| R-P1-8 | P1 | 金额单位与铁律冲突:模型 `Numeric(12,2)`(元)+ 服务层 `float()` 聚合 | 待处理 | 报告 §3.3 W4;评估「分(整数)」迁移路径(涉对接/前端兼容)或人工审核登记豁免 |
| R-P1-9 | P1 | Gunicorn worker 数矛盾:`Dockerfile.patch:17` `-w 4` vs compose/Dockerfile `-w 1` | ✅ 已完成(#212,2026-10-07) | 报告 §3.3 W1;核实部署路径:compose `build:./backend` 且 command 显式 `-w 1`,patch 不被 compose 引用(应急仅覆盖代码),但被直接 `docker run` 时其内嵌 CMD 生效。已把 patch 改为 `-w 1` 并在 Dockerfile/compose/patch 三处补注释(llm_scheduler 进程级 Condition 门闸 + SyncTaskManager/rate_limit 进程内状态依赖单 worker,多 worker 须先迁 Redis,v0.7 立此存照不做);顺手纠正 rate_limit.py 陈旧「-w 4」注释。全仓 grep 三处启动点唯一 `-w 1`,compose YAML 合法、ruff 全绿 |
| R-P1-10 | P1 | async 路由内阻塞事件循环(`deps.py` 同步 ORM;`uploads.py` 同步 pandas/ingest) | 待处理 | 报告 §3.3 W2/W3;改同步 `def` 或 `run_in_threadpool` |
| R-P1-10 | P1 | async 路由内阻塞事件循环(`deps.py` 同步 ORM;`uploads.py` 同步 pandas/ingest) | ✅ 已完成(#213,2026-10-07) | 报告 §3.3 W2/W3;AST 盘点 app/api:9 个「假 async(无 await)」+22 真 async。处置:① 纯同步阻塞且不依赖 loop 的 5 处改同步 `def`(FastAPI 自动线程池)——`get_current_user`/`require_role._check`(全站最高频依赖)、uploads `confirm_import`/`preview_diff`、staff `import_snapshot`;② 4 处虽无 await 但内部 `asyncio.get_running_loop().create_task` 挂后台任务,**必须留事件循环线程**,保留 `async def` 并把唯一同步阻塞点包 `run_in_threadpool`——ai_onboarding `create_draft`(占位行插入)、governance `create_scan`/`create_judge`(占位行插入)、fetch `start_schedule`(仅内存态无阻塞);③ uploads `parse_upload` 保留 async(需 await file.read),`save_upload_file`/`parse_file`(pandas) 入线程池。验证:ruff app+tests 全绿;test_onboarding_router + test_governance_router 44 项定向通过;全量 pytest `PYTEST_EXIT=0`。不做全仓 async 化(守不做项) |
| R-P1-11 | P1 | 运行时破坏性操作未过 `db_guard`(治理执行 DELETE/DROP、保留期清理 DELETE) | 待处理 | 报告 §3.3 W5 / §3.5 W9;补显式授权或登记豁免 |
| R-P1-12 | P1 | 顶层 `scripts/` 23 脚本绕过 `bootstrap()`,2 处 DELETE 未过守卫 | 待处理 | 报告 §3.5 W5 / §3.6 W2;迁移 + 补 bootstrap/守卫 |
| R-P1-13 | P1 | lint 门禁仅 `tests/`(`app/` 未纳入);gitleaks 整体放行 workflows 与 backend/tests | 待处理 | 报告 §3.5 W2/W4;扩大 ruff 范围、收窄 allowlist |