xsl
cb8a1e3c0f
CI / lint / typecheck / import-linter / tests / secrets (push) Canceled after 0s
FE↔后端联调实测暴露并修复： 1. 问卷键名三处错位（FE height_cm/weight_kg/psych_style vs 文档 height/weight/psyche_style）——vision._bmi 多键取值、diagnosis 维度 装配与规则上下文双键或运算，两侧谁改键名都不再炸 2. celery worker asyncio.run 每投递起新 loop + 模块级引擎缓存 → asyncpg 连接跨 loop（同进程第 3 个报告任务必现 RuntimeError: attached to a different loop；集成测试单循环掩盖）。新增 apps/worker/loop.py 持久 loop，report/ping 两任务接入 E2E 终态： psyche_style/height_weight 恢复、BMI 体型确定值、ok 26/32、 真实 LLM 文案通过校验 degraded=False。
27 lines
1.1 KiB
Python
27 lines
1.1 KiB
Python
"""Celery worker 的持久事件循环。
|
||
|
||
``asyncio.run`` 每次投递都新建并随即关闭事件循环，而
|
||
``packages.shared.db`` 的引擎是模块级缓存——asyncpg 连接绑定在创建它的
|
||
loop 上，后续投递从池里拿到旧 loop 的连接就会
|
||
``RuntimeError: attached to a different loop``（live E2E 实测：同进程第
|
||
三个报告任务必现，集成测试单循环跑掩盖了它）。
|
||
|
||
worker 进程内复用一个 loop，让连接池与 loop 生命周期一致，问题从根上
|
||
消失。prefork 模型下每个子进程各有一份模块状态，互不影响。集成测试
|
||
不受影响：它们在 pytest 自己的 loop 里直调任务内芯，不经过本模块。
|
||
"""
|
||
|
||
import asyncio
|
||
from collections.abc import Coroutine
|
||
from typing import Any
|
||
|
||
_loop: asyncio.AbstractEventLoop | None = None
|
||
|
||
|
||
def run(coro: Coroutine[Any, Any, Any]) -> Any:
|
||
"""在 worker 进程唯一的持久 loop 上执行协程并返回结果。"""
|
||
global _loop
|
||
if _loop is None or _loop.is_closed():
|
||
_loop = asyncio.new_event_loop()
|
||
return _loop.run_until_complete(coro)
|