fix(persistence): separate retryable rejection errors (#6423)

app/api/endpoints/plugin.py

@@ -57,7 +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.schemas.exception import DatabaseWorkerOverloadedError
+from app.schemas.exception import PersistenceUnavailableError
 from app.runtime.log import logger
 from app.schemas.types import SystemConfigKey
 from app.api.context import get_background_task_registry, resolve_background_task_registry
@@ -893,7 +893,7 @@ async def save_plugin_folders(
             folders,
         )
         return _SchemaResponse(success=True)
-    except DatabaseWorkerOverloadedError:
+    except PersistenceUnavailableError:
         raise
     except Exception as e:
         logger.error(f"[文件夹API] 保存文件夹配置失败: {str(e)}")

app/application/messaging/chat.py

@@ -12,10 +12,7 @@ from app.application.database import (
     AsyncDatabaseExecutor,
 )
 from app.schemas.agent import AgentChatSessionDetail, AgentChatSessionSummary
-from app.schemas.exception import (
-    DatabaseWorkerClosedError,
-    DatabaseWorkerOverloadedError,
-)
+from app.schemas.exception import AgentChatPersistenceUnavailableError
 from app.runtime.observability import record_metric
 
 
@@ -364,14 +361,16 @@ class AgentChatPersistenceService:
         """在线程 worker 内完成同步写入并丢弃仓储对象返回值。"""
         # 同时限制全局和单会话等待量，避免一个热点会话占满总 admission 后饿死其他会话。
         if self._closing:
-            raise DatabaseWorkerClosedError("AgentChat 持久化服务当前不可接收任务")
+            raise AgentChatPersistenceUnavailableError(
+                "AgentChat 持久化服务当前不可接收任务"
+            )
         session_pending = self._pending_by_session.get(session_id, 0)
         if (
             self._pending_writes >= self._capacity
             or session_pending >= self._session_capacity
         ):
             record_metric("agent.chat.persistence.rejected")
-            raise DatabaseWorkerOverloadedError(
+            raise AgentChatPersistenceUnavailableError(
                 f"AgentChat 写入容量已用尽（全局上限 {self._capacity}，"
                 f"单会话上限 {self._session_capacity}）"
             )

app/application/plugin/install.py

@@ -7,7 +7,7 @@ from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 from typing import Any, Optional
 
-from app.schemas.exception import DatabaseWorkerOverloadedError
+from app.schemas.exception import PersistenceUnavailableError
 from app.application.plugin.lifecycle import plugin_lifecycle
 from app.runtime.log import logger
 
@@ -196,7 +196,7 @@ class PluginInstallCommand:
                 message=str(err),
                 package_installed=False,
             )
-            if isinstance(err, DatabaseWorkerOverloadedError):
+            if isinstance(err, PersistenceUnavailableError):
                 raise
             return result
         if not package_installed:
@@ -228,7 +228,7 @@ class PluginInstallCommand:
                     package_installed=True,
                     installed_list_persisted=state.installed_list_touched,
                 )
-                if isinstance(err, DatabaseWorkerOverloadedError):
+                if isinstance(err, PersistenceUnavailableError):
                     raise
                 return result
 
@@ -247,7 +247,7 @@ class PluginInstallCommand:
                 installed_list_persisted=installed_list_persisted,
                 runtime_touched=True,
             )
-            if isinstance(err, DatabaseWorkerOverloadedError):
+            if isinstance(err, PersistenceUnavailableError):
                 raise
             return result
 
@@ -267,7 +267,7 @@ class PluginInstallCommand:
                 runtime_touched=True,
                 registrations_touched=True,
             )
-            if isinstance(err, DatabaseWorkerOverloadedError):
+            if isinstance(err, PersistenceUnavailableError):
                 raise
             return result
 
@@ -412,7 +412,7 @@ class PluginInstallCommand:
                 failure_stage=failure_stage,
                 rollback=rollback,
             )
-            if isinstance(err, DatabaseWorkerOverloadedError):
+            if isinstance(err, PersistenceUnavailableError):
                 raise
             return result
 

app/factory.py

@@ -15,8 +15,7 @@ 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.schemas.exception import (
-    DatabaseWorkerClosedError,
-    DatabaseWorkerOverloadedError,
+    PersistenceUnavailableError,
 )
 from app.adapters.web.security.access import (
     configure_token_codec,
@@ -235,11 +234,11 @@ async def localized_http_exception_handler(
     )
 
 
-async def database_worker_overloaded_handler(
+async def persistence_unavailable_handler(
         request: Request,
-        _exc: DatabaseWorkerClosedError | DatabaseWorkerOverloadedError,
+        _exc: PersistenceUnavailableError,
 ) -> JSONResponse:
-    """将数据库 worker 暂不可用映射为可重试的 503 响应。"""
+    """将持久化能力暂不可用映射为可重试的 503 响应。"""
     return await localized_http_exception_handler(
         request,
         HTTPException(
@@ -335,12 +334,8 @@ 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(
-        DatabaseWorkerClosedError,
-        database_worker_overloaded_handler,
+        PersistenceUnavailableError,
+        persistence_unavailable_handler,
     )
     _app.add_exception_handler(
         RequestValidationError,

app/schemas/exception.py

@@ -58,14 +58,22 @@ class StorageQueryError(Exception):
     pass
 
 
-class DatabaseWorkerClosedError(RuntimeError):
+class PersistenceUnavailableError(RuntimeError):
+    """持久化执行能力暂时拒绝新操作，调用方可在稍后重试。"""
+
+
+class DatabaseWorkerClosedError(PersistenceUnavailableError):
     """数据库执行器尚未启动或已经停止。"""
 
 
-class DatabaseWorkerOverloadedError(RuntimeError):
+class DatabaseWorkerOverloadedError(PersistenceUnavailableError):
     """数据库执行器的运行与排队容量已经用尽。"""
 
 
+class AgentChatPersistenceUnavailableError(PersistenceUnavailableError):
+    """AgentChat 持久化服务因关闭或自身容量限制拒绝新写入。"""
+
+
 class TMDbException(Exception):
     """
     用于表示TheMovieDB数据源请求失败的跨层异常契约。

tests/test_agent_chat_persistence.py

@@ -12,7 +12,7 @@ import pytest
 from sqlalchemy import delete, select
 
 from app.schemas.exception import (
-    DatabaseWorkerClosedError,
+    AgentChatPersistenceUnavailableError,
     DatabaseWorkerOverloadedError,
 )
 from app.application.messaging.chat import AgentChatPersistenceService, AgentChatService
@@ -302,7 +302,7 @@ async def test_agent_chat_persistence_bounds_session_waiters_and_releases_cancel
             messages=[],
         )
     )
-    with pytest.raises(DatabaseWorkerOverloadedError):
+    with pytest.raises(AgentChatPersistenceUnavailableError):
         await third
     second.cancel()
     with pytest.raises(asyncio.CancelledError):
@@ -347,7 +347,7 @@ async def test_agent_chat_persistence_session_admission_is_fair() -> None:
         )
     )
     await asyncio.sleep(0)
-    with pytest.raises(DatabaseWorkerOverloadedError):
+    with pytest.raises(AgentChatPersistenceUnavailableError):
         await service.async_save_agent_messages(
             session_id="hot-session", user_id="1", messages=[]
         )
@@ -393,7 +393,7 @@ async def test_agent_chat_persistence_shutdown_drains_active_writes() -> None:
     shutdown = asyncio.create_task(service.shutdown())
     await asyncio.sleep(0)
     assert not shutdown.done()
-    with pytest.raises(DatabaseWorkerClosedError):
+    with pytest.raises(AgentChatPersistenceUnavailableError):
         await service.async_save_agent_messages(
             session_id="new-session", user_id="1", messages=[]
         )

tests/test_api_response.py

@@ -18,14 +18,16 @@ 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,
+    persistence_unavailable_handler,
 )
 from app.schemas.exception import (
+    AgentChatPersistenceUnavailableError,
     DatabaseWorkerClosedError,
     DatabaseWorkerOverloadedError,
+    PersistenceUnavailableError,
 )
 from app.runtime.localization import LocaleHelper
 from app.runtime.config import settings
@@ -67,12 +69,8 @@ def api_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,
-    )
-    app.add_exception_handler(
-        DatabaseWorkerClosedError,
-        database_worker_overloaded_handler,
+        PersistenceUnavailableError,
+        persistence_unavailable_handler,
     )
     from fastapi.exceptions import RequestValidationError
 
@@ -137,6 +135,11 @@ def api_app() -> FastAPI:
         """模拟数据库 worker 在关闭态拒绝新任务。"""
         raise DatabaseWorkerClosedError("worker closed")
 
+    @app.get("/agent-chat-persistence-unavailable")
+    async def get_agent_chat_persistence_unavailable() -> None:
+        """模拟 AgentChat 自身 admission 拒绝新写入。"""
+        raise AgentChatPersistenceUnavailableError("agent persistence full")
+
     @app.get("/native", response_model=None)
     async def get_native_response() -> dict[str, bool]:
         """返回显式旁路的原生 JSON 协议。"""
@@ -247,14 +250,30 @@ async def test_database_worker_closed_is_retryable_service_unavailable(
     }
 
 
-def test_create_app_registers_closed_database_worker_handler() -> None:
-    """生产组合根必须为 worker 关闭态登记 503 处理器。"""
+async def test_agent_chat_persistence_rejection_is_retryable_service_unavailable(
+        api_app: FastAPI,
+) -> None:
+    """AgentChat 自身 admission 拒绝也应返回可重试的 503。"""
+    async with make_client(api_app) as client:
+        response = await client.get("/agent-chat-persistence-unavailable")
+
+    assert response.status_code == 503
+    assert response.headers["retry-after"] == "1"
+    assert response.json() == {
+        "success": False,
+        "message": "服务当前繁忙，请稍后重试",
+        "data": None,
+    }
+
+
+def test_create_app_registers_persistence_unavailable_handler() -> None:
+    """生产组合根必须为持久化暂不可用登记统一 503 处理器。"""
     from app.factory import create_app
 
     app = create_app()
 
-    assert app.exception_handlers[DatabaseWorkerClosedError] is (
-        database_worker_overloaded_handler
+    assert app.exception_handlers[PersistenceUnavailableError] is (
+        persistence_unavailable_handler
     )
 
 
@@ -283,7 +302,7 @@ async def test_database_worker_overload_preserves_retry_after_for_native_protoco
         "client": ("testclient", 123),
         "root_path": "",
     }
-    response = await database_worker_overloaded_handler(
+    response = await persistence_unavailable_handler(
         Request(scope),
         DatabaseWorkerOverloadedError("worker full"),
     )

tests/test_plugin_install_command.py

@@ -3,7 +3,11 @@ from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 
-from app.schemas.exception import DatabaseWorkerOverloadedError
+from app.schemas.exception import (
+    DatabaseWorkerClosedError,
+    DatabaseWorkerOverloadedError,
+    PersistenceUnavailableError,
+)
 from app.application.plugin.install import PluginInstallCommand
 
 
@@ -249,17 +253,23 @@ async def test_persistence_exception_after_write_restores_installed_list():
 
 
 @pytest.mark.asyncio
-async def test_database_worker_overload_rolls_back_and_reaches_api_boundary():
-    """配置 worker 背压完成补偿后继续抛出，交由 API 映射为 503。"""
+@pytest.mark.parametrize(
+    "error_type",
+    [DatabaseWorkerClosedError, DatabaseWorkerOverloadedError],
+)
+async def test_persistence_unavailable_rolls_back_and_reaches_api_boundary(
+        error_type: type[PersistenceUnavailableError],
+) -> None:
+    """持久化能力暂不可用时完成补偿并交由 API 映射为 503。"""
     checkpoint = object()
     rollback = AsyncMock()
     command = _command(
         checkpointer=AsyncMock(return_value=checkpoint),
-        writer=AsyncMock(side_effect=DatabaseWorkerOverloadedError("worker full")),
+        writer=AsyncMock(side_effect=error_type("persistence unavailable")),
         rollback=rollback,
     )
 
-    with pytest.raises(DatabaseWorkerOverloadedError):
+    with pytest.raises(error_type):
         await command.execute(
             plugin_id="DemoPlugin",
             repo_url="https://github.com/demo/plugins",