From 39b1806ba40af0c4d1a53a2b608a55272846f20e Mon Sep 17 00:00:00 2001 From: "Daniel A. Wozniak" Date: Tue, 1 Sep 2026 15:55:43 -0700 Subject: [PATCH 1/2] Fix unclosed PublishServer / _TCPPubServerPublisher leak on minion shutdown MinionManager.destroy() (invoked from cli.daemons.Minion.shutdown on KeyboardInterrupt / SaltSystemExit / early-exit and from __del__ on GC) was missing the close/destroy chain for the local event_publisher PublishServer graph and the event SaltEvent that MinionManager._bind creates. Only the SIGTERM stop_async path closed them, so any non-SIGTERM shutdown leaked the graph and surfaced the three-warning cascade reported in #70175 (unclosed publish server / SyncWrapper / publisher client). --- changelog/70175.fixed.md | 1 + salt/minion.py | 27 +++++++++++++++++ tests/pytests/unit/test_minion.py | 50 +++++++++++++++++++++++++++++++ 3 files changed, 78 insertions(+) create mode 100644 changelog/70175.fixed.md diff --git a/changelog/70175.fixed.md b/changelog/70175.fixed.md new file mode 100644 index 000000000000..4a531142e036 --- /dev/null +++ b/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. diff --git a/salt/minion.py b/salt/minion.py index 8292806508a9..1951fbe79e2b 100644 --- a/salt/minion.py +++ b/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, diff --git a/tests/pytests/unit/test_minion.py b/tests/pytests/unit/test_minion.py index de28616b4165..f87932ad1c60 100644 --- a/tests/pytests/unit/test_minion.py +++ b/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 `` + - ``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. From 668cfcd5bc14202d4937cc309d76993156f20be0 Mon Sep 17 00:00:00 2001 From: "Daniel A. Wozniak" Date: Tue, 8 Sep 2026 14:08:25 -0700 Subject: [PATCH 2/2] Close cached publishers in PublishServer._async_pub_by_loop on close (#70175) PublishServer.close iterated the per-loop publisher cache but called stream.close() on each cached _TCPPubServerPublisher rather than pub.close(). The stream FD was released (Bug 1 fix) but pub._closing stayed False, so _TCPPubServerPublisher.__del__ still fired the "unclosed publisher client" ResourceWarning at GC -- the third warning of the cascade twangboy reported on 3008.2+506. Round 1 of this PR closed the outer PublishServer + pub_sock SyncWrapper via MinionManager.destroy (silences warnings 1 and 2); this round covers the raw cached publishers created in the async-context bypass at tcp.py:2242/:2281. _TCPPubServerPublisher.close is idempotent and subsumes the previous stream-only close. --- changelog/70175.fixed.md | 2 +- salt/transport/tcp.py | 23 +++++-- tests/pytests/unit/transport/test_tcp.py | 83 ++++++++++++++++++++++++ 3 files changed, 101 insertions(+), 7 deletions(-) diff --git a/changelog/70175.fixed.md b/changelog/70175.fixed.md index 4a531142e036..ebf8b2962e79 100644 --- a/changelog/70175.fixed.md +++ b/changelog/70175.fixed.md @@ -1 +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. +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). diff --git a/salt/transport/tcp.py b/salt/transport/tcp.py index 6853270efc81..f343f4f36aee 100644 --- a/salt/transport/tcp.py +++ b/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 diff --git a/tests/pytests/unit/transport/test_tcp.py b/tests/pytests/unit/transport/test_tcp.py index 34432a08d657..e72c2f8f5bfe 100644 --- a/tests/pytests/unit/transport/test_tcp.py +++ b/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