fix(plugin): 修分身恢复的占用判定、空模板配置重建、自动分配探测上限与预留落库回滚 (#6692)

This commit is contained in:
Aqr-K
2026-09-14 16:06:02 +08:00
committed by GitHub
parent bc99355b2a
commit c5d2fd2ef4
4 changed files with 298 additions and 61 deletions

View File

@@ -4,7 +4,7 @@ from __future__ import annotations
import threading
from collections.abc import Callable
from typing import Any, NamedTuple, Optional
from typing import Any, NamedTuple, Optional, Protocol
from pydantic import ValidationError
@@ -12,8 +12,8 @@ from app.schemas.plugin import PluginInstance, PluginRuntimeStatus
# 自动分配的后缀从 2 起算:源插件本体在用户眼里就是第 1 个实例,分身接着往下排
_FIRST_AUTO_SUFFIX = 2
# 探测次数上限:占用判据若因端口故障恒为真,没有上限会让请求线程原地打转并一直扣着占位锁
_MAX_AUTO_SUFFIX_PROBES = 1000
# 在「已登记分身数」之上再多探测的号数,取值见 PluginCloneService._allocate_suffix
_AUTO_SUFFIX_PROBE_SLACK = 1000
def _first_validation_message(error: ValidationError) -> str:
@@ -22,6 +22,24 @@ def _first_validation_message(error: ValidationError) -> str:
return str(details[0].get("msg")) if details else str(error)
class InstanceIdTaken(Protocol):
"""候选实例 ID 的占用判据。"""
def __call__(
self,
instance_id: str,
*,
ignore_instance_row: bool = False,
) -> bool:
"""判断该 ID 是否已被某个插件身份占住。
:param instance_id: 候选实例 ID
:param ignore_instance_row: 是否把「实例表里已有这一行」排除在占用之外。恢复
一个已停用的分身时为真:那一行正是本次要拿回来的东西,它自己不构成冲突
:return: 该 ID 是否已被占用
"""
class _Reservation(NamedTuple):
"""占位阶段已经定下来的事实,供后续加载与回滚判断来历。"""
@@ -29,6 +47,13 @@ class _Reservation(NamedTuple):
restoring: bool
class _ClonePlan(NamedTuple):
"""落库之前定下的分身方案:写哪一行,以及这一行是新建还是恢复。"""
reservation: _Reservation
instance: PluginInstance
class PluginCloneService:
"""创建共享源码的虚拟插件实例并协调失败回滚。"""
@@ -36,8 +61,9 @@ class PluginCloneService:
self,
*,
plugin_class: Callable[[str], Optional[Any]],
instance_id_taken: Callable[[str], bool],
instance_id_taken: InstanceIdTaken,
get_instance: Callable[[str], Optional[PluginInstance]],
instances_for_source: Callable[[str], list[PluginInstance]],
source_plugin_id: Callable[[str], str],
save_instance: Callable[[PluginInstance], Any],
delete_instance: Callable[[str], bool],
@@ -53,6 +79,7 @@ class PluginCloneService:
self._plugin_class = plugin_class
self._instance_id_taken = instance_id_taken
self._get_instance = get_instance
self._instances_for_source = instances_for_source
self._source_plugin_id = source_plugin_id
self._save_instance = save_instance
self._delete_instance = delete_instance
@@ -96,22 +123,43 @@ class PluginCloneService:
if rejection:
return False, rejection
with self._reservation_lock:
reservation, message = self._reserve(
# 实例行落库、备配置与首次加载同属一次创建,共用这一个异常与清理边界:落库留在
# 边界之外时,写到一半失败既不会按约定回报 (False, 原因),也不会触发回滚,一行
# 没有配置、也没有运行态的实例就此留在表里
reservation: Optional[_Reservation] = None
try:
with self._reservation_lock:
plan, message = self._plan(
plugin_id=plugin_id,
suffix=suffix,
name=name,
description=description,
icon=icon,
restore_previous=restore_previous,
)
if plan is None:
return False, message
# 来历要赶在落库之前记下:写入抛异常时,回滚靠它判断该抹掉整行还是只
# 把启用位退回停用
reservation = plan.reservation
self._save_instance(plan.instance)
self._activate(
reservation,
plugin_id=plugin_id,
suffix=suffix,
name=name,
description=description,
icon=icon,
restore_previous=restore_previous,
)
if reservation is None:
return False, message
return self._activate(
reservation,
plugin_id=plugin_id,
restore_previous=restore_previous,
)
action = "恢复" if reservation.restoring else "创建"
self._logger.info(f"插件分身 {reservation.clone_id} {action}成功")
return True, reservation.clone_id
except Exception as error: # noqa: BLE001
# 占位之前失败的话没有任何东西落下来,无需也无从回滚
if reservation is not None:
self._rollback(
reservation.clone_id,
purge_instance=not reservation.restoring,
)
self._logger.error(f"创建插件分身失败:{error}")
return False, f"创建插件分身失败:{error}"
def _reject_invalid_source(self, plugin_id: str) -> str:
"""判断给定插件能否作为分身来源,可以时返回空串。
@@ -132,7 +180,7 @@ class PluginCloneService:
)
return ""
def _reserve(
def _plan(
self,
*,
plugin_id: str,
@@ -141,11 +189,11 @@ class PluginCloneService:
description: str,
icon: Optional[str],
restore_previous: bool,
) -> tuple[Optional[_Reservation], str]:
"""定下分身 ID 并把实例行落库,返回占位结果或拒绝原因。
) -> tuple[Optional[_ClonePlan], str]:
"""定下分身 ID 与要写出的实例行,或给出拒绝原因;本方法自身不写任何东西。
调用方须持有占位锁:本方法内部先读后写,两个并发创建各自读到「未占用」就会
落到同一行上。
调用方须持有占位锁:判存是「先读后写」的前半段,两个并发创建各自读到「未占用」
就会落到同一行上。
"""
resolved_suffix = (suffix or "").strip().lower() or self._allocate_suffix(plugin_id)
if not resolved_suffix:
@@ -159,8 +207,15 @@ class PluginCloneService:
and not previous.is_enabled
and previous.source_plugin_id == plugin_id
)
if not restoring and self._instance_id_taken(clone_id):
return None, f"分身插件 {clone_id} 已存在"
# 恢复只豁免待恢复的那一行自身。停用从不删行,那一行可以在表里躺很久,其间同名
# 的真实插件完全可能出现(装进类注册表、进安装清单、或在磁盘上留下插件包);此时
# 放行等于让分身顶掉一个真实插件的身份,那个插件此后连装都装不回来
if self._instance_id_taken(clone_id, ignore_instance_row=restoring):
return None, (
f"分身插件 {clone_id} 的 ID 已被其他插件占用,无法恢复"
if restoring
else f"分身插件 {clone_id} 已存在"
)
instance, message = self._build_instance(
clone_id=clone_id,
@@ -173,8 +228,10 @@ class PluginCloneService:
)
if instance is None:
return None, message
self._save_instance(instance)
return _Reservation(clone_id=clone_id, restoring=restoring), ""
return _ClonePlan(
reservation=_Reservation(clone_id=clone_id, restoring=restoring),
instance=instance,
), ""
def _allocate_suffix(self, plugin_id: str) -> str:
"""为新分身分配一个最小可用的数字后缀,无号可用时返回空串。
@@ -182,11 +239,14 @@ class PluginCloneService:
占用判据与显式指定后缀走的是同一个 ``instance_id_taken``,自动分配因此不可能
挑中一个手填时会被判成「已存在」的 ID。已停用的分身同样占位:重用它的 ID 是
「恢复」而不是新建,用户没点恢复就不该凭空拿到上一个分身留下的配置。
探测上限随该插件已登记的分身数一起抬高。定成一个常数的话,分身数量越过它之后
每次自动分配都报「后缀已耗尽」,而下一个号明明空着;按鸽巢原理,N 个已登记分身
最多占掉 N 个号,再多探测一段余量必然能撞上空位。上限仍然有限:占用判据若因端口
故障恒为真,无界的循环会让请求线程原地打转并一直扣着占位锁。
"""
for index in range(
_FIRST_AUTO_SUFFIX,
_FIRST_AUTO_SUFFIX + _MAX_AUTO_SUFFIX_PROBES,
):
probe_limit = len(self._instances_for_source(plugin_id)) + _AUTO_SUFFIX_PROBE_SLACK
for index in range(_FIRST_AUTO_SUFFIX, _FIRST_AUTO_SUFFIX + probe_limit):
if not self._instance_id_taken(f"{plugin_id}{index}"):
return str(index)
return ""
@@ -238,31 +298,36 @@ class PluginCloneService:
*,
plugin_id: str,
restore_previous: bool,
) -> tuple[bool, str]:
"""准备配置并完成首次加载,失败时按来历回滚。"""
clone_id = reservation.clone_id
) -> None:
"""准备配置并完成首次加载,失败一律抛出,由 :meth:`clone` 统一回滚。"""
# 恢复留存配置时不得用源插件模板盖掉它,那正是用户要拿回来的东西
keep_previous_config = reservation.restoring and restore_previous
try:
if not keep_previous_config:
original_config = self._read_config(plugin_id)
if original_config:
clone_config = dict(original_config)
clone_config["enable"] = False
clone_config["enabled"] = False
if not self._save_config(clone_id, clone_config):
raise RuntimeError("虚拟实例配置保存失败")
if not (reservation.restoring and restore_previous):
self._rebuild_config(reservation.clone_id, source_plugin_id=plugin_id)
if self._reload_plugin(reservation.clone_id) is PluginRuntimeStatus.LOAD_FAILED:
raise RuntimeError("虚拟实例加载失败")
status = self._reload_plugin(clone_id)
if status is PluginRuntimeStatus.LOAD_FAILED:
raise RuntimeError("虚拟实例加载失败")
action = "恢复" if reservation.restoring else "创建"
self._logger.info(f"插件分身 {clone_id} {action}成功")
return True, clone_id
except Exception as error: # noqa: BLE001
self._rollback(clone_id, purge_instance=not reservation.restoring)
self._logger.error(f"创建插件分身失败:{error}")
return False, f"创建插件分身失败:{error}"
def _rebuild_config(self, clone_id: str, *, source_plugin_id: str) -> None:
"""按源插件当前的配置模板重建分身配置,模板为空时清掉分身原有的那一份。
源插件没有配置与源插件配置为空字典都落在「模板里没有可继承的业务参数」这一档,
两者都必须清除:只是跳过写入的话,已停用分身留存的旧参数会原样活下来并在重载后
继续生效,用户要的「按模板重建一份」于是变成了「沿用旧配置」,且毫无提示。
:param clone_id: 分身实例 ID
:param source_plugin_id: 提供配置模板的源插件 ID
:raise RuntimeError: 配置写入被持久化层拒绝
"""
template = self._read_config(source_plugin_id)
if not template:
self._delete_config(clone_id)
return
clone_config = dict(template)
# 分身是照着源插件的参数建出来的,业务开关一律置假:它还没被用户配过,直接开跑
# 等于拿着别人的参数上线
clone_config["enable"] = False
clone_config["enabled"] = False
if not self._save_config(clone_id, clone_config):
raise RuntimeError("虚拟实例配置保存失败")
def _rollback(self, clone_id: str, *, purge_instance: bool) -> None:
"""逐项清理失败实例,单个清理错误不得阻断其余回滚。

View File

@@ -355,7 +355,11 @@ def build_plugin_runtime(
) or []
return plugin_id in installed
def instance_id_taken(instance_id: str) -> bool:
def instance_id_taken(
instance_id: str,
*,
ignore_instance_row: bool = False,
) -> bool:
"""判断一个候选实例 ID 是否已被占用。
创建分身的判存与自动分配后缀共用这一个判据,自动分配因此不可能挑中一个手填
@@ -374,14 +378,19 @@ def build_plugin_runtime(
报告不存在,因而这里直接看包目录。
:param instance_id: 候选实例 ID
:param ignore_instance_row: 是否把实例行这一类占用者排除在外。恢复一个已停用的
分身时为真:那一行正是本次要拿回来的东西,它自己不构成冲突。一个 ID 在实例
表里至多一行,忽略这一类即等于只豁免待恢复的那一行,另外三类仍然要挡——否则
恢复会绕开全部判存,把一个真实插件的身份顶掉
:return: 该 ID 是否已被占用
"""
if registry.plugin_class(instance_id) is not None:
return True
if instances.get(instance_id) is not None:
return True
if instances.get_host(instance_id) is not None:
return True
if not ignore_instance_row:
if instances.get(instance_id) is not None:
return True
if instances.get_host(instance_id) is not None:
return True
installed = environment.storage().read(
SystemConfigKey.UserInstalledPlugins
) or []
@@ -393,6 +402,7 @@ def build_plugin_runtime(
plugin_class=registry.plugin_class,
instance_id_taken=instance_id_taken,
get_instance=instances.get,
instances_for_source=instances.for_source,
source_plugin_id=source_plugin_id,
save_instance=instances.save,
delete_instance=instances.delete,

View File

@@ -57,6 +57,7 @@ class _World:
self.removed: list[str] = []
self.status = PluginRuntimeStatus.ACTIVE
self.probe_delay = 0.0
self.save_failure: Optional[Exception] = None
self.runtime: Optional[PluginRuntime] = None
def clone(self, **kwargs: Any) -> tuple[bool, str]:
@@ -112,6 +113,12 @@ def _build_world(
world.rows[instance_id] = record.model_copy(update={"is_enabled": is_enabled})
return True
def _save_row(instance: PluginInstance) -> None:
"""写入实例行,可按需让这一步抛异常以模拟持久化故障。"""
if world.save_failure is not None:
raise world.save_failure
world.rows[instance.instance_id] = instance
storage = PluginStorage(
read=values.get,
write=values.__setitem__,
@@ -130,7 +137,7 @@ def _build_world(
for record in world.rows.values()
if record.source_plugin_id == source_plugin_id
],
save=lambda instance: world.rows.__setitem__(instance.instance_id, instance),
save=_save_row,
delete=lambda instance_id: world.rows.pop(instance_id, None) is not None,
list_enabled=lambda: [
record for record in world.rows.values() if record.is_enabled
@@ -280,6 +287,23 @@ def test_auto_allocated_suffix_skips_a_currently_loaded_plugin_class():
assert instance_id == "DemoPlugin3"
def test_auto_allocation_keeps_looking_past_a_long_run_of_taken_numbers():
"""已登记分身多到超过固定探测窗口时,仍要挑出真正最小的可用序号。
探测次数若定死成一个常数,分身数量越过它之后每次自动分配都报「后缀已耗尽」,
可实际上下一个号就空着;用户此时只能改为手填一个号,而手填走的又是同一套判据。
"""
taken = {
f"DemoPlugin{index}": _disabled_clone(f"DemoPlugin{index}", is_enabled=True)
for index in range(2, 1002)
}
world = _build_world(rows=taken)
success, instance_id = world.clone(plugin_id="DemoPlugin")
assert (success, instance_id) == (True, "DemoPlugin1002")
@pytest.mark.parametrize("blank", ["", " ", None])
def test_blank_suffix_falls_back_to_automatic_allocation(blank):
"""后缀留空的几种写法都走自动分配,而不是拼出一个等于源插件 ID 的实例。"""
@@ -436,6 +460,72 @@ def test_restore_previous_false_rebuilds_the_config_from_the_source_template():
assert world.rows["DemoPlugin2"].plugin_name == "全新"
@pytest.mark.parametrize("template", [None, {}])
def test_restore_previous_false_clears_the_old_config_when_the_source_has_none(template):
"""源插件没有配置或配置为空时,按模板重建同样要把旧配置清掉。
两种情形下模板里都没有可继承的业务参数;若只是跳过写入,用户明确要求的「重建一份
全新配置」会变成「原样沿用旧配置」,那些旧参数在重载后继续生效,且毫无提示。
"""
configs: dict[str, dict] = {"DemoPlugin2": {"token": "该被丢弃"}}
if template is not None:
configs["DemoPlugin"] = template
world = _build_world(
rows={"DemoPlugin2": _disabled_clone("DemoPlugin2")},
configs=configs,
)
success, _instance_id = world.clone(
plugin_id="DemoPlugin",
suffix="2",
restore_previous=False,
)
assert success is True
assert "DemoPlugin2" not in world.configs
def _world_with_a_disabled_clone_and_a_foreign_occupant(kind: str, tmp_path: Path) -> _World:
"""建出「停用分身行还在,同一个 ID 又被某个真实插件占住」的世界。
停用从不删行,那一行可以在表里躺很久;这段时间里同名的真实插件完全可能出现——
磁盘上被放进一个同名插件包、安装清单里多出一条、或者它已经装载进类注册表。
"""
rows = {"DemoPlugin2": _disabled_clone("DemoPlugin2", plugin_name="夜间任务")}
configs = {"DemoPlugin2": {"token": "必须留着"}}
if kind == "disk-package":
(tmp_path / "demoplugin2").mkdir()
return _build_world(rows=rows, configs=configs, plugins_root=tmp_path)
if kind == "installed-entry":
return _build_world(
rows=rows,
configs=configs,
installed=["DemoPlugin", "DemoPlugin2"],
)
world = _build_world(rows=rows, configs=configs)
world.runtime.registry.classes["DemoPlugin2"] = DemoPlugin
return world
@pytest.mark.parametrize("kind", ["disk-package", "installed-entry", "loaded-class"])
def test_restoring_still_refuses_an_id_a_real_plugin_already_holds(kind, tmp_path):
"""恢复只豁免待恢复的那一行自身,另外三类占用者仍然要挡。
豁免写成「只要停用行在就全放行」的话,恢复会绕过全部判存:该 ID 上压着的真实
插件身份被分身顶掉,那个插件此后既装不回来,实例行的归属列也对不上。
"""
world = _world_with_a_disabled_clone_and_a_foreign_occupant(kind, tmp_path)
success, message = world.clone(plugin_id="DemoPlugin", suffix="2")
assert success is False
assert "占用" in message
# 被拒绝的恢复不得改动那一行,也不得把它拉起来跑
assert world.rows["DemoPlugin2"].is_enabled is False
assert world.configs["DemoPlugin2"] == {"token": "必须留着"}
assert world.reloaded == []
def test_failed_restore_only_puts_the_row_back_to_disabled():
"""恢复失败只把启用位退回停用,不得连同用户留存的配置一起毁掉。
@@ -471,6 +561,43 @@ def test_failed_fresh_creation_still_removes_the_row_it_created():
assert "DemoPluginwork" not in world.configs
def test_a_failed_row_write_is_reported_and_cleaned_up_like_a_failed_load():
"""实例行落库失败按失败回执返回并清理,不把异常抛给调用方。
落库是本次创建写出的第一样东西,它与备配置、首次加载同属一次创建;把它留在异常
边界之外,写到一半失败时既没有回滚、也没有 ``(False, 原因)``,只有一个异常冒到
上层,而半写的实例行仍留在表里。
"""
world = _build_world(configs={"DemoPlugin": {"token": "源插件的"}})
world.save_failure = RuntimeError("实例表不可写")
success, message = world.clone(plugin_id="DemoPlugin", suffix="Work")
assert success is False
assert "实例表不可写" in message
assert world.rows == {}
assert world.configs == {"DemoPlugin": {"token": "源插件的"}}
assert world.reloaded == []
assert world.removed == ["DemoPluginwork"]
def test_a_failed_row_write_while_restoring_keeps_the_stored_settings():
"""恢复途中落库失败同样只把启用位退回停用,用户留存的配置不受牵连。"""
world = _build_world(
rows={"DemoPlugin2": _disabled_clone("DemoPlugin2", plugin_name="夜间任务")},
configs={"DemoPlugin2": {"token": "必须留着"}},
)
world.save_failure = RuntimeError("实例表不可写")
success, message = world.clone(plugin_id="DemoPlugin", suffix="2")
assert success is False
assert "实例表不可写" in message
assert world.rows["DemoPlugin2"].is_enabled is False
assert world.rows["DemoPlugin2"].plugin_name == "夜间任务"
assert world.configs["DemoPlugin2"] == {"token": "必须留着"}
# --------------------------------------------------------------------------- #
# 校验在提交前拦截
# --------------------------------------------------------------------------- #

View File

@@ -660,8 +660,9 @@ def test_clone_service_persists_descriptor_without_copying_source_package():
service = PluginCloneService(
plugin_class=lambda plugin_id: DemoPlugin if plugin_id == "DemoPlugin" else None,
instance_id_taken=lambda plugin_id: plugin_id in instances,
instance_id_taken=lambda plugin_id, **_kwargs: plugin_id in instances,
get_instance=instances.get,
instances_for_source=lambda _plugin_id: list(instances.values()),
source_plugin_id=lambda plugin_id: plugin_id,
save_instance=lambda instance: instances.__setitem__(
instance.instance_id,
@@ -699,6 +700,39 @@ def test_clone_service_persists_descriptor_without_copying_source_package():
assert removed == []
def test_clone_service_gives_up_allocating_when_no_id_is_ever_reported_free():
"""占用判据恒为真时自动分配必须有限次放弃,不能扣着占位锁原地打转。
探测上限随已登记分身数抬高是为了不把仍可用的号误报成耗尽;上限本身不能因此取消,
判据一旦因端口故障恒为真,无界的循环会让请求线程一直占着创建入口不放。
"""
class DemoPlugin:
"""提供探测上限用例需要的最小源插件类。"""
service = PluginCloneService(
plugin_class=lambda _plugin_id: DemoPlugin,
instance_id_taken=lambda _plugin_id, **_kwargs: True,
get_instance=lambda _plugin_id: None,
instances_for_source=lambda _plugin_id: [],
source_plugin_id=lambda plugin_id: plugin_id,
save_instance=lambda _instance: None,
delete_instance=lambda _plugin_id: False,
disable_instance=lambda _plugin_id: False,
read_config=lambda _plugin_id: {},
save_config=lambda _plugin_id, _config: True,
delete_config=lambda _plugin_id: False,
reload_plugin=lambda _plugin_id: PluginRuntimeStatus.ACTIVE,
remove_plugin=lambda _plugin_id: None,
log=_logger(),
)
success, message = service.clone(plugin_id="DemoPlugin", name="", description="")
assert success is False
assert "耗尽" in message
def test_clone_service_rolls_back_descriptor_and_config_after_load_failure():
"""实例首次加载失败时不留下不可见描述或孤立配置。"""
instances: dict[str, PluginInstance] = {}
@@ -710,8 +744,9 @@ def test_clone_service_rolls_back_descriptor_and_config_after_load_failure():
service = PluginCloneService(
plugin_class=lambda _plugin_id: DemoPlugin,
instance_id_taken=lambda plugin_id: plugin_id in instances,
instance_id_taken=lambda plugin_id, **_kwargs: plugin_id in instances,
get_instance=instances.get,
instances_for_source=lambda _plugin_id: list(instances.values()),
source_plugin_id=lambda plugin_id: plugin_id,
save_instance=lambda instance: instances.__setitem__(
instance.instance_id,