Python async/await 静默失败防护
陷阱机制(为什么致命)
Python 中调用 async def 函数但忘记 await 时:
- 函数体完全不执行,只返回一个 coroutine 对象
- 不会抛异常,程序继续往下跑
- 仅在 coroutine 被垃圾回收时打印 RuntimeWarning: coroutine 'xxx' was never awaited(默认静默)
- 结果:接口返回 200、状态机标了 done,但数据库没写、消息没发、标签没存——典型的"假成功"
这是 2026-08-05 同事会项目 V2.3.472 故障的根因:_apply_ext_to_log(async)在 intake 内联路径、后台异步分析路径被调用时漏 await,AI 实体/标签从未写入库,日志却显示"AI 已分析"。
一、静态检测(机器防线,不依赖人)
项目 requirements.txt 只含 pytest/pytest-asyncio,没有 ruff。因此落地方案是零依赖 AST 检查,已固化进仓库:
scripts/async_await_lint.py(零依赖,仅标准库)
检测「调用本文件定义的 async def X,裸名调用且未 await 且非调度器参数」的情况。
设计要点(踩坑后定型):
- 逐文件作用域:只匹配【同一文件】内定义的 async 函数名。若全局收集,项目里某模块定义的 async def set/async def main 会让全仓所有 set()/main()/Depends(get_current_user) 引用误报(曾产生 160+ 条噪音)。逐文件作用域把误报降为 0。
- 跳过调度器参数:若 X(...) 的直接父调用是已知协程调度器(asyncio.run/wait_for/gather/create_task/ensure_future/shield/to_thread/loop.create_task/loop.run_until_complete ...),视为正确写法跳过。否则 asyncio.run(main())、asyncio.wait_for(extract(...))、asyncio.create_task(resume(...)) 这类正确写法会被误报。
- 排除内置名:set/map/list 等即便被定义为 async def,也按内置对待,防误报。
- 白名单:行尾加 # async-lint-ignore 可跳过(极端情况兜底)。
scripts/backend-pre-deploy.sh(后端部署前闸门)
在 scp 上传后端代码前物理保证:
- G1 py_compile 全量:拦截语法错误 / async 作用域错误(曾 V2.3.412 致全站 502)。
- G2 AST 检查:调用上面的 lint,拦截"调用 async 函数漏 await"的假成功。
- 失败即 exit 1,禁止继续部署。
- 路径处理:用 pwd -W 输出 Windows 风格 E:/...,避免 Git Bash 下 /e/... 被 Python 误读成 E:\e\...;py_compile 用相对路径 find 规避路径转换问题。
ruff(若未来引入)
如引入 ruff,ASYNC 规则组(ASYNC910/ASYNC100)可替代上面手写 AST;但当前以 async_await_lint.py 为准。
二、编码模式(消灭分叉)
收敛单一入口,不要多处各写一遍
坏味道:同一段"写 AI 结果到日志"逻辑在 intake、_analyze_logs_async、_run_reanalyze 三处各手写 tags=/entities= 赋值 → 改一处漏两处(V2.3.472 即此)。
好模式:_apply_ext_to_log(log, ext, db, uid, ...) 作为唯一落库入口,所有路径都 await 它。
命名约定降低误用
- async 函数加
async_前缀,或在 docstring 首行写(coroutine) 必须 await。
三、回归测试断言"实际副作用"(消灭伪自测)
发版前不仅测"接口返回 200",必须查数据库实际状态。已落地 scripts/test_ai_tagging_regression.py(走 API,不直连 DB;自动经 ssh tongshihui 生成 2h 调试 token):
- 路径1 intake 异步:
POST /api/intake创建日志 → 轮询GET /api/logs/{id}直到ai_status=done→ 断言tags非空、ai_classified=1。 - 路径2 reanalyze 补分析:
POST /api/logs创建空标签日志 →POST /api/logs/reanalyze(mode=all,指定 log_ids) → 轮询reanalyze/status/{tid}→GET日志断言tags非空、ai_classified=1。 - 注:
create_log本身仅做 10s 内联抽取、无后台兜底,AI 慢时tags可能为空,故不作为确定性断言路径(这是一个已知产品缺口,非本防护范围)。 - 最后清理两条测试数据。
- 运行:
PYTHONUNBUFFERED=1 <venv>/python -u scripts/test_ai_tagging_regression.py(需 httpx;已在托管 venvbinaries/python/envs/default安装)。
四、流程铁律(本项目已立,强化执行)
任何改动若触及被多处调用的共享函数(如 _apply_ext_to_log):
1. 先 grep -rn "函数名(" --include=*.py . 列出所有调用点
2. 逐处核对调用方式(async 是否 await、参数是否一致)
3. 改一处即同步所有调用点,不要"先改主线看效果"
关联项目铁律:
- 框架复用铁律:改码前 grep 同类实现逐行 diff(本故障即违反此条——V2.3.471 只改了 create_log/update_log,漏了 intake/异步两处调用点)
- 诊断铁律/伪自测兄弟条款:自测不只看"接口通/有返回",必须核对副作用是否真实发生
五、部署接线(让闸门成为硬门禁)
scripts/deploy-backend.sh 串联:后端部署前闸门(backend-pre-deploy.sh)→ pre-deploy-gate.sh(gitee 一致性)→ tar(排除 .env/venv/__pycache__/.git)→ scp(经 tongshihui 别名,禁止 root@IP)→ systemctl restart tongshihui-python → 可选运行回归测试。后端 scp 前必跑本闸门,失败禁止部署。
六、快速自查清单(改完 async 代码问自己)
- [ ] 每个
async def调用前面都有await?(静态闸门已拦则划掉) - [ ] 有没有 coroutine 被当作同步值赋值/返回?
- [ ] 状态字段(
ai_status=done)的设置,是否保证在副作用(写库)之后? - [ ] 测试是否断言了 DB 实际值,而非只看 HTTP 响应?
真实案例锚点
- 项目:同事会
tongshihui-python-server/app/api/v1/briefing.py - 版本:V2.3.472-backend(commit 0b1f444)
- 现象:日志显示"AI 已分析"但
tags=[]、ai_classified=0 - 修复:补两处
await _apply_ext_to_log(...)+ 补分析任务同步写 tags - 防护:本 skill +
scripts/async_await_lint.py+scripts/backend-pre-deploy.sh+scripts/test_ai_tagging_regression.py
