refactor: unify database readiness lifecycle

app/adapters/web/health.py

@@ -0,0 +1,45 @@
+"""面向编排器的最小公开健康探针。"""
+
+from fastapi import FastAPI, Request, status
+from fastapi.responses import JSONResponse
+from fastapi.routing import APIRoute
+
+from app.runtime.health import get_application_health
+
+
+async def liveness() -> JSONResponse:
+    """确认进程和当前事件循环能够处理请求，不访问任何外部依赖。"""
+    return JSONResponse(content={"status": "alive"})
+
+
+async def readiness(request: Request) -> JSONResponse:
+    """仅公开可否接流量，不泄露数据库、插件或启动异常细节。"""
+    ready = get_application_health(request.app).is_ready
+    return JSONResponse(
+        status_code=(
+            status.HTTP_200_OK
+            if ready
+            else status.HTTP_503_SERVICE_UNAVAILABLE
+        ),
+        content={"status": "ready" if ready else "not_ready"},
+    )
+
+
+def install_health_routes(app: FastAPI) -> None:
+    """把公开探针装到 API 版本前缀之外，供容器和反向代理使用。"""
+    app.router.add_api_route(
+        "/health/live",
+        liveness,
+        methods=["GET"],
+        response_class=JSONResponse,
+        include_in_schema=False,
+        route_class_override=APIRoute,
+    )
+    app.router.add_api_route(
+        "/health/ready",
+        readiness,
+        methods=["GET"],
+        response_class=JSONResponse,
+        include_in_schema=False,
+        route_class_override=APIRoute,
+    )

app/db/engine.py

@@ -92,7 +92,8 @@ def _get_sqlite_engine(is_async: bool = False, pooled: bool = False):
         # 设置WAL模式。
         # 这是引擎构建里唯一的阻塞 I/O，且发生在 get_engine() 的创建锁内——异步侧因此
         # 移除了对称的那一段（见下方 else 分支）。同步侧保留是因为 journal_mode 必须有人
-        # 设置一次，而同步引擎的首次创建由 init_db() 在启动期单线程完成，不存在一群线程
+        # 设置一次，而同步引擎的首次创建由 lifespan 数据库准备组件中的 init_db() 完成，
+        # 不存在一群线程
         # 等在锁上的场面；即便退化到运行期首次访问，阻塞的也只是本地 SQLite 的一次 PRAGMA。
         _journal_mode = "WAL" if settings.DB_WAL_ENABLE else "DELETE"
         with engine.connect() as connection:

app/factory.py

@@ -9,6 +9,7 @@ from starlette.exceptions import HTTPException
 
 from app.api.response import ResponseAPIRoute
 from app.adapters.web.plugin.routes import FastAPIDynamicRouteRegistry
+from app.adapters.web.health import install_health_routes
 from app.application.plugin.routes import configure_plugin_routes
 from app.adapters.web.security.access import (
     configure_token_codec,
@@ -305,6 +306,8 @@ def create_app() -> FastAPI:
     _app.add_exception_handler(Exception, localized_unhandled_exception_handler)
     # 主程序静态路由统一使用 ResponseAPIRoute；动态插件注册时会显式覆盖为原生 APIRoute。
     _app.router.route_class = ResponseAPIRoute
+    # 编排器探针使用原生 APIRoute 和最小响应，不进入业务响应包络或版本前缀。
+    install_health_routes(_app)
 
     # 配置 CORS 中间件
     _app.add_middleware(

app/main.py

@@ -60,7 +60,6 @@ from app.runtime.topology import (
     UnsupportedProcessTopologyError,
     validate_process_topology,
 )
-from app.startup.database_initializer import prepare_database
 
 setproctitle.setproctitle(settings.PROJECT_NAME)
 
@@ -189,7 +188,6 @@ def run_application() -> None:
     signal.signal(signal.SIGINT, signal_handler)
 
     start_tray()
-    prepare_database()
     run_api_server()
 
 

app/runtime/health.py

@@ -0,0 +1,60 @@
+"""进程内应用健康状态。"""
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+from fastapi import FastAPI
+
+
+class ReadinessPhase(StrEnum):
+    """应用生命周期对公开 readiness 探针暴露的最小阶段。"""
+
+    STARTING = "starting"
+    READY = "ready"
+    STOPPING = "stopping"
+    FAILED = "failed"
+
+
+@dataclass(slots=True)
+class ApplicationHealth:
+    """保存单个 FastAPI 实例的数据库和生命周期就绪状态。"""
+
+    phase: ReadinessPhase = ReadinessPhase.STARTING
+    database_ready: bool = False
+
+    @property
+    def is_ready(self) -> bool:
+        """仅在数据库和完整生命周期均成功后返回就绪。"""
+        return self.phase is ReadinessPhase.READY and self.database_ready
+
+    def begin_startup(self) -> None:
+        """重置一次 lifespan 启动尝试的状态。"""
+        self.phase = ReadinessPhase.STARTING
+        self.database_ready = False
+
+    def mark_database_ready(self) -> None:
+        """记录数据库迁移和 head 校验已经完成。"""
+        self.database_ready = True
+
+    def mark_ready(self) -> None:
+        """记录所有 fail-fast 生命周期组件已经启动。"""
+        if not self.database_ready:
+            raise RuntimeError("数据库尚未就绪，不能发布应用 ready 状态")
+        self.phase = ReadinessPhase.READY
+
+    def mark_failed(self) -> None:
+        """记录启动存在不可恢复失败。"""
+        self.phase = ReadinessPhase.FAILED
+
+    def mark_stopping(self) -> None:
+        """在资源关闭前撤销 readiness。"""
+        self.phase = ReadinessPhase.STOPPING
+
+
+def get_application_health(app: FastAPI) -> ApplicationHealth:
+    """读取应用级健康状态；为直接构造的测试应用补齐默认状态。"""
+    health = getattr(app.state, "moviepilot_health", None)
+    if health is None:
+        health = ApplicationHealth()
+        app.state.moviepilot_health = health
+    return health

app/startup/database_initializer.py

@@ -11,7 +11,7 @@ from sqlalchemy import inspect
 from sqlalchemy.engine import Engine
 
 from app.runtime.config import settings
-from app.db import Base
+from app.db.base import Base
 from app.db.engine import get_engine
 from app.db.models import load_all_models
 from app.runtime.log import logger
@@ -121,6 +121,18 @@ def prepare_database(*, before_alembic: Callable[[], None] | None = None) -> Non
     update_db(alembic_cfg)
 
 
+def verify_database_revision() -> None:
+    """确认活动数据库已位于当前唯一 Alembic head，否则阻止 readiness。"""
+    engine = get_engine()
+    alembic_cfg = _build_alembic_config(engine)
+    _, current_heads, target_heads = _migration_state(engine, alembic_cfg)
+    if set(current_heads) != set(target_heads):
+        raise RuntimeError(
+            "数据库迁移完成后 revision 仍未到达当前 head："
+            f"current={current_heads}, target={target_heads}"
+        )
+
+
 def init_db():
     """
     初始化数据库

app/startup/lifecycle/__init__.py

@@ -27,6 +27,7 @@ except Exception:
 from app.chain.system import SystemChain
 from app.application.plugin.runtime import get_plugin_manager
 from app.runtime.config import global_vars, settings
+from app.runtime.health import get_application_health
 from app.runtime.topology import validate_process_topology
 from app.adapters.external.server import MoviePilotServerHelper
 from app.runtime.state import SystemHelper
@@ -133,14 +134,34 @@ def prepare_plugin_restore() -> None:
     SystemChain().restore_plugins()
 
 
+def prepare_database_component(app: FastAPI) -> None:
+    """完成数据库建表、迁移与 head 校验后发布数据库就绪状态。"""
+    # Alembic 及全部 ORM 元数据只在 lifespan 真正启动时加载，create_app/import 阶段
+    # 继续保持不建库、不加载迁移运行时的纯 ASGI 结构语义。
+    from app.startup.database_initializer import (
+        prepare_database,
+        verify_database_revision,
+    )
+
+    prepare_database()
+    verify_database_revision()
+    get_application_health(app).mark_database_ready()
+
+
 def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]:
     """按现有顺序构建应用组件清单，回调在每次 lifespan 启动时重新绑定。"""
     return (
+        LifecycleComponent(
+            name="数据库准备",
+            start=lambda: prepare_database_component(app),
+            start_order=10,
+            start_timeout_seconds=300,
+        ),
         LifecycleComponent(
             name="HTTP 基础能力",
             start=lambda: configure_default_user_agent(settings.USER_AGENT),
             stop=aclose_shared_async_transports,
-            start_order=10,
+            start_order=20,
             stop_order=80,
             start_timeout_seconds=30,
             stop_timeout_seconds=120,
@@ -149,28 +170,28 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]:
             name="领域依赖装配",
             dependencies=("HTTP 基础能力",),
             start=configure_domain_dependencies,
-            start_order=20,
+            start_order=30,
             start_timeout_seconds=30,
         ),
         LifecycleComponent(
             name="数据库引擎预热",
-            dependencies=("领域依赖装配",),
+            dependencies=("数据库准备", "领域依赖装配"),
             start=lambda: (get_engine(), get_global_async_engine()),
-            start_order=30,
+            start_order=40,
             start_timeout_seconds=120,
         ),
         LifecycleComponent(
             name="数据库连接预算",
             dependencies=("数据库引擎预热",),
             start=check_connection_budget,
-            start_order=40,
+            start_order=50,
             start_timeout_seconds=30,
         ),
         LifecycleComponent(
             name="路由",
             dependencies=("数据库连接预算",),
             start=lambda: init_routers(app),
-            start_order=50,
+            start_order=60,
             start_timeout_seconds=30,
         ),
         LifecycleComponent(
@@ -178,7 +199,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]:
             dependencies=("路由",),
             start=init_modules,
             stop=stop_modules,
-            start_order=60,
+            start_order=70,
             stop_order=70,
             start_timeout_seconds=300,
             stop_timeout_seconds=300,
@@ -188,7 +209,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]:
             dependencies=("模块服务",),
             mode=LifecycleMode.NORMAL_ONLY,
             start=prepare_plugin_restore,
-            start_order=70,
+            start_order=80,
             start_timeout_seconds=300,
         ),
         LifecycleComponent(
@@ -197,7 +218,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]:
             mode=LifecycleMode.NORMAL_ONLY,
             start=init_plugins,
             stop=stop_plugins,
-            start_order=80,
+            start_order=90,
             stop_order=60,
             start_timeout_seconds=300,
             stop_timeout_seconds=300,
@@ -208,7 +229,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]:
             mode=LifecycleMode.NORMAL_ONLY,
             start=init_scheduler,
             stop=stop_scheduler,
-            start_order=90,
+            start_order=100,
             stop_order=50,
             start_timeout_seconds=120,
             stop_timeout_seconds=120,
@@ -219,7 +240,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]:
             mode=LifecycleMode.NORMAL_ONLY,
             start=init_monitor,
             stop=stop_monitor,
-            start_order=100,
+            start_order=110,
             stop_order=40,
             start_timeout_seconds=120,
             stop_timeout_seconds=120,
@@ -229,7 +250,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]:
             dependencies=("监控器",),
             mode=LifecycleMode.NORMAL_ONLY,
             start=replay_pending_transfers,
-            start_order=110,
+            start_order=120,
             start_timeout_seconds=30,
         ),
         LifecycleComponent(
@@ -238,7 +259,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]:
             mode=LifecycleMode.NORMAL_ONLY,
             start=init_command,
             stop=stop_command,
-            start_order=120,
+            start_order=130,
             stop_order=30,
             start_timeout_seconds=120,
             stop_timeout_seconds=120,
@@ -249,7 +270,7 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]:
             mode=LifecycleMode.NORMAL_ONLY,
             start=init_workflow,
             stop=stop_workflow,
-            start_order=130,
+            start_order=140,
             stop_order=20,
             start_timeout_seconds=120,
             stop_timeout_seconds=120,
@@ -278,60 +299,51 @@ async def lifespan(app: FastAPI):
     """
     定义应用的生命周期事件
     """
-    validate_process_topology(
-        workers=settings.API_WORKERS,
-        safe_mode=settings.MOVIEPILOT_SAFE_MODE,
-    )
-    print("Starting up...")
-    # 存储当前循环
-    global_vars.set_loop(asyncio.get_event_loop())
-    # 同步与异步引擎各预热一次。引擎改为惰性创建后，两者的首次创建时机都不再由启动路径
-    # 决定，这一步把它们拉回来。必须排在所有 init_* 之前，两个理由：
-    #
-    # 其一，fail-fast 的落点。异步驱动缺失、异步 URL 拼错这类问题若不在这里暴露，会一路
-    # 推迟到第一个异步查询——表现为用户请求 500 或调度任务静默失败，而不是启动即崩。
-    # 故意不 try/except：起不来就该起不来，吞掉它等于把 fail-fast 又还回去了。而既然会抛，
-    # 就必须抛在 init_routers / init_modules 之前——下面的 try/finally 关停块要到 yield 处
-    # 才开始，在它之后抛异常，已经初始化好的模块就拿不到 stop_modules() 了。
-    #
-    # 其二，同步引擎的首次创建要落在单线程期。init_db() 会顺带预热它，但那只对
-    # run_application() 入口成立；外部 supervisor 直挂 ASGI app（如
-    # `gunicorn -k uvicorn.workers.UvicornWorker app.factory:app`）时 init_db() 根本不执行，
-    # 首次创建便退到运行期——而那时 init_scheduler() / init_monitor() 已经放出上百个线程，
-    # 引擎构建里那段 PRAGMA journal_mode 会让它们一起堵在创建锁上。
-    #
-    # 代价：异步侧几乎为零，create_async_engine 只校验 URL 与驱动导入、不建立连接；同步侧
-    # 会连一次库、设一遍 journal mode，在事件循环上阻塞一小会儿——但那一次本来就免不了，
-    # 放在这里至少还独占着单线程，而且此刻 uvicorn 尚未开始接请求。
-    components = build_lifecycle_components(app)
-    enabled_components = tuple(
-        component
-        for component in components
-        if component.enabled(settings.MOVIEPILOT_SAFE_MODE)
-    )
-    logger.info(
-        "启用生命周期组件：%s",
-        ", ".join(component.name for component in enabled_components),
-    )
-    for component in sorted(
-        (item for item in enabled_components if item.start is not None),
-        key=lambda item: item.start_order or 0,
-    ):
-        await run_startup_step(
-            component.name,
-            component.start,
-            component.start_timeout_seconds,
+    health = get_application_health(app)
+    health.begin_startup()
+    try:
+        validate_process_topology(
+            workers=settings.API_WORKERS,
+            safe_mode=settings.MOVIEPILOT_SAFE_MODE,
         )
-    if settings.MOVIEPILOT_SAFE_MODE:
-        print("MoviePilot safe mode enabled: skip plugins, scheduler, monitor, commands and workflow.")
-    # 插件同步到本地
-    sync_plugins_task = asyncio.create_task(
-        run_startup_step("插件同步与启动收尾", init_extra)
-    )
+        print("Starting up...")
+        # 存储当前循环
+        global_vars.set_loop(asyncio.get_event_loop())
+        components = build_lifecycle_components(app)
+        enabled_components = tuple(
+            component
+            for component in components
+            if component.enabled(settings.MOVIEPILOT_SAFE_MODE)
+        )
+        logger.info(
+            "启用生命周期组件：%s",
+            ", ".join(component.name for component in enabled_components),
+        )
+        for component in sorted(
+            (item for item in enabled_components if item.start is not None),
+            key=lambda item: item.start_order or 0,
+        ):
+            await run_startup_step(
+                component.name,
+                component.start,
+                component.start_timeout_seconds,
+            )
+        if settings.MOVIEPILOT_SAFE_MODE:
+            print("MoviePilot safe mode enabled: skip plugins, scheduler, monitor, commands and workflow.")
+        # 插件同步到本地
+        sync_plugins_task = asyncio.create_task(
+            run_startup_step("插件同步与启动收尾", init_extra)
+        )
+        health.mark_ready()
+    except BaseException:
+        # Uvicorn 在 lifespan 抛错时不会开始接流量；状态仍需供嵌入式入口和测试诊断。
+        health.mark_failed()
+        raise
     try:
         # 在此处 yield，表示应用已经启动，控制权交回 FastAPI 主事件循环
         yield
     finally:
+        health.mark_stopping()
         print("Shutting down...")
         global_vars.stop_system()
         # 插件恢复会在线程池中修改源码与依赖，必须完成后再进入资源关闭阶段。