Roll back plugin registration when historic replay raises

changelog/761.bugfix.rst

@@ -0,0 +1,3 @@
+Roll back a plugin's registration if replaying a historic hook or its result
+callback raises. Previously, the plugin and some of its hook implementations
+could remain registered after the failure.

src/pluggy/_manager.py

@@ -238,18 +238,24 @@ def register(self, plugin: _Plugin, name: str | None = None) -> str | None:
                 hookimpls.append((hook_name, hookimpl, verified))
 
         self._name2plugin[plugin_name] = plugin
-        for hook_name, hookimpl, verified in hookimpls:
-            hook = getattr(self.hook, hook_name, None)
-            if hook is None:
-                hook = HookCaller(hook_name, self._hookexec)
-                setattr(self.hook, hook_name, hook)
-            elif not verified and hook.has_spec():
-                # An earlier impl's historic replay added this spec after the
-                # first pass. Failing here leaves the plugin partially
-                # registered, but only user code during replay can get here.
-                self._verify_hook(hook, hookimpl)
-            hook._maybe_apply_history(hookimpl)
-            hook._add_hookimpl(hookimpl)
+        try:
+            for hook_name, hookimpl, verified in hookimpls:
+                hook = getattr(self.hook, hook_name, None)
+                if hook is None:
+                    hook = HookCaller(hook_name, self._hookexec)
+                    setattr(self.hook, hook_name, hook)
+                elif not verified and hook.has_spec():
+                    # An earlier impl's historic replay added this spec after
+                    # the first pass, so validate against it now.
+                    self._verify_hook(hook, hookimpl)
+                hook._maybe_apply_history(hookimpl)
+                hook._add_hookimpl(hookimpl)
+        except BaseException:
+            # Replay runs user code, including result callbacks. Remove the
+            # failed plugin without disturbing other plugins registered by it.
+            if self._name2plugin.get(plugin_name) is plugin:
+                self.unregister(name=plugin_name)
+            raise
         return plugin_name
 
     def parse_hookimpl_opts(self, plugin: _Plugin, name: str) -> HookimplOpts | None:

testing/test_pluginmanager.py

@@ -427,6 +427,114 @@ def late(self, qlwkje):
     with pytest.raises(PluginValidationError, match="qlwkje"):
         pm.register(hello())
 
+    assert pm.list_name_plugin() == []
+    assert pm.hook.configure.get_hookimpls() == []
+    assert pm.hook.late.get_hookimpls() == []
+
+
+@pytest.mark.parametrize("exception_type", [RuntimeError, KeyboardInterrupt])
+@pytest.mark.parametrize("fail_in_callback", [False, True])
+def test_register_rolls_back_failed_historic_replay(
+    pm: PluginManager, exception_type: type[BaseException], fail_in_callback: bool
+) -> None:
+    class Hooks:
+        @hookspec
+        def a_regular(self): ...
+
+        @hookspec(historic=True)
+        def b_historic(self): ...
+
+        @hookspec
+        def c_regular(self): ...
+
+    class ExistingPlugin:
+        @hookimpl
+        def a_regular(self):
+            return "existing"
+
+    error = exception_type("historic replay failed")
+    fail = True
+
+    def callback(result):
+        if fail and fail_in_callback:
+            raise error
+
+    class Plugin:
+        @hookimpl
+        def a_regular(self):
+            return "new"
+
+        @hookimpl
+        def b_historic(self):
+            if fail and not fail_in_callback:
+                raise error
+            return "historic"
+
+        @hookimpl
+        def c_regular(self):
+            return "last"
+
+    pm.add_hookspecs(Hooks)
+    existing = ExistingPlugin()
+    pm.register(existing, name="existing")
+    pm.hook.b_historic.call_historic(result_callback=callback)
+    plugin = Plugin()
+
+    with pytest.raises(exception_type) as excinfo:
+        pm.register(plugin, name="new")
+
+    assert excinfo.value is error
+    assert pm.list_name_plugin() == [("existing", existing)]
+    assert pm.hook.a_regular() == ["existing"]
+    assert pm.hook.b_historic.get_hookimpls() == []
+    assert pm.hook.c_regular() == []
+
+    fail = False
+    assert pm.register(plugin, name="new") == "new"
+    assert pm.hook.a_regular() == ["new", "existing"]
+    assert pm.hook.c_regular() == ["last"]
+
+
+def test_register_keeps_replacement_after_historic_replay_failure(
+    pm: PluginManager,
+) -> None:
+    class Hooks:
+        @hookspec
+        def a_regular(self): ...
+
+        @hookspec(historic=True)
+        def b_historic(self): ...
+
+    class Replacement:
+        @hookimpl
+        def a_regular(self):
+            return "replacement"
+
+    replacement = Replacement()
+    error = RuntimeError("historic replay failed")
+
+    class Plugin:
+        @hookimpl
+        def a_regular(self):
+            return "failed"  # pragma: no cover
+
+        @hookimpl
+        def b_historic(self):
+            pm.unregister(name="plugin")
+            pm.register(replacement, name="plugin")
+            raise error
+
+    pm.add_hookspecs(Hooks)
+    pm.hook.b_historic.call_historic()
+
+    with pytest.raises(RuntimeError) as excinfo:
+        pm.register(Plugin(), name="plugin")
+
+    assert excinfo.value is error
+    assert pm.get_plugin("plugin") is replacement
+    assert pm.hook.a_regular() == ["replacement"]
+    assert pm.hook.b_historic.get_hookimpls() == []
+
 
 def test_register_hookwrapper_not_a_generator_function(he_pm: PluginManager) -> None:
     class hello: