nas/Auto_Bangumi
1
0
Fork 0
mirror of https://github.com/EstrellaXD/Auto_Bangumi.git synced 2026-11-03 22:24:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(4.0): 插件运行时的发现、setup 失败、熔断与状态修复

Browse Source 
- 非 UTF-8 的 plugin.toml、pip 插件顶层导入失败记为发现错误，不再使启动失败
- setup 失败或超时后调用 teardown，释放已创建的任务与连接
- 事件订阅成功时清零熔断计数，只按连续失败熔断
- 关闭插件后清除旧的失败记录，状态显示为 disabled
- 熔断禁用只作用于触发熔断的实例，不影响重新加载后的新实例
This commit is contained in:
Estrella Pan
2026-10-08 16:04:47 +02:00
parent 83527c342b
commit 2c36970d91
5 changed files with 132 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/src/ab_sdk/manifest.py
View File

@@ -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))

15
backend/src/module/plugin/bus.py
View File

@@ -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)

8
backend/src/module/plugin/loader.py
View File

@@ -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]]:

39
backend/src/module/plugin/manager.py
View File

@@ -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()

83
backend/src/test/test_plugin_runtime.py
View File

@@ -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",