From fc677a2bba098904c42a16eda0395fd9b50d33b1 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 01:53:00 +0800 Subject: [PATCH] feat(database): bound async configuration writes --- app/adapters/external/server.py | 34 ++++++++++++++ app/api/endpoints/plugin.py | 3 ++ app/application/configuration.py | 5 ++- app/application/database.py | 8 ++++ app/application/server/report.py | 43 ++++++++++++++++++ app/db/engine.py | 5 ++- app/db/worker.py | 20 +++++---- app/factory.py | 20 +++++++++ app/startup/lifecycle/__init__.py | 5 ++- app/startup/modules_initializer.py | 21 ++++++--- scripts/architecture/baseline.py | 35 ++++++++++++++- scripts/local_setup.py | 12 +++++ .../configuration-debt-baseline.json | 3 +- .../architecture/dependency-baseline.json | 39 ++++++++-------- tests/test_agent_lifecycle.py | 6 +-- tests/test_api_response.py | 27 +++++++++++ tests/test_architecture_contract_baseline.py | 1 + tests/test_architecture_dependencies.py | 4 +- tests/test_cache_system.py | 4 +- tests/test_configuration_initializer.py | 36 +++++++++++++++ tests/test_database_migration_startup.py | 30 +++++++++++++ tests/test_db_engine_postgresql.py | 11 +++++ tests/test_server_helper.py | 3 ++ tests/test_server_report_service.py | 45 ++++++++++++++++++- tests/test_userconfig_oper.py | 36 +++++++++++++++ 25 files changed, 408 insertions(+), 48 deletions(-) diff --git a/app/adapters/external/server.py b/app/adapters/external/server.py index c90547650..633aa9aae 100644 --- a/app/adapters/external/server.py +++ b/app/adapters/external/server.py @@ -359,6 +359,15 @@ class MoviePilotServerHelper: reporter=cls.sub_report, ) + @classmethod + async def async_init_subscribe_report(cls) -> None: + """异步初始化订阅统计标记。""" + await cls._report_service().async_init_report( + enabled=settings.SUBSCRIBE_STATISTIC_SHARE, + state_key=SystemConfigKey.SubscribeReport, + reporter=cls.async_sub_report, + ) + @classmethod def init_plugin_report(cls) -> None: """ @@ -370,6 +379,15 @@ class MoviePilotServerHelper: reporter=cls.install_plugin_report, ) + @classmethod + async def async_init_plugin_report(cls) -> None: + """异步初始化插件统计标记。""" + await cls._report_service().async_init_report( + enabled=settings.PLUGIN_STATISTIC_SHARE, + state_key=SystemConfigKey.PluginInstallReport, + reporter=cls.async_install_plugin_report, + ) + @staticmethod def _handle_list_response(res) -> List[dict]: """ @@ -677,6 +695,15 @@ class MoviePilotServerHelper: timeout=5, ) + @classmethod + async def async_subscribe_report(cls, subscribes: List[Dict[str, Any]]): + """异步批量上报存量订阅统计。""" + return await cls._async_post_json( + cls._server_url(cls._SUBSCRIBE_REPORT_PATH), + {"subscribes": subscribes}, + timeout=10, + ) + @classmethod def subscribe_report(cls, subscribes: List[Dict[str, Any]]): """ @@ -953,6 +980,13 @@ class MoviePilotServerHelper: enabled=settings.SUBSCRIBE_STATISTIC_SHARE, ) + @classmethod + async def async_sub_report(cls) -> bool: + """异步上报存量订阅统计。""" + return await cls._report_service().async_report_subscribes( + enabled=settings.SUBSCRIBE_STATISTIC_SHARE, + ) + @classmethod def sub_share( cls, diff --git a/app/api/endpoints/plugin.py b/app/api/endpoints/plugin.py index 76fd14fd5..951d23455 100644 --- a/app/api/endpoints/plugin.py +++ b/app/api/endpoints/plugin.py @@ -57,6 +57,7 @@ from app.api.dependencies.plugin import ( from app.adapters.external.server import MoviePilotServerHelper from app.adapters.external.market import PluginHelper from app.adapters.system.plugin.package import PluginPackageManager +from app.application.database import DatabaseWorkerOverloadedError from app.runtime.log import logger from app.schemas.types import SystemConfigKey @@ -879,6 +880,8 @@ async def save_plugin_folders( folders, ) return _SchemaResponse(success=True) + except DatabaseWorkerOverloadedError: + raise except Exception as e: logger.error(f"[文件夹API] 保存文件夹配置失败: {str(e)}") return _SchemaResponse(success=False, message=str(e)) diff --git a/app/application/configuration.py b/app/application/configuration.py index 01dd656ff..62969360d 100644 --- a/app/application/configuration.py +++ b/app/application/configuration.py @@ -6,7 +6,7 @@ from collections.abc import Callable from dataclasses import dataclass from functools import partial from pathlib import Path -from typing import Any, Optional, Protocol +from typing import Any, Optional, Protocol, cast from app.application.database import AsyncDatabaseExecutor from app.schemas.types import MediaType @@ -315,7 +315,8 @@ class SystemConfigService: """异步写入配置，并等待数据库提交或回滚完成。""" if self._async_executor is None: raise RuntimeError("系统配置异步数据库执行端口尚未配置") - return await self._async_executor.run(partial(self._writer.set, key, value)) + result = await self._async_executor.run(partial(self._writer.set, key, value)) + return cast(bool | None, result) def delete(self, key: Any) -> Any: """删除配置。""" diff --git a/app/application/database.py b/app/application/database.py index d69414d27..78e6dafa5 100644 --- a/app/application/database.py +++ b/app/application/database.py @@ -18,6 +18,14 @@ DatabaseProbe = Callable[[], Optional[str]] T = TypeVar("T") +class DatabaseWorkerClosedError(RuntimeError): + """数据库执行器尚未启动或已经停止。""" + + +class DatabaseWorkerOverloadedError(RuntimeError): + """数据库执行器的运行与排队容量已经用尽。""" + + class AsyncDatabaseExecutor(Protocol): """让异步业务调用同步短事务而不阻塞事件循环。""" diff --git a/app/application/server/report.py b/app/application/server/report.py index eded84acc..b60490aca 100644 --- a/app/application/server/report.py +++ b/app/application/server/report.py @@ -24,19 +24,25 @@ class ServerReportService: config_writer: Callable[[Any, Any], Any], installed_plugins_provider: Callable[[], list[str]], subscribes_provider: Callable[[], list[Any]], + async_subscribes_provider: Callable[[], Awaitable[list[Any]]] | None = None, plugin_report_sender: Callable[[list[dict]], Any], async_plugin_report_sender: Callable[[list[dict]], Awaitable[Any]], subscribe_report_sender: Callable[[list[dict]], Any], repo_url_sanitizer: Callable[[Optional[str]], Optional[str]], + async_subscribe_report_sender: Callable[[list[dict]], Awaitable[Any]] | None = None, + async_config_writer: Callable[[Any, Any], Awaitable[Any]] | None = None, ) -> None: """保存本地读取端口和只负责 I/O 的中心服务发送端口。""" self._config_reader = config_reader self._config_writer = config_writer self._installed_plugins_provider = installed_plugins_provider self._subscribes_provider = subscribes_provider + self._async_subscribes_provider = async_subscribes_provider self._plugin_report_sender = plugin_report_sender self._async_plugin_report_sender = async_plugin_report_sender self._subscribe_report_sender = subscribe_report_sender + self._async_subscribe_report_sender = async_subscribe_report_sender + self._async_config_writer = async_config_writer self._repo_url_sanitizer = repo_url_sanitizer def init_report( @@ -50,6 +56,22 @@ class ServerReportService: if enabled and not self._config_reader(state_key) and reporter(): self._config_writer(state_key, "1") + async def async_init_report( + self, + *, + enabled: bool, + state_key: Any, + reporter: Callable[[], Awaitable[bool]], + ) -> None: + """异步完成首次上报，并通过异步配置端口持久化完成标记。""" + if not enabled or self._config_reader(state_key): + return + if not await reporter(): + return + if self._async_config_writer is None: + raise RuntimeError("中心服务上报未配置异步配置写入端口") + await self._async_config_writer(state_key, "1") + def build_subscribe_payload(self, item: Optional[dict]) -> Optional[dict]: """构造中心服务订阅统计载荷并移除本地运行字段。""" if not isinstance(item, dict): @@ -132,3 +154,24 @@ class ServerReportService: return False response = await self._async_plugin_report_sender(payload) return bool(response is not None and response.status_code == 200) + + async def async_report_subscribes(self, *, enabled: bool) -> bool: + """异步上报当前全部有效订阅的公开统计字段。""" + if not enabled: + return False + if self._async_subscribe_report_sender is None: + raise RuntimeError("中心服务上报未配置异步订阅发送端口") + if self._async_subscribes_provider is None: + raise RuntimeError("中心服务未配置异步订阅读取端口") + subscribes = await self._async_subscribes_provider() + if not subscribes: + return True + payloads = [ + payload + for subscribe in subscribes + if (payload := self.build_subscribe_payload(subscribe.to_dict())) + ] + if not payloads: + return True + response = await self._async_subscribe_report_sender(payloads) + return bool(response is not None and response.status_code == 200) diff --git a/app/db/engine.py b/app/db/engine.py index c7f2cc373..0967c453a 100644 --- a/app/db/engine.py +++ b/app/db/engine.py @@ -14,6 +14,7 @@ from sqlalchemy.pool import Pool from app.runtime.config import settings from app.db.diagnostics import _register_database_error_logging +from app.db.worker import DATABASE_WORKER_MAX_WORKERS from app.runtime.log import logger from app.runtime.observability import record_metric @@ -304,8 +305,8 @@ def connection_budget() -> Dict[str, int]: else: sync_max = settings.DB_SQLITE_POOL_SIZE + settings.DB_SQLITE_MAX_OVERFLOW if settings.DB_POOL_TYPE == "NullPool": - # 同步侧也可能被配成 NullPool，此时同样无界，用线程池规模作为可观测的上限估计 - sync_max = settings.CONF.threadpool + # 未池化连接由通用线程池和专属数据库 worker 共同创建，二者都要计入上限估计。 + sync_max = settings.CONF.threadpool + DATABASE_WORKER_MAX_WORKERS async_max = (settings.DB_ASYNC_POOL_SIZE + settings.DB_ASYNC_MAX_OVERFLOW if _async_pool_enabled() else 0) fallback = settings.DB_ASYNC_FALLBACK_LIMIT if _async_pool_enabled() else settings.CONF.scheduler diff --git a/app/db/worker.py b/app/db/worker.py index b99f0883c..b00ee45ae 100644 --- a/app/db/worker.py +++ b/app/db/worker.py @@ -10,18 +10,17 @@ from contextvars import copy_context from dataclasses import dataclass from typing import Callable, TypeVar +from app.application.database import ( + DatabaseWorkerClosedError, + DatabaseWorkerOverloadedError, +) from app.runtime.observability import record_metric T = TypeVar("T") - -class DatabaseWorkerClosedError(RuntimeError): - """数据库执行器尚未启动或已经停止。""" - - -class DatabaseWorkerOverloadedError(RuntimeError): - """数据库执行器的运行与排队容量已经用尽。""" +DATABASE_WORKER_MAX_WORKERS = 4 +DATABASE_WORKER_CAPACITY = 32 @dataclass(frozen=True, slots=True) @@ -47,7 +46,12 @@ class _WorkItem: class DatabaseWorker: """以有限线程和队列执行不能原生异步化的数据库短事务。""" - def __init__(self, *, max_workers: int = 4, capacity: int = 32) -> None: + def __init__( + self, + *, + max_workers: int = DATABASE_WORKER_MAX_WORKERS, + capacity: int = DATABASE_WORKER_CAPACITY, + ) -> None: """保存容量配置，线程只在显式启动后创建。""" if max_workers < 1: raise ValueError("数据库 worker 线程数必须大于 0") diff --git a/app/factory.py b/app/factory.py index ffe39fe6e..4ab628db1 100644 --- a/app/factory.py +++ b/app/factory.py @@ -14,6 +14,7 @@ from app.adapters.observability.otel import build_observation_port 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.application.database import DatabaseWorkerOverloadedError from app.adapters.web.security.access import ( configure_token_codec, verify_apikey, @@ -222,6 +223,21 @@ async def localized_http_exception_handler( ) +async def database_worker_overloaded_handler( + request: Request, + _exc: DatabaseWorkerOverloadedError, +) -> JSONResponse: + """将数据库短事务背压映射为可重试的 503 响应。""" + return await localized_http_exception_handler( + request, + HTTPException( + status_code=503, + detail="服务当前繁忙，请稍后重试", + headers={"Retry-After": "1"}, + ), + ) + + async def localized_validation_exception_handler( request: Request, exc: RequestValidationError, @@ -306,6 +322,10 @@ def create_app() -> FastAPI: ) _app.add_exception_handler(HTTPException, localized_http_exception_handler) + _app.add_exception_handler( + DatabaseWorkerOverloadedError, + database_worker_overloaded_handler, + ) _app.add_exception_handler( RequestValidationError, localized_validation_exception_handler, diff --git a/app/startup/lifecycle/__init__.py b/app/startup/lifecycle/__init__.py index 68705f3b6..b23a10f6e 100644 --- a/app/startup/lifecycle/__init__.py +++ b/app/startup/lifecycle/__init__.py @@ -135,7 +135,10 @@ async def initialize_modules_component(app: FastAPI) -> None: except BaseException: from app.startup.modules_initializer import stop_database_worker - await stop_database_worker() + try: + await stop_database_worker() + except Exception as cleanup_error: # noqa: BLE001 保留原始启动异常 + logger.error(f"启动失败后的数据库任务清理失败：{cleanup_error}") raise if runtime is not None: app.state.host_runtime = runtime diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index 598da1dc7..8063e259a 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -165,9 +165,9 @@ async def stop_database_worker() -> None: """停止当前进程的数据库短事务 worker。""" global _database_worker worker = _database_worker - _database_worker = None if worker is not None: await worker.shutdown() + _database_worker = None async def _initialize_configuration_services( @@ -237,15 +237,20 @@ def configure_runtime_data_providers() -> None: report_service=ServerReportService( config_reader=lambda key: get_configured_system_config().get(key), config_writer=lambda key, value: get_configured_system_config().set(key, value), + async_config_writer=lambda key, value: get_configured_system_config().async_set( + key, value + ), installed_plugins_provider=lambda: get_configured_system_config().get( SystemConfigKey.UserInstalledPlugins ) or [], subscribes_provider=lambda: SubscribeOper().list(), + async_subscribes_provider=lambda: SubscribeOper().async_list(), plugin_report_sender=MoviePilotServerHelper.plugin_install_report, async_plugin_report_sender=( MoviePilotServerHelper.async_plugin_install_report ), subscribe_report_sender=MoviePilotServerHelper.subscribe_report, + async_subscribe_report_sender=MoviePilotServerHelper.async_subscribe_report, repo_url_sanitizer=MoviePilotServerHelper.sanitize_plugin_repo_url, ), sharing_service=ServerSharingService( @@ -542,7 +547,10 @@ async def stop_modules(): await run_step("Redis缓存连接", lambda: RedisHelper().close()) await run_step("异步Redis缓存连接", lambda: AsyncRedisHelper().close()) await run_step("数据库任务", stop_database_worker) - await run_step("数据库连接", close_database) + if _database_worker is None: + await run_step("数据库连接", close_database) + else: + logger.error("数据库任务未收敛，跳过数据库连接关闭以避免运行中事务使用已释放连接") await run_step("前端服务", stop_frontend) await run_step("临时文件", clear_temp) @@ -567,7 +575,10 @@ async def init_modules() -> HostRuntime: try: await _initialize_configuration_services(database_worker) except BaseException: - await stop_database_worker() + try: + await stop_database_worker() + except Exception as cleanup_error: # noqa: BLE001 保留原始启动异常 + logger.error(f"启动失败后的数据库任务清理失败：{cleanup_error}") raise # 数据访问能力统一在启动组合根注入，Runtime 和 Adapter 不再直接依赖 Oper。 api_data = ApiDataPorts( @@ -731,8 +742,8 @@ async def init_modules() -> HostRuntime: # 启动事件消费 EventManager().start() # 初始化共享服务端状态 - MoviePilotServerHelper.init_plugin_report() - MoviePilotServerHelper.init_subscribe_report() + await MoviePilotServerHelper.async_init_plugin_report() + await MoviePilotServerHelper.async_init_subscribe_report() MoviePilotServerHelper.get_user_uuid() MoviePilotServerHelper.get_github_user() # 初始化AI智能体 diff --git a/scripts/architecture/baseline.py b/scripts/architecture/baseline.py index fa2442f8c..3aa8693db 100644 --- a/scripts/architecture/baseline.py +++ b/scripts/architecture/baseline.py @@ -63,6 +63,7 @@ CONFIGURATION_EXCLUDED_ROOTS = ( APP_ROOT / "plugins", APP_ROOT / "sdk", APP_ROOT / "runtime" / "compat", + APP_ROOT / "testing", ) @@ -85,6 +86,37 @@ def parse_source(path: Path) -> ast.Module: return ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) +def _is_type_checking_test(test: ast.expr) -> bool: + """判断条件是否只在静态类型检查阶段成立。""" + return ( + isinstance(test, ast.Name) + and test.id == "TYPE_CHECKING" + ) or ( + isinstance(test, ast.Attribute) + and isinstance(test.value, ast.Name) + and test.value.id == "typing" + and test.attr == "TYPE_CHECKING" + ) + + +def iter_runtime_import_nodes(tree: ast.AST): + """遍历运行期导入，排除 ``if TYPE_CHECKING`` 内的仅类型依赖。""" + parents: dict[ast.AST, ast.AST] = {} + for parent in ast.walk(tree): + for child in ast.iter_child_nodes(parent): + parents[child] = parent + for node in ast.walk(tree): + if not isinstance(node, (ast.Import, ast.ImportFrom)): + continue + parent = parents.get(node) + while parent is not None: + if isinstance(parent, ast.If) and _is_type_checking_test(parent.test): + break + parent = parents.get(parent) + else: + yield node + + def collect_configuration_debt_baseline() -> dict[str, Any]: """收集宿主 canonical 代码直接读取 settings 和构造数据库配置适配器的债务。""" settings_files: list[str] = [] @@ -124,6 +156,7 @@ def collect_configuration_debt_baseline() -> dict[str, Any]: "app/plugins", "app/sdk", "app/runtime/compat", + "app/testing", ], }, "settings_imports": { @@ -144,7 +177,7 @@ def iter_import_candidates( """提取模块导入候选，第二项记录 from-import 的具体符号。""" package = module_name if path.name == "__init__.py" else module_name.rpartition(".")[0] candidates: list[tuple[str, Optional[str]]] = [] - for node in ast.walk(parse_source(path)): + for node in iter_runtime_import_nodes(parse_source(path)): if isinstance(node, ast.Import): candidates.extend((alias.name, None) for alias in node.names) continue diff --git a/scripts/local_setup.py b/scripts/local_setup.py index 86d19ced8..1c0dcaa16 100644 --- a/scripts/local_setup.py +++ b/scripts/local_setup.py @@ -2383,11 +2383,23 @@ def _apply_local_system_config_inner(config_payload: dict[str, Any]) -> None: from app.startup.database_initializer import prepare_database from app.db.oper.systemconfig import SystemConfigOper from app.schemas.types import SystemConfigKey + from app.db.session import SessionFactory, async_session_scope + from app.db.uow import configure_transaction_runners + from app.startup.transaction import TransactionalWriteRunner except ModuleNotFoundError as exc: raise RuntimeError( "当前环境尚未安装 MoviePilot 运行依赖，请先执行 moviepilot install deps 或 moviepilot setup" ) from exc + transaction_runner = TransactionalWriteRunner( + sync_session=SessionFactory, + async_session=async_session_scope, + ) + configure_transaction_runners( + sync=transaction_runner.sync, + async_=transaction_runner.async_, + ) + generated_password = None def prepare_superuser_password() -> None: diff --git a/tests/fixtures/architecture/configuration-debt-baseline.json b/tests/fixtures/architecture/configuration-debt-baseline.json index dc1e86d8c..8ec3b9b75 100644 --- a/tests/fixtures/architecture/configuration-debt-baseline.json +++ b/tests/fixtures/architecture/configuration-debt-baseline.json @@ -4,7 +4,8 @@ "excluded": [ "app/plugins", "app/sdk", - "app/runtime/compat" + "app/runtime/compat", + "app/testing" ], "root": "app" }, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 27928e3be..5ec5adf11 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -2054,6 +2054,7 @@ "app.api.endpoints.plugin -> app.application", "app.api.endpoints.plugin -> app.application.commands", "app.api.endpoints.plugin -> app.application.configuration", + "app.api.endpoints.plugin -> app.application.database", "app.api.endpoints.plugin -> app.application.plugin", "app.api.endpoints.plugin -> app.application.plugin.config", "app.api.endpoints.plugin -> app.application.plugin.folders", @@ -2468,13 +2469,12 @@ "app.application.chain.durable_events -> app.schemas.file", "app.application.chain.durable_events -> app.schemas.transfer", "app.application.chain.durable_events -> app.schemas.types", + "app.application.configuration -> app.application", + "app.application.configuration -> app.application.database", "app.application.configuration -> app.schemas", "app.application.configuration -> app.schemas.types", "app.application.dashboard -> app.schemas", "app.application.dashboard -> app.schemas.dashboard", - "app.application.database -> app.application", - "app.application.database -> app.application.backup", - "app.application.database -> app.application.maintenance", "app.application.directory -> app.adapters", "app.application.directory -> app.adapters.system", "app.application.directory -> app.adapters.system.host", @@ -2714,6 +2714,8 @@ "app.application.security.url -> app.runtime", "app.application.security.url -> app.runtime.coalesce", "app.application.security.url -> app.runtime.log", + "app.application.security.userconfig -> app.application", + "app.application.security.userconfig -> app.application.database", "app.application.servarr -> app.schemas", "app.application.servarr -> app.schemas.types", "app.application.server.report -> app.schemas", @@ -3454,6 +3456,7 @@ "app.db.diagnostics -> app.runtime.log", "app.db.engine -> app.db", "app.db.engine -> app.db.diagnostics", + "app.db.engine -> app.db.worker", "app.db.engine -> app.runtime", "app.db.engine -> app.runtime.config", "app.db.engine -> app.runtime.log", @@ -3562,23 +3565,6 @@ "app.db.models.workflow -> app.db", "app.db.models.workflow -> app.db.base", "app.db.models.workflow -> app.db.decorators", - "app.db.oper -> app.db", - "app.db.oper -> app.db.oper.agentchat", - "app.db.oper -> app.db.oper.agenttask", - "app.db.oper -> app.db.oper.downloadfailure", - "app.db.oper -> app.db.oper.downloadhistory", - "app.db.oper -> app.db.oper.mediaserver", - "app.db.oper -> app.db.oper.message", - "app.db.oper -> app.db.oper.plugindata", - "app.db.oper -> app.db.oper.site", - "app.db.oper -> app.db.oper.subscribe", - "app.db.oper -> app.db.oper.subscribehistory", - "app.db.oper -> app.db.oper.systemconfig", - "app.db.oper -> app.db.oper.transferhistory", - "app.db.oper -> app.db.oper.transferpending", - "app.db.oper -> app.db.oper.user", - "app.db.oper -> app.db.oper.userconfig", - "app.db.oper -> app.db.oper.workflow", "app.db.oper.agentchat -> app.db", "app.db.oper.agentchat -> app.db.base", "app.db.oper.agentchat -> app.db.models", @@ -3682,6 +3668,10 @@ "app.db.session -> app.runtime.config", "app.db.session -> app.runtime.log", "app.db.session -> app.runtime.observability", + "app.db.worker -> app.application", + "app.db.worker -> app.application.database", + "app.db.worker -> app.runtime", + "app.db.worker -> app.runtime.observability", "app.doctor.checks -> app.adapters", "app.doctor.checks -> app.adapters.system", "app.doctor.checks -> app.adapters.system.backup", @@ -3805,6 +3795,7 @@ "app.factory -> app.api", "app.factory -> app.api.response", "app.factory -> app.application", + "app.factory -> app.application.database", "app.factory -> app.application.plugin", "app.factory -> app.application.plugin.routes", "app.factory -> app.application.security", @@ -6137,6 +6128,7 @@ "app.startup.modules_initializer -> app.db.oper.workflow", "app.startup.modules_initializer -> app.db.session", "app.startup.modules_initializer -> app.db.uow", + "app.startup.modules_initializer -> app.db.worker", "app.startup.modules_initializer -> app.runtime", "app.startup.modules_initializer -> app.runtime.cache", "app.startup.modules_initializer -> app.runtime.config", @@ -6259,6 +6251,10 @@ "app.testing -> app.testing.stub", "app.testing.bootstrap -> app.application", "app.testing.bootstrap -> app.application.site", + "app.testing.bootstrap -> app.db", + "app.testing.bootstrap -> app.db.oper", + "app.testing.bootstrap -> app.db.oper.systemconfig", + "app.testing.bootstrap -> app.db.oper.userconfig", "app.testing.bootstrap -> app.startup", "app.testing.bootstrap -> app.startup.cache_initializer", "app.testing.bootstrap -> app.startup.database_initializer", @@ -6438,7 +6434,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 796, + "module_count": 797, "modules": [ "app", "app.adapters", @@ -6862,6 +6858,7 @@ "app.db.oper.workflow", "app.db.session", "app.db.uow", + "app.db.worker", "app.doctor", "app.doctor.checks", "app.doctor.formatters", diff --git a/tests/test_agent_lifecycle.py b/tests/test_agent_lifecycle.py index fe4e42b76..f5519c85b 100644 --- a/tests/test_agent_lifecycle.py +++ b/tests/test_agent_lifecycle.py @@ -123,15 +123,15 @@ async def test_agent_initialization_failure_does_not_stop_module_startup( monkeypatch.setattr(modules_initializer, "user_auth", MagicMock()) monkeypatch.setattr(modules_initializer.EventManager, "start", MagicMock()) for name in ( - "init_plugin_report", - "init_subscribe_report", + "async_init_plugin_report", + "async_init_subscribe_report", "get_user_uuid", "get_github_user", ): monkeypatch.setattr( modules_initializer.MoviePilotServerHelper, name, - MagicMock(), + AsyncMock() if name.startswith("async_") else MagicMock(), ) start_frontend = MagicMock() check_auth = MagicMock() diff --git a/tests/test_api_response.py b/tests/test_api_response.py index 6dba78121..c4f48c476 100644 --- a/tests/test_api_response.py +++ b/tests/test_api_response.py @@ -17,10 +17,12 @@ from app.api.response import ( ResponseAPIRouter, ) from app.factory import ( + database_worker_overloaded_handler, localized_http_exception_handler, localized_unhandled_exception_handler, localized_validation_exception_handler, ) +from app.application.database import DatabaseWorkerOverloadedError from app.runtime.localization import LocaleHelper from app.schemas.common import JsonData from app.schemas.response import Response @@ -59,6 +61,10 @@ def api_app() -> FastAPI: app = FastAPI() app.router.route_class = ResponseAPIRoute app.add_exception_handler(HTTPException, localized_http_exception_handler) + app.add_exception_handler( + DatabaseWorkerOverloadedError, + database_worker_overloaded_handler, + ) from fastapi.exceptions import RequestValidationError app.add_exception_handler( @@ -112,6 +118,11 @@ def api_app() -> FastAPI: """抛出需要隐藏内部细节的未捕获异常。""" raise RuntimeError("private failure detail") + @app.get("/database-busy") + async def get_database_busy() -> None: + """模拟数据库短事务容量耗尽。""" + raise DatabaseWorkerOverloadedError("worker full") + @app.get("/native", response_model=None) async def get_native_response() -> dict[str, bool]: """返回显式旁路的原生 JSON 协议。""" @@ -190,6 +201,22 @@ async def test_accept_language_localizes_success_and_http_error(api_app: FastAPI assert zh_error_response.json()["message"] == "用户名或密码错误" +async def test_database_worker_overload_is_retryable_service_unavailable( + api_app: FastAPI, +): + """数据库 worker 背压应返回 503，而不是伪装成未知错误。""" + async with make_client(api_app) as client: + response = await client.get("/database-busy") + + assert response.status_code == 503 + assert response.headers["retry-after"] == "1" + assert response.json() == { + "success": False, + "message": "服务当前繁忙，请稍后重试", + "data": None, + } + + async def test_validation_error_uses_unified_model(api_app: FastAPI): """请求参数校验失败应返回统一协议和明确的错误项结构。""" async with make_client(api_app) as client: diff --git a/tests/test_architecture_contract_baseline.py b/tests/test_architecture_contract_baseline.py index 5ac76c595..e00474714 100644 --- a/tests/test_architecture_contract_baseline.py +++ b/tests/test_architecture_contract_baseline.py @@ -180,6 +180,7 @@ def test_configuration_debt_baseline_tracks_canonical_direct_access() -> None: "app/plugins", "app/sdk", "app/runtime/compat", + "app/testing", ] assert baseline["settings_imports"]["count"] == len( baseline["settings_imports"]["files"] diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index 1b448ae29..5b3936e0b 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -2,6 +2,8 @@ import ast from functools import lru_cache from pathlib import Path +from scripts.architecture.baseline import iter_runtime_import_nodes + PROJECT_ROOT = Path(__file__).parents[1] APP_ROOT = PROJECT_ROOT / "app" LEGACY_ROOTS = ("app.core", "app.helper", "app.utils") @@ -156,7 +158,7 @@ def _resolve_imports( tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) package = module_name if path.name == "__init__.py" else module_name.rpartition(".")[0] dependencies: set[str] = set() - for node in ast.walk(tree): + for node in iter_runtime_import_nodes(tree): candidates: list[str] = [] if isinstance(node, ast.Import): candidates.extend(alias.name for alias in node.names) diff --git a/tests/test_cache_system.py b/tests/test_cache_system.py index aa38f68e2..2241cb5fb 100644 --- a/tests/test_cache_system.py +++ b/tests/test_cache_system.py @@ -155,8 +155,8 @@ def test_init_modules_does_not_clear_package_tool_cache(monkeypatch): monkeypatch.setattr(modules_initializer, "user_auth", lambda: None) monkeypatch.setattr(modules_initializer, "ModuleManager", lambda: None) monkeypatch.setattr(modules_initializer.EventManager, "start", lambda self: None) - monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "init_plugin_report", lambda: None) - monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "init_subscribe_report", lambda: None) + monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "async_init_plugin_report", AsyncMock()) + monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "async_init_subscribe_report", AsyncMock()) monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "get_user_uuid", lambda: None) monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "get_github_user", lambda: None) init_agent = AsyncMock() diff --git a/tests/test_configuration_initializer.py b/tests/test_configuration_initializer.py index 3f3be0f5a..84aaae367 100644 --- a/tests/test_configuration_initializer.py +++ b/tests/test_configuration_initializer.py @@ -102,3 +102,39 @@ async def test_modules_startup_failure_stops_database_worker(monkeypatch) -> Non await initialize_modules_component(object()) stop_worker.assert_awaited_once_with() + + +@pytest.mark.asyncio +async def test_modules_startup_failure_preserves_original_error_when_cleanup_fails( + monkeypatch, +) -> None: + """数据库任务清理失败时仍向上层保留原始启动异常。""" + monkeypatch.setattr( + "app.startup.lifecycle.init_modules", + AsyncMock(side_effect=RuntimeError("startup failed")), + ) + monkeypatch.setattr( + modules_initializer, + "stop_database_worker", + AsyncMock(side_effect=RuntimeError("cleanup failed")), + ) + + with pytest.raises(RuntimeError, match="startup failed"): + await initialize_modules_component(object()) + + +@pytest.mark.asyncio +async def test_database_worker_owner_is_retained_when_shutdown_fails(monkeypatch) -> None: + """数据库 worker 关闭失败时保留 owner，允许后续重试或诊断。""" + + class _FailingWorker: + async def shutdown(self): + raise RuntimeError("shutdown failed") + + worker = _FailingWorker() + monkeypatch.setattr(modules_initializer, "_database_worker", worker) + + with pytest.raises(RuntimeError, match="shutdown failed"): + await modules_initializer.stop_database_worker() + + assert modules_initializer._database_worker is worker diff --git a/tests/test_database_migration_startup.py b/tests/test_database_migration_startup.py index 033c01ef5..8e6f9c8bc 100644 --- a/tests/test_database_migration_startup.py +++ b/tests/test_database_migration_startup.py @@ -31,6 +31,7 @@ from app.startup import database_initializer as db_init from app.startup import database as startup_database from app.startup import lifecycle from app.runtime.health import get_application_health +from app.db.models.systemconfig import SystemConfig LOCAL_SETUP_PATH = ( @@ -782,3 +783,32 @@ def test_local_setup_returns_failure_when_database_migration_fails( assert module.main() == 1 assert "migration failed" in capsys.readouterr().err + + +def test_local_setup_apply_config_registers_offline_transaction_runner( + monkeypatch, + tmp_path: Path, + db, +) -> None: + """离线 apply-config 写入配置前必须装配同步事务执行器。""" + db.watermark(SystemConfig) + module = _load_local_setup_module() + monkeypatch.setattr(db_init, "prepare_database", lambda **_kwargs: None) + monkeypatch.setattr(module, "_ensure_superuser_account_inner", lambda: None) + payload = { + "directories": [{ + "name": "offline-config", + "download_path": str(tmp_path / "downloads"), + "library_path": str(tmp_path / "library"), + "priority": 0, + }], + } + + module._apply_local_system_config_inner(payload) + + persisted = SystemConfig.get_by_key( + db.session, + "Directories", + ) + assert persisted is not None + assert persisted.value[0]["name"] == "offline-config" diff --git a/tests/test_db_engine_postgresql.py b/tests/test_db_engine_postgresql.py index 02c519659..e9b4734df 100644 --- a/tests/test_db_engine_postgresql.py +++ b/tests/test_db_engine_postgresql.py @@ -103,6 +103,17 @@ def test_budget_uses_sqlite_pool_for_sqlite(monkeypatch): assert engine_module.connection_budget()["sync"] == 7 +def test_budget_counts_database_worker_with_sync_nullpool(monkeypatch): + """同步 NullPool 需要同时计入通用线程池和专属数据库 worker。""" + monkeypatch.setattr(settings, "DB_TYPE", "postgresql", raising=False) + monkeypatch.setattr(settings, "DB_POOL_TYPE", "NullPool", raising=False) + threadpool_size = settings.CONF.threadpool + + budget = engine_module.connection_budget() + + assert budget["sync"] == threadpool_size + 4 + + # --------------------------------------------------------------------------- # # 额度校验（PostgreSQL 路径） # --------------------------------------------------------------------------- # diff --git a/tests/test_server_helper.py b/tests/test_server_helper.py index 41ff73eb2..85150b4da 100644 --- a/tests/test_server_helper.py +++ b/tests/test_server_helper.py @@ -29,9 +29,12 @@ class MoviePilotServerHelperTests(unittest.TestCase): config_writer=Mock(), installed_plugins_provider=Mock(return_value=[]), subscribes_provider=Mock(return_value=[]), + async_subscribes_provider=AsyncMock(return_value=[]), plugin_report_sender=Mock(), async_plugin_report_sender=AsyncMock(), subscribe_report_sender=Mock(), + async_subscribe_report_sender=AsyncMock(), + async_config_writer=AsyncMock(), repo_url_sanitizer=MoviePilotServerHelper.sanitize_plugin_repo_url, ), sharing_service=ServerSharingService( diff --git a/tests/test_server_report_service.py b/tests/test_server_report_service.py index 37f408645..49bfb01f1 100644 --- a/tests/test_server_report_service.py +++ b/tests/test_server_report_service.py @@ -1,5 +1,7 @@ from types import SimpleNamespace -from unittest.mock import Mock +from unittest.mock import AsyncMock, Mock + +import pytest from app.application.server.report import ServerReportService @@ -11,6 +13,7 @@ def _service(**overrides) -> ServerReportService: "config_writer": Mock(), "installed_plugins_provider": Mock(return_value=[]), "subscribes_provider": Mock(return_value=[]), + "async_subscribes_provider": AsyncMock(return_value=[]), "plugin_report_sender": Mock( return_value=SimpleNamespace(status_code=200) ), @@ -18,6 +21,8 @@ def _service(**overrides) -> ServerReportService: "subscribe_report_sender": Mock( return_value=SimpleNamespace(status_code=200) ), + "async_subscribe_report_sender": AsyncMock(), + "async_config_writer": AsyncMock(), "repo_url_sanitizer": lambda value: value, } defaults.update(overrides) @@ -62,6 +67,44 @@ def test_initial_report_marker_is_written_only_after_success(): writer.assert_not_called() +@pytest.mark.asyncio +async def test_async_initial_report_marker_uses_async_writer_after_success(): + """异步首次上报成功后只通过异步配置端口写完成标记。""" + sync_writer = Mock() + async_writer = AsyncMock() + reporter = AsyncMock(return_value=True) + service = _service( + config_writer=sync_writer, + async_config_writer=async_writer, + ) + + await service.async_init_report( + enabled=True, + state_key="report", + reporter=reporter, + ) + + reporter.assert_awaited_once_with() + async_writer.assert_awaited_once_with("report", "1") + sync_writer.assert_not_called() + + +@pytest.mark.asyncio +async def test_async_subscribe_report_uses_async_reader(): + """异步订阅上报通过异步读取端口获取数据，不在事件循环内查同步库。""" + sync_reader = Mock(side_effect=AssertionError("不应调用同步订阅读取")) + async_reader = AsyncMock(return_value=[]) + service = _service( + subscribes_provider=sync_reader, + async_subscribes_provider=async_reader, + ) + + assert await service.async_report_subscribes(enabled=True) is True + + sync_reader.assert_not_called() + async_reader.assert_awaited_once_with() + + def test_plugin_report_sanitizes_explicit_sources_before_transport(): """插件统计载荷在进入传输适配器前完成来源脱敏。""" sender = Mock(return_value=SimpleNamespace(status_code=200)) diff --git a/tests/test_userconfig_oper.py b/tests/test_userconfig_oper.py index 1221307b4..b48bba25c 100644 --- a/tests/test_userconfig_oper.py +++ b/tests/test_userconfig_oper.py @@ -91,3 +91,39 @@ async def test_async_write_uses_same_repository_rule() -> None: username="async-user", key="theme", ).value == "dark" + + +def test_existing_falsey_value_is_removed_from_db_but_kept_until_reload(db) -> None: + """已有用户配置写入假值时删除记录，当前快照仍保留该假值直到重载。""" + db.watermark(UserConfig) + oper = _fresh_oper() + + oper.set("falsey-user", "enabled", True) + oper.set("falsey-user", "enabled", False) + + assert UserConfig.get_by_key( + oper._db, + username="falsey-user", + key="enabled", + ) is None + assert oper.get("falsey-user", "enabled") is False + + oper.load_snapshot() + assert oper.get("falsey-user", "enabled") is None + + +def test_falsey_value_without_existing_row_is_persisted(db) -> None: + """不存在的用户配置写入假值时保留记录，兼容历史写入规则。""" + db.watermark(UserConfig) + oper = _fresh_oper() + + oper.set("new-falsey-user", "enabled", False) + + persisted = UserConfig.get_by_key( + oper._db, + username="new-falsey-user", + key="enabled", + ) + assert persisted is not None + assert persisted.value is False + assert oper.get("new-falsey-user", "enabled") is False