fix(plugin): 落库失败的清理留在占位锁内，不再误删他人的实例 (#6695)

app/runtime/extensions/plugin/clone.py

@@ -139,10 +139,10 @@ class PluginCloneService:
                 )
                 if plan is None:
                     return False, message
-                # 来历要赶在落库之前记下：写入抛异常时，回滚靠它判断该抹掉整行还是只
-                # 把启用位退回停用
+                self._claim(plan)
+                # 记下来历的时机就是占位成功的时机：此后这一行确定归本次所有，锁外的
+                # 清理不会误伤别人。落库失败的清理由 _claim 在锁内自己做完
                 reservation = plan.reservation
-                self._save_instance(plan.instance)
             self._activate(
                 reservation,
                 plugin_id=plugin_id,
@@ -152,7 +152,7 @@ class PluginCloneService:
             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,
@@ -233,6 +233,26 @@ class PluginCloneService:
             instance=instance,
         ), ""
 
+    def _claim(self, plan: _ClonePlan) -> None:
+        """把实例行写出去占住这个 ID，写入失败时就地清理并把异常抛出。
+
+        调用方须持有占位锁，清理也必须在锁内做完：锁一放开，另一个同后缀请求立刻就能
+        把这个 ID 占下来并成功落库，而按 ID 清理分不出那一行是谁写的，删掉的会是它的
+        行、配置和运行态。占位成功之后则不必再持锁——实例行此时已经挡住同 ID 的后续
+        请求（判存认这一行），这个 ID 确定归本次所有，直到本次自己把它清掉。
+
+        :param plan: 本次要写出的实例行与它的来历
+        :raise Exception: 原样抛出持久化层的写入错误，交由 :meth:`clone` 统一回报
+        """
+        try:
+            self._save_instance(plan.instance)
+        except Exception:
+            self._rollback(
+                plan.reservation.clone_id,
+                purge_instance=not plan.reservation.restoring,
+            )
+            raise
+
     def _allocate_suffix(self, plugin_id: str) -> str:
         """为新分身分配一个最小可用的数字后缀，无号可用时返回空串。
 

tests/test_plugin_clone_experience.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import threading
 import time
+from collections.abc import Callable
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any, Optional
@@ -58,6 +59,7 @@ class _World:
         self.status = PluginRuntimeStatus.ACTIVE
         self.probe_delay = 0.0
         self.save_failure: Optional[Exception] = None
+        self.on_remove: Optional[Callable[[], None]] = None
         self.runtime: Optional[PluginRuntime] = None
 
     def clone(self, **kwargs: Any) -> tuple[bool, str]:
@@ -114,11 +116,22 @@ def _build_world(
         return True
 
     def _save_row(instance: PluginInstance) -> None:
-        """写入实例行，可按需让这一步抛异常以模拟持久化故障。"""
-        if world.save_failure is not None:
-            raise world.save_failure
+        """写入实例行；预置的故障只对下一次写入生效。
+
+        一次性是为了并发用例：同一个世界里只让先到的那个请求落库失败，后到的请求
+        必须能正常写进去，才谈得上验证前者的清理有没有越权删掉后者的行。
+        """
+        failure, world.save_failure = world.save_failure, None
+        if failure is not None:
+            raise failure
         world.rows[instance.instance_id] = instance
 
+    def _remove_runtime(instance_id: str) -> None:
+        """摘掉运行态，并在回滚的第一步给用例一个确定的交错点。"""
+        world.removed.append(instance_id)
+        if world.on_remove is not None:
+            world.on_remove()
+
     storage = PluginStorage(
         read=values.get,
         write=values.__setitem__,
@@ -169,7 +182,7 @@ def _build_world(
     runtime = build_plugin_runtime(
         SimpleNamespace(
             reload_plugin=_reload,
-            remove_plugin=world.removed.append,
+            remove_plugin=_remove_runtime,
             get_plugin_remote_entry=lambda _plugin_id, _page: "",
             _run_file_watcher=lambda: None,
             get_plugins_from_market=lambda *_args, **_kwargs: None,
@@ -598,6 +611,61 @@ def test_a_failed_row_write_while_restoring_keeps_the_stored_settings():
     assert world.configs["DemoPlugin2"] == {"token": "必须留着"}
 
 
+def test_a_failed_row_write_never_rolls_back_another_requests_instance():
+    """落库失败的清理只许清掉自己这次的预留，不得抹掉另一个请求刚建好的同 ID 实例。
+
+    清理一旦跑在占位锁之外，判据就只剩「ID 相同」：先到的请求落库失败、锁随异常释放，
+    后到的同后缀请求立刻把这个 ID 占下来并成功落库，随后前者按同一个 ID 执行回滚，
+    删掉的是后者的实例行、配置与运行态。
+    """
+    world = _build_world(configs={"DemoPlugin": {"token": "源插件的"}})
+    # 只让先到的请求落库失败，后到的请求必须能正常建成
+    world.save_failure = RuntimeError("实例表不可写")
+    rollback_started = threading.Event()
+    later_request_done = threading.Event()
+    outcomes: dict[str, tuple[bool, str]] = {}
+
+    def _hold_until_the_later_request_lands() -> None:
+        """卡在先到请求的回滚第一步，直到后到的请求把自己的行写完。
+
+        交错由事件定序，不靠抢跑：清理若在锁外，后者此刻必然能落库，这一等确定等得到；
+        清理留在锁内，后者进不来，这一等走超时返回，两种实现下的顺序都是确定的。
+        """
+        rollback_started.set()
+        later_request_done.wait(timeout=1)
+
+    world.on_remove = _hold_until_the_later_request_lands
+
+    def _first() -> None:
+        """先到的请求：落库失败，随后执行清理。"""
+        outcomes["first"] = world.clone(plugin_id="DemoPlugin", suffix="2")
+
+    def _later() -> None:
+        """后到的请求：同一个后缀，落库成功。"""
+        outcomes["later"] = world.clone(plugin_id="DemoPlugin", suffix="2")
+        later_request_done.set()
+
+    first = threading.Thread(target=_first)
+    first.start()
+    assert rollback_started.wait(timeout=10)
+    later = threading.Thread(target=_later)
+    later.start()
+    for thread in (later, first):
+        thread.join(timeout=20)
+
+    assert outcomes["first"][0] is False
+    assert "实例表不可写" in outcomes["first"][1]
+    assert outcomes["later"] == (True, "DemoPlugin2")
+    # 后到请求建出来的这一行不是前者的产物，前者的清理无权碰它
+    assert "DemoPlugin2" in world.rows
+    assert world.rows["DemoPlugin2"].is_enabled is True
+    assert world.configs["DemoPlugin2"] == {
+        "enable": False,
+        "enabled": False,
+        "token": "源插件的",
+    }
+
+
 # --------------------------------------------------------------------------- #
 # 校验在提交前拦截
 # --------------------------------------------------------------------------- #