[3008.x] Fix minion PublishServer / _TCPPubServerPublisher unclosed leaks (#70175)

changelog/70175.fixed.md

@@ -0,0 +1 @@
+Fixed leak of the minion's local ``PublishServer`` graph (``event_publisher`` -> ``pub_sock`` SyncWrapper -> ``_TCPPubServerPublisher``) when the minion exits through ``cli.daemons.Minion.shutdown`` (KeyboardInterrupt, SaltSystemExit, early-exit guards) or ``MinionManager`` GC. ``MinionManager.destroy`` now closes ``event_publisher`` and destroys ``event`` -- previously only the SIGTERM ``stop_async`` path did, so non-SIGTERM shutdown paths triggered the three-warning cascade in issue #70175. ``PublishServer.close`` also now calls ``pub.close()`` on every ``_TCPPubServerPublisher`` cached in ``_async_pub_by_loop`` (previously it closed the underlying stream only, leaving ``_closing = False`` and letting the publisher's ``__del__`` emit the "unclosed publisher client" warning at GC).

salt/minion.py

@@ -1553,6 +1553,33 @@ def destroy(self):
                 if hasattr(minion, "destroy"):
                     minion.destroy()
             self.minions = []
+        # Close the local event publisher and event bus.  ``stop_async``
+        # (invoked from the SIGTERM signal handler) already does this,
+        # but ``destroy`` is *also* reached from
+        # ``cli.daemons.Minion.shutdown`` (KeyboardInterrupt / SaltSystemExit
+        # / early-exit ``shutdown(1)`` guards) and from ``__del__`` on GC.
+        # Without this the ``PublishServer`` graph created in ``_bind``
+        # (``event_publisher`` -> ``pub_sock`` SyncWrapper ->
+        # ``_TCPPubServerPublisher``) leaks at process exit, surfacing as
+        # the three-warning cascade in issue #70175.
+        if getattr(self, "event_publisher", None) is not None:
+            try:
+                self.event_publisher.close()
+            except Exception:  # pylint: disable=broad-except
+                log.debug(
+                    "Error closing event_publisher during MinionManager.destroy",
+                    exc_info=True,
+                )
+            self.event_publisher = None
+        if getattr(self, "event", None) is not None:
+            try:
+                self.event.destroy()
+            except Exception:  # pylint: disable=broad-except
+                log.debug(
+                    "Error destroying event during MinionManager.destroy",
+                    exc_info=True,
+                )
+            self.event = None
 
     def _create_minion_object(
         self,

salt/transport/tcp.py

@@ -2310,15 +2310,26 @@ def close(self):
         # minion's local event bus (~450 leaked pull.ipc client FDs
         # under sustained stress -> ulimit trip).  Close every cached
         # publisher we still hold before dropping the map.
+        #
+        # PATCH (#70175 round 2): call ``pub.close()`` rather than reaching
+        # into ``pub.stream`` directly.  The stream-only close released
+        # the socket FD but never flipped ``pub._closing = True``, so
+        # every cached publisher tripped ``_TCPPubServerPublisher.__del__``
+        # at GC and emitted the "unclosed publisher client"
+        # ``ResourceWarning`` -- the third warning of the cascade the
+        # user reported on 3008.2+506 (round 1 closed the outer
+        # ``PublishServer`` + ``pub_sock`` SyncWrapper via
+        # ``MinionManager.destroy``; the raw cached publishers were
+        # still leaking their own warning).  ``_TCPPubServerPublisher.close``
+        # is idempotent (early-return on ``_closing``) and subsumes the
+        # stream close.
         per_loop = getattr(self, "_async_pub_by_loop", None)
         if per_loop is not None:
             for pub, _lock in list(per_loop.values()):
-                stream = getattr(pub, "stream", None)
-                if stream is not None and not stream.closed():
-                    try:
-                        stream.close()
-                    except Exception:  # pylint: disable=broad-except
-                        pass
+                try:
+                    pub.close()
+                except Exception:  # pylint: disable=broad-except
+                    pass
             try:
                 per_loop.clear()
             except Exception:  # pylint: disable=broad-except

tests/pytests/unit/test_minion.py

@@ -2045,6 +2045,56 @@ async def test_minion_manager_async_stop(io_loop, minion_opts, tmp_path):
     assert mm.event is None
 
 
+async def test_minion_manager_destroy_closes_event_publisher(
+    io_loop, minion_opts, tmp_path
+):
+    """
+    Regression test for issue #70175.
+
+    ``MinionManager.destroy()`` is invoked from
+    ``cli.daemons.Minion.shutdown()`` (KeyboardInterrupt, SaltSystemExit,
+    the ``shutdown(1)`` guard in ``prepare()``) and from
+    ``MinionManager.__del__`` on GC.  It must close the ``event_publisher``
+    ``PublishServer`` graph -- otherwise the three-warning cascade
+    from #70175 fires at interpreter shutdown:
+
+      - ``unclosed publish server <PublishServer>``
+      - ``unclosed SyncWrapper for cls=<_TCPPubServerPublisher>``
+      - ``unclosed publisher client <_TCPPubServerPublisher>``
+
+    Only the ``stop_async`` shutdown path (invoked from the SIGTERM
+    signal handler) used to close these; ``destroy()`` did not, so any
+    non-SIGTERM exit leaked them.
+    """
+    minion_opts["sock_dir"] = str(tmp_path / "sock")
+    os.makedirs(minion_opts["sock_dir"])
+
+    mm = salt.minion.MinionManager(minion_opts)
+    mm._bind()
+    assert mm.event_publisher is not None
+    assert mm.event is not None
+
+    # Wait for pub server to bind so the underlying PublishServer graph
+    # is fully constructed.
+    while not list(pathlib.Path(minion_opts["sock_dir"]).glob("*")):
+        await tornado.gen.sleep(0.1)
+
+    ep = mm.event_publisher
+    ev = mm.event
+
+    # Call destroy directly (the buggy path).  Post-fix it must close
+    # both resources and null the references.
+    mm.destroy()
+
+    assert mm.event_publisher is None
+    assert mm.event is None
+    # PublishServer.close() sets _closing=True so __del__ won't warn.
+    assert ep._closing is True
+    # SaltEvent.destroy() closes pusher / subscriber and clears them.
+    assert ev.subscriber is None
+    assert ev.pusher is None
+
+
 def test_minion_io_loop_is_asyncio_loop(minion_opts):
     """
     Test that Minion io_loop is converted to asyncio.AbstractEventLoop.

tests/pytests/unit/transport/test_tcp.py

@@ -1218,6 +1218,89 @@ def _new_publisher(*args, **kwargs):
         server.close()
 
 
+async def test_publish_server_close_closes_cached_publishers(master_opts):
+    """
+    ``PublishServer.close()`` must call ``pub.close()`` on every publisher
+    cached in ``_async_pub_by_loop`` -- not just close the underlying
+    stream -- so ``_TCPPubServerPublisher._closing`` gets flipped to
+    ``True`` and the object's ``__del__`` does not emit the
+    "unclosed publisher client" ``ResourceWarning``.
+
+    Regression guard for issue #70175 round 2.  Pre-fix,
+    ``PublishServer.close`` did ``stream.close()`` directly on each
+    cached publisher, which released the socket FD (round-1 Bug 1 fix)
+    but left ``_closing = False`` on the publisher object.  When GC
+    reaped the cached publisher, its finalizer emitted the third
+    warning of the three-warning cascade the user reported on
+    3008.2+506.  Round-1 PR #70206 closed the outer PublishServer +
+    pub_sock SyncWrapper via MinionManager.destroy (silences warnings
+    1 and 2); round-2 must call ``pub.close()`` here (silences warning
+    3).
+    """
+    opts = dict(master_opts)
+
+    server = salt.transport.tcp.PublishServer(
+        opts,
+        pub_host="127.0.0.1",
+        pub_port=5151,
+        pull_host="127.0.0.1",
+        pull_port=5152,
+    )
+
+    # Populate the per-loop cache with two real ``_TCPPubServerPublisher``
+    # objects (no connect() -- we only need instances whose
+    # ``__del__`` will fire if ``close()`` is not called on them).
+    pub_a = salt.transport.tcp._TCPPubServerPublisher("127.0.0.1", 5152, None)
+    pub_b = salt.transport.tcp._TCPPubServerPublisher("127.0.0.1", 5152, None)
+    loop = asyncio.get_running_loop()
+    server._async_pub_by_loop = weakref.WeakKeyDictionary()
+    # Two distinct dummy loop keys so both cache slots are exercised.
+    key_a = asyncio.new_event_loop()
+    key_b = asyncio.new_event_loop()
+    try:
+        server._async_pub_by_loop[key_a] = (pub_a, asyncio.Lock())
+        server._async_pub_by_loop[key_b] = (pub_b, asyncio.Lock())
+
+        assert pub_a._closing is False
+        assert pub_b._closing is False
+
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            server.close()
+
+            # After close, every cached publisher must have been
+            # ``close()``-d (i.e. ``_closing`` flipped True) and the
+            # cache map dropped.
+            assert pub_a._closing is True
+            assert pub_b._closing is True
+            assert server._async_pub_by_loop is None
+
+            # Drop remaining strong refs and force GC to run
+            # ``_TCPPubServerPublisher.__del__`` for both cached pubs.
+            # With the fix in place their ``__del__`` sees
+            # ``_closing = True`` and returns silently -- no
+            # ``ResourceWarning`` emitted.
+            del pub_a
+            del pub_b
+            gc.collect()
+
+        unclosed_publisher_warnings = [
+            w
+            for w in caught
+            if issubclass(w.category, ResourceWarning)
+            and "unclosed publisher client" in str(w.message)
+        ]
+        assert unclosed_publisher_warnings == [], (
+            "PublishServer.close did not close the cached "
+            "_TCPPubServerPublisher instances -- their __del__ still "
+            "emits unclosed publisher client warnings.  This is the "
+            "third warning of the #70175 cascade."
+        )
+    finally:
+        key_a.close()
+        key_b.close()
+
+
 async def test_pub_server_paths_no_perms(master_opts, io_loop):
     def publish_payload(payload):
         return payload