GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-09-05 15:38:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(database): bound async configuration writes

Browse Source
This commit is contained in:
InfinityPacer
2026-08-23 02:29:40 +08:00
parent 3d7fb44dc7
commit fc677a2bba
25 changed files with 408 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/adapters/external/server.py
Vendored
+34
View File
@@ -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
+3
View File
@@ -57,6 +57,7 @@ from app.api.dependencies.plugin import (
from app.adapters.external.server import MoviePilotServerHelper
from app.adapters.external.market import PluginHelper
from app.adapters.system.plugin.package import PluginPackageManager
from app.application.database import DatabaseWorkerOverloadedError
from app.runtime.log import logger
from app.schemas.types import SystemConfigKey
@@ -879,6 +880,8 @@ async def save_plugin_folders(
folders,
)
return _SchemaResponse(success=True)
except DatabaseWorkerOverloadedError:
raise
except Exception as e:
logger.error(f"[文件夹API] 保存文件夹配置失败: {str(e)}")
return _SchemaResponse(success=False, message=str(e))
app/application/configuration.py
+3 -2
View File
@@ -6,7 +6,7 @@ from collections.abc import Callable
from dataclasses import dataclass
from functools import partial
from pathlib import Path
from typing import Any, Optional, Protocol
from typing import Any, Optional, Protocol, cast
from app.application.database import AsyncDatabaseExecutor
from app.schemas.types import MediaType
@@ -315,7 +315,8 @@ class SystemConfigService:
"""异步写入配置，并等待数据库提交或回滚完成。"""
if self._async_executor is None:
raise RuntimeError("系统配置异步数据库执行端口尚未配置")
return await self._async_executor.run(partial(self._writer.set, key, value))
result = await self._async_executor.run(partial(self._writer.set, key, value))
return cast(bool | None, result)
def delete(self, key: Any) -> Any:
"""删除配置。"""
app/application/database.py
+8
View File
@@ -18,6 +18,14 @@ DatabaseProbe = Callable[[], Optional[str]]
T = TypeVar("T")
class DatabaseWorkerClosedError(RuntimeError):
"""数据库执行器尚未启动或已经停止。"""
class DatabaseWorkerOverloadedError(RuntimeError):
"""数据库执行器的运行与排队容量已经用尽。"""
class AsyncDatabaseExecutor(Protocol):
"""让异步业务调用同步短事务而不阻塞事件循环。"""
app/application/server/report.py
+43
View File
@@ -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
+3 -2
View File
@@ -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/worker.py
+12 -8
View File
@@ -10,18 +10,17 @@ from contextvars import copy_context
from dataclasses import dataclass
from typing import Callable, TypeVar
from app.application.database import (
DatabaseWorkerClosedError,
DatabaseWorkerOverloadedError,
)
from app.runtime.observability import record_metric
T = TypeVar("T")
class DatabaseWorkerClosedError(RuntimeError):
"""数据库执行器尚未启动或已经停止。"""
class DatabaseWorkerOverloadedError(RuntimeError):
"""数据库执行器的运行与排队容量已经用尽。"""
DATABASE_WORKER_MAX_WORKERS = 4
DATABASE_WORKER_CAPACITY = 32
@dataclass(frozen=True, slots=True)
@@ -47,7 +46,12 @@ class _WorkItem:
class DatabaseWorker:
"""以有限线程和队列执行不能原生异步化的数据库短事务。"""
def __init__(self, *, max_workers: int = 4, capacity: int = 32) -> None:
def __init__(
self,
*,
max_workers: int = DATABASE_WORKER_MAX_WORKERS,
capacity: int = DATABASE_WORKER_CAPACITY,
) -> None:
"""保存容量配置，线程只在显式启动后创建。"""
if max_workers < 1:
raise ValueError("数据库 worker 线程数必须大于 0")
app/factory.py
+20
View File
@@ -14,6 +14,7 @@ from app.adapters.observability.otel import build_observation_port
from app.adapters.web.plugin.routes import FastAPIDynamicRouteRegistry
from app.adapters.web.health import install_health_routes
from app.application.plugin.routes import configure_plugin_routes
from app.application.database import DatabaseWorkerOverloadedError
from app.adapters.web.security.access import (
configure_token_codec,
verify_apikey,
@@ -222,6 +223,21 @@ async def localized_http_exception_handler(
)
async def database_worker_overloaded_handler(
request: Request,
_exc: DatabaseWorkerOverloadedError,
) -> JSONResponse:
"""将数据库短事务背压映射为可重试的 503 响应。"""
return await localized_http_exception_handler(
request,
HTTPException(
status_code=503,
detail="服务当前繁忙，请稍后重试",
headers={"Retry-After": "1"},
),
)
async def localized_validation_exception_handler(
request: Request,
exc: RequestValidationError,
@@ -306,6 +322,10 @@ def create_app() -> FastAPI:
)
_app.add_exception_handler(HTTPException, localized_http_exception_handler)
_app.add_exception_handler(
DatabaseWorkerOverloadedError,
database_worker_overloaded_handler,
)
_app.add_exception_handler(
RequestValidationError,
localized_validation_exception_handler,
app/startup/lifecycle/__init__.py
+4 -1
View File
@@ -135,7 +135,10 @@ async def initialize_modules_component(app: FastAPI) -> None:
except BaseException:
from app.startup.modules_initializer import stop_database_worker
await stop_database_worker()
try:
await stop_database_worker()
except Exception as cleanup_error: # noqa: BLE001 保留原始启动异常
logger.error(f"启动失败后的数据库任务清理失败：{cleanup_error}")
raise
if runtime is not None:
app.state.host_runtime = runtime
app/startup/modules_initializer.py
+16 -5
View File
@@ -165,9 +165,9 @@ async def stop_database_worker() -> None:
"""停止当前进程的数据库短事务 worker。"""
global _database_worker
worker = _database_worker
_database_worker = None
if worker is not None:
await worker.shutdown()
_database_worker = None
async def _initialize_configuration_services(
@@ -237,15 +237,20 @@ def configure_runtime_data_providers() -> None:
report_service=ServerReportService(
config_reader=lambda key: get_configured_system_config().get(key),
config_writer=lambda key, value: get_configured_system_config().set(key, value),
async_config_writer=lambda key, value: get_configured_system_config().async_set(
key, value
),
installed_plugins_provider=lambda: get_configured_system_config().get(
SystemConfigKey.UserInstalledPlugins
) or [],
subscribes_provider=lambda: SubscribeOper().list(),
async_subscribes_provider=lambda: SubscribeOper().async_list(),
plugin_report_sender=MoviePilotServerHelper.plugin_install_report,
async_plugin_report_sender=(
MoviePilotServerHelper.async_plugin_install_report
),
subscribe_report_sender=MoviePilotServerHelper.subscribe_report,
async_subscribe_report_sender=MoviePilotServerHelper.async_subscribe_report,
repo_url_sanitizer=MoviePilotServerHelper.sanitize_plugin_repo_url,
),
sharing_service=ServerSharingService(
@@ -542,7 +547,10 @@ async def stop_modules():
await run_step("Redis缓存连接", lambda: RedisHelper().close())
await run_step("异步Redis缓存连接", lambda: AsyncRedisHelper().close())
await run_step("数据库任务", stop_database_worker)
await run_step("数据库连接", close_database)
if _database_worker is None:
await run_step("数据库连接", close_database)
else:
logger.error("数据库任务未收敛，跳过数据库连接关闭以避免运行中事务使用已释放连接")
await run_step("前端服务", stop_frontend)
await run_step("临时文件", clear_temp)
@@ -567,7 +575,10 @@ async def init_modules() -> HostRuntime:
try:
await _initialize_configuration_services(database_worker)
except BaseException:
await stop_database_worker()
try:
await stop_database_worker()
except Exception as cleanup_error: # noqa: BLE001 保留原始启动异常
logger.error(f"启动失败后的数据库任务清理失败：{cleanup_error}")
raise
# 数据访问能力统一在启动组合根注入，Runtime 和 Adapter 不再直接依赖 Oper。
api_data = ApiDataPorts(
@@ -731,8 +742,8 @@ async def init_modules() -> HostRuntime:
# 启动事件消费
EventManager().start()
# 初始化共享服务端状态
MoviePilotServerHelper.init_plugin_report()
MoviePilotServerHelper.init_subscribe_report()
await MoviePilotServerHelper.async_init_plugin_report()
await MoviePilotServerHelper.async_init_subscribe_report()
MoviePilotServerHelper.get_user_uuid()
MoviePilotServerHelper.get_github_user()
# 初始化AI智能体