diff --git a/backend/src/ab_sdk/manifest.py b/backend/src/ab_sdk/manifest.py index 2ba7bc42..b4eebd2e 100644 --- a/backend/src/ab_sdk/manifest.py +++ b/backend/src/ab_sdk/manifest.py @@ -186,7 +186,7 @@ def load_manifest(plugin_dir: Path) -> PluginManifest: path = plugin_dir / MANIFEST_NAME try: text = path.read_text(encoding="utf-8") - except OSError as e: + except (OSError, UnicodeDecodeError) as e: raise ManifestError(f"{path}: 无法读取：{e}") from e return parse_manifest(text, str(path)) diff --git a/backend/src/module/plugin/bus.py b/backend/src/module/plugin/bus.py index ef956da4..c0513d2c 100644 --- a/backend/src/module/plugin/bus.py +++ b/backend/src/module/plugin/bus.py @@ -16,6 +16,7 @@ logger = logging.getLogger(__name__) Handler = Callable[[Event], Awaitable[None] | None] ErrorCallback = Callable[[str, str], None] +SuccessCallback = Callable[[str], None] ALL_EVENTS = "*" DEFAULT_HANDLER_TIMEOUT = 30.0 @@ -31,12 +32,14 @@ class _Subscriber: timeout: float, queue_size: int, on_error: ErrorCallback | None, + on_success: SuccessCallback | None, ) -> None: self.kind = kind self.handler = handler self.owner = owner self._timeout = timeout self._on_error = on_error + self._on_success = on_success self._queue: asyncio.Queue[Event] = asyncio.Queue(maxsize=queue_size) self._worker: asyncio.Task | None = None @@ -80,6 +83,9 @@ class _Subscriber: ) if self._on_error is not None and self.owner is not None: self._on_error(self.owner, reason) + else: + if self._on_success is not None and self.owner is not None: + self._on_success(self.owner) finally: self._queue.task_done() @@ -103,12 +109,14 @@ class EventBus: handler_timeout: float = DEFAULT_HANDLER_TIMEOUT, queue_size: int = DEFAULT_QUEUE_SIZE, on_error: ErrorCallback | None = None, + on_success: SuccessCallback | None = None, ) -> None: self._timeout = handler_timeout self._queue_size = queue_size self._subscribers: list[_Subscriber] = [] - # 订阅者失败回调（owner, reason），宿主用它驱动插件熔断 + # 订阅者失败 / 成功回调，宿主用它驱动插件熔断（成功时清零连续失败计数） self.on_error = on_error + self.on_success = on_success def subscribe( self, @@ -125,6 +133,7 @@ class EventBus: timeout if timeout is not None else self._timeout, self._queue_size, self._report_error, + self._report_success, ) self._subscribers.append(sub) @@ -157,3 +166,7 @@ class EventBus: def _report_error(self, owner: str, reason: str) -> None: if self.on_error is not None: self.on_error(owner, reason) + + def _report_success(self, owner: str) -> None: + if self.on_success is not None: + self.on_success(owner) diff --git a/backend/src/module/plugin/loader.py b/backend/src/module/plugin/loader.py index ee858458..bd10793e 100644 --- a/backend/src/module/plugin/loader.py +++ b/backend/src/module/plugin/loader.py @@ -222,10 +222,14 @@ def _unload_package(package: str, plugin_dir: Path) -> None: def _entry_point_candidate(ep: importlib.metadata.EntryPoint) -> PluginCandidate: top_package = ep.module.split(".", 1)[0] try: + # 取包内文件会执行插件包顶层代码，语法错误、缺依赖等都在这里出现 package_files = importlib.resources.files(top_package) + except Exception as e: + raise PluginLoadError(f"导入 {top_package} 失败：{e}") from e + try: text = (package_files / MANIFEST_NAME).read_text(encoding="utf-8") - except (ModuleNotFoundError, FileNotFoundError) as e: - raise PluginLoadError(f"{top_package} 包内缺少 {MANIFEST_NAME}") from e + except (OSError, UnicodeDecodeError) as e: + raise PluginLoadError(f"{top_package} 包内无法读取 {MANIFEST_NAME}：{e}") from e manifest = parse_manifest(text, f"{top_package}/{MANIFEST_NAME}") def load() -> type[Plugin[Any]]: diff --git a/backend/src/module/plugin/manager.py b/backend/src/module/plugin/manager.py index 73054a00..ba3e3179 100644 --- a/backend/src/module/plugin/manager.py +++ b/backend/src/module/plugin/manager.py @@ -94,7 +94,10 @@ class PluginManager: self._data_root = data_root self.registry = registry if registry is not None else ExtensionRegistry() self.breaker = CircuitBreaker(on_trip=self._on_trip) - self.bus = EventBus(on_error=self.breaker.record_failure) + self.bus = EventBus( + on_error=self.breaker.record_failure, + on_success=self.breaker.record_success, + ) self.runner = HookRunner( self.registry, self.breaker, @@ -155,7 +158,11 @@ class PluginManager: ): await self._deactivate(plugin_id) for plugin_id, candidate in self._candidates.items(): - if plugin_id in self._active or self._blocked_reason(candidate): + if plugin_id in self._active: + continue + if self._blocked_reason(candidate): + # 已被关闭的插件不再显示旧的失败原因，重新开启时也会重试 + self._failed.pop(plugin_id, None) continue failed = self._failed.get(plugin_id) if failed is not None and failed[0] == self._snapshot(plugin_id): @@ -353,6 +360,7 @@ class PluginManager: manifest = candidate.manifest plugin_id = manifest.id snapshot = self._snapshot(plugin_id) + instance: Plugin | None = None try: if not manifest.sdk_compatible(): raise PluginLoadError( @@ -373,6 +381,9 @@ class PluginManager: ) self.registry.remove_plugin(plugin_id) await self.bus.close_owner(plugin_id) + if instance is not None: + # 释放 setup 失败前已创建的任务、连接等 + await self._teardown(plugin_id, instance) candidate.unload() self._failed[plugin_id] = (snapshot, reason) logger.error("[Plugin:%s] 加载失败：%s", plugin_id, reason) @@ -432,25 +443,31 @@ class PluginManager: return self.registry.remove_plugin(plugin_id) await self.bus.close_owner(plugin_id) - try: - await asyncio.wait_for(active.instance.teardown(), self._setup_timeout) - except Exception as e: - logger.warning("[Plugin:%s] teardown 失败：%s", plugin_id, e) + await self._teardown(plugin_id, active.instance) active.candidate.unload() logger.info("[Plugin:%s] 已停用", plugin_id) + async def _teardown(self, plugin_id: str, instance: Plugin) -> None: + try: + await asyncio.wait_for(instance.teardown(), self._setup_timeout) + except Exception as e: + logger.warning("[Plugin:%s] teardown 失败：%s", plugin_id, e) + def _on_trip(self, plugin_id: str, reason: str) -> None: - task = asyncio.create_task(self._disable(plugin_id, reason)) + # 记下触发熔断的实例：禁用任务拿到锁前插件可能已被重新加载 + tripped = self._active.get(plugin_id) + task = asyncio.create_task(self._disable(plugin_id, reason, tripped)) self._pending.add(task) task.add_done_callback(self._pending.discard) - async def _disable(self, plugin_id: str, reason: str) -> None: + async def _disable( + self, plugin_id: str, reason: str, tripped: _Active | None + ) -> None: async with self._lock: - active = self._active.get(plugin_id) - if active is None: + if tripped is None or self._active.get(plugin_id) is not tripped: return await self._deactivate(plugin_id) - self._failed[plugin_id] = (active.snapshot, reason) + self._failed[plugin_id] = (tripped.snapshot, reason) logger.error("[Plugin:%s] %s", plugin_id, reason) self.bus.publish(PluginDisabled(plugin_id=plugin_id, reason=reason)) await self._notify_change() diff --git a/backend/src/test/test_plugin_runtime.py b/backend/src/test/test_plugin_runtime.py index 33891cdf..f35fb777 100644 --- a/backend/src/test/test_plugin_runtime.py +++ b/backend/src/test/test_plugin_runtime.py @@ -555,6 +555,38 @@ class TestLoader: with pytest.raises(PluginLoadError, match=message): candidates[0].load() + def test_undecodable_manifest_is_reported(self, tmp_path): + local = tmp_path / "local" + (local / "demo").mkdir(parents=True) + (local / "demo" / "plugin.toml").write_bytes(b"\xff\xfe[plugin]\n") + candidates, errors = discover( + builtin_root=tmp_path / "none", + local_root=local, + entry_point_group="ab-test-none", + ) + assert candidates == [] + assert len(errors) == 1 + + def test_entry_point_import_failure_is_reported(self, tmp_path, monkeypatch): + import importlib.metadata + + package = tmp_path / "ab_broken_ep" + package.mkdir() + (package / "__init__.py").write_text("from os import ab_missing_name\n") + monkeypatch.syspath_prepend(str(tmp_path)) + ep = importlib.metadata.EntryPoint( + "broken", "ab_broken_ep.plugin:BrokenPlugin", "ab-test-broken" + ) + monkeypatch.setattr(importlib.metadata, "entry_points", lambda group: [ep]) + candidates, errors = discover( + builtin_root=tmp_path / "none", + local_root=tmp_path / "none", + entry_point_group="ab-test-broken", + ) + assert candidates == [] + [error] = errors + assert "ab_missing_name" in error.error + # ---------------------------------------------------------------- manager @@ -771,18 +803,38 @@ class TestManager: await manager.stop() async def test_failed_setup_rolls_back_registrations(self, tmp_path): + recorder = Recorder() + class Broken(Plugin): @hook("torrent.filter") def check(self, x): return True async def setup(self): + recorder.events.append("setup") raise RuntimeError("cannot connect") + async def teardown(self): + recorder.events.append("teardown") + manager, _ = make_manager(candidate_for(Broken), tmp_path=tmp_path) await manager.start() assert manager.registry.hooks("torrent.filter") == [] assert manager.statuses()[0].error == "cannot connect" + # setup 中途创建的资源由 teardown 释放 + assert recorder.events == ["setup", "teardown"] + + async def test_disabling_failed_plugin_shows_disabled(self, tmp_path): + class Broken(Plugin): + async def setup(self): + raise RuntimeError("cannot connect") + + manager, settings_obj = make_manager(candidate_for(Broken), tmp_path=tmp_path) + await manager.start() + settings_obj.plugins.enabled = {"demo": False} + await manager.apply_settings() + [status] = manager.statuses() + assert (status.state, status.error) == ("disabled", "未启用") async def test_unknown_extension_point_fails_load(self, tmp_path): class Typo(Plugin): @@ -848,6 +900,37 @@ class TestManager: assert len(seen) == 1 await manager.stop() + async def test_subscriber_success_resets_failure_count(self, tmp_path): + class Sometimes(Plugin): + @subscribe("demo.happened") + def on_demo(self, event): + if event.value < 0: + raise RuntimeError("transient") + + manager, _ = make_manager(candidate_for(Sometimes), tmp_path=tmp_path) + await manager.start() + for value in [-1] * 4 + [1] + [-1] * 4: + manager.bus.publish(DemoEvent(value)) + await manager.bus.drain() + await asyncio.gather(*manager._pending) + assert manager.statuses()[0].state == "active" + await manager.stop() + + async def test_stale_breaker_trip_spares_reloaded_instance(self, tmp_path): + recorder = Recorder() + manager, _ = make_manager( + candidate_for(build_plugin(recorder)), tmp_path=tmp_path + ) + await manager.start() + # 熔断任务尚未拿到锁时，插件已被重新加载（安装升级 / dev_mode） + manager._on_trip("demo", "连续失败 5 次") + await manager.reload("demo") + await asyncio.gather(*manager._pending) + [status] = manager.statuses() + assert status.state == "active" + assert recorder.events == ["setup:0", "teardown", "setup:0"] + await manager.stop() + async def test_end_to_end_from_directory(self, tmp_path): write_plugin( tmp_path / "builtin",