From 20dddc151ec012eb935d8d80244fbd25ebeb92dc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Mon, 12 Oct 2026 19:37:51 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E8=A3=85=E9=85=8D=E5=88=87=E6=8D=A2?= =?UTF-8?q?=E5=AE=A1=E6=9F=A5=E4=BF=AE=E5=A4=8D=EF=BC=88=E5=90=AF=E5=8A=A8?= =?UTF-8?q?=E5=BC=82=E5=B8=B8=E8=B7=AF=E5=BE=84=E9=9F=A7=E6=80=A7/?= =?UTF-8?q?=E6=B8=85=E6=89=AB=E9=9A=94=E7=A6=BB/=E6=96=87=E6=A1=88?= =?UTF-8?q?=E4=B8=8E=E5=A4=B9=E5=85=B7=E5=8D=AB=E7=94=9F=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - main.lifespan: 预热+清扫挪进 try,finally dispose_all 覆盖启动半途 失败(幂等于更少引擎,堵 engine 泄漏);清扫循环体 per-tenant try/except + logger.exception 后继续——一个租户坏库不炸平台启动, resolve/load_from_dir 保持 fail-fast - routes/runtime.py: RuntimeError 文案补 registry_reload dispose 旧 runtime 成因;TenantRuntime 注解挪 TYPE_CHECKING(engine 包运行期 零 import platform) - conftest client: teardown 置空 app.state.registry(关残留窗); rmtree 先正常删,失败 warnings.warn 留痕后再 ignore_errors 兜底 Co-Authored-By: Claude Code --- app/engine/routes/runtime.py | 12 +++++++--- app/main.py | 45 +++++++++++++++++++++++------------- tests/conftest.py | 12 ++++++++-- 3 files changed, 48 insertions(+), 21 deletions(-) diff --git a/app/engine/routes/runtime.py b/app/engine/routes/runtime.py index b0c5426..326e0ac 100644 --- a/app/engine/routes/runtime.py +++ b/app/engine/routes/runtime.py @@ -9,12 +9,17 @@ Task 7 的 Host 中间件上线前,没有人为请求挂 ``state.tenant``。本 热切换后 default_runtime 仍指向旧 runtime——过渡期不修,Task 7 中间件 按请求 Host 重新 resolve 后本模块整体删除。 """ +from typing import TYPE_CHECKING + from fastapi import Request, WebSocket -from app.platform.registry import TenantRuntime +if TYPE_CHECKING: + # 仅类型注解用: engine 包运行期零 import platform(依赖方向保持 + # platform → engine,反向只在类型层) + from app.platform.registry import TenantRuntime -def get_runtime(request_or_ws: Request | WebSocket) -> TenantRuntime: +def get_runtime(request_or_ws: Request | WebSocket) -> "TenantRuntime": """state.tenant 优先,缺省落 app.state.default_runtime(Task 7 删)。""" runtime = getattr(request_or_ws.state, "tenant", None) if runtime is not None: @@ -22,5 +27,6 @@ def get_runtime(request_or_ws: Request | WebSocket) -> TenantRuntime: runtime = getattr(request_or_ws.app.state, "default_runtime", None) if runtime is None or runtime.session_factory is None: raise RuntimeError( - "默认租户运行时未就绪——lifespan 未运行或未预热 default_runtime") + "默认租户运行时未就绪——lifespan 未运行、未预热 default_runtime," + "或 registry_reload 已 dispose 旧 runtime(过渡期限制,Task 7 解决)") return runtime diff --git a/app/main.py b/app/main.py index ed119ca..f69f25a 100644 --- a/app/main.py +++ b/app/main.py @@ -1,5 +1,6 @@ from contextlib import asynccontextmanager from pathlib import Path +import logging from fastapi import FastAPI from fastapi.staticfiles import StaticFiles @@ -7,6 +8,8 @@ from fastapi.staticfiles import StaticFiles from app.engine.config import STATIC_DIR from app.platform.registry import TenantRegistry +logger = logging.getLogger(__name__) + # 路径锚定(不随 CWD 漂移):seed_profiles 与 tenants 都在仓库根 BASE_DIR = Path(__file__).resolve().parents[1] SEED_DIR = BASE_DIR / "seed_profiles" @@ -25,25 +28,35 @@ async def lifespan(app: FastAPI): registry = TenantRegistry.load_from_dir(SEED_DIR, data_dir=TENANTS_DIR) app.state.registry = registry - # 预热默认租户(过渡兜底读 app.state.default_runtime)。engine 建在本 - # lifespan 的事件循环里——连接与 loop 绑定,别处另起 asyncio.run 复用 - # 同一 engine 会跨 loop 死锁(测试直连造数走 conftest.async_db) - app.state.default_runtime = await registry.resolve(DEFAULT_TENANT_HOST) - - # 询盘推送(设计 8.2): 清扫崩溃遗留的 stale pending → failed。逐租户、 - # 用该租户自己的 session_factory;db 文件不存在的租户跳过——全新库 - # 无遗留行,保持 engine 懒建纪律(预热的默认租户文件已建,照常扫)。 - from app.engine.services.lead_push_service import sweep_stale_pending - for rt in registry.runtimes(): - if rt.config.status != "active" or not rt.db_path.exists(): - continue - ready = await registry.ensure_ready(rt) - await sweep_stale_pending(session_factory=ready.session_factory) - try: + # 预热默认租户(过渡兜底读 app.state.default_runtime)。engine 建在 + # 本 lifespan 的事件循环里——连接与 loop 绑定,别处另起 asyncio.run + # 复用同一 engine 会跨 loop 死锁(测试直连造数走 conftest.async_db)。 + # 预热/加载保持 fail-fast:默认租户起不来 = 过渡期全站无租户。 + app.state.default_runtime = await registry.resolve(DEFAULT_TENANT_HOST) + + # 询盘推送(设计 8.2): 清扫崩溃遗留的 stale pending → failed。逐租户、 + # 用该租户自己的 session_factory;db 文件不存在的租户跳过——全新库 + # 无遗留行,保持 engine 懒建纪律(预热的默认租户文件已建,照常扫)。 + # 清扫失败只记日志继续:一个租户的坏库不炸整个平台启动(其余租户 + # 照常服务),该租户下轮重启还会重扫。 + from app.engine.services.lead_push_service import sweep_stale_pending + for rt in registry.runtimes(): + if rt.config.status != "active" or not rt.db_path.exists(): + continue + try: + ready = await registry.ensure_ready(rt) + await sweep_stale_pending(session_factory=ready.session_factory) + except Exception: + logger.exception( + "启动清扫失败,跳过该租户(slug=%s, db=%s)", + rt.config.slug, rt.db_path) + yield finally: - await registry.dispose_all() # Windows 文件锁: 不关则库文件删不掉 + # 幂等于"更少的引擎":启动半途失败也把已建好的 engine 关掉 + # (Windows 文件锁: 不关则库文件删不掉) + await registry.dispose_all() app = FastAPI(title="HESUPIPE AI 客服", lifespan=lifespan) diff --git a/tests/conftest.py b/tests/conftest.py index 4b6dc52..eed8e91 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,6 +2,7 @@ import asyncio import os import shutil import tempfile +import warnings from pathlib import Path import pytest @@ -72,7 +73,9 @@ def client(tmp_path): 线程需要直接造数时不得用该 engine——asyncio.run 另起循环,跨循环 复用常驻连接会死锁;直连断言走 async_db,或经真实路由(WS/REST)。 - 退出: lifespan finally 里 registry.dispose_all()(Windows 文件锁, - 不关则 tmp 目录删不掉),随后清理 tmp 目录。 + 不关则 tmp 目录删不掉);随后置空 app.state.registry(关残留窗—— + 不让已 dispose 的测试 Registry 被后续无夹具的 lifespan 误用)并 + 清理 tmp 目录(删不掉告警留痕,不静默吞)。 """ from fastapi.testclient import TestClient @@ -82,4 +85,9 @@ def client(tmp_path): with TestClient(app) as c: yield c finally: - shutil.rmtree(tmp_path, ignore_errors=True) + app.state.registry = None + try: + shutil.rmtree(tmp_path) + except OSError as exc: + warnings.warn(f"client 夹具清理 tmp 目录失败(文件锁?): {exc}") + shutil.rmtree(tmp_path, ignore_errors=True)