Restore __del__ safety-net cleanup on SyncWrapper + PublishServer + _TCPPubServerPublisher (#70175)

Extends the "warn + fall back to close()" pattern from commit 9955b89
(``salt.utils.event.SaltEvent.__del__``) to the three sub-classes ``SaltEvent``
composes with:

* ``salt.utils.asynchronous.SyncWrapper.__del__``
* ``salt.transport.tcp.PublishServer.__del__``
* ``salt.transport.tcp._TCPPubServerPublisher.__del__``

Each ``__del__`` still emits the ``ResourceWarning`` via
``salt.utils.resource_warnings.warn_until_close`` (so leaky callers keep
surfacing for pre-Potassium tracking), then falls back to ``close()``
wrapped in try/except so a finalizer never propagates.  For
``PublishServer.close`` the individual sub-resource close steps
(``pub_sock``, ``pub_server``, ``pull_sock``, ``io_loop.stop``,
``io_loop.close``) are additionally guarded, because they can raise
during GC-time execution when the io_loop is in a partially torn-down
state -- which is exactly the failure mode driving ~50 MB/hr RSS
growth on the minion under sustained event traffic.

Companion to sibling branches ``dwoz/fix/70175-pubserver-perjob-leak``,
``dwoz/fix/70175-saltevent-caller-close``,
``dwoz/fix/70175-receive-path-saltevent`` and to shutdown-path fix #70206.
The ``master`` (Potassium) branch drops the ``close()`` fallback and
requires explicit ``close()`` / context-manager use; the loud
``ResourceWarning`` here is the migration signal for that change.

Regression tests:

* tests/pytests/unit/utils/test_asynchronous.py::test_syncwrapper_del_safety_net_calls_close_70175
* tests/pytests/unit/transport/test_tcp.py::test_publish_server_del_safety_net_calls_close_70175
* tests/pytests/unit/transport/test_tcp.py::test_tcppubserverpublisher_del_safety_net_calls_close_70175

Each test:
  1. Instantiates the class without ``with``, drops the reference, forces GC
  2. Asserts the ``ResourceWarning`` still fires (behavior preserved)
  3. Asserts a class-appropriate observable that only ``close()`` would
     set (asyncio_loop.is_closed() for SyncWrapper; sub-resource
     ``close()`` mock calls for PublishServer; stream+socket close for
     _TCPPubServerPublisher)

All three tests fail on pre-patch origin/3008.x with only the
``ResourceWarning`` firing, and pass with the safety-net restored.

Refs #70175, #70206.

Author: dwoz

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. ``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). ``PubServer._discard_on_close`` now cancels the pending ``_stream_read`` Task and force-closes the Subscriber so the coroutine frame (with its 1 MiB msgpack Unpacker buffer) is released the moment a subscriber disconnects, closing the per-job leak path on the steady-state per-job path. Fire-and-forget ``MinionEvent`` callers in ``salt.modules.event.fire_master`` and ``salt.modules.mine._mine_send`` are now wrapped in ``with`` blocks so the ``PublishServer`` / ``_TCPPubServerPublisher`` / ``SyncWrapper`` triad is torn down synchronously instead of leaking one bundle per invocation onto the minion event bus. ``salt.utils.error.fire_exception`` is also wrapped in a ``with`` block so the temporary ``SaltEvent`` it constructs closes both pusher and subscriber synchronously -- previously the helper (called from ``salt/minion.py:_thread_return`` job-exception path and from ``salt/metaproxy/{proxy,deltaproxy}.py``) dropped the ``SaltEvent`` reference immediately after ``fire_event``, so cleanup ran only when GC invoked ``SaltEvent.__del__`` and each finalization emitted the three-warning triad from the underlying transport chain.
+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). ``PubServer._discard_on_close`` now cancels the pending ``_stream_read`` Task and force-closes the Subscriber so the coroutine frame (with its 1 MiB msgpack Unpacker buffer) is released the moment a subscriber disconnects, closing the per-job leak path on the steady-state per-job path. Fire-and-forget ``MinionEvent`` callers in ``salt.modules.event.fire_master`` and ``salt.modules.mine._mine_send`` are now wrapped in ``with`` blocks so the ``PublishServer`` / ``_TCPPubServerPublisher`` / ``SyncWrapper`` triad is torn down synchronously instead of leaking one bundle per invocation onto the minion event bus. ``salt.utils.error.fire_exception`` is also wrapped in a ``with`` block so the temporary ``SaltEvent`` it constructs closes both pusher and subscriber synchronously -- previously the helper (called from ``salt/minion.py:_thread_return`` job-exception path and from ``salt/metaproxy/{proxy,deltaproxy}.py``) dropped the ``SaltEvent`` reference immediately after ``fire_event``, so cleanup ran only when GC invoked ``SaltEvent.__del__`` and each finalization emitted the three-warning triad from the underlying transport chain. ``salt.utils.asynchronous.SyncWrapper.__del__``, ``salt.transport.tcp.PublishServer.__del__``, and ``salt.transport.tcp._TCPPubServerPublisher.__del__`` also now fall back to ``close()`` as a GC-time safety net -- mirroring the pattern on ``salt.utils.event.SaltEvent.__del__`` -- while still emitting the ``ResourceWarning`` so leaky callers can be surfaced for tracking pre-Potassium.

salt/transport/tcp.py

@@ -2341,13 +2341,18 @@ async def publish(
     def close(self):
         self._closing = True
         if self.pub_sock:
-            # pub_sock is a SyncWrapper - need to call close() on the wrapper itself
+            # pub_sock is a SyncWrapper - need to call close() on the wrapper itself.
+            # Guarded because ``__del__`` may drive this during GC when the
+            # SyncWrapper's io_loop / asyncio_loop is in a torn-down state.
             import salt.utils.asynchronous
 
-            if isinstance(self.pub_sock, salt.utils.asynchronous.SyncWrapper):
-                salt.utils.asynchronous.SyncWrapper.close(self.pub_sock)
-            else:
-                self.pub_sock.close()
+            try:
+                if isinstance(self.pub_sock, salt.utils.asynchronous.SyncWrapper):
+                    salt.utils.asynchronous.SyncWrapper.close(self.pub_sock)
+                else:
+                    self.pub_sock.close()
+            except Exception:  # pylint: disable=broad-except
+                pass
             self.pub_sock = None
         # PATCH: Bug 1's async-context bypass caches a raw
         # ``_TCPPubServerPublisher`` per running loop in
@@ -2383,14 +2388,32 @@ def close(self):
                 pass
             self._async_pub_by_loop = None
         if self.pub_server:
-            self.pub_server.close()
+            try:
+                self.pub_server.close()
+            except Exception:  # pylint: disable=broad-except
+                pass
             self.pub_server = None
         if self.pull_sock:
-            self.pull_sock.close()
+            try:
+                self.pull_sock.close()
+            except Exception:  # pylint: disable=broad-except
+                pass
             self.pull_sock = None
         if self.io_loop:
-            self.io_loop.stop()
-            self.io_loop.close(all_fds=True)
+            # Each io_loop step can raise during GC-time close when the
+            # loop has already been half-torn-down (e.g. by a
+            # ``__del__``-driven cleanup on a partially freed C
+            # extension).  Guard each step individually so a failure in
+            # ``stop()`` does not leak the underlying fds that
+            # ``close(all_fds=True)`` would otherwise release.
+            try:
+                self.io_loop.stop()
+            except Exception:  # pylint: disable=broad-except
+                pass
+            try:
+                self.io_loop.close(all_fds=True)
+            except Exception:  # pylint: disable=broad-except
+                pass
             self.io_loop = None
         # Drop the multiprocessing.Event reference so its internal pipe FDs can be
         # released when no other references remain.
@@ -2399,10 +2422,45 @@ def close(self):
 
     # pylint: disable=W1701
     def __del__(self):
-        if not self._closing:
-            salt.utils.resource_warnings.warn_until_close(
-                f"unclosed publish server {self!r}", source=self, log=log
-            )
+        # On this LTS branch ``__del__`` both surfaces the leak via
+        # ``warn_until_close`` (loud WARNING-level log record and
+        # ``ResourceWarning``) AND falls back to calling ``close()`` as
+        # a safety net, so callers that historically relied on GC-time
+        # cleanup (typically ``MinionManager`` teardown paths that
+        # bypass explicit destroy) do not silently leak the pub/pull
+        # sockets, the per-loop cached ``_TCPPubServerPublisher`` map,
+        # the pub_server ``IOStream``, and the io_loop backing them.
+        #
+        # The companion change on ``master`` (Potassium) drops the
+        # ``close()`` fallback and requires callers to use a context
+        # manager or explicit ``close()``; the loud warning here is the
+        # migration signal for that change.
+        #
+        # Python's ``__del__`` runs during GC (may be delayed, may skip
+        # on reference cycles) and during interpreter shutdown (when
+        # the world is already tearing down and closing an
+        # ``IOStream`` or an io_loop can raise from a partially-freed C
+        # extension).  The ``close()`` call chain below is guarded so a
+        # finalizer never propagates an exception.
+        try:
+            already_closed = getattr(self, "_closing", True)
+        except Exception:  # pylint: disable=broad-except
+            return
+        if already_closed:
+            return
+        salt.utils.resource_warnings.warn_until_close(
+            f"unclosed publish server {self!r}", source=self, log=log
+        )
+        try:
+            self.close()
+        except Exception:  # pylint: disable=broad-except
+            # Finalizer must never raise.  ``close()`` walks pub_sock,
+            # ``_async_pub_by_loop`` cached publishers, pub_server,
+            # pull_sock, io_loop -- each step is guarded individually
+            # inside ``close()``.  This outer handler catches any
+            # residual failure from a partially-freed C extension
+            # during interpreter shutdown.
+            pass
 
     # pylint: enable=W1701
 
@@ -2591,10 +2649,49 @@ def close(self):
 
     # pylint: disable=W1701
     def __del__(self):
-        if not self._closing:
-            salt.utils.resource_warnings.warn_until_close(
-                f"unclosed publisher client {self!r}", source=self, log=log
-            )
+        # On this LTS branch ``__del__`` both surfaces the leak via
+        # ``warn_until_close`` (loud WARNING-level log record and
+        # ``ResourceWarning``) AND falls back to calling ``close()`` as
+        # a safety net, so callers that historically relied on GC-time
+        # cleanup do not silently leak the underlying ``IOStream``
+        # socket FD and the in-flight ``_connecting_future``.
+        #
+        # Motivation: raw ``_TCPPubServerPublisher`` instances get
+        # cached per-loop in ``PublishServer._async_pub_by_loop`` on
+        # the minion; a botched outer teardown that never calls
+        # ``PublishServer.close()`` (see the sibling fixes referenced
+        # from issue #70175) leaves these publishers unclosed and one
+        # ``pull.ipc`` client FD leaks per instance.
+        #
+        # The companion change on ``master`` (Potassium) drops the
+        # ``close()`` fallback and requires callers to use a context
+        # manager or explicit ``close()``; the loud warning here is the
+        # migration signal for that change.
+        #
+        # Python's ``__del__`` runs during GC (may be delayed, may skip
+        # on reference cycles) and during interpreter shutdown (when
+        # the world is already tearing down and touching a tornado
+        # ``IOStream`` can raise from a partially-freed C extension).
+        # The ``close()`` call chain is guarded so a finalizer never
+        # propagates an exception.
+        try:
+            already_closed = getattr(self, "_closing", True)
+        except Exception:  # pylint: disable=broad-except
+            return
+        if already_closed:
+            return
+        salt.utils.resource_warnings.warn_until_close(
+            f"unclosed publisher client {self!r}", source=self, log=log
+        )
+        try:
+            self.close()
+        except Exception:  # pylint: disable=broad-except
+            # Finalizer must never raise.  ``close()`` handles the
+            # ``_connecting_future`` and ``stream.close()`` steps
+            # individually with try/except-pass; this outer handler is
+            # a last resort for partially-freed C extensions during
+            # interpreter shutdown.
+            pass
 
     # pylint: enable=W1701
 

salt/utils/asynchronous.py

@@ -423,16 +423,14 @@ def __exit__(self, exc_type, exc_val, tb):
 
     # pylint: disable=W1701
     def __del__(self):
-        # PATCH: mirror ``SaltEvent.__del__`` at ``salt/utils/event.py``
-        # -- deliberately do NOT close the wrapped ``obj`` / io_loop /
-        # asyncio_loop from ``__del__``.  ``__del__`` fires during GC
-        # (may be arbitrarily delayed, may skip on reference cycles)
-        # and during interpreter shutdown, when the world is already
-        # tearing down and touching a tornado/asyncio loop can raise
-        # from a partially-freed C extension.  Instead, emit a
-        # ``ResourceWarning`` so callers that missed ``close()`` /
-        # context-manager surface loudly in tests / sentry / log
-        # aggregators.
+        # On this LTS branch ``__del__`` both surfaces the leak via
+        # ``warn_until_close`` (loud WARNING-level log record and
+        # ``ResourceWarning``) AND falls back to calling ``close()`` as
+        # a safety net, so callers that historically relied on GC-time
+        # cleanup do not silently leak a whole ``asyncio`` event loop,
+        # its tornado IOLoop, and the ZMQ context / socketpairs backing
+        # the wrapped async object (typically ``AsyncReqChannel``,
+        # ``AsyncPubChannel`` or ``AsyncEventPublisher``).
         #
         # Motivation: ``SyncWrapper``-owned asyncio loops are the
         # dominant leak surface on the minion under sustained
@@ -442,6 +440,18 @@ def __del__(self):
         # leaked socketpairs (~902 fds) per minion, tripping the
         # 1024-file ulimit critical threshold and the minion's own
         # sock-throttle logic.
+        #
+        # The companion change on ``master`` (Potassium) drops the
+        # ``close()`` fallback and requires callers to use a context
+        # manager or explicit ``close()``; the loud warning here is the
+        # migration signal for that change.
+        #
+        # Python's ``__del__`` runs during GC (may be delayed, may skip
+        # on reference cycles) and during interpreter shutdown (when the
+        # world is already tearing down and touching a tornado/asyncio
+        # loop can raise from a partially-freed C extension).  The
+        # ``close()`` call chain below is guarded so a finalizer never
+        # propagates an exception.
         try:
             unclosed = getattr(self, "obj", None) is not None or (
                 getattr(self, "asyncio_loop", None) is not None
@@ -458,5 +468,15 @@ def __del__(self):
             source=self,
             log=log,
         )
+        try:
+            self.close()
+        except Exception:  # pylint: disable=broad-except
+            # Finalizer must never raise.  ``close()`` is itself heavily
+            # guarded at each step (see the try/except-pass around every
+            # ``run_until_complete`` / ``io_loop.close`` call) so we do
+            # not expect to reach this outer handler in normal flow --
+            # it is a last resort for partially-freed C extensions
+            # during interpreter shutdown.
+            pass
 
     # pylint: enable=W1701

tests/pytests/unit/transport/test_tcp.py

@@ -2384,3 +2384,110 @@ def connect(self, timeout=None):
     assert captured["cls"] is salt.transport.tcp._TCPPubServerPublisher
     assert captured["kwargs"] == {"max_write_buffer_size": 4321}
     assert captured.get("connect_called") is True
+
+
+def test_publish_server_del_safety_net_calls_close_70175(master_opts):
+    """
+    Regression test for the __del__ safety-net cleanup extension of #70175.
+
+    When a caller drops the last reference to a ``PublishServer`` without
+    invoking ``close()`` first (typical of shutdown paths that skip
+    ``MinionManager.destroy``), the ``__del__`` finalizer must:
+
+    1. Emit the ``ResourceWarning`` so the leaky caller still surfaces
+       for tracking (behavior preserved from the warn-only revision).
+    2. Fall back to ``close()`` so the ``pub_sock`` / ``pub_server`` /
+       ``pull_sock`` / io_loop / per-loop cached publishers are
+       released, converting a ~50 MB/hr RSS leak into a bounded per-GC
+       cleanup.
+    """
+    server = salt.transport.tcp.PublishServer(
+        master_opts,
+        pub_host="127.0.0.1",
+        pub_port=1,
+        pull_host="127.0.0.1",
+        pull_port=2,
+    )
+    assert server._closing is False
+
+    # Wire fake sub-resources so we can observe that close() actually
+    # traversed them. Each mock records whether ``close()`` was called.
+    saved_pub_sock = MagicMock()
+    saved_pub_server = MagicMock()
+    saved_pull_sock = MagicMock()
+    saved_io_loop = MagicMock()
+    stale_pub = MagicMock()
+    stale_pub.close = MagicMock()
+    server.pub_sock = saved_pub_sock
+    server.pub_server = saved_pub_server
+    server.pull_sock = saved_pull_sock
+    server.io_loop = saved_io_loop
+    server._async_pub_by_loop = {"loop-key": (stale_pub, MagicMock())}
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        del server
+        gc.collect()
+
+    # 1. ResourceWarning still fires (behavior preserved).
+    resource_warnings = [w for w in caught if issubclass(w.category, ResourceWarning)]
+    assert resource_warnings, (
+        "expected ResourceWarning from PublishServer.__del__; got "
+        f"{[(w.category, str(w.message)) for w in caught]}"
+    )
+    assert any("unclosed publish server" in str(w.message) for w in resource_warnings)
+
+    # 2. Safety-net close() ran -- observed via the sub-resource mocks
+    #    (each ``.close()`` was invoked exactly once by ``PublishServer.close``).
+    saved_pub_sock.close.assert_called_once()
+    saved_pub_server.close.assert_called_once()
+    saved_pull_sock.close.assert_called_once()
+    # 3. io_loop had stop() + close() driven.
+    saved_io_loop.stop.assert_called_once()
+    saved_io_loop.close.assert_called_once_with(all_fds=True)
+    # 4. Per-loop cached publisher was drained.
+    stale_pub.close.assert_called_once()
+
+
+def test_tcppubserverpublisher_del_safety_net_calls_close_70175():
+    """
+    Regression test for the __del__ safety-net cleanup extension of #70175.
+
+    When a caller drops the last reference to a
+    ``_TCPPubServerPublisher`` without invoking ``close()`` first, the
+    ``__del__`` finalizer must both emit the ``ResourceWarning`` and
+    call ``close()`` so ``_closing`` flips True and the underlying
+    ``IOStream`` / socket FD are released rather than lingering as a
+    slow leak.
+    """
+    io_loop = tornado.ioloop.IOLoop()
+    publisher = salt.transport.tcp._TCPPubServerPublisher(
+        host="127.0.0.1", port=4511, path=None, io_loop=io_loop
+    )
+    # Install a fake stream so close() has something observable to
+    # close.  Its ``closed()`` returns False so ``close()`` walks the
+    # stream branch.
+    fake_stream = MagicMock()
+    fake_stream.closed.return_value = False
+    fake_stream.socket = MagicMock()
+    publisher.stream = fake_stream
+    publisher._connecting_future = tornado.concurrent.Future()
+    assert publisher._closing is False
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        del publisher
+        gc.collect()
+
+    resource_warnings = [w for w in caught if issubclass(w.category, ResourceWarning)]
+    assert resource_warnings, (
+        "expected ResourceWarning from _TCPPubServerPublisher.__del__; got "
+        f"{[(w.category, str(w.message)) for w in caught]}"
+    )
+    assert any("unclosed publisher client" in str(w.message) for w in resource_warnings)
+
+    # Safety-net close() ran: stream + socket were closed.
+    fake_stream.close.assert_called_once()
+    fake_stream.socket.close.assert_called_once()
+
+    io_loop.close()