Destroy MinionManager event resources deterministically

changelog/70175.fixed.md

@@ -0,0 +1 @@
+Fixed the minion logging "unclosed publish server", "unclosed SyncWrapper", and "unclosed publisher client" WARNING messages after a failed master reconnect. ``MinionManager``'s event-bus resources (``event_publisher``/``event``) are now destroyed deterministically instead of relying on ``__del__``'s GC-time safety net.

salt/cli/daemons.py

@@ -352,6 +352,17 @@ def start(self):
             try:
                 self._real_start()
             except SaltClientError as exc:
+                # The minion lost its master connection and tune_in()
+                # returned without ever going through shutdown(). Destroy
+                # the MinionManager's resources (event_publisher, event,
+                # minions) deterministically here -- both before retrying
+                # and before falling through to the final break/return --
+                # instead of leaving them for __del__'s GC-time safety net,
+                # which is what logs the "unclosed publish server"/
+                # "unclosed SyncWrapper"/"unclosed publisher client"
+                # warnings. See #70175.
+                if hasattr(self.minion, "destroy"):
+                    self.minion.destroy()
                 # Restart for multi_master failover when daemonized
                 if self.options.daemon:
                     continue

salt/minion.py

@@ -1553,6 +1553,19 @@ def destroy(self):
                 if hasattr(minion, "destroy"):
                     minion.destroy()
             self.minions = []
+        # Mirror the event_publisher/event teardown ``stop_async`` performs
+        # on graceful (SIGTERM) shutdown. ``destroy`` is what ``__del__``
+        # falls back to, so it must be able to reclaim these deterministically
+        # on its own -- otherwise they're only ever closed by ``__del__``'s
+        # GC-time safety net, which is what logs the "unclosed publish
+        # server"/"unclosed SyncWrapper"/"unclosed publisher client" warnings.
+        # See #70175.
+        if hasattr(self, "event_publisher") and self.event_publisher is not None:
+            self.event_publisher.close()
+            self.event_publisher = None
+        if hasattr(self, "event") and self.event is not None:
+            self.event.destroy()
+            self.event = None
 
     def _create_minion_object(
         self,

tests/pytests/unit/cli/test_daemons.py

@@ -6,6 +6,7 @@
 import multiprocessing
 
 import salt.cli.daemons
+from salt.exceptions import SaltClientError
 from tests.support.mock import MagicMock, patch
 
 log = logging.getLogger(__name__)
@@ -409,3 +410,51 @@ def test_master_prepare_cluster(tmp_path):
     assert (cluster_dir / "minions_denied").exists()
     assert (cluster_dir / "minions_autosign").exists()
     assert (cluster_dir / "minions_rejected").exists()
+
+
+def test_minion_start_destroys_minion_manager_on_lost_master():
+    """
+    ``Minion.start()`` must destroy the MinionManager whenever
+    ``_real_start()`` raises ``SaltClientError`` (i.e. the minion lost its
+    master connection and isn't retrying), instead of returning with
+    ``event_publisher``/``event`` left for ``__del__``'s GC-time safety net
+    to reclaim. See #70175.
+    """
+    minion = salt.cli.daemons.Minion()
+    minion.options = MagicMock(daemon=False)
+    minion.minion = MagicMock()
+
+    with patch("salt.utils.parsers.DaemonMixIn.start", MagicMock()), patch.object(
+        minion, "_real_start", side_effect=SaltClientError("Minion could not connect")
+    ):
+        minion.start()
+
+    minion.minion.destroy.assert_called_once()
+
+
+def test_minion_start_destroys_minion_manager_before_daemon_retry():
+    """
+    Same as above, but for the daemonized multi-master failover retry
+    branch -- the MinionManager must be destroyed before ``continue``
+    re-enters ``_real_start()``, so the next ``tune_in()``/``_bind()``
+    doesn't leak the old ``event_publisher``/``event``. See #70175.
+    """
+    minion = salt.cli.daemons.Minion()
+    minion.options = MagicMock(daemon=True)
+    minion.minion = MagicMock()
+
+    calls = {"count": 0}
+
+    def _real_start_side_effect():
+        calls["count"] += 1
+        if calls["count"] == 1:
+            raise SaltClientError("Minion could not connect")
+        # Second attempt "succeeds" (returns normally).
+
+    with patch("salt.utils.parsers.DaemonMixIn.start", MagicMock()), patch.object(
+        minion, "_real_start", side_effect=_real_start_side_effect
+    ):
+        minion.start()
+
+    assert calls["count"] == 2
+    minion.minion.destroy.assert_called_once()

tests/pytests/unit/test_minion.py

@@ -2441,3 +2441,37 @@ async def _instant_sleep(_):
         # code path; the .destroy() call would try to tear down channels
         # we never created. A best-effort close is enough.
         pass
+
+
+def test_minion_manager_destroy_closes_event_resources(minion_opts):
+    """
+    ``MinionManager.destroy()`` is what ``__del__`` falls back to, so it
+    must reclaim ``event_publisher``/``event`` deterministically on its
+    own -- not just when ``stop_async`` happens to run first. Otherwise
+    they're only ever closed by ``__del__``'s GC-time safety net, which
+    is what logs the "unclosed publish server"/"unclosed SyncWrapper"/
+    "unclosed publisher client" warnings. See #70175.
+    """
+    manager = salt.minion.MinionManager(minion_opts)
+    try:
+        fake_event_publisher = MagicMock()
+        fake_event = MagicMock()
+        manager.event_publisher = fake_event_publisher
+        manager.event = fake_event
+
+        manager.destroy()
+
+        fake_event_publisher.close.assert_called_once()
+        fake_event.destroy.assert_called_once()
+        assert manager.event_publisher is None
+        assert manager.event is None
+
+        # destroy() must be idempotent: calling it again (e.g. once from
+        # the daemon retry path and again from __del__) must not attempt
+        # to close the already-closed resources a second time.
+        manager.destroy()
+        fake_event_publisher.close.assert_called_once()
+        fake_event.destroy.assert_called_once()
+    finally:
+        manager.event_publisher = None
+        manager.event = None