GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-10 07:54:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(lifecycle): harden application shutdown (#6125)

Browse Source
This commit is contained in:
InfinityPacer
2026-07-16 06:24:31 +08:00
committed by GitHub
parent 107ba37834
commit a4335fe753
9 changed files with 476 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
app/api/endpoints/system.py
View File

@@ -1339,12 +1339,7 @@ def restart_system(_: User = Depends(get_current_active_superuser)):
"""
if not SystemHelper.can_restart():
return schemas.Response(success=False, message="当前运行环境不支持重启操作！")
# 标识停止事件
global_vars.stop_system()
# 执行重启
ret, msg = SystemHelper.restart()
if not ret:
global_vars.resume_system()
return schemas.Response(success=ret, message=msg)
@@ -1362,11 +1357,7 @@ def upgrade_system(
if not SystemHelper.can_restart():
return schemas.Response(success=False, message="当前运行环境不支持升级操作！")
# 标识停止事件
global_vars.stop_system()
ret, msg = SystemHelper.upgrade(mode=mode or "release")
if not ret:
global_vars.resume_system()
return schemas.Response(success=ret, message=msg)

4
app/chain/system.py
View File

@@ -38,8 +38,6 @@ class SystemChain(ChainBase):
"""
重启系统
"""
from app.core.config import global_vars
if channel and userid:
self.post_message(Notification(
channel=channel,
@@ -54,8 +52,6 @@ class SystemChain(ChainBase):
}, self._restart_file)
# 主动备份一次插件
self.backup_plugins()
# 设置停止标志，通知所有模块准备停止
global_vars.stop_system()
# 重启
SystemHelper.restart()

6
app/core/config.py
View File

@@ -1211,12 +1211,6 @@ class GlobalVar(object):
"""
self.STOP_EVENT.set()
def resume_system(self):
"""
恢复系统运行标记。
"""
self.STOP_EVENT.clear()
@property
def is_system_stopped(self):
"""

34
app/main.py
View File

@@ -30,16 +30,31 @@ elif SystemUtils.is_frozen():
sys.stderr = open(os.devnull, 'w')
from app.factory import app
from app.core.config import settings
from app.core.config import global_vars, settings
from app.db.init import init_db, update_db
# 设置进程名
setproctitle.setproctitle(settings.PROJECT_NAME)
class MoviePilotServer(uvicorn.Server):
"""在 Uvicorn 开始优雅退出前发布应用协作停止标志"""
def handle_exit(self, sig, frame) -> None:
global_vars.stop_system()
super().handle_exit(sig, frame)
# uvicorn服务
Server = uvicorn.Server(Config(app, host=settings.HOST, port=settings.PORT,
reload=settings.DEV, workers=multiprocessing.cpu_count() * 2 + 1,
timeout_graceful_shutdown=60))
Server = MoviePilotServer(Config(app, host=settings.HOST, port=settings.PORT,
reload=settings.DEV, workers=multiprocessing.cpu_count() * 2 + 1,
timeout_graceful_shutdown=60))
def request_shutdown() -> None:
"""发布协作停止标志并请求 Uvicorn 退出"""
global_vars.stop_system()
Server.should_exit = True
def start_tray():
@@ -64,8 +79,8 @@ def start_tray():
"""
退出程序
"""
request_shutdown()
TrayIcon.stop()
Server.should_exit = True
import pystray
@@ -93,10 +108,11 @@ def signal_handler(signum, frame):
信号处理函数，用于优雅停止服务
"""
print(f"收到信号 {signum}，开始优雅停止服务...")
Server.should_exit = True
request_shutdown()
if __name__ == '__main__':
def run_application() -> None:
"""初始化进程并启动 API 服务"""
# 注册信号处理器
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
@@ -109,3 +125,7 @@ if __name__ == '__main__':
update_db()
# 启动API服务
Server.run()
if __name__ == '__main__':
run_application()

1
app/modules/thetvdb/__init__.py
View File

@@ -114,7 +114,6 @@ class TheTvDbModule(_ModuleBase):
return 4
def stop(self):
logger.info("TheTvDbModule 停止。正在清除 TVDB 会话。")
with self.__auth_lock:
self.tvdb = None

44
app/startup/lifecycle.py
View File

@@ -1,5 +1,7 @@
import asyncio
import inspect
from contextlib import asynccontextmanager
from typing import Callable
from fastapi import FastAPI
@@ -20,7 +22,7 @@ from app.chain.system import SystemChain
from app.core.config import global_vars, settings
from app.helper.server import MoviePilotServerHelper
from app.helper.system import SystemHelper
from app.log import LoggerManager
from app.log import logger, LoggerManager
from app.startup.command_initializer import init_command, stop_command, restart_command
from app.startup.modules_initializer import init_modules, stop_modules
from app.startup.monitor_initializer import stop_monitor, init_monitor
@@ -56,6 +58,16 @@ async def init_extra():
await MoviePilotServerHelper.async_report_usage()
async def run_shutdown_step(name: str, callback: Callable[[], object]) -> None:
"""隔离单个关闭阶段的异常，确保后续资源仍有机会释放"""
try:
result = callback()
if inspect.isawaitable(result):
await result
except Exception as err:
logger.error(f"关闭{name}失败：{err}")
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
@@ -90,6 +102,7 @@ async def lifespan(app: FastAPI):
yield
finally:
print("Shutting down...")
global_vars.stop_system()
# 取消同步插件任务
try:
sync_plugins_task.cancel()
@@ -100,22 +113,19 @@ async def lifespan(app: FastAPI):
print(str(e))
try:
if not settings.MOVIEPILOT_SAFE_MODE:
# 备份插件
SystemChain().backup_plugins()
# 停止工作流
stop_workflow()
# 停止命令
stop_command()
# 停止监控器
stop_monitor()
# 停止定时器
stop_scheduler()
# 停止插件
stop_plugins()
# 停止模块
await stop_modules()
# 关闭共享的异步 HTTP 连接池，释放底层连接资源
await aclose_shared_async_transports()
await run_shutdown_step(
"插件备份", lambda: SystemChain().backup_plugins()
)
await run_shutdown_step("工作流", stop_workflow)
await run_shutdown_step("命令服务", stop_command)
await run_shutdown_step("监控器", stop_monitor)
await run_shutdown_step("定时器", stop_scheduler)
await run_shutdown_step("插件", stop_plugins)
await run_shutdown_step("模块服务", stop_modules)
await run_shutdown_step(
"共享异步 HTTP 连接池",
aclose_shared_async_transports,
)
finally:
# 日志最后关闭，确保其他组件的收尾信息已写入文件
LoggerManager.shutdown()

46
app/startup/modules_initializer.py
View File

@@ -1,4 +1,6 @@
import inspect
import sys
from typing import Callable
from app.helper.redis import RedisHelper, AsyncRedisHelper
@@ -129,29 +131,27 @@ async def stop_modules():
"""
服务关闭
"""
# 停止AI智能体
await stop_agent()
# 停止模块
ModuleManager().stop()
# 停止事件消费
EventManager().stop()
# 停止虚拟显示
DisplayHelper().stop()
# 停止 DoH 服务
DohHelper().shutdown()
# 停止线程池
ThreadHelper().shutdown()
# 停止消息服务
stop_message()
# 关闭Redis缓存连接
RedisHelper().close()
await AsyncRedisHelper().close()
# 停止数据库连接
await close_database()
# 停止前端服务
stop_frontend()
# 清理临时文件
clear_temp()
async def run_step(name: str, callback: Callable[[], object]) -> None:
"""单个模块资源关闭失败时继续执行后续阶段"""
try:
result = callback()
if inspect.isawaitable(result):
await result
except Exception as err:
logger.error(f"关闭{name}失败：{err}")
await run_step("AI智能体", stop_agent)
await run_step("模块", lambda: ModuleManager().stop())
await run_step("事件消费", lambda: EventManager().stop())
await run_step("虚拟显示", lambda: DisplayHelper().stop())
await run_step("DoH服务", lambda: DohHelper().shutdown())
await run_step("线程池", lambda: ThreadHelper().shutdown())
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_frontend)
await run_step("临时文件", clear_temp)
def init_modules():

33
app/utils/http.py
View File

@@ -76,6 +76,14 @@ _REQUESTS_RETRY_IDEMPOTENT_METHODS = ("GET", "HEAD", "OPTIONS")
_pending_eviction_tasks: set[asyncio.Task] = set()
def _discard_pending_eviction_task(task: asyncio.Task) -> None:
"""从跨线程共享集合移除已完成的 transport 关闭任务"""
with _shared_async_transports_lock:
_pending_eviction_tasks.discard(task)
if not task.cancelled() and (error := task.exception()):
logger.debug(f"LRU 淘汰共享 transport 时关闭失败: {error!r}")
def _get_shared_async_transport(
proxy: Optional[str],
verify: Union[bool, str],
@@ -140,8 +148,9 @@ def _get_shared_async_transport(
try:
task = loop.create_task(evicted_transport.aclose())
# 强引用避免 task 仅被 loop 弱持有而触发 "Task was destroyed but pending"
_pending_eviction_tasks.add(task)
task.add_done_callback(_pending_eviction_tasks.discard)
with _shared_async_transports_lock:
_pending_eviction_tasks.add(task)
task.add_done_callback(_discard_pending_eviction_task)
except Exception as e: # pragma: no cover - 防御性
logger.debug(f"LRU 淘汰共享 transport 时调度关闭失败: {e!r}")
@@ -160,17 +169,27 @@ async def aclose_shared_async_transports() -> None:
# 弹出而非 get+clear，避免外层 dict 残留空 OrderedDict 占位
with _shared_async_transports_lock:
per_loop = _shared_async_transports.pop(loop, None)
if not per_loop:
pending_evictions = [
task
for task in _pending_eviction_tasks
if task.get_loop() is loop
]
transports = list(per_loop.values()) if per_loop else []
if per_loop:
per_loop.clear()
if not transports and not pending_evictions:
return
transports = list(per_loop.values())
per_loop.clear()
# 并行关闭：每个 transport 的 TLS close_notify 各占一个 RTT，
# 顺序等待会线性放大 shutdown 耗时；return_exceptions 让单点失败
# 不影响其他 transport 的释放
results = await asyncio.gather(
*(t.aclose() for t in transports), return_exceptions=True
*pending_evictions,
*(t.aclose() for t in transports),
return_exceptions=True,
)
for result in results:
with _shared_async_transports_lock:
_pending_eviction_tasks.difference_update(pending_evictions)
for result in results[len(pending_evictions):]:
if isinstance(result, BaseException):
logger.debug(f"关闭共享 AsyncHTTPTransport 失败: {result!r}")

387
tests/test_lifecycle_shutdown.py
View File

@@ -1,16 +1,28 @@
import asyncio
import signal
import threading
from unittest.mock import AsyncMock, MagicMock
import pytest
from fastapi import FastAPI
from app.startup import lifecycle
from app.startup import lifecycle, modules_initializer
from app.utils import http as http_utils
def test_lifespan_closes_logger_when_early_shutdown_step_fails(monkeypatch):
"""前置关闭步骤失败时仍应关闭 Logger"""
def _assert_completed_once(mock: MagicMock) -> None:
if isinstance(mock, AsyncMock):
mock.assert_awaited_once_with()
else:
mock.assert_called_once_with()
def _patch_lifespan(monkeypatch, *, failing_step: str | None = None) -> dict:
"""隔离 lifespan 的外部依赖，并按名称注入一个关闭失败"""
monkeypatch.setattr(lifecycle.settings, "MOVIEPILOT_SAFE_MODE", False)
monkeypatch.setattr(lifecycle.global_vars, "set_loop", MagicMock())
monkeypatch.setattr(lifecycle.global_vars, "stop_system", MagicMock())
for name in (
"init_routers",
"init_modules",
@@ -19,28 +31,375 @@ def test_lifespan_closes_logger_when_early_shutdown_step_fails(monkeypatch):
"init_monitor",
"init_command",
"init_workflow",
):
monkeypatch.setattr(lifecycle, name, MagicMock())
system_chain = MagicMock()
monkeypatch.setattr(lifecycle, "SystemChain", MagicMock(return_value=system_chain))
monkeypatch.setattr(lifecycle, "init_extra", AsyncMock())
shutdown_steps = {
"backup_plugins": system_chain.backup_plugins,
"stop_workflow": MagicMock(),
"stop_command": MagicMock(),
"stop_monitor": MagicMock(),
"stop_scheduler": MagicMock(),
"stop_plugins": MagicMock(),
"stop_modules": AsyncMock(),
"close_http": AsyncMock(),
}
for name in (
"stop_workflow",
"stop_command",
"stop_monitor",
"stop_scheduler",
"stop_plugins",
):
monkeypatch.setattr(lifecycle, name, MagicMock())
monkeypatch.setattr(lifecycle, name, shutdown_steps[name])
monkeypatch.setattr(lifecycle, "stop_modules", shutdown_steps["stop_modules"])
monkeypatch.setattr(
lifecycle,
"aclose_shared_async_transports",
shutdown_steps["close_http"],
)
if failing_step:
shutdown_steps[failing_step].side_effect = RuntimeError(
f"{failing_step} failed"
)
system_chain = MagicMock()
system_chain.backup_plugins.side_effect = RuntimeError("backup failed")
monkeypatch.setattr(lifecycle, "SystemChain", MagicMock(return_value=system_chain))
monkeypatch.setattr(lifecycle, "init_extra", AsyncMock())
monkeypatch.setattr(lifecycle, "stop_modules", AsyncMock())
monkeypatch.setattr(lifecycle, "aclose_shared_async_transports", AsyncMock())
logger_shutdown = MagicMock()
monkeypatch.setattr(lifecycle.LoggerManager, "shutdown", logger_shutdown)
shutdown_steps["logger"] = logger_shutdown
return shutdown_steps
@pytest.mark.parametrize(
"failing_step",
[
"backup_plugins",
"stop_workflow",
"stop_command",
"stop_monitor",
"stop_scheduler",
"stop_plugins",
"stop_modules",
"close_http",
],
)
def test_lifespan_continues_after_each_shutdown_owner_failure(
monkeypatch,
failing_step,
):
"""任一关闭阶段失败都不能跳过后续资源所有者"""
shutdown_steps = _patch_lifespan(monkeypatch, failing_step=failing_step)
async def run_lifespan():
with pytest.raises(RuntimeError, match="backup failed"):
async with lifecycle.lifespan(FastAPI()):
pass
async with lifecycle.lifespan(FastAPI()):
pass
asyncio.run(run_lifespan())
logger_shutdown.assert_called_once_with()
lifecycle.global_vars.stop_system.assert_called_once_with()
for step in shutdown_steps.values():
_assert_completed_once(step)
def test_uvicorn_signal_publishes_stop_before_server_exit(monkeypatch):
"""Uvicorn 接管系统信号时必须先发布协作停止标志"""
from app import main
calls = []
monkeypatch.setattr(main.global_vars, "stop_system", lambda: calls.append("stop"))
monkeypatch.setattr(
main.uvicorn.Server,
"handle_exit",
lambda _self, _sig, _frame: calls.append("uvicorn"),
)
server = object.__new__(main.MoviePilotServer)
server.handle_exit(signal.SIGTERM, None)
assert calls == ["stop", "uvicorn"]
def test_application_preserves_stop_requested_before_startup(monkeypatch):
"""启动流程不能清除初始化前已经发布的退出请求"""
from app import main
stop_event = threading.Event()
stop_event.set()
monkeypatch.setattr(main.global_vars, "STOP_EVENT", stop_event)
calls = []
monkeypatch.setattr(
main.signal,
"signal",
lambda *_args: calls.append("signal"),
)
monkeypatch.setattr(main, "start_tray", lambda: calls.append("tray"))
monkeypatch.setattr(main, "init_db", lambda: calls.append("init_db"))
monkeypatch.setattr(main, "update_db", lambda: calls.append("update_db"))
monkeypatch.setattr(main.Server, "run", lambda: calls.append("server"))
main.run_application()
assert stop_event.is_set()
assert calls == [
"signal",
"signal",
"tray",
"init_db",
"update_db",
"server",
]
def test_uvicorn_preserves_stop_requested_before_serve(monkeypatch):
"""Uvicorn 启动不能清除数据库初始化阶段已经发布的停止请求"""
from app import main
stop_event = threading.Event()
monkeypatch.setattr(main.global_vars, "STOP_EVENT", stop_event)
main.global_vars.stop_system()
async def serve(_self, sockets=None):
assert main.global_vars.is_system_stopped
monkeypatch.setattr(main.uvicorn.Server, "serve", serve)
server = object.__new__(main.MoviePilotServer)
asyncio.run(server.serve())
@pytest.mark.parametrize("endpoint_name", ["restart_system", "upgrade_system"])
@pytest.mark.parametrize(
"initially_stopped",
[False, True],
ids=["running", "stopping"],
)
def test_restart_endpoint_failure_preserves_stop_state(
monkeypatch,
endpoint_name,
initially_stopped,
):
"""重启或升级失败不能发布或撤销停止请求"""
from app.api.endpoints import system
stop_event = threading.Event()
if initially_stopped:
stop_event.set()
monkeypatch.setattr(system.global_vars, "STOP_EVENT", stop_event)
monkeypatch.setattr(system.SystemHelper, "can_restart", MagicMock(return_value=True))
monkeypatch.setattr(
system.SystemHelper,
"restart" if endpoint_name == "restart_system" else "upgrade",
MagicMock(return_value=(False, "restart failed")),
)
if endpoint_name == "restart_system":
response = system.restart_system(None)
else:
response = system.upgrade_system(None, None)
assert not response.success
assert stop_event.is_set() is initially_stopped
def test_command_restart_failure_does_not_publish_stop_request(monkeypatch):
"""命令重启失败时进程仍在运行，不能提前发布停止请求"""
from app.chain.system import SystemChain
from app.core.config import global_vars
stop_event = threading.Event()
monkeypatch.setattr(global_vars, "STOP_EVENT", stop_event)
monkeypatch.setattr(SystemChain, "backup_plugins", MagicMock())
restart = MagicMock(return_value=(False, "restart failed"))
monkeypatch.setattr("app.chain.system.SystemHelper.restart", restart)
chain = object.__new__(SystemChain)
chain.restart(channel=None, userid=None)
restart.assert_called_once_with()
assert not stop_event.is_set()
def test_stop_modules_continues_after_internal_owner_failures(monkeypatch):
"""模块关闭编排中的多个失败不能阻断其余清理"""
stop_agent = AsyncMock(side_effect=RuntimeError("agent failed"))
monkeypatch.setattr(modules_initializer, "stop_agent", stop_agent)
dependencies = _patch_module_shutdown_dependencies(monkeypatch)
dependencies["module"].side_effect = RuntimeError("module failed")
asyncio.run(modules_initializer.stop_modules())
stop_agent.assert_awaited_once_with()
for dependency in dependencies.values():
_assert_completed_once(dependency)
def _patch_module_shutdown_dependencies(monkeypatch) -> dict:
"""替换 stop_modules 的资源所有者，避免测试启动真实后台服务"""
dependencies = {}
for name, method_name in (
("ModuleManager", "stop"),
("EventManager", "stop"),
("DisplayHelper", "stop"),
("DohHelper", "shutdown"),
("ThreadHelper", "shutdown"),
("RedisHelper", "close"),
):
instance = MagicMock()
setattr(instance, method_name, MagicMock())
monkeypatch.setattr(
modules_initializer,
name,
MagicMock(return_value=instance),
)
key = name.removesuffix("Helper").removesuffix("Manager").lower()
dependencies[key] = getattr(instance, method_name)
for name in ("stop_message", "stop_frontend", "clear_temp"):
dependency = MagicMock()
monkeypatch.setattr(modules_initializer, name, dependency)
dependencies[name] = dependency
async_redis = MagicMock()
async_redis.close = AsyncMock()
monkeypatch.setattr(
modules_initializer,
"AsyncRedisHelper",
MagicMock(return_value=async_redis),
)
dependencies["async_redis"] = async_redis.close
close_database = AsyncMock()
monkeypatch.setattr(modules_initializer, "close_database", close_database)
dependencies["close_database"] = close_database
return dependencies
def test_shared_http_close_waits_for_real_lru_eviction(monkeypatch):
"""最终 HTTP 关闭必须等待真实 LRU 淘汰任务并消费其异常"""
class FakeTransport:
created = []
def __init__(self, **_kwargs):
self.close_started = asyncio.Event()
self.release_close = asyncio.Event()
self.closed = False
self.fail_on_close = not self.created
if not self.fail_on_close:
self.release_close.set()
self.created.append(self)
async def aclose(self):
self.close_started.set()
await self.release_close.wait()
self.closed = True
if self.fail_on_close:
raise RuntimeError("eviction close failed")
monkeypatch.setattr(http_utils, "_MAX_SHARED_TRANSPORTS_PER_LOOP", 1)
monkeypatch.setattr(http_utils.httpx, "AsyncHTTPTransport", FakeTransport)
debug = MagicMock()
monkeypatch.setattr(http_utils.logger, "debug", debug)
async def run_test():
transport_kwargs = {
"proxy": None,
"verify": True,
"http2": False,
"max_keepalive_connections": 1,
"max_connections": 1,
}
evicted_transport = http_utils._get_shared_async_transport(
**transport_kwargs,
keepalive_expiry=1,
)
active_transport = http_utils._get_shared_async_transport(
**transport_kwargs,
keepalive_expiry=2,
)
await asyncio.wait_for(evicted_transport.close_started.wait(), timeout=1)
loop = asyncio.get_running_loop()
with http_utils._shared_async_transports_lock:
eviction_tasks = [
task
for task in http_utils._pending_eviction_tasks
if task.get_loop() is loop
]
assert len(eviction_tasks) == 1
close_task = asyncio.create_task(http_utils.aclose_shared_async_transports())
await asyncio.sleep(0)
try:
assert not close_task.done()
evicted_transport.release_close.set()
await close_task
await asyncio.sleep(0)
assert eviction_tasks[0].done()
assert evicted_transport.closed
assert active_transport.closed
with http_utils._shared_async_transports_lock:
assert not any(
task.get_loop() is loop
for task in http_utils._pending_eviction_tasks
)
finally:
evicted_transport.release_close.set()
active_transport.release_close.set()
await asyncio.gather(close_task, return_exceptions=True)
await http_utils.aclose_shared_async_transports()
asyncio.run(run_test())
debug.assert_any_call(
"LRU 淘汰共享 transport 时关闭失败: "
"RuntimeError('eviction close failed')"
)
def test_shared_http_close_ignores_eviction_from_other_loop():
"""当前事件循环关闭不能等待其他循环持有的淘汰任务"""
ready = threading.Event()
release = threading.Event()
failures = []
state = {}
def run_foreign_loop():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
async def delayed_close():
while not release.is_set():
await asyncio.sleep(0.01)
task = loop.create_task(delayed_close())
state["task"] = task
with http_utils._shared_async_transports_lock:
http_utils._pending_eviction_tasks.add(task)
task.add_done_callback(http_utils._discard_pending_eviction_task)
ready.set()
try:
loop.run_until_complete(task)
loop.run_until_complete(asyncio.sleep(0))
except BaseException as err:
failures.append(err)
finally:
with http_utils._shared_async_transports_lock:
http_utils._pending_eviction_tasks.discard(task)
loop.close()
thread = threading.Thread(target=run_foreign_loop)
thread.start()
try:
assert ready.wait(timeout=2)
asyncio.run(http_utils.aclose_shared_async_transports())
assert thread.is_alive()
assert not state["task"].done()
finally:
release.set()
thread.join(timeout=2)
assert not thread.is_alive()
assert not failures