From c0158b1f6331a8327f1bc3b067fbee30bb7ca611 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 12 Sep 2026 20:38:03 +0800 Subject: [PATCH] fix(plugin): isolate startup recovery failures --- app/adapters/system/plugin/package.py | 46 ++++++++++++++++++---- app/application/plugin/recovery.py | 34 ++++++++++------ docs/docker-startup.md | 7 +++- tests/test_plugin_installation_recovery.py | 41 +++++++++++-------- tests/test_plugin_package_manager.py | 34 ++++++++++++++++ 5 files changed, 126 insertions(+), 36 deletions(-) diff --git a/app/adapters/system/plugin/package.py b/app/adapters/system/plugin/package.py index 3565b3581..841cc48b9 100644 --- a/app/adapters/system/plugin/package.py +++ b/app/adapters/system/plugin/package.py @@ -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，不覆盖现有备份。""" diff --git a/app/application/plugin/recovery.py b/app/application/plugin/recovery.py index ebeef41bb..3455df2f1 100644 --- a/app/application/plugin/recovery.py +++ b/app/application/plugin/recovery.py @@ -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( diff --git a/docs/docker-startup.md b/docs/docker-startup.md index 23c776a56..72d9a053b 100644 --- a/docs/docker-startup.md +++ b/docs/docker-startup.md @@ -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` 只表示进程和事件循环仍可响应。 diff --git a/tests/test_plugin_installation_recovery.py b/tests/test_plugin_installation_recovery.py index 31d0652a1..f38efe5bc 100644 --- a/tests/test_plugin_installation_recovery.py +++ b/tests/test_plugin_installation_recovery.py @@ -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() diff --git a/tests/test_plugin_package_manager.py b/tests/test_plugin_package_manager.py index 1cff91503..c66cc6019 100644 --- a/tests/test_plugin_package_manager.py +++ b/tests/test_plugin_package_manager.py @@ -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)