fanhuadesenlinnn/MoviePilot
1
0
Fork 0
mirror of https://hubproxy.babadafafafafa.cn/https://github.com/jxxghp/MoviePilot.git synced 2026-09-20 08:03:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(plugin): isolate startup recovery failures

Browse Source
This commit is contained in:
jxxghp
2026-09-12 20:38:03 +08:00
parent b3f748ffa8
commit c0158b1f63
5 changed files with 126 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
app/adapters/system/plugin/package.py
View File

@@ -3,6 +3,7 @@
from __future__ import annotations
import asyncio
import errno
import hashlib
import io
import re
@@ -447,32 +448,63 @@ class PluginPackageManager:
existed: bool,
label: str,
) -> None:
"""用同级 staging 替换目录，失败时保留替换前的当前目录。"""
"""用同级 staging 替换目录，并兼容 overlayfs 的跨设备替换。"""
if existed and not snapshot.is_dir():
raise FileNotFoundError(f"{label}补偿快照不存在：{snapshot}")
target.parent.mkdir(parents=True, exist_ok=True)
staging = target.parent / f".{target.name}.restore-{uuid.uuid4().hex}"
previous = target.parent / f".{target.name}.previous-{uuid.uuid4().hex}"
previous_available = False
published = False
try:
if existed:
shutil.copytree(snapshot, staging)
if target.exists():
target.replace(previous)
try:
target.replace(previous)
except OSError as error:
if error.errno != errno.EXDEV:
raise
# overlayfs 可能拒绝把镜像层目录直接 rename 到可写层，
# 先复制旧目标保留回滚材料，再删除旧目录继续发布快照。
if target.is_dir():
shutil.copytree(target, previous, symlinks=True)
else:
shutil.copy2(target, previous, follow_symlinks=False)
previous_available = True
PluginPackageManager.__remove_snapshot_path(target)
else:
previous_available = True
if existed:
staging.replace(target)
if previous.exists():
shutil.rmtree(previous)
published = True
except Exception:
if not target.exists() and previous.exists():
previous.replace(target)
if previous_available and not published:
try:
PluginPackageManager.__remove_snapshot_path(target)
previous.replace(target)
previous_available = False
except Exception as rollback_error:
logger.error(
f"恢复{label}旧目录失败，已保留恢复材料 {previous}: "
f"{rollback_error}"
)
raise
finally:
if staging.exists():
shutil.rmtree(staging, ignore_errors=True)
if target.exists() and previous.exists():
if published and previous.exists():
shutil.rmtree(previous, ignore_errors=True)
@staticmethod
def __remove_snapshot_path(path: Path) -> None:
"""删除待替换的当前路径，保留快照材料供失败回滚。"""
if path.is_dir() and not path.is_symlink():
shutil.rmtree(path)
elif path.exists() or path.is_symlink():
path.unlink()
@classmethod
def stage_persistent_backup(cls, checkpoint: PluginPackageCheckpoint) -> None:
"""把新载荷复制到持久配置目录的独立 staging，不覆盖现有备份。"""

34
app/application/plugin/recovery.py
View File

@@ -43,11 +43,12 @@ class PluginRecoveryPackagePort(PluginPackageTransactionPort, Protocol):
@dataclass(frozen=True, slots=True)
class PluginInstallationRecoveryResult:
"""启动恢复批次的 PREPARED 回滚和 COMMITTED 收尾数量。"""
"""启动恢复批次的成功、待清理和按插件隔离失败数量。"""
restored: int = 0
finalized: int = 0
cleanup_pending: int = 0
failed: int = 0
class PluginInstallationRecoveryService:
@@ -64,24 +65,35 @@ class PluginInstallationRecoveryService:
self.__packages = packages
async def replay(self) -> PluginInstallationRecoveryResult:
"""按创建顺序恢复全部 journal；关键事实不一致时阻止插件启动。"""
"""按创建顺序恢复 journal，隔离单个插件失败并保留其记录重试。"""
restored = 0
finalized = 0
cleanup_pending = 0
failed = 0
for record in await self.__persistence.list_installations():
checkpoint = self.__checkpoint(record)
if record.phase is PluginInstallationPhase.PREPARED:
await self.__restore_prepared(record, checkpoint)
restored += 1
continue
if await self.__finish_committed(record, checkpoint):
finalized += 1
else:
cleanup_pending += 1
try:
checkpoint = self.__checkpoint(record)
if record.phase is PluginInstallationPhase.PREPARED:
await self.__restore_prepared(record, checkpoint)
restored += 1
continue
if await self.__finish_committed(record, checkpoint):
finalized += 1
else:
cleanup_pending += 1
except Exception as error: # noqa: BLE001 - 单插件故障不得阻断宿主
failed += 1
logger.error(
"插件 %s 的安装恢复失败，已跳过本条恢复并保留 journal 供重试：%s",
record.plugin_id,
error,
exc_info=True,
)
return PluginInstallationRecoveryResult(
restored=restored,
finalized=finalized,
cleanup_pending=cleanup_pending,
failed=failed,
)
def __checkpoint(

7
docs/docker-startup.md
View File

@@ -383,6 +383,10 @@ Transfer、Workflow 和 MoviePilot Server 服务，注册站点资源版本读
`MOVIEPILOT_SAFE_MODE=true` 时跳过标记为 `NORMAL_ONLY` 的插件、调度器、监控、命令和工作流等组件，
但数据库、路由、核心模块服务和后台诊断入口仍会启动。
插件安装 journal 的启动恢复按插件逐条处理：单个插件的恢复或事实校验失败会记录日志、保留 journal
供后续重试，并跳过本条恢复，不阻断其他插件和宿主启动；数据库读取或插件恢复服务本身的全局故障仍按
生命周期组件失败处理。
### 8.2 数据库就绪边界
数据库准备先读取当前 revision 和代码唯一 head：
@@ -397,7 +401,8 @@ Transfer、Workflow 和 MoviePilot Server 服务，注册站点资源版本读
### 8.3 readiness
所有启用的 fail-fast 启动组件成功后，lifespan 才把应用标记为 `ready`。
所有启用的 fail-fast 启动组件成功后，lifespan 才把应用标记为 `ready`。插件恢复中的单插件失败已在
组件内部隔离，不会把该组件升级为宿主级启动失败。
Dockerfile 的 `HEALTHCHECK` 每 30 秒请求 `http://127.0.0.1:${PORT}/health/ready`：数据库迁移和完整 lifespan 成功后返回 200；启动、
失败或关停阶段返回 503。`/health/live` 只表示进程和事件循环仍可响应。

41
tests/test_plugin_installation_recovery.py
View File

@@ -14,10 +14,7 @@ from app.application.plugin.identity import (
PluginPayloadSourceType,
TrustedPluginSourceType,
)
from app.application.plugin.recovery import (
PluginInstallationRecoveryError,
PluginInstallationRecoveryService,
)
from app.application.plugin.recovery import PluginInstallationRecoveryService
from app.application.plugin.transaction import (
PluginInstallationPhase,
PluginInstallationRecord,
@@ -170,8 +167,8 @@ async def test_prepared_delete_failure_keeps_replayable_journal() -> None:
packages=packages,
)
with pytest.raises(PluginInstallationRecoveryError, match="未提交安装恢复失败"):
await service.replay()
first = await service.replay()
assert first.failed == 1
assert "txn-demo" in persistence.records
packages.async_cleanup.assert_not_awaited()
@@ -184,23 +181,33 @@ async def test_prepared_delete_failure_keeps_replayable_journal() -> None:
@pytest.mark.asyncio
async def test_prepared_restore_failure_blocks_plugin_import() -> None:
"""旧载荷无法恢复时必须保留 journal，并让启动阶段失败。"""
persistence = _Persistence([_record(phase=PluginInstallationPhase.PREPARED)])
async def test_prepared_restore_failure_does_not_block_other_plugins() -> None:
"""单个旧载荷无法恢复时保留 journal，但继续处理其他插件。"""
persistence = _Persistence(
[
_record(phase=PluginInstallationPhase.PREPARED, transaction_id="txn-bad"),
_record(phase=PluginInstallationPhase.PREPARED, transaction_id="txn-good"),
]
)
packages = _packages(
async_restore=AsyncMock(side_effect=RuntimeError("snapshot missing"))
async_restore=AsyncMock(
side_effect=[RuntimeError("snapshot missing"), None]
)
)
service = PluginInstallationRecoveryService(
persistence=persistence,
packages=packages,
)
with pytest.raises(PluginInstallationRecoveryError, match="snapshot missing"):
await service.replay()
result = await service.replay()
assert "txn-demo" in persistence.records
assert persistence.delete_calls == []
packages.async_cleanup.assert_not_awaited()
assert result.restored == 1
assert result.failed == 1
assert set(persistence.records) == {"txn-bad"}
assert persistence.delete_calls == [
("txn-good", PluginInstallationPhase.PREPARED)
]
packages.async_cleanup.assert_awaited_once()
@pytest.mark.asyncio
@@ -254,9 +261,9 @@ async def test_committed_fact_mismatch_blocks_plugin_import(
packages=packages,
)
with pytest.raises(PluginInstallationRecoveryError, match=message):
await service.replay()
result = await service.replay()
assert result.failed == 1
assert "txn-demo" in persistence.records
packages.async_finalize_persistent_backup.assert_not_awaited()
packages.async_commit.assert_not_awaited()

34
tests/test_plugin_package_manager.py
View File

@@ -1,3 +1,4 @@
import errno
import shutil
from pathlib import Path
from types import SimpleNamespace
@@ -45,6 +46,39 @@ def test_checkpoint_rollback_restores_existing_package(monkeypatch, tmp_path):
assert not checkpoint.transaction_dir.exists()
def test_checkpoint_rollback_handles_cross_device_plugin_directory(
monkeypatch,
tmp_path,
):
"""overlayfs 拒绝目录 rename 时仍应恢复旧插件并清理事务。"""
manager = _manager(monkeypatch, tmp_path)
plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
plugin_dir.mkdir(parents=True)
source_file = plugin_dir / "__init__.py"
source_file.write_text("old", encoding="utf-8")
checkpoint = manager.checkpoint("DemoPlugin")
source_file.write_text("new", encoding="utf-8")
(plugin_dir / "partial.py").write_text("partial", encoding="utf-8")
original_replace = Path.replace
def replace(self: Path, target: Path) -> Path:
"""只模拟运行目录跨设备 rename 失败，保留其他替换操作。"""
if self == plugin_dir and target.name.startswith(
".demoplugin.previous-"
):
raise OSError(errno.EXDEV, "Invalid cross-device link")
return original_replace(self, target)
monkeypatch.setattr(Path, "replace", replace)
manager.rollback(checkpoint)
assert source_file.read_text(encoding="utf-8") == "old"
assert not (plugin_dir / "partial.py").exists()
assert not checkpoint.transaction_dir.exists()
def test_checkpoint_uses_injected_plugin_root(monkeypatch, tmp_path):
"""显式装配的插件根目录必须覆盖全局运行目录设置。"""
_manager(monkeypatch, tmp_path)