From 3d7fb44dc7fdceadd55ee30cb90d9add066ba821 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 01:10:26 +0800 Subject: [PATCH 1/7] refactor(database): isolate configuration transactions --- app/api/endpoints/plugin.py | 10 +- app/api/endpoints/site.py | 4 +- app/api/endpoints/user.py | 12 +- app/application/configuration.py | 22 +-- app/application/database.py | 24 ++- app/application/security/userconfig.py | 21 ++- app/db/oper/systemconfig.py | 133 +++++++------- app/db/oper/userconfig.py | 92 ++++++---- app/db/worker.py | 234 ++++++++++++++++++++++++ app/runtime/observability/__init__.py | 5 + app/startup/lifecycle/__init__.py | 8 +- app/startup/modules_initializer.py | 48 ++++- app/testing/bootstrap.py | 7 +- scripts/local_setup.py | 1 + tests/conftest.py | 32 +++- tests/test_agent_lifecycle.py | 5 +- tests/test_cache_system.py | 8 +- tests/test_configuration_initializer.py | 104 +++++++++++ tests/test_configuration_ports.py | 42 ++++- tests/test_database_worker.py | 140 ++++++++++++++ tests/test_systemconfig_oper.py | 125 ++++++++++++- tests/test_user_config_endpoint.py | 54 ++++++ tests/test_userconfig_oper.py | 93 ++++++++++ 23 files changed, 1071 insertions(+), 153 deletions(-) create mode 100644 app/db/worker.py create mode 100644 tests/test_configuration_initializer.py create mode 100644 tests/test_database_worker.py create mode 100644 tests/test_user_config_endpoint.py create mode 100644 tests/test_userconfig_oper.py diff --git a/app/api/endpoints/plugin.py b/app/api/endpoints/plugin.py index 41957db8b..76fd14fd5 100644 --- a/app/api/endpoints/plugin.py +++ b/app/api/endpoints/plugin.py @@ -874,7 +874,10 @@ async def save_plugin_folders( 保存插件文件夹分组配置 """ try: - get_configured_system_config().set(SystemConfigKey.PluginFolders, folders) + await get_configured_system_config().async_set( + SystemConfigKey.PluginFolders, + folders, + ) return _SchemaResponse(success=True) except Exception as e: logger.error(f"[文件夹API] 保存文件夹配置失败: {str(e)}") @@ -893,7 +896,10 @@ async def create_plugin_folder( folders = get_configured_system_config().get(SystemConfigKey.PluginFolders) or {} if folder_name not in folders: folders[folder_name] = [] - get_configured_system_config().set(SystemConfigKey.PluginFolders, folders) + await get_configured_system_config().async_set( + SystemConfigKey.PluginFolders, + folders, + ) return _SchemaResponse( success=True, message=f"文件夹 '{folder_name}' 创建成功" ) diff --git a/app/api/endpoints/site.py b/app/api/endpoints/site.py index ebd257279..57c6abe79 100644 --- a/app/api/endpoints/site.py +++ b/app/api/endpoints/site.py @@ -188,8 +188,8 @@ async def reset( 清空所有站点数据并重新同步CookieCloud站点信息 """ result = await command.reset() - get_configured_system_config().set(SystemConfigKey.IndexerSites, []) - get_configured_system_config().set(SystemConfigKey.RssSites, []) + await get_configured_system_config().async_set(SystemConfigKey.IndexerSites, []) + await get_configured_system_config().async_set(SystemConfigKey.RssSites, []) # 启动定时服务 Scheduler().start("cookiecloud", manual=True) # 插件站点删除 diff --git a/app/api/endpoints/user.py b/app/api/endpoints/user.py index 9f986aba1..94163a1f8 100644 --- a/app/api/endpoints/user.py +++ b/app/api/endpoints/user.py @@ -16,7 +16,6 @@ from app.application.security.user import UserService from app.api.dependencies.auth import ( get_current_active_superuser_async, get_current_active_user_async, - get_current_active_user, get_user_service, ) from app.application.security.userconfig import get_configured_user_configuration @@ -140,7 +139,10 @@ async def upload_avatar( summary="查询用户配置", response_model=_SchemaResponse[_SchemaValueData], ) -def get_config(key: str, current_user: Any = Depends(get_current_active_user)): +async def get_config( + key: str, + current_user: Any = Depends(get_current_active_user_async), +): """ 查询用户配置 """ @@ -149,15 +151,15 @@ def get_config(key: str, current_user: Any = Depends(get_current_active_user)): @router.post("/config/{key}", summary="更新用户配置", response_model=_SchemaResponse[None]) -def set_config( +async def set_config( key: str, value: Annotated[Union[list, dict, bool, int, str] | None, Body()] = None, - current_user: Any = Depends(get_current_active_user), + current_user: Any = Depends(get_current_active_user_async), ): """ 更新用户配置 """ - get_configured_user_configuration().set( + await get_configured_user_configuration().async_set( username=current_user.name, key=key, value=value, diff --git a/app/application/configuration.py b/app/application/configuration.py index 59f62d669..01dd656ff 100644 --- a/app/application/configuration.py +++ b/app/application/configuration.py @@ -4,9 +4,11 @@ from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass +from functools import partial from pathlib import Path from typing import Any, Optional, Protocol +from app.application.database import AsyncDatabaseExecutor from app.schemas.types import MediaType @@ -16,9 +18,6 @@ class SystemConfigReader(Protocol): def get(self, key: Any = None) -> Any: """读取配置。""" - async def async_get(self, key: Any = None) -> Any: - """异步读取配置。""" - class SystemConfigWriter(Protocol): """持久化用户配置的最小写入端口。""" @@ -26,9 +25,6 @@ class SystemConfigWriter(Protocol): def set(self, key: Any, value: Any) -> bool | None: """写入配置。""" - async def async_set(self, key: Any, value: Any) -> bool | None: - """异步写入配置。""" - def delete(self, key: Any) -> Any: """删除配置。""" @@ -296,14 +292,16 @@ class SystemConfigService: *, reader: SystemConfigReader | None = None, writer: SystemConfigWriter | None = None, + async_executor: AsyncDatabaseExecutor | None = None, ) -> None: - """注入可分离的读写端口，并兼容旧的单仓储装配参数。""" + """注入读写端口及可选的异步事务执行能力。""" resolved_reader = reader or repository resolved_writer = writer or repository if resolved_reader is None or resolved_writer is None: raise ValueError("系统配置服务必须同时提供 reader 与 writer") self._reader = resolved_reader self._writer = resolved_writer + self._async_executor = async_executor def get(self, key: Any = None) -> Any: """读取配置。""" @@ -313,13 +311,11 @@ class SystemConfigService: """写入配置。""" return self._writer.set(key, value) - async def async_get(self, key: Any = None) -> Any: - """异步读取配置。""" - return await self._reader.async_get(key) - async def async_set(self, key: Any, value: Any) -> bool | None: - """异步写入配置。""" - return await self._writer.async_set(key, value) + """异步写入配置，并等待数据库提交或回滚完成。""" + if self._async_executor is None: + raise RuntimeError("系统配置异步数据库执行端口尚未配置") + return await self._async_executor.run(partial(self._writer.set, key, value)) def delete(self, key: Any) -> Any: """删除配置。""" diff --git a/app/application/database.py b/app/application/database.py index 511350f1d..d69414d27 100644 --- a/app/application/database.py +++ b/app/application/database.py @@ -3,17 +3,27 @@ from __future__ import annotations from collections.abc import Callable -from typing import Any, Optional +from typing import TYPE_CHECKING, Any, Optional, Protocol, TypeVar -from app.application.backup import ( - BackupArtifact, - BackupVerification, - DatabaseBackupService, -) -from app.application.maintenance import DataCleanupService +if TYPE_CHECKING: + from app.application.backup import ( + BackupArtifact, + BackupVerification, + DatabaseBackupService, + ) + from app.application.maintenance import DataCleanupService DatabaseProbe = Callable[[], Optional[str]] +T = TypeVar("T") + + +class AsyncDatabaseExecutor(Protocol): + """让异步业务调用同步短事务而不阻塞事件循环。""" + + async def run(self, operation: Callable[[], T]) -> T: + """等待数据库操作完成提交或回滚，并返回执行结果。""" + ... class DatabaseHealthService: diff --git a/app/application/security/userconfig.py b/app/application/security/userconfig.py index 09617ee53..739656363 100644 --- a/app/application/security/userconfig.py +++ b/app/application/security/userconfig.py @@ -2,8 +2,11 @@ from __future__ import annotations +from functools import partial from typing import Any, Protocol +from app.application.database import AsyncDatabaseExecutor + class UserConfigurationRepository(Protocol): """用户配置数据端口。""" @@ -18,9 +21,15 @@ class UserConfigurationRepository(Protocol): class UserConfigurationService: """编排用户个性化配置读写。""" - def __init__(self, repository: UserConfigurationRepository) -> None: - """注入用户配置数据端口。""" + def __init__( + self, + repository: UserConfigurationRepository, + *, + async_executor: AsyncDatabaseExecutor | None = None, + ) -> None: + """注入用户配置数据端口及可选的异步事务执行能力。""" self._repository = repository + self._async_executor = async_executor def get(self, username: str, key: str) -> Any: """读取用户配置。""" @@ -30,6 +39,14 @@ class UserConfigurationService: """写入用户配置。""" return self._repository.set(username=username, key=key, value=value) + async def async_set(self, username: str, key: str, value: Any) -> Any: + """异步写入用户配置，并等待数据库提交或回滚完成。""" + if self._async_executor is None: + raise RuntimeError("用户配置异步数据库执行端口尚未配置") + return await self._async_executor.run( + partial(self._repository.set, username=username, key=key, value=value) + ) + _configured_user_configuration: UserConfigurationService | None = None diff --git a/app/db/oper/systemconfig.py b/app/db/oper/systemconfig.py index df0212bac..63be40ce0 100644 --- a/app/db/oper/systemconfig.py +++ b/app/db/oper/systemconfig.py @@ -1,4 +1,3 @@ -import asyncio import copy import threading from typing import Any, Optional, Union @@ -14,15 +13,38 @@ class SystemConfigOper(DbOper, metaclass=Singleton): 系统配置管理 """ def __init__(self): - """ - 加载配置到内存 - """ + """初始化空快照，数据库加载由启动组合根显式执行。""" super().__init__() self.__SYSTEMCONF = {} - self._rlock = threading.RLock() - self._alock = asyncio.Lock() - for item in SystemConfig.list(self._db): - self.__SYSTEMCONF[item.key] = item.value + self._snapshot_lock = threading.RLock() + self._write_lock = threading.RLock() + self._loaded = False + + def load_snapshot(self) -> None: + """从数据库加载完整配置，并一次性发布新的内存快照。""" + with self._write_lock: + snapshot = { + item.key: copy.deepcopy(item.value) + for item in SystemConfig.list(self._db) + } + with self._snapshot_lock: + self.__SYSTEMCONF = snapshot + self._loaded = True + + def _require_loaded(self) -> None: + """阻止消费者读取尚未完成启动加载的半成品快照。""" + if not self._loaded: + raise RuntimeError("系统配置快照尚未加载") + + def _publish_value(self, key: str, value: Any) -> None: + """在事务成功后短暂持锁发布单项配置。""" + with self._snapshot_lock: + self.__SYSTEMCONF[key] = copy.deepcopy(value) + + def _publish_delete(self, key: str) -> None: + """在事务成功后短暂持锁移除单项配置。""" + with self._snapshot_lock: + self.__SYSTEMCONF.pop(key, None) def set(self, key: Union[str, SystemConfigKey], value: Any) -> Optional[bool]: """ @@ -33,59 +55,25 @@ class SystemConfigOper(DbOper, metaclass=Singleton): """ if isinstance(key, SystemConfigKey): key = key.value - with self._rlock: - # 旧值 - old_value = self.__SYSTEMCONF.get(key) - # 更新内存(deepcopy避免内存共享) - self.__SYSTEMCONF[key] = copy.deepcopy(value) - conf = SystemConfig.get_by_key(self._db, key) - if conf: - if old_value != value: - # 假值（False/0/None/空容器）同样落库而不是删除记录： - # 读取端以「无记录」表示未配置并回落默认值，删除会使布尔开关的关闭态无法持久化 - self._stage_update(conf, {"value": value}) - return True - return None - else: - conf = SystemConfig(key=key, value=value) - self._stage_create(conf) + self._require_loaded() + with self._write_lock: + + def write(db): + """在当前事务中创建或更新配置记录。""" + conf = SystemConfig.get_by_key(db, key) + if conf: + if conf.value == value: + return None + # 假值同样是有效配置；删除记录会使读取端错误回落默认值。 + conf.value = copy.deepcopy(value) + else: + db.add(SystemConfig(key=key, value=copy.deepcopy(value))) return True - async def async_set(self, key: Union[str, SystemConfigKey], value: Any) -> Optional[bool]: - """ - 异步设置系统设置 - :param key: 配置键 - :param value: 配置值 - :return: 是否设置成功（True 成功/False 失败/None 无需更新） - """ - if isinstance(key, SystemConfigKey): - key = key.value - async with self._alock: - conf = await SystemConfig.async_get_by_key(self._db, key) - # 确定是否需要更新数据库 - needs_db_update = False - if conf: - if conf.value != value: - needs_db_update = True - else: # 记录不存在，总是需要创建/更新 - needs_db_update = True - if not needs_db_update: - # 即使数据库值相同，也要确保缓存同步 - with self._rlock: - self.__SYSTEMCONF[key] = copy.deepcopy(value) - return None - # 执行数据库更新 - if conf: - # 假值（False/0/None/空容器）同样落库而不是删除记录： - # 读取端以「无记录」表示未配置并回落默认值，删除会使布尔开关的关闭态无法持久化 - await self._stage_async_update(conf, {"value": value}) - else: - conf = SystemConfig(key=key, value=value) - await self._stage_async_create(conf) - # 数据库更新成功后，再更新缓存 - with self._rlock: - self.__SYSTEMCONF[key] = copy.deepcopy(value) - return True + result = self._execute_sync_write(write) + # 数据库操作返回时事务已经提交，读取方不会看到尚未持久化的配置。 + self._publish_value(key, value) + return result def get(self, key: Optional[Union[str, SystemConfigKey]] = None) -> Any: """ @@ -95,7 +83,8 @@ class SystemConfigOper(DbOper, metaclass=Singleton): key = key.value if not key: return self.all() - with self._rlock: + with self._snapshot_lock: + self._require_loaded() # 避免将__SYSTEMCONF内的值引用出去，会导致set时误判没有变动 return copy.deepcopy(self.__SYSTEMCONF.get(key)) @@ -107,7 +96,8 @@ class SystemConfigOper(DbOper, metaclass=Singleton): :param step: 递增步长 :return: 递增后的整数值 """ - with self._rlock: + self._require_loaded() + with self._write_lock: value = int(self.get(key) or 0) + step self.set(key, value) return value @@ -116,7 +106,8 @@ class SystemConfigOper(DbOper, metaclass=Singleton): """ 获取所有系统设置 """ - with self._rlock: + with self._snapshot_lock: + self._require_loaded() # 避免将__SYSTEMCONF内的值引用出去，会导致set时误判没有变动 return copy.deepcopy(self.__SYSTEMCONF) @@ -126,11 +117,15 @@ class SystemConfigOper(DbOper, metaclass=Singleton): """ if isinstance(key, SystemConfigKey): key = key.value - with self._rlock: - # 更新内存 - self.__SYSTEMCONF.pop(key, None) - # 写入数据库 - conf = SystemConfig.get_by_key(self._db, key) - if conf: - self._stage_delete(SystemConfig, conf.id) + self._require_loaded() + with self._write_lock: + + def delete(db): + """在当前事务中删除配置记录。""" + conf = SystemConfig.get_by_key(db, key) + if conf: + db.delete(conf) + + self._execute_sync_write(delete) + self._publish_delete(key) return True diff --git a/app/db/oper/userconfig.py b/app/db/oper/userconfig.py index 5edfc3b97..7b1656bdc 100644 --- a/app/db/oper/userconfig.py +++ b/app/db/oper/userconfig.py @@ -1,3 +1,5 @@ +import copy +import threading from typing import Any, Union, Dict, Optional from app.db.base import DbOper @@ -11,13 +13,30 @@ class UserConfigOper(DbOper, metaclass=Singleton): 用户配置管理 """ def __init__(self): - """ - 加载配置到内存 - """ + """初始化空快照，数据库加载由启动组合根显式执行。""" super().__init__() self.__USERCONF = {} - for item in UserConfig.list(self._db): - self.__set_config_cache(username=item.username, key=item.key, value=item.value) + self._snapshot_lock = threading.RLock() + self._write_lock = threading.RLock() + self._loaded = False + + def load_snapshot(self) -> None: + """从数据库加载完整用户配置，并一次性发布新的内存快照。""" + with self._write_lock: + snapshot: dict[str, dict[str, Any]] = {} + for item in UserConfig.list(self._db): + if item.username and item.key: + snapshot.setdefault(item.username, {})[item.key] = copy.deepcopy( + item.value + ) + with self._snapshot_lock: + self.__USERCONF = snapshot + self._loaded = True + + def _require_loaded(self) -> None: + """阻止消费者读取尚未完成启动加载的半成品快照。""" + if not self._loaded: + raise RuntimeError("用户配置快照尚未加载") def set(self, username: str, key: Union[str, UserConfigKey], value: Any): """ @@ -25,30 +44,43 @@ class UserConfigOper(DbOper, metaclass=Singleton): """ if isinstance(key, UserConfigKey): key = key.value - # 更新内存 - self.__set_config_cache(username=username, key=key, value=value) - # 写入数据库 - conf = UserConfig.get_by_key(db=self._db, username=username, key=key) - if conf: - if value: - self._stage_update(conf, {"value": value}) - else: - self._stage_delete(UserConfig, conf.id) - else: - conf = UserConfig(username=username, key=key, value=value) - self._stage_create(conf) + self._require_loaded() + with self._write_lock: + + def write(db): + """在当前事务中按用户配置的假值规则写入记录。""" + conf = UserConfig.get_by_key(db=db, username=username, key=key) + if conf: + if value: + conf.value = copy.deepcopy(value) + else: + db.delete(conf) + else: + db.add( + UserConfig( + username=username, + key=key, + value=copy.deepcopy(value), + ) + ) + + self._execute_sync_write(write) + # 既有运行时语义会保留刚写入的假值，即使其数据库记录被删除。 + self.__set_config_cache(username=username, key=key, value=value) def get(self, username: str, key: Optional[Union[str, UserConfigKey]] = None) -> Any: """ 获取用户配置 """ - if not username: - return self.__USERCONF - if isinstance(key, UserConfigKey): - key = key.value - if not key: - return self.__get_config_caches(username=username) - return self.__get_config_cache(username=username, key=key) + with self._snapshot_lock: + self._require_loaded() + if not username: + return copy.deepcopy(self.__USERCONF) + if isinstance(key, UserConfigKey): + key = key.value + if not key: + return copy.deepcopy(self.__get_config_caches(username=username)) + return copy.deepcopy(self.__get_config_cache(username=username, key=key)) def __set_config_cache(self, username: str, key: str, value: Any): """ @@ -56,15 +88,9 @@ class UserConfigOper(DbOper, metaclass=Singleton): """ if not username or not key: return - cache = self.__USERCONF - if not cache: - cache = {} - user_cache = cache.get(username) - if not user_cache: - user_cache = {} - cache[username] = user_cache - user_cache[key] = value - self.__USERCONF = cache + with self._snapshot_lock: + user_cache = self.__USERCONF.setdefault(username, {}) + user_cache[key] = copy.deepcopy(value) def __get_config_caches(self, username: str) -> Optional[Dict[str, Any]]: """ diff --git a/app/db/worker.py b/app/db/worker.py new file mode 100644 index 000000000..b99f0883c --- /dev/null +++ b/app/db/worker.py @@ -0,0 +1,234 @@ +"""同步数据库短事务的异步执行器。""" + +from __future__ import annotations + +import asyncio +import threading +import time +from concurrent.futures import Future, ThreadPoolExecutor +from contextvars import copy_context +from dataclasses import dataclass +from typing import Callable, TypeVar + +from app.runtime.observability import record_metric + + +T = TypeVar("T") + + +class DatabaseWorkerClosedError(RuntimeError): + """数据库执行器尚未启动或已经停止。""" + + +class DatabaseWorkerOverloadedError(RuntimeError): + """数据库执行器的运行与排队容量已经用尽。""" + + +@dataclass(frozen=True, slots=True) +class DatabaseWorkerStats: + """数据库执行器当前的容量和任务数量。""" + + max_workers: int + capacity: int + queued: int + running: int + rejected: int + closing: bool + + +@dataclass(slots=True) +class _WorkItem: + """记录一个任务的排队时间和执行状态。""" + + submitted_at: float + started_at: float | None = None + + +class DatabaseWorker: + """以有限线程和队列执行不能原生异步化的数据库短事务。""" + + def __init__(self, *, max_workers: int = 4, capacity: int = 32) -> None: + """保存容量配置，线程只在显式启动后创建。""" + if max_workers < 1: + raise ValueError("数据库 worker 线程数必须大于 0") + if capacity < max_workers: + raise ValueError("数据库 worker 总容量不能小于线程数") + self._max_workers = max_workers + self._capacity = capacity + self._state_lock = threading.Lock() + self._loop: asyncio.AbstractEventLoop | None = None + self._executor: ThreadPoolExecutor | None = None + self._futures: dict[ + Future[object], tuple[asyncio.Future[object], _WorkItem] + ] = {} + self._queued = 0 + self._running = 0 + self._rejected = 0 + self._closing = False + + async def start(self) -> None: + """绑定当前事件循环并准备专属线程池。""" + if self._executor is not None: + if self._loop is not asyncio.get_running_loop(): + raise RuntimeError("数据库 worker 不能跨事件循环复用") + return + self._loop = asyncio.get_running_loop() + self._executor = ThreadPoolExecutor( + max_workers=self._max_workers, + thread_name_prefix="moviepilot-db", + ) + self._closing = False + self._record_depth() + + def snapshot(self) -> DatabaseWorkerStats: + """返回无需访问任务对象的低基数运行快照。""" + with self._state_lock: + return DatabaseWorkerStats( + max_workers=self._max_workers, + capacity=self._capacity, + queued=self._queued, + running=self._running, + rejected=self._rejected, + closing=self._closing, + ) + + async def run(self, operation: Callable[[], T]) -> T: + """执行短事务，取消时仍等待已开始的事务取得最终结果。""" + loop = asyncio.get_running_loop() + executor = self._executor + if executor is None or self._loop is not loop or self._closing: + raise DatabaseWorkerClosedError("数据库 worker 当前不可接收任务") + + with self._state_lock: + if self._queued + self._running >= self._capacity: + self._rejected += 1 + record_metric("db.worker.rejected") + raise DatabaseWorkerOverloadedError( + f"数据库 worker 容量已用尽（上限 {self._capacity}）" + ) + self._queued += 1 + + item = _WorkItem(submitted_at=time.perf_counter()) + context = copy_context() + try: + future = executor.submit(self._execute, item, context.run, operation) + except BaseException: + with self._state_lock: + self._queued -= 1 + self._record_depth() + raise + + wrapped = asyncio.wrap_future(future, loop=loop) + with self._state_lock: + self._futures[future] = (wrapped, item) + future.add_done_callback( + lambda completed: self._schedule_completion(completed, item) + ) + self._record_depth() + + try: + return await asyncio.shield(wrapped) + except asyncio.CancelledError: + if not future.cancel(): + await self._wait_until_done(wrapped) + if wrapped.done() and not wrapped.cancelled(): + wrapped.exception() + raise + + def _execute( + self, + item: _WorkItem, + context_run: Callable[..., T], + operation: Callable[[], T], + ) -> T: + """在线程中标记任务开始并保留提交时的上下文。""" + item.started_at = time.perf_counter() + loop = self._loop + if loop is not None: + try: + loop.call_soon_threadsafe(self._mark_running, item) + except RuntimeError: + pass + return context_run(operation) + + def _mark_running(self, item: _WorkItem) -> None: + """把任务从排队状态移入运行状态。""" + with self._state_lock: + self._queued -= 1 + self._running += 1 + started_at = item.started_at or time.perf_counter() + record_metric("db.worker.wait", started_at - item.submitted_at) + self._record_depth() + + def _schedule_completion( + self, + future: Future[object], + item: _WorkItem, + ) -> None: + """把线程完成通知安全地回投到所属事件循环。""" + loop = self._loop + if loop is None: + return + try: + loop.call_soon_threadsafe(self._complete, future, item) + except RuntimeError: + pass + + def _complete(self, future: Future[object], item: _WorkItem) -> None: + """释放 admission，并记录任务的最终结果。""" + with self._state_lock: + self._futures.pop(future, None) + if future.running() or future.done() and not future.cancelled(): + self._running -= 1 + else: + self._queued -= 1 + outcome = "cancelled" if future.cancelled() else "success" + if not future.cancelled(): + try: + future.result() + except BaseException: + outcome = "error" + started_at = item.started_at or item.submitted_at + record_metric( + "db.worker.duration", + time.perf_counter() - started_at, + outcome=outcome, + ) + self._record_depth() + + async def _wait_until_done(self, future: asyncio.Future[object]) -> None: + """忽略后续取消请求，直到线程内事务结束。""" + while not future.done(): + try: + await asyncio.shield(future) + except asyncio.CancelledError: + continue + except BaseException: + break + + async def shutdown(self) -> None: + """拒绝新任务，取消排队任务并等待运行中事务结束。""" + executor = self._executor + if executor is None: + return + if self._loop is not asyncio.get_running_loop(): + raise RuntimeError("数据库 worker 必须在所属事件循环中停止") + self._closing = True + with self._state_lock: + futures = tuple(self._futures.items()) + for future, _state in futures: + future.cancel() + for future, (wrapped, _item) in futures: + if not future.cancelled(): + await self._wait_until_done(wrapped) + executor.shutdown(wait=True, cancel_futures=True) + while self.snapshot().queued or self.snapshot().running: + await asyncio.sleep(0) + self._executor = None + self._record_depth() + + def _record_depth(self) -> None: + """记录当前排队量与运行量。""" + stats = self.snapshot() + record_metric("db.worker.queue.depth", stats.queued) + record_metric("db.worker.active", stats.running) diff --git a/app/runtime/observability/__init__.py b/app/runtime/observability/__init__.py index cf7f198b3..1da6e1d10 100644 --- a/app/runtime/observability/__init__.py +++ b/app/runtime/observability/__init__.py @@ -35,6 +35,11 @@ METRIC_SPECS = { MetricSpec("db.pool.wait", MetricKind.HISTOGRAM, frozenset({"backend", "outcome"})), MetricSpec("db.pool.checked_out", MetricKind.GAUGE, frozenset({"backend"})), MetricSpec("db.pool.timeout", MetricKind.COUNTER, frozenset({"backend"})), + MetricSpec("db.worker.wait", MetricKind.HISTOGRAM, frozenset()), + MetricSpec("db.worker.duration", MetricKind.HISTOGRAM, frozenset({"outcome"})), + MetricSpec("db.worker.queue.depth", MetricKind.GAUGE, frozenset()), + MetricSpec("db.worker.active", MetricKind.GAUGE, frozenset()), + MetricSpec("db.worker.rejected", MetricKind.COUNTER, frozenset()), MetricSpec("event.queue.depth", MetricKind.GAUGE, frozenset({"delivery"})), MetricSpec("event.handler.duration", MetricKind.HISTOGRAM, frozenset({"event_type", "handler_type", "outcome"})), MetricSpec("module.provider.duration", MetricKind.HISTOGRAM, frozenset({"method", "provider_type", "outcome"})), diff --git a/app/startup/lifecycle/__init__.py b/app/startup/lifecycle/__init__.py index 4f3bf3bd6..68705f3b6 100644 --- a/app/startup/lifecycle/__init__.py +++ b/app/startup/lifecycle/__init__.py @@ -130,7 +130,13 @@ async def run_startup_step( async def initialize_modules_component(app: FastAPI) -> None: """启动模块并把其类型化运行时发布到当前 FastAPI AppState。""" - runtime = await init_modules() + try: + runtime = await init_modules() + except BaseException: + from app.startup.modules_initializer import stop_database_worker + + await stop_database_worker() + 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 52260d4d7..598da1dc7 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -89,6 +89,7 @@ from app.db.session import ( get_async_db, get_db, ) +from app.db.worker import DatabaseWorker from app.db.uow import ( SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork, @@ -157,6 +158,40 @@ from app.runtime.extensions.service_config import ( ) +_database_worker: DatabaseWorker | None = None + + +async def stop_database_worker() -> None: + """停止当前进程的数据库短事务 worker。""" + global _database_worker + worker = _database_worker + _database_worker = None + if worker is not None: + await worker.shutdown() + + +async def _initialize_configuration_services( + database_worker: DatabaseWorker, +) -> None: + """加载完整配置快照后发布系统与用户配置服务。""" + system_config = SystemConfigOper() + user_config = UserConfigOper() + await database_worker.run(system_config.load_snapshot) + await database_worker.run(user_config.load_snapshot) + configure_system_config( + SystemConfigService( + repository=system_config, + async_executor=database_worker, + ) + ) + configure_user_configuration( + UserConfigurationService( + repository=user_config, + async_executor=database_worker, + ) + ) + + async def _async_get_subscribe(subscribe_id: int): """通过数据库操作器异步读取订阅，供服务端共享用例使用。""" return await SubscribeOper().async_get(subscribe_id) @@ -506,6 +541,7 @@ async def stop_modules(): await run_step("消息服务", stop_message) 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) await run_step("前端服务", stop_frontend) await run_step("临时文件", clear_temp) @@ -515,6 +551,7 @@ async def init_modules() -> HostRuntime: """ 启动模块并返回本次 lifespan 唯一的类型化 HostRuntime。 """ + global _database_worker # 兼容 Oper 的无 Session 写入口仍由组合根持有事务，避免模型恢复自动提交。 transaction_runner = TransactionalWriteRunner( sync_session=SessionFactory, @@ -524,6 +561,14 @@ async def init_modules() -> HostRuntime: sync=transaction_runner.sync, async_=transaction_runner.async_, ) + database_worker = DatabaseWorker() + await database_worker.start() + _database_worker = database_worker + try: + await _initialize_configuration_services(database_worker) + except BaseException: + await stop_database_worker() + raise # 数据访问能力统一在启动组合根注入，Runtime 和 Adapter 不再直接依赖 Oper。 api_data = ApiDataPorts( sync_session=get_db, @@ -599,8 +644,6 @@ async def init_modules() -> HostRuntime: configure_runtime_settings(host_runtime.settings) configure_runtime_setting_provider(lambda key: getattr(settings, key)) configure_token_runtime_config(lambda: build_token_runtime_config(settings)) - # 先发布系统配置服务，后续启动组合步骤统一复用同一配置端口。 - configure_system_config(SystemConfigService(repository=SystemConfigOper())) # 旧 app.api.data 导入只保留 ABI 转发，正式 API 依赖全部读取 HostRuntime。 configure_api_data_runtime(api_data) configure_runtime_data_providers() @@ -640,7 +683,6 @@ async def init_modules() -> HostRuntime: ) ) configure_passkey_service(PasskeyService(repository=PassKeyOper())) - configure_user_configuration(UserConfigurationService(repository=UserConfigOper())) configure_transfer_history_provider(lambda: TransferHistoryOper()) configure_site_query_service(SiteQueryService(repository=SiteOper())) configure_site_health_service(SiteHealthService(repository=SiteOper())) diff --git a/app/testing/bootstrap.py b/app/testing/bootstrap.py index 67c453a95..5c4e7a030 100644 --- a/app/testing/bootstrap.py +++ b/app/testing/bootstrap.py @@ -171,7 +171,7 @@ def ensure_optional_stub(name: str, **attrs) -> None: def prepare_backend() -> None: - """隔离 CONFIG_DIR、补 sites 垫片并建表（后端须已在 ``sys.path`` 上）。 + """隔离 CONFIG_DIR、补 sites 垫片、建表并加载配置快照。 主程序中后端即当前包；插件仓由其 ``tests/_bootstrap.py`` shim 在 import 本模块前 先把后端目录注入 ``sys.path``。顺序固定：先隔离 CONFIG_DIR，再补 ``app.application.site.sites`` 垫片， @@ -182,6 +182,11 @@ def prepare_backend() -> None: ensure_sites_stub() from app.startup.database_initializer import init_db init_db() + from app.db.oper.systemconfig import SystemConfigOper + from app.db.oper.userconfig import UserConfigOper + + SystemConfigOper().load_snapshot() + UserConfigOper().load_snapshot() # 缓存装饰器在测试模块导入时即创建后端，先装配隔离配置对应的适配器。 from app.startup.cache_initializer import configure_cache_dependencies configure_cache_dependencies() diff --git a/scripts/local_setup.py b/scripts/local_setup.py index 8035e10f8..86d19ced8 100644 --- a/scripts/local_setup.py +++ b/scripts/local_setup.py @@ -2400,6 +2400,7 @@ def _apply_local_system_config_inner(config_payload: dict[str, Any]) -> None: print_step(f"超级管理员初始密码：{generated_password}") system_config = SystemConfigOper() + system_config.load_snapshot() directory_items = config_payload.get("directories") or [] if directory_items: current_directories = system_config.get(SystemConfigKey.Directories) or [] diff --git a/tests/conftest.py b/tests/conftest.py index 8584ed70c..4092f5a7e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,6 +3,7 @@ 引导与网络守卫均复用 ``app/testing`` 的共享 harness（与插件仓 conftest 同源）， 引导逻辑只在 ``app/testing`` 维护一处。 """ +import asyncio import sys import pytest @@ -20,6 +21,14 @@ prepare_backend() from app.testing.network_guard import block_real_network # noqa: E402,F401 +class _TestDatabaseExecutor: + """让绕过完整 lifespan 的测试仍通过线程执行同步数据库写入。""" + + async def run(self, operation): + """在线程中执行测试事务。""" + return await asyncio.to_thread(operation) + + @pytest.fixture(autouse=True) def configure_plugin_system_services(): """为绕过完整启动流程的单元测试装配真实插件系统适配器。""" @@ -61,6 +70,11 @@ def configure_plugin_system_services(): configure_transaction_runners, ) from app.db.oper.systemconfig import SystemConfigOper + from app.db.oper.userconfig import UserConfigOper + from app.application.security.userconfig import ( + UserConfigurationService, + configure_user_configuration, + ) configure_token_codec(create_access_token, decode_access_token) configure_runtime_configuration( @@ -73,7 +87,23 @@ def configure_plugin_system_services(): configure_runtime_settings(RuntimeSettingsService(settings)) configure_runtime_setting_provider(lambda key: getattr(settings, key)) configure_token_runtime_config(lambda: build_token_runtime_config(settings)) - configure_system_config(SystemConfigService(repository=SystemConfigOper())) + database_executor = _TestDatabaseExecutor() + system_config = SystemConfigOper() + system_config.load_snapshot() + user_config = UserConfigOper() + user_config.load_snapshot() + configure_system_config( + SystemConfigService( + repository=system_config, + async_executor=database_executor, + ) + ) + configure_user_configuration( + UserConfigurationService( + repository=user_config, + async_executor=database_executor, + ) + ) configure_transfer_retry_config( lambda: TransferRetryConfig( max_failed_retries=settings.TRANSFER_MAX_FAILED_RETRIES, diff --git a/tests/test_agent_lifecycle.py b/tests/test_agent_lifecycle.py index 2ec843909..fe4e42b76 100644 --- a/tests/test_agent_lifecycle.py +++ b/tests/test_agent_lifecycle.py @@ -138,7 +138,10 @@ async def test_agent_initialization_failure_does_not_stop_module_startup( monkeypatch.setattr(modules_initializer, "start_frontend", start_frontend) monkeypatch.setattr(modules_initializer, "check_auth", check_auth) - await modules_initializer.init_modules() + try: + await modules_initializer.init_modules() + finally: + await modules_initializer.stop_database_worker() manager.initialize.assert_awaited_once_with() start_frontend.assert_called_once_with() diff --git a/tests/test_cache_system.py b/tests/test_cache_system.py index 95016906a..aa38f68e2 100644 --- a/tests/test_cache_system.py +++ b/tests/test_cache_system.py @@ -164,7 +164,13 @@ def test_init_modules_does_not_clear_package_tool_cache(monkeypatch): monkeypatch.setattr(modules_initializer, "start_frontend", lambda: None) monkeypatch.setattr(modules_initializer, "check_auth", lambda: None) - asyncio.run(modules_initializer.init_modules()) + async def initialize_modules() -> None: + try: + await modules_initializer.init_modules() + finally: + await modules_initializer.stop_database_worker() + + asyncio.run(initialize_modules()) assert called is False init_agent.assert_awaited_once_with() diff --git a/tests/test_configuration_initializer.py b/tests/test_configuration_initializer.py new file mode 100644 index 000000000..3f3be0f5a --- /dev/null +++ b/tests/test_configuration_initializer.py @@ -0,0 +1,104 @@ +"""配置快照启动顺序测试。""" + +from unittest.mock import AsyncMock + +import pytest + +from app.startup import modules_initializer +from app.startup.lifecycle import initialize_modules_component + + +class _InlineWorker: + """按提交顺序执行配置加载操作。""" + + async def run(self, operation): + """执行并返回操作结果。""" + return operation() + + +@pytest.mark.asyncio +async def test_configuration_services_publish_after_both_snapshots_load( + monkeypatch, +) -> None: + """两个完整快照加载成功前不发布任一配置服务。""" + events = [] + + class _SystemConfig: + def load_snapshot(self): + events.append("load-system") + + class _UserConfig: + def load_snapshot(self): + events.append("load-user") + + monkeypatch.setattr(modules_initializer, "SystemConfigOper", _SystemConfig) + monkeypatch.setattr(modules_initializer, "UserConfigOper", _UserConfig) + monkeypatch.setattr( + modules_initializer, + "configure_system_config", + lambda _service: events.append("publish-system"), + ) + monkeypatch.setattr( + modules_initializer, + "configure_user_configuration", + lambda _service: events.append("publish-user"), + ) + + await modules_initializer._initialize_configuration_services(_InlineWorker()) + + assert events == [ + "load-system", + "load-user", + "publish-system", + "publish-user", + ] + + +@pytest.mark.asyncio +async def test_configuration_load_failure_does_not_publish_partial_service( + monkeypatch, +) -> None: + """任一快照加载失败时不发布半套配置服务。""" + published = [] + + class _SystemConfig: + def load_snapshot(self): + return None + + class _UserConfig: + def load_snapshot(self): + raise RuntimeError("load failed") + + monkeypatch.setattr(modules_initializer, "SystemConfigOper", _SystemConfig) + monkeypatch.setattr(modules_initializer, "UserConfigOper", _UserConfig) + monkeypatch.setattr( + modules_initializer, + "configure_system_config", + lambda service: published.append(service), + ) + monkeypatch.setattr( + modules_initializer, + "configure_user_configuration", + lambda service: published.append(service), + ) + + with pytest.raises(RuntimeError, match="load failed"): + await modules_initializer._initialize_configuration_services(_InlineWorker()) + + assert published == [] + + +@pytest.mark.asyncio +async def test_modules_startup_failure_stops_database_worker(monkeypatch) -> None: + """模块启动失败时立即关闭已创建的数据库 worker。""" + monkeypatch.setattr( + "app.startup.lifecycle.init_modules", + AsyncMock(side_effect=RuntimeError("startup failed")), + ) + stop_worker = AsyncMock() + monkeypatch.setattr(modules_initializer, "stop_database_worker", stop_worker) + + with pytest.raises(RuntimeError, match="startup failed"): + await initialize_modules_component(object()) + + stop_worker.assert_awaited_once_with() diff --git a/tests/test_configuration_ports.py b/tests/test_configuration_ports.py index ac0f1b46e..b45a6cb45 100644 --- a/tests/test_configuration_ports.py +++ b/tests/test_configuration_ports.py @@ -2,7 +2,7 @@ import asyncio from dataclasses import FrozenInstanceError -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import MagicMock import pytest @@ -19,6 +19,15 @@ from app.application.configuration import ( get_api_runtime_config_snapshot, get_transfer_retry_config, ) +from app.application.security.userconfig import UserConfigurationService + + +class _InlineDatabaseExecutor: + """同步执行测试操作，并保留异步应用端口的调用形态。""" + + async def run(self, operation): + """执行并返回操作结果。""" + return operation() class _MutableSettings: @@ -66,23 +75,44 @@ def test_system_config_service_supports_separate_reader_and_writer() -> None: """应用服务可以分别注入只读与写入适配器。""" reader = MagicMock() reader.get.return_value = "old" - reader.async_get = AsyncMock(return_value="async-old") writer = MagicMock() writer.set.return_value = True - writer.async_set = AsyncMock(return_value=True) - service = SystemConfigService(reader=reader, writer=writer) + service = SystemConfigService( + reader=reader, + writer=writer, + async_executor=_InlineDatabaseExecutor(), + ) assert service.get("key") == "old" assert service.set("key", "new") is True - assert asyncio.run(service.async_get("key")) == "async-old" assert asyncio.run(service.async_set("key", "new")) is True service.delete("key") reader.get.assert_called_once_with("key") - writer.set.assert_called_once_with("key", "new") + assert writer.set.call_args_list == [ + (("key", "new"), {}), + (("key", "new"), {}), + ] writer.delete.assert_called_once_with("key") +def test_user_configuration_service_supports_sync_and_async_writes() -> None: + """用户配置服务的同步与异步入口执行同一仓储方法。""" + repository = MagicMock() + repository.set.return_value = True + service = UserConfigurationService( + repository, + async_executor=_InlineDatabaseExecutor(), + ) + + assert service.set("alice", "theme", "dark") is True + assert asyncio.run(service.async_set("alice", "theme", "light")) is True + assert repository.set.call_args_list == [ + ((), {"username": "alice", "key": "theme", "value": "dark"}), + ((), {"username": "alice", "key": "theme", "value": "light"}), + ] + + def test_transfer_retry_provider_returns_frozen_snapshot_per_call() -> None: """配置工厂在每次用例入口创建新快照，旧快照不受 reload 后状态影响。""" state = {"value": 2} diff --git a/tests/test_database_worker.py b/tests/test_database_worker.py new file mode 100644 index 000000000..563079d29 --- /dev/null +++ b/tests/test_database_worker.py @@ -0,0 +1,140 @@ +"""数据库短事务 worker 的容量、取消与关闭合同测试。""" + +import asyncio +import threading + +import pytest + +from app.db.worker import ( + DatabaseWorker, + DatabaseWorkerClosedError, + DatabaseWorkerOverloadedError, +) + + +@pytest.mark.asyncio +async def test_worker_requires_explicit_start() -> None: + """构造对象不会隐式创建可执行线程池。""" + worker = DatabaseWorker(max_workers=1, capacity=1) + + with pytest.raises(DatabaseWorkerClosedError): + await worker.run(lambda: None) + + +@pytest.mark.asyncio +async def test_worker_rejects_work_beyond_running_and_queue_capacity() -> None: + """运行与排队任务达到总容量后立即拒绝新任务。""" + worker = DatabaseWorker(max_workers=1, capacity=2) + await worker.start() + started = threading.Event() + release = threading.Event() + + def block() -> None: + started.set() + release.wait(1) + + running = asyncio.create_task(worker.run(block)) + await asyncio.to_thread(started.wait) + queued = asyncio.create_task(worker.run(lambda: None)) + await asyncio.sleep(0) + + with pytest.raises(DatabaseWorkerOverloadedError): + await worker.run(lambda: None) + + assert worker.snapshot().running == 1 + assert worker.snapshot().queued == 1 + assert worker.snapshot().rejected == 1 + + release.set() + await asyncio.gather(running, queued) + await worker.shutdown() + + +@pytest.mark.asyncio +async def test_cancelling_queued_work_prevents_execution() -> None: + """尚未取得线程的任务取消后不得执行数据库操作。""" + worker = DatabaseWorker(max_workers=1, capacity=2) + await worker.start() + started = threading.Event() + release = threading.Event() + queued_executed = threading.Event() + + def block() -> None: + started.set() + release.wait(1) + + running = asyncio.create_task(worker.run(block)) + await asyncio.to_thread(started.wait) + queued = asyncio.create_task(worker.run(queued_executed.set)) + await asyncio.sleep(0) + queued.cancel() + with pytest.raises(asyncio.CancelledError): + await queued + + release.set() + await running + await worker.shutdown() + + assert queued_executed.is_set() is False + assert worker.snapshot().queued == 0 + assert worker.snapshot().running == 0 + + +@pytest.mark.asyncio +async def test_cancelling_running_work_waits_for_transaction_terminal_state() -> None: + """线程内操作开始后，取消结果必须晚于操作的最终状态。""" + worker = DatabaseWorker(max_workers=1, capacity=1) + await worker.start() + started = threading.Event() + release = threading.Event() + completed = threading.Event() + + def operation() -> None: + started.set() + release.wait(1) + completed.set() + + task = asyncio.create_task(worker.run(operation)) + await asyncio.to_thread(started.wait) + task.cancel() + await asyncio.sleep(0.01) + + assert task.done() is False + assert completed.is_set() is False + + release.set() + with pytest.raises(asyncio.CancelledError): + await task + + assert completed.is_set() is True + await worker.shutdown() + + +@pytest.mark.asyncio +async def test_shutdown_rejects_new_work_and_waits_for_running_work() -> None: + """关闭期间不接收新任务，并等待已开始的操作结束。""" + worker = DatabaseWorker(max_workers=1, capacity=2) + await worker.start() + started = threading.Event() + release = threading.Event() + + def operation() -> None: + started.set() + release.wait(1) + + running = asyncio.create_task(worker.run(operation)) + await asyncio.to_thread(started.wait) + shutdown = asyncio.create_task(worker.shutdown()) + await asyncio.sleep(0) + + with pytest.raises(DatabaseWorkerClosedError): + await worker.run(lambda: None) + assert shutdown.done() is False + + release.set() + await running + await shutdown + + assert worker.snapshot().closing is True + assert worker.snapshot().queued == 0 + assert worker.snapshot().running == 0 diff --git a/tests/test_systemconfig_oper.py b/tests/test_systemconfig_oper.py index 8d80cf415..5e6b77d61 100644 --- a/tests/test_systemconfig_oper.py +++ b/tests/test_systemconfig_oper.py @@ -1,6 +1,7 @@ import threading import uuid from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace import pytest @@ -16,15 +17,105 @@ def _unique_key() -> str: def _fresh_oper() -> SystemConfigOper: - """重置单例并从数据库重新加载配置缓存。""" + """重置单例并显式加载数据库配置快照。""" Singleton._instances.pop((SystemConfigOper, (), frozenset()), None) - return SystemConfigOper() + oper = SystemConfigOper() + oper.load_snapshot() + return oper + + +def test_constructor_does_not_query_database(monkeypatch): + """构造配置对象时不打开数据库会话。""" + Singleton._instances.pop((SystemConfigOper, (), frozenset()), None) + monkeypatch.setattr( + SystemConfig, + "list", + lambda _db: pytest.fail("构造阶段不应查询数据库"), + ) + + oper = SystemConfigOper() + + with pytest.raises(RuntimeError, match="快照尚未加载"): + oper.get("key") + + +def test_load_snapshot_publishes_complete_dictionary(monkeypatch): + """重新加载期间读取方只会看到完整旧快照或完整新快照。""" + Singleton._instances.pop((SystemConfigOper, (), frozenset()), None) + oper = SystemConfigOper() + values = [SimpleNamespace(key="key", value="old")] + entered = threading.Event() + release = threading.Event() + + monkeypatch.setattr(SystemConfig, "list", lambda _db: values) + oper.load_snapshot() + + def load_new_snapshot(_db): + entered.set() + release.wait(1) + return [SimpleNamespace(key="key", value="new")] + + monkeypatch.setattr(SystemConfig, "list", load_new_snapshot) + thread = threading.Thread(target=oper.load_snapshot) + thread.start() + assert entered.wait(1) + + assert oper.get("key") == "old" + + release.set() + thread.join(1) + assert thread.is_alive() is False + assert oper.get("key") == "new" + + +def test_read_does_not_wait_for_slow_write_transaction(monkeypatch): + """数据库写入期间，内存读取仍返回最近一次已提交值。""" + oper = _fresh_oper() + key = _unique_key() + oper.set(key, "old") + entered = threading.Event() + release = threading.Event() + + def slow_write(_operation): + entered.set() + release.wait(1) + return True + + monkeypatch.setattr(oper, "_execute_sync_write", slow_write) + thread = threading.Thread(target=lambda: oper.set(key, "new")) + thread.start() + assert entered.wait(1) + + assert oper.get(key) == "old" + + release.set() + thread.join(1) + assert thread.is_alive() is False + assert oper.get(key) == "new" + + +def test_failed_write_keeps_committed_snapshot(monkeypatch): + """事务失败时内存快照保持最近一次已提交值。""" + oper = _fresh_oper() + key = _unique_key() + oper.set(key, "old") + + def fail_write(_operation): + raise RuntimeError("write failed") + + monkeypatch.setattr(oper, "_execute_sync_write", fail_write) + + with pytest.raises(RuntimeError, match="write failed"): + oper.set(key, "new") + assert oper.get(key) == "old" def test_increment_serializes_concurrent_counter_updates(monkeypatch): """并发递增系统计数时不应丢失更新。""" oper = object.__new__(SystemConfigOper) - oper._rlock = threading.RLock() + oper._snapshot_lock = threading.RLock() + oper._write_lock = threading.RLock() + oper._loaded = True stored_value = {"value": 0} monkeypatch.setattr(oper, "get", lambda _key: stored_value["value"]) @@ -51,7 +142,9 @@ def test_increment_serializes_concurrent_counter_updates(monkeypatch): def test_increment_supports_custom_step(monkeypatch): """整数系统计数应支持指定递增步长。""" oper = object.__new__(SystemConfigOper) - oper._rlock = threading.RLock() + oper._snapshot_lock = threading.RLock() + oper._write_lock = threading.RLock() + oper._loaded = True stored_value = {"value": 4} monkeypatch.setattr(oper, "get", lambda _key: stored_value["value"]) @@ -95,7 +188,17 @@ async def test_async_set_persists_falsy_value_on_existing_record(): oper = _fresh_oper() oper.set(key, True) - assert await oper.async_set(key, False) is True + from app.application.configuration import SystemConfigService + + class _InlineDatabaseExecutor: + async def run(self, operation): + return operation() + + service = SystemConfigService( + repository=oper, + async_executor=_InlineDatabaseExecutor(), + ) + assert await service.async_set(key, False) is True assert oper.get(key) is False assert SystemConfig.get_by_key(oper._db, key).value is False @@ -106,7 +209,17 @@ async def test_async_set_creates_record_for_falsy_value(): key = _unique_key() oper = _fresh_oper() - assert await oper.async_set(key, 0) is True + from app.application.configuration import SystemConfigService + + class _InlineDatabaseExecutor: + async def run(self, operation): + return operation() + + service = SystemConfigService( + repository=oper, + async_executor=_InlineDatabaseExecutor(), + ) + assert await service.async_set(key, 0) is True assert oper.get(key) == 0 assert SystemConfig.get_by_key(oper._db, key).value == 0 diff --git a/tests/test_user_config_endpoint.py b/tests/test_user_config_endpoint.py new file mode 100644 index 000000000..112993435 --- /dev/null +++ b/tests/test_user_config_endpoint.py @@ -0,0 +1,54 @@ +"""用户配置 API 的异步应用端口测试。""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.api.endpoints import user as user_endpoint + + +@pytest.mark.asyncio +async def test_set_config_waits_for_async_configuration_write(monkeypatch) -> None: + """更新接口等待用户配置事务完成后再返回成功。""" + service = MagicMock() + service.async_set = AsyncMock(return_value=True) + monkeypatch.setattr( + user_endpoint, + "get_configured_user_configuration", + lambda: service, + ) + + response = await user_endpoint.set_config( + "theme", + "dark", + current_user=SimpleNamespace(name="alice"), + ) + + assert response.success is True + service.async_set.assert_awaited_once_with( + username="alice", + key="theme", + value="dark", + ) + + +@pytest.mark.asyncio +async def test_get_config_reads_loaded_snapshot(monkeypatch) -> None: + """查询接口直接读取已加载的用户配置快照。""" + service = MagicMock() + service.get.return_value = "dark" + monkeypatch.setattr( + user_endpoint, + "get_configured_user_configuration", + lambda: service, + ) + + response = await user_endpoint.get_config( + "theme", + current_user=SimpleNamespace(name="alice"), + ) + + assert response.success is True + assert response.data == {"value": "dark"} + service.get.assert_called_once_with(username="alice", key="theme") diff --git a/tests/test_userconfig_oper.py b/tests/test_userconfig_oper.py new file mode 100644 index 000000000..1221307b4 --- /dev/null +++ b/tests/test_userconfig_oper.py @@ -0,0 +1,93 @@ +"""用户配置快照与异步写入合同测试。""" + +import asyncio +import threading +from types import SimpleNamespace + +import pytest + +from app.application.security.userconfig import UserConfigurationService +from app.db.models.userconfig import UserConfig +from app.db.oper.userconfig import UserConfigOper +from app.foundation.singleton import Singleton + + +class _ThreadDatabaseExecutor: + """在线程中执行测试事务。""" + + async def run(self, operation): + """执行并返回操作结果。""" + return await asyncio.to_thread(operation) + + +def _fresh_oper() -> UserConfigOper: + """重置单例并显式加载用户配置快照。""" + Singleton._instances.pop((UserConfigOper, (), frozenset()), None) + oper = UserConfigOper() + oper.load_snapshot() + return oper + + +def test_constructor_does_not_query_database(monkeypatch): + """构造用户配置对象时不打开数据库会话。""" + Singleton._instances.pop((UserConfigOper, (), frozenset()), None) + monkeypatch.setattr( + UserConfig, + "list", + lambda _db: pytest.fail("构造阶段不应查询数据库"), + ) + + oper = UserConfigOper() + + with pytest.raises(RuntimeError, match="快照尚未加载"): + oper.get("alice", "theme") + + +def test_load_snapshot_publishes_complete_dictionary(monkeypatch): + """重新加载期间读取方不会看到逐项构造的用户配置。""" + Singleton._instances.pop((UserConfigOper, (), frozenset()), None) + oper = UserConfigOper() + monkeypatch.setattr( + UserConfig, + "list", + lambda _db: [SimpleNamespace(username="alice", key="theme", value="old")], + ) + oper.load_snapshot() + entered = threading.Event() + release = threading.Event() + + def load_new_snapshot(_db): + entered.set() + release.wait(1) + return [SimpleNamespace(username="alice", key="theme", value="new")] + + monkeypatch.setattr(UserConfig, "list", load_new_snapshot) + thread = threading.Thread(target=oper.load_snapshot) + thread.start() + assert entered.wait(1) + + assert oper.get("alice", "theme") == "old" + + release.set() + thread.join(1) + assert thread.is_alive() is False + assert oper.get("alice", "theme") == "new" + + +@pytest.mark.asyncio +async def test_async_write_uses_same_repository_rule() -> None: + """异步入口提交后同步读取立即看到相同结果。""" + oper = _fresh_oper() + service = UserConfigurationService( + oper, + async_executor=_ThreadDatabaseExecutor(), + ) + + await service.async_set("async-user", "theme", "dark") + + assert service.get("async-user", "theme") == "dark" + assert UserConfig.get_by_key( + oper._db, + username="async-user", + key="theme", + ).value == "dark" From fc677a2bba098904c42a16eda0395fd9b50d33b1 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 01:53:00 +0800 Subject: [PATCH 2/7] 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 From 58352c2eb7bfd34ed32e12e54e548e869998789d Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 02:00:29 +0800 Subject: [PATCH 3/7] test(db): ignore non-orm dataclasses in declarative guard --- tests/test_db_declarative_2_0.py | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/tests/test_db_declarative_2_0.py b/tests/test_db_declarative_2_0.py index 334997947..56d153736 100644 --- a/tests/test_db_declarative_2_0.py +++ b/tests/test_db_declarative_2_0.py @@ -93,12 +93,19 @@ def _class_level_annotations(py_file: Path): 只取 ClassDef 直接子语句中的 AnnAssign：函数体内的局部注解、模块级注解都不算 类级注解；``if TYPE_CHECKING:`` 块里的注解运行期根本不存在，声明式系统也看不到， - 同样不在此列。 + 同样不在此列。独立的 dataclass 是 worker、DTO 等运行时数据结构，不参与 SQLAlchemy + 声明式映射，也不属于本守卫的范围。 """ tree = ast.parse(py_file.read_text(encoding="utf-8")) for node in ast.walk(tree): if not isinstance(node, ast.ClassDef): continue + if any( + ast.unparse(decorator).split("(", maxsplit=1)[0].split(".")[-1] + == "dataclass" + for decorator in node.decorator_list + ): + continue for stmt in node.body: if isinstance(stmt, ast.AnnAssign): yield (node.name, ast.unparse(stmt.target), @@ -130,7 +137,7 @@ def test_allow_unmapped_is_not_set(): def test_no_unmapped_class_level_annotations_in_db_package(): """ - app/db 内不存在非 Mapped[] 的类级注解——这是移除 __allow_unmapped__ 的前提。 + app/db 内的 ORM 声明不存在非 Mapped[] 的类级注解——这是移除 __allow_unmapped__ 的前提。 上一条用例断言标志不在，本条断言仓内确实不需要它。两者缺一不可：只断言标志不在， 则某天有人补进一条 legacy 注解、发现 import 就炸、顺手把标志加回来，上一条用例 @@ -140,11 +147,11 @@ def test_no_unmapped_class_level_annotations_in_db_package(): 「现有 22 个模型仍是 legacy Column() 写法」，在 329 列全部迁移完之后仍原样留了 很久，主动误导读者。 - 扫全部类而非只扫 Base 子类：判定 Base 子类要么靠运行期 Base.__subclasses__()， + 扫全部 ORM 类而非只扫 Base 子类：判定 Base 子类要么靠运行期 Base.__subclasses__()， 要么靠 AST 解析基类名。前者会漏掉「新增了模型文件但还没接进 app/db/models/__init__.py」 的情况——恰恰是最可能带进 legacy 注解的场景；后者一遇 mixin 或跨文件继承就不准。 - 纯静态扫全部类没有这个盲区，而且当下不需要任何白名单：DbOper 这类非 ORM 类本身 - 就没有类级注解，天然不受影响。 + 纯静态扫描仍保留这个盲区保护；独立 dataclass 已在扫描函数中排除，其他非 ORM 类 + 若声明类级注解仍会被报告，避免通过装饰器名称伪装模型。 变红时怎么办（二选一，别直接把用例删了）： 1. 常见情况——新模型忘了用 2.0 写法，把它改成 mapped_column() + Mapped[] 即可； From b79f927fcfc94b3aae40f17f472def3934d74e36 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 02:26:04 +0800 Subject: [PATCH 4/7] fix(database): harden worker shutdown and overload propagation --- app/application/plugin/install.py | 27 ++++++-- app/db/worker.py | 28 +++++++-- app/factory.py | 7 +++ .../architecture/dependency-baseline.json | 2 + tests/test_api_response.py | 36 +++++++++++ tests/test_database_worker.py | 62 +++++++++++++++++++ tests/test_plugin_install_command.py | 21 +++++++ 7 files changed, 172 insertions(+), 11 deletions(-) diff --git a/app/application/plugin/install.py b/app/application/plugin/install.py index 22419a70b..90eddd7c7 100644 --- a/app/application/plugin/install.py +++ b/app/application/plugin/install.py @@ -6,6 +6,8 @@ from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from typing import Any, Optional +from app.application.database import DatabaseWorkerOverloadedError + InstalledPluginsReader = Callable[[], list[str]] InstalledPluginsWriter = Callable[[list[str]], Awaitable[object]] @@ -127,7 +129,7 @@ class PluginInstallCommand: force, ) except Exception as err: - return await self._failure( + result = await self._failure( plugin_id=plugin_id, original_plugins=installed_plugins, checkpoint=checkpoint, @@ -135,6 +137,9 @@ class PluginInstallCommand: message=str(err), package_installed=False, ) + if isinstance(err, DatabaseWorkerOverloadedError): + raise + return result if not state: return await self._failure( plugin_id=plugin_id, @@ -152,7 +157,7 @@ class PluginInstallCommand: await self._installed_plugins_writer(updated_plugins) installed_list_persisted = True except Exception as err: - return await self._failure( + result = await self._failure( plugin_id=plugin_id, original_plugins=installed_plugins, checkpoint=checkpoint, @@ -160,11 +165,14 @@ class PluginInstallCommand: message=str(err), package_installed=True, ) + if isinstance(err, DatabaseWorkerOverloadedError): + raise + return result try: await self._plugin_reloader(plugin_id) except Exception as err: - return await self._failure( + result = await self._failure( plugin_id=plugin_id, original_plugins=installed_plugins, checkpoint=checkpoint, @@ -174,11 +182,14 @@ class PluginInstallCommand: installed_list_persisted=installed_list_persisted, runtime_touched=True, ) + if isinstance(err, DatabaseWorkerOverloadedError): + raise + return result try: await self._registration_refresher(plugin_id) except Exception as err: - return await self._failure( + result = await self._failure( plugin_id=plugin_id, original_plugins=installed_plugins, checkpoint=checkpoint, @@ -189,6 +200,9 @@ class PluginInstallCommand: runtime_touched=True, registrations_touched=True, ) + if isinstance(err, DatabaseWorkerOverloadedError): + raise + return result checkpoint_cleanup_error = "" try: @@ -262,7 +276,7 @@ class PluginInstallCommand: registrations_restored = True except Exception as rollback_err: rollback_errors.append(f"路由和服务注册恢复失败：{rollback_err}") - return PluginInstallResult( + result = PluginInstallResult( success=False, message=f"刷新插件运行态失败：{err}", refreshed_only=True, @@ -275,6 +289,9 @@ class PluginInstallCommand: errors=tuple(rollback_errors), ), ) + if isinstance(err, DatabaseWorkerOverloadedError): + raise + return result reported = False report_error = "" diff --git a/app/db/worker.py b/app/db/worker.py index b00ee45ae..1690df663 100644 --- a/app/db/worker.py +++ b/app/db/worker.py @@ -67,6 +67,8 @@ class DatabaseWorker: ] = {} self._queued = 0 self._running = 0 + self._reported_queued = 0 + self._reported_running = 0 self._rejected = 0 self._closing = False @@ -200,12 +202,19 @@ class DatabaseWorker: ) self._record_depth() - async def _wait_until_done(self, future: asyncio.Future[object]) -> None: - """忽略后续取消请求，直到线程内事务结束。""" + async def _wait_until_done( + self, + future: asyncio.Future[object], + *, + interruptible: bool = False, + ) -> None: + """等待线程内事务结束，并按调用场景决定是否响应外层取消。""" while not future.done(): try: await asyncio.shield(future) except asyncio.CancelledError: + if interruptible: + raise continue except BaseException: break @@ -224,7 +233,8 @@ class DatabaseWorker: future.cancel() for future, (wrapped, _item) in futures: if not future.cancelled(): - await self._wait_until_done(wrapped) + # 关停超时必须能返回并保留 owner；已开始的数据库事务继续由线程完成。 + await self._wait_until_done(wrapped, interruptible=True) executor.shutdown(wait=True, cancel_futures=True) while self.snapshot().queued or self.snapshot().running: await asyncio.sleep(0) @@ -232,7 +242,13 @@ class DatabaseWorker: self._record_depth() def _record_depth(self) -> None: - """记录当前排队量与运行量。""" + """以状态变化量记录队列和运行中的任务数量。""" stats = self.snapshot() - record_metric("db.worker.queue.depth", stats.queued) - record_metric("db.worker.active", stats.running) + queued_delta = stats.queued - self._reported_queued + running_delta = stats.running - self._reported_running + if queued_delta: + record_metric("db.worker.queue.depth", queued_delta) + if running_delta: + record_metric("db.worker.active", running_delta) + self._reported_queued = stats.queued + self._reported_running = stats.running diff --git a/app/factory.py b/app/factory.py index 4ab628db1..5e7d77f3a 100644 --- a/app/factory.py +++ b/app/factory.py @@ -79,6 +79,7 @@ def _native_ai_error_response( protocol: str, status_code: int, message: str, + headers: dict[str, str] | None = None, ) -> JSONResponse: """按 OpenAI 或 Anthropic 兼容协议构造原生错误响应。""" if protocol == "openai": @@ -98,6 +99,7 @@ def _native_ai_error_response( code=error_type, ) ).model_dump(mode="json"), + headers=headers, ) error_type = ( @@ -112,6 +114,7 @@ def _native_ai_error_response( content=AnthropicErrorResponse( error=AnthropicErrorDetail(type=error_type, message=message) ).model_dump(mode="json"), + headers=headers, ) @@ -119,6 +122,7 @@ def _mcp_jsonrpc_error_response( status_code: int, code: int, message: str, + headers: dict[str, str] | None = None, ) -> JSONResponse: """构造带 HTTP 状态码的 MCP JSON-RPC 原生错误响应。""" return JSONResponse( @@ -128,6 +132,7 @@ def _mcp_jsonrpc_error_response( id=None, error=McpJsonRpcErrorDetail(code=code, message=message), ).model_dump(mode="json"), + headers=headers, ) @@ -202,6 +207,7 @@ async def localized_http_exception_handler( protocol=native_ai_protocol, status_code=exc.status_code, message=message, + headers=exc.headers, ) if _is_mcp_jsonrpc_request(request): error_codes = { @@ -215,6 +221,7 @@ async def localized_http_exception_handler( status_code=exc.status_code, code=error_codes.get(exc.status_code, -32000), message=message, + headers=exc.headers, ) return JSONResponse( status_code=exc.status_code, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 5ec5adf11..23382aed7 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -2640,6 +2640,8 @@ "app.application.plugin.folders -> app.runtime.log", "app.application.plugin.folders -> app.schemas", "app.application.plugin.folders -> app.schemas.types", + "app.application.plugin.install -> app.application", + "app.application.plugin.install -> app.application.database", "app.application.recognition -> app.application", "app.application.recognition -> app.application.configuration", "app.application.recognition -> app.schemas", diff --git a/tests/test_api_response.py b/tests/test_api_response.py index c4f48c476..c4419d92e 100644 --- a/tests/test_api_response.py +++ b/tests/test_api_response.py @@ -7,6 +7,7 @@ from fastapi import APIRouter, Depends, FastAPI, HTTPException from fastapi.responses import JSONResponse from fastapi.routing import APIRoute from pydantic import BaseModel, ValidationError +from starlette.requests import Request from starlette.responses import Response as StarletteResponse from starlette.responses import StreamingResponse @@ -24,6 +25,7 @@ from app.factory import ( ) from app.application.database import DatabaseWorkerOverloadedError from app.runtime.localization import LocaleHelper +from app.runtime.config import settings from app.schemas.common import JsonData from app.schemas.response import Response @@ -217,6 +219,40 @@ async def test_database_worker_overload_is_retryable_service_unavailable( } +@pytest.mark.parametrize( + "path", + [ + f"{settings.API_V1_STR}/openai/v1/chat/completions", + f"{settings.API_V1_STR}/anthropic/v1/messages", + f"{settings.API_V1_STR}/mcp", + ], +) +async def test_database_worker_overload_preserves_retry_after_for_native_protocols( + path: str, +): + """OpenAI、Anthropic 和 MCP 的原生 503 也必须保留重试提示。""" + scope = { + "type": "http", + "http_version": "1.1", + "method": "GET", + "scheme": "http", + "path": path, + "raw_path": path.encode(), + "query_string": b"", + "headers": [], + "server": ("testserver", 80), + "client": ("testclient", 123), + "root_path": "", + } + response = await database_worker_overloaded_handler( + Request(scope), + DatabaseWorkerOverloadedError("worker full"), + ) + + assert response.status_code == 503 + assert response.headers["retry-after"] == "1" + + async def test_validation_error_uses_unified_model(api_app: FastAPI): """请求参数校验失败应返回统一协议和明确的错误项结构。""" async with make_client(api_app) as client: diff --git a/tests/test_database_worker.py b/tests/test_database_worker.py index 563079d29..156862610 100644 --- a/tests/test_database_worker.py +++ b/tests/test_database_worker.py @@ -2,6 +2,7 @@ import asyncio import threading +from unittest.mock import patch import pytest @@ -138,3 +139,64 @@ async def test_shutdown_rejects_new_work_and_waits_for_running_work() -> None: assert worker.snapshot().closing is True assert worker.snapshot().queued == 0 assert worker.snapshot().running == 0 + + +@pytest.mark.asyncio +async def test_shutdown_timeout_keeps_running_owner_until_transaction_finishes() -> None: + """关闭超时应返回给生命周期编排，并保留执行器等待事务收敛。""" + worker = DatabaseWorker(max_workers=1, capacity=1) + await worker.start() + started = threading.Event() + release = threading.Event() + + def operation() -> None: + started.set() + release.wait(1) + + running = asyncio.create_task(worker.run(operation)) + await asyncio.to_thread(started.wait) + + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(worker.shutdown(), timeout=0.01) + + assert worker.snapshot().closing is True + assert worker._executor is not None + + release.set() + await running + await asyncio.sleep(0) + await worker.shutdown() + assert worker._executor is None + + +@pytest.mark.asyncio +async def test_worker_depth_metrics_emit_deltas_and_return_to_zero() -> None: + """队列和运行量指标按增减量上报，不能把绝对值累加成漂移。""" + worker = DatabaseWorker(max_workers=1, capacity=1) + started = threading.Event() + release = threading.Event() + + def operation() -> None: + started.set() + release.wait(1) + + with patch("app.db.worker.record_metric") as record_metric: + await worker.start() + running = asyncio.create_task(worker.run(operation)) + await asyncio.to_thread(started.wait) + release.set() + await running + await worker.shutdown() + + queue_values = [ + call.args[1] + for call in record_metric.call_args_list + if call.args[0] == "db.worker.queue.depth" + ] + active_values = [ + call.args[1] + for call in record_metric.call_args_list + if call.args[0] == "db.worker.active" + ] + assert queue_values == [1.0, -1.0] + assert active_values == [1.0, -1.0] diff --git a/tests/test_plugin_install_command.py b/tests/test_plugin_install_command.py index c0c559d00..8b989d222 100644 --- a/tests/test_plugin_install_command.py +++ b/tests/test_plugin_install_command.py @@ -2,6 +2,7 @@ from unittest.mock import AsyncMock, Mock import pytest +from app.application.database import DatabaseWorkerOverloadedError from app.application.plugin.install import PluginInstallCommand @@ -181,6 +182,26 @@ async def test_persistence_failure_restores_package_without_touching_runtime(): reloader.assert_not_awaited() +@pytest.mark.asyncio +async def test_database_worker_overload_rolls_back_and_reaches_api_boundary(): + """配置 worker 背压完成补偿后继续抛出，交由 API 映射为 503。""" + checkpoint = object() + rollback = AsyncMock() + command = _command( + checkpointer=AsyncMock(return_value=checkpoint), + writer=AsyncMock(side_effect=DatabaseWorkerOverloadedError("worker full")), + rollback=rollback, + ) + + with pytest.raises(DatabaseWorkerOverloadedError): + await command.execute( + plugin_id="DemoPlugin", + repo_url="https://github.com/demo/plugins", + ) + + rollback.assert_awaited_once_with(checkpoint) + + @pytest.mark.asyncio async def test_reload_failure_restores_list_files_and_previous_runtime(): """重载失败时依次恢复已安装列表、包文件和旧运行态。""" From 14e92555906f38f1ecf5cd07e4388343397dc664 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 02:31:00 +0800 Subject: [PATCH 5/7] chore(architecture): refresh dependency baseline --- tests/fixtures/architecture/dependency-baseline.json | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 23382aed7..78fc37996 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6421, - "edge_sha256": "889385bbfbe3634711d921dc76afe38bbfc3ea7c672663e67ae6453aad19af4c", + "edge_count": 6419, + "edge_sha256": "58a002380592d49f9b78e05b6f03b1950b8622213396e160917d7e2f73887527", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", From aa65e3ace78517dc435d373c7ca40115ab94299a Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 04:40:04 +0800 Subject: [PATCH 6/7] fix(ci): align architecture baseline and test imports --- tests/fixtures/architecture/dependency-baseline.json | 5 +++-- tests/test_agent_lifecycle.py | 2 +- tests/test_db_declarative_2_0.py | 2 +- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 78fc37996..fd34e1312 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6419, - "edge_sha256": "58a002380592d49f9b78e05b6f03b1950b8622213396e160917d7e2f73887527", + "edge_count": 6420, + "edge_sha256": "0e5b5cc00f36ed87ec8e6ae8d09f7a3756d8488302907e12db239bb981c98d5c", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -3260,6 +3260,7 @@ "app.chain.subscribe -> app.application.configuration", "app.chain.subscribe -> app.application.mediaserver", "app.chain.subscribe -> app.application.messaging", + "app.chain.subscribe -> app.application.messaging.message", "app.chain.subscribe -> app.application.messaging.subscribe", "app.chain.subscribe -> app.application.subscription", "app.chain.subscribe -> app.application.subscription.complete", diff --git a/tests/test_agent_lifecycle.py b/tests/test_agent_lifecycle.py index f5519c85b..8c6209fa7 100644 --- a/tests/test_agent_lifecycle.py +++ b/tests/test_agent_lifecycle.py @@ -4,9 +4,9 @@ from unittest.mock import AsyncMock, MagicMock import pytest import app.agent.orchestrator as agent_module -from app.agent import AgentManager from app.agent.orchestrator import ( AGENT_SESSION_QUEUE_MAX_SIZE, + AgentManager, AgentManagerQueueFullError, AgentManagerUnavailableError, ) diff --git a/tests/test_db_declarative_2_0.py b/tests/test_db_declarative_2_0.py index 56d153736..55959ec6b 100644 --- a/tests/test_db_declarative_2_0.py +++ b/tests/test_db_declarative_2_0.py @@ -16,7 +16,7 @@ from pathlib import Path import pytest from sqlalchemy.orm import DeclarativeBase -from app.db import Base +from app.db.base import Base from app.db.models import load_all_models From 4145979e22219dd1a9c0c02998c0bf74fa2697a9 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 04:57:12 +0800 Subject: [PATCH 7/7] fix(config): bind runtime service to legacy settings --- app/startup/modules_initializer.py | 7 +++- tests/test_configuration_initializer.py | 44 +++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 1 deletion(-) diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index 60bcdeca5..04cff61f1 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -195,6 +195,11 @@ async def _initialize_configuration_services( ) +def _build_runtime_settings_service() -> RuntimeSettingsService: + """将可变部署配置实现注入管理服务，避免把兼容代理再次包装。""" + return RuntimeSettingsService(legacy_settings) + + async def _async_get_subscribe(subscribe_id: int): """通过数据库操作器异步读取订阅，供服务端共享用例使用。""" return await SubscribeOper().async_get(subscribe_id) @@ -630,7 +635,7 @@ async def init_modules() -> HostRuntime: scheduler=lambda: build_scheduler_runtime_config(settings), chain=lambda: build_chain_runtime_config(settings), ) - runtime_settings = RuntimeSettingsService(settings) + runtime_settings = _build_runtime_settings_service() host_runtime = HostRuntime( agent_chat=AgentChatRuntime( async_session=get_async_db, diff --git a/tests/test_configuration_initializer.py b/tests/test_configuration_initializer.py index 84aaae367..24a765f90 100644 --- a/tests/test_configuration_initializer.py +++ b/tests/test_configuration_initializer.py @@ -6,6 +6,8 @@ import pytest from app.startup import modules_initializer from app.startup.lifecycle import initialize_modules_component +from app.application.configuration import configure_runtime_settings +from app.runtime.settings import RuntimeSettingsCompat class _InlineWorker: @@ -16,6 +18,48 @@ class _InlineWorker: return operation() +class _MutableSettings: + """提供运行时配置代理回归测试所需的最小可变设置实现。""" + + def __init__(self) -> None: + """初始化一项可读写的部署设置。""" + self.VALUE = "before" + + def model_dump(self, *, include=None, exclude=None): + """导出测试设置快照。""" + values = {"VALUE": self.VALUE} + if include is not None: + values = {key: value for key, value in values.items() if key in include} + if exclude is not None: + values = {key: value for key, value in values.items() if key not in exclude} + return values + + def update_settings(self, env): + """批量更新测试设置。""" + for key, value in env.items(): + setattr(self, key, value) + return {key: (True, "") for key in env} + + def update_setting(self, key, value): + """更新单项测试设置。""" + setattr(self, key, value) + return True, "" + + +def test_runtime_settings_compat_uses_legacy_settings_from_startup_root(monkeypatch) -> None: + """组合根装配的兼容代理应读写原始部署配置而不是自身。""" + legacy_settings = _MutableSettings() + monkeypatch.setattr(modules_initializer, "legacy_settings", legacy_settings) + + service = modules_initializer._build_runtime_settings_service() + configure_runtime_settings(service) + compat = RuntimeSettingsCompat() + + assert compat.model_dump(include={"VALUE"}) == {"VALUE": "before"} + assert compat.update_setting("VALUE", "after") == (True, "") + assert compat.model_dump(include={"VALUE"}) == {"VALUE": "after"} + + @pytest.mark.asyncio async def test_configuration_services_publish_after_both_snapshots_load( monkeypatch,