Merge pull request #6401 from InfinityPacer/codex/arch/config-database-worker

feat(database): 配置快照异步端口与有界 worker

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,

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
 
@@ -874,8 +875,13 @@ 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 DatabaseWorkerOverloadedError:
+        raise
     except Exception as e:
         logger.error(f"[文件夹API] 保存文件夹配置失败: {str(e)}")
         return _SchemaResponse(success=False, message=str(e))
@@ -893,7 +899,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}' 创建成功"
         )

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)
     # 插件站点删除

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,

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 typing import Any, Optional, Protocol, cast
 
+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,12 @@ 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("系统配置异步数据库执行端口尚未配置")
+        result = await self._async_executor.run(partial(self._writer.set, key, value))
+        return cast(bool | None, result)
 
     def delete(self, key: Any) -> Any:
         """删除配置。"""

app/application/database.py

@@ -3,17 +3,35 @@
 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 DatabaseWorkerClosedError(RuntimeError):
+    """数据库执行器尚未启动或已经停止。"""
+
+
+class DatabaseWorkerOverloadedError(RuntimeError):
+    """数据库执行器的运行与排队容量已经用尽。"""
+
+
+class AsyncDatabaseExecutor(Protocol):
+    """让异步业务调用同步短事务而不阻塞事件循环。"""
+
+    async def run(self, operation: Callable[[], T]) -> T:
+        """等待数据库操作完成提交或回滚，并返回执行结果。"""
+        ...
 
 
 class DatabaseHealthService:

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 = ""

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
 

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)

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

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

app/db/oper/userconfig.py

@@ -1,13 +1,11 @@
-from typing import Any, Union, Dict, Optional, List
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session
+import copy
+import threading
+from typing import Any, Union, Dict, Optional
 
 from app.db.base import DbOper
 from app.db.models.userconfig import UserConfig
 from app.schemas.types import UserConfigKey
 from app.foundation.singleton import Singleton
-from app.db.decorators import run_legacy_sync_query
 
 
 class UserConfigOper(DbOper, metaclass=Singleton):
@@ -15,33 +13,30 @@ class UserConfigOper(DbOper, metaclass=Singleton):
     用户配置管理
     """
     def __init__(self):
-        """
-        加载配置到内存
-        """
+        """初始化空快照，数据库加载由启动组合根显式执行。"""
         super().__init__()
         self.__USERCONF = {}
-        for item in self._list_configs():
-            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 _with_sync_session(self, operation):
-        """在显式会话或兼容查询会话中执行只读操作。"""
-        if isinstance(self._db, Session):
-            return operation(self._db)
-        return run_legacy_sync_query(operation)
+    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 _list_configs(self) -> List[UserConfig]:
-        """读取全部用户配置，避免把 None 会话传入已显式化的 Model。"""
-        return self._with_sync_session(
-            lambda session: list(session.execute(select(UserConfig)).scalars().all())
-        )
-
-    def _get_by_key(self, username: str, key: str) -> Optional[UserConfig]:
-        """按用户名和键读取配置，复用调用方事务或一次性兼容会话。"""
-        return self._with_sync_session(
-            lambda session: UserConfig.get_by_key(
-                db=session, username=username, key=key
-            )
-        )
+    def _require_loaded(self) -> None:
+        """阻止消费者读取尚未完成启动加载的半成品快照。"""
+        if not self._loaded:
+            raise RuntimeError("用户配置快照尚未加载")
 
     def set(self, username: str, key: Union[str, UserConfigKey], value: Any):
         """
@@ -49,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 = self._get_by_key(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):
         """
@@ -80,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]]:
         """

app/db/worker.py

@@ -0,0 +1,254 @@
+"""同步数据库短事务的异步执行器。"""
+
+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.application.database import (
+    DatabaseWorkerClosedError,
+    DatabaseWorkerOverloadedError,
+)
+from app.runtime.observability import record_metric
+
+
+T = TypeVar("T")
+
+DATABASE_WORKER_MAX_WORKERS = 4
+DATABASE_WORKER_CAPACITY = 32
+
+
+@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 = DATABASE_WORKER_MAX_WORKERS,
+        capacity: int = DATABASE_WORKER_CAPACITY,
+    ) -> 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._reported_queued = 0
+        self._reported_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],
+        *,
+        interruptible: bool = False,
+    ) -> None:
+        """等待线程内事务结束，并按调用场景决定是否响应外层取消。"""
+        while not future.done():
+            try:
+                await asyncio.shield(future)
+            except asyncio.CancelledError:
+                if interruptible:
+                    raise
+                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():
+                # 关停超时必须能返回并保留 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)
+        self._executor = None
+        self._record_depth()
+
+    def _record_depth(self) -> None:
+        """以状态变化量记录队列和运行中的任务数量。"""
+        stats = self.snapshot()
+        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

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,
@@ -80,6 +81,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":
@@ -99,6 +101,7 @@ def _native_ai_error_response(
                     code=error_type,
                 )
             ).model_dump(mode="json"),
+            headers=headers,
         )
 
     error_type = (
@@ -113,6 +116,7 @@ def _native_ai_error_response(
         content=AnthropicErrorResponse(
             error=AnthropicErrorDetail(type=error_type, message=message)
         ).model_dump(mode="json"),
+        headers=headers,
     )
 
 
@@ -120,6 +124,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(
@@ -129,6 +134,7 @@ def _mcp_jsonrpc_error_response(
             id=None,
             error=McpJsonRpcErrorDetail(code=code, message=message),
         ).model_dump(mode="json"),
+        headers=headers,
     )
 
 
@@ -203,6 +209,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 = {
@@ -216,6 +223,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,
@@ -224,6 +232,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,
@@ -308,6 +331,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,

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"})),

app/startup/lifecycle/__init__.py

@@ -133,7 +133,16 @@ 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
+
+        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
 

app/startup/modules_initializer.py

@@ -92,6 +92,7 @@ from app.db.session import (
     get_async_db,
     get_db,
 )
+from app.db.worker import DatabaseWorker
 from app.db.uow import (
     SqlAlchemyAsyncUnitOfWork,
     SqlAlchemyUnitOfWork,
@@ -160,6 +161,45 @@ 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
+    if worker is not None:
+        await worker.shutdown()
+        _database_worker = None
+
+
+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,
+        )
+    )
+
+
+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)
@@ -205,15 +245,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(
@@ -525,7 +570,11 @@ 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("数据库连接", close_database)
+    await run_step("数据库任务", stop_database_worker)
+    if _database_worker is None:
+        await run_step("数据库连接", close_database)
+    else:
+        logger.error("数据库任务未收敛，跳过数据库连接关闭以避免运行中事务使用已释放连接")
     await run_step("前端服务", stop_frontend)
     await run_step("临时文件", clear_temp)
 
@@ -534,6 +583,7 @@ async def init_modules() -> HostRuntime:
     """
     启动模块并返回本次 lifespan 唯一的类型化 HostRuntime。
     """
+    global _database_worker
     # 兼容 Oper 的无 Session 写入口仍由组合根持有事务，避免模型恢复自动提交。
     transaction_runner = TransactionalWriteRunner(
         sync_session=SessionFactory,
@@ -543,6 +593,17 @@ 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:
+        try:
+            await stop_database_worker()
+        except Exception as cleanup_error:  # noqa: BLE001  保留原始启动异常
+            logger.error(f"启动失败后的数据库任务清理失败：{cleanup_error}")
+        raise
     # 数据访问能力统一在启动组合根注入，Runtime 和 Adapter 不再直接依赖 Oper。
     api_data = ApiDataPorts(
         sync_session=get_db,
@@ -574,9 +635,7 @@ async def init_modules() -> HostRuntime:
         scheduler=lambda: build_scheduler_runtime_config(settings),
         chain=lambda: build_chain_runtime_config(settings),
     )
-    # RuntimeSettingsService 必须持有真实 Settings；RuntimeSettingsCompat 是旧 ABI 代理，
-    # 将代理自身注入服务会让 model_dump() 在代理和服务之间无限递归。
-    runtime_settings = RuntimeSettingsService(legacy_settings)
+    runtime_settings = _build_runtime_settings_service()
     host_runtime = HostRuntime(
         agent_chat=AgentChatRuntime(
             async_session=get_async_db,
@@ -620,8 +679,6 @@ async def init_modules() -> HostRuntime:
     configure_runtime_settings(host_runtime.settings)
     configure_runtime_setting_provider(lambda key: getattr(legacy_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()
@@ -661,7 +718,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()))
@@ -710,8 +766,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智能体

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()

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] = []
@@ -125,6 +157,7 @@ def collect_configuration_debt_baseline() -> dict[str, Any]:
                 "app/plugins",
                 "app/sdk",
                 "app/runtime/compat",
+                "app/testing",
             ],
         },
         "settings_imports": {
@@ -145,7 +178,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

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:
@@ -2400,6 +2412,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 []

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,

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"
   },

tests/fixtures/architecture/dependency-baseline.json

@@ -13,8 +13,8 @@
     "runtime_to_db": [],
     "workflow_to_db": []
   },
-  "edge_count": 6432,
-  "edge_sha256": "96bead9b5bbbf10d49a5be50d7d9dc33f47e23010ef28d02a17aebfc3b250e4f",
+  "edge_count": 6429,
+  "edge_sha256": "664681a500ba1c3d273568829b1b860e4dae7fe7a6c053c566583341fee7f903",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -2052,6 +2052,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",
@@ -2466,15 +2467,14 @@
     "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.runtime",
     "app.application.configuration -> app.runtime.settings",
     "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",
@@ -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",
@@ -2714,6 +2716,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",
@@ -3456,6 +3460,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",
@@ -3563,23 +3568,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",
@@ -3667,7 +3655,6 @@
     "app.db.oper.user -> app.db.models.user",
     "app.db.oper.userconfig -> app.db",
     "app.db.oper.userconfig -> app.db.base",
-    "app.db.oper.userconfig -> app.db.decorators",
     "app.db.oper.userconfig -> app.db.models",
     "app.db.oper.userconfig -> app.db.models.userconfig",
     "app.db.oper.userconfig -> app.foundation",
@@ -3684,6 +3671,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",
@@ -3807,6 +3798,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",
@@ -6147,6 +6139,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",
@@ -6270,6 +6263,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",
@@ -6449,7 +6446,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",
@@ -6873,6 +6870,7 @@
     "app.db.oper.workflow",
     "app.db.session",
     "app.db.uow",
+    "app.db.worker",
     "app.doctor",
     "app.doctor.checks",
     "app.doctor.formatters",

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,
 )
@@ -123,22 +123,25 @@ 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()
     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()

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
 
@@ -17,11 +18,14 @@ 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.runtime.config import settings
 from app.schemas.common import JsonData
 from app.schemas.response import Response
 
@@ -59,6 +63,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 +120,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 +203,56 @@ 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,
+    }
+
+
+@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:

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"]

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)

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()
@@ -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()

tests/test_configuration_initializer.py

@@ -0,0 +1,184 @@
+"""配置快照启动顺序测试。"""
+
+from unittest.mock import AsyncMock
+
+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:
+    """按提交顺序执行配置加载操作。"""
+
+    async def run(self, operation):
+        """执行并返回操作结果。"""
+        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,
+) -> 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()
+
+
+@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

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,9 +19,18 @@ from app.application.configuration import (
     get_api_runtime_config_snapshot,
     get_transfer_retry_config,
 )
+from app.application.security.userconfig import UserConfigurationService
 from app.runtime.settings import RuntimeSettingsCompat, configure_runtime_settings_compat
 
 
+class _InlineDatabaseExecutor:
+    """同步执行测试操作，并保留异步应用端口的调用形态。"""
+
+    async def run(self, operation):
+        """执行并返回操作结果。"""
+        return operation()
+
+
 class _MutableSettings:
     """记录管理设置服务的读取和更新操作。"""
 
@@ -77,23 +86,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}

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"

tests/test_database_worker.py

@@ -0,0 +1,202 @@
+"""数据库短事务 worker 的容量、取消与关闭合同测试。"""
+
+import asyncio
+import threading
+from unittest.mock import patch
+
+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
+
+
+@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]

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
 
 
@@ -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[] 即可；

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 路径）
 # --------------------------------------------------------------------------- #

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():
     """重载失败时依次恢复已安装列表、包文件和旧运行态。"""

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(

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))

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
 

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")

tests/test_userconfig_oper.py

@@ -0,0 +1,129 @@
+"""用户配置快照与异步写入合同测试。"""
+
+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(db) -> 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(
+        db.session,
+        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(
+        db.session,
+        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(
+        db.session,
+        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