[wip] close async audit 3008.x

salt/channel/client.py

@@ -445,7 +445,17 @@ def __exit__(self, *args):
         self.close()
 
     async def __aenter__(self):
-        await self.transport.connect()
+        # Match the sync ``__enter__`` pattern: return without eagerly
+        # calling ``transport.connect()``.  ``transport.send()`` does its
+        # own lazy connect on the first send, so all real callers still
+        # get a live socket when they need one.  The eager-connect variant
+        # would allocate a ZMQ ``Context`` (and register the
+        # ``weakref.finalize`` from PR-70315) for every ``async with``
+        # even when no send followed -- an issue in tests that mock the
+        # inner call (e.g. AsyncAuth._authenticate with a mocked
+        # sign_in) but let the real ``AsyncReqChannel.factory`` build
+        # the transport.  Those Contexts accumulated across the run and
+        # eventually wedged pytest under Python 3.14.
         return self
 
     async def __aexit__(self, *_):
@@ -643,7 +653,7 @@ async def connect_callback(self, result):
                     "data": data,
                     "tag": tag,
                 }
-                with AsyncReqChannel.factory(self.opts) as channel:
+                async with AsyncReqChannel.factory(self.opts) as channel:
                     try:
                         await channel.send(load, timeout=60)
                     except salt.exceptions.SaltReqTimeoutError:

salt/client/__init__.py

@@ -2217,7 +2217,7 @@ async def pub_async(
             + str(self.opts["ret_port"])
         )
 
-        with salt.channel.client.AsyncReqChannel.factory(
+        async with salt.channel.client.AsyncReqChannel.factory(
             self.opts, io_loop=io_loop, crypt="clear", master_uri=master_uri
         ) as channel:
             try:

salt/crypt.py

@@ -1225,7 +1225,7 @@ async def _authenticate(self):
             acceptance_wait_time_max = acceptance_wait_time
         creds = None
 
-        with salt.channel.client.AsyncReqChannel.factory(
+        async with salt.channel.client.AsyncReqChannel.factory(
             self.opts, crypt="clear", io_loop=self.io_loop
         ) as channel:
             error = None
@@ -1407,7 +1407,19 @@ async def sign_in(self, timeout=60, safe=True, tries=1, channel=None):
                 )
         finally:
             if close_channel:
-                channel.close()
+                # Prefer ``close_async`` when the channel exposes it
+                # (``AsyncReqChannel`` does) so the underlying transport's
+                # ``_send_recv`` task drains its shutdown sentinel and
+                # releases the socket ref before we drop our reference.
+                # The sync ``close`` fallback keeps compatibility with
+                # third-party channel subclasses that predate
+                # ``close_async``.  See PR-70316 for the wedge this
+                # ordering avoids.
+                close_async = getattr(channel, "close_async", None)
+                if close_async is not None:
+                    await close_async()
+                else:
+                    channel.close()
         return self.handle_signin_response(sign_in_payload, payload)
 
     def handle_signin_response(self, sign_in_payload, payload):

salt/minion.py

@@ -2079,7 +2079,11 @@ async def _post_master_init(self, master):
                 pillarenv=self.opts.get("pillarenv"),
             )
             self.opts["pillar"] = await async_pillar.compile_pillar()
-            async_pillar.destroy()
+            # Async aclose awaits the transport's send/recv exit
+            # future before releasing the socket + context, avoiding
+            # the pyzmq Context.__del__ wedge that sync ``destroy``
+            # can leave behind (see PR-70316 trace).
+            await async_pillar.aclose()
             # _setup_core uses _load_modules only — unlike gen_modules it does not
             # run _discover_resources().  tune_in schedules _register_resources_with_master
             # right after connect; without this, the master registry gets {} until an
@@ -4347,7 +4351,9 @@ async def pillar_refresh(self, force_refresh=False, clean_cache=False):
                 self.opts["resources"] = self._discover_resources()
                 await self._register_resources_with_master()
             finally:
-                async_pillar.destroy()
+                # See ``_post_master_init`` for why aclose is preferred
+                # over sync destroy on ioloop-owning callers.
+                await async_pillar.aclose()
         self.matchers_refresh()
         self.beacons_refresh()
         # Fire the completion event synchronously on the minion event bus.

salt/pillar/__init__.py

@@ -282,6 +282,38 @@ async def compile_pillar(self):
         ret_pillar = salt.utils.secret.hide(ret_pillar)
         return ret_pillar
 
+    async def aclose(self):
+        """Async-aware teardown.
+
+        ``AsyncRemotePillar`` owns an ``AsyncReqChannel`` and is only
+        ever constructed on an active asyncio loop (via
+        ``get_async_pillar`` from ``async def post_master_init`` /
+        ``pillar_refresh`` / the master's ``_pillar`` handler).  Sync
+        ``destroy`` calls ``channel.close()``, whose same-thread +
+        loop-running fallback cannot await the transport's running
+        ``_send_recv`` task -- so the socket reference the task holds
+        keeps the underlying ``zmq.asyncio.Context`` alive until GC
+        later finalizes it from an ioloop callback and wedges the loop
+        in ``zmq_ctx_term()`` (see the PR-70316 wedge trace).
+
+        Callers that hold an ``AsyncRemotePillar`` on an ioloop must
+        prefer ``await pillar.aclose()`` over ``pillar.destroy()`` so
+        the send/recv task drains its shutdown sentinel and releases
+        the socket before we drop our reference to the channel.  A
+        ``getattr`` guard keeps this compatible with third-party
+        channel subclasses that only expose sync ``close`` -- the
+        same shape ``Minion.connect_master`` and
+        ``Minion.handle_event`` use.
+        """
+        if self._closing:
+            return
+        self._closing = True
+        close_async = getattr(self.channel, "close_async", None)
+        if close_async is not None:
+            await close_async()
+        else:
+            self.channel.close()
+
     def destroy(self):
         if self._closing:
             return
@@ -291,6 +323,20 @@ def destroy(self):
 
     # pylint: disable=W1701
     def __del__(self):
+        # Kept as a defensive net for third-party consumers that
+        # never migrated to ``aclose``.  Async callsites now go
+        # through ``aclose`` first, which flips ``_closing`` and
+        # makes this ``destroy`` a no-op -- so the sync
+        # ``channel.close()`` teardown does not fire from ``__del__``
+        # under normal use.  Emit a debug log if we do reach this
+        # path so a leaked ``AsyncRemotePillar`` is visible in the
+        # logs rather than silent.
+        if not getattr(self, "_closing", False):
+            log.debug(
+                "AsyncRemotePillar reached __del__ without prior aclose(); "
+                "falling back to sync destroy -- caller should await "
+                "aclose() from its ioloop instead."
+            )
         self.destroy()
 
     # pylint: enable=W1701

salt/transport/zeromq.py

@@ -41,7 +41,7 @@
 import salt.utils.stringutils
 import salt.utils.zeromq
 from salt._compat import ipaddress
-from salt.exceptions import SaltException, SaltReqTimeoutError
+from salt.exceptions import SaltClientError, SaltException, SaltReqTimeoutError
 from salt.utils.zeromq import LIBZMQ_VERSION_INFO, ZMQ_VERSION_INFO, zmq
 
 try:
@@ -2120,9 +2120,30 @@ def __init__(self, opts, io_loop, linger=0):  # pylint: disable=W0231
 
     async def connect(self):  # pylint: disable=invalid-overridden-method
         async with self._connect_lock:
+            if self._closing:
+                # A closed ``RequestClient`` must not silently resurrect
+                # itself here.  ``close_async`` cleared ``self.socket`` and
+                # ``self.context`` and destroyed the underlying ZMQ Context
+                # deterministically; the previous version of ``connect``
+                # then unconditionally reset ``self._closing = False`` and
+                # ran ``_init_socket``, which allocated a fresh Context
+                # and registered a new ``weakref.finalize`` against a
+                # RequestClient that no live caller was tracking any more.
+                # When that RequestClient eventually got GC'd the new
+                # finalizer fired from an ioloop callback and blocked in
+                # ``zmq_ctx_term()``, wedging the minion.  Refuse the
+                # reconnect and force the caller to construct a fresh
+                # ``AsyncReqChannel`` if it needs one.  This scenario shows
+                # up in practice when a coroutine captured a reference to
+                # ``self.req_channel`` across an ``await`` and the
+                # reconnect path (``connect_master`` / ``handle_event``
+                # master-changed) swapped in a new channel before that
+                # captured coroutine's next ``send`` fires.
+                raise SaltClientError(
+                    "RequestClient is closed; construct a new one to reconnect."
+                )
             if self.socket is None:
                 self._connect_called = True
-                self._closing = False
                 # wire up sockets
                 self._init_socket()
 

tests/pytests/functional/channel/test_client.py

@@ -1,10 +1,15 @@
 import salt.channel.client
-from tests.support.mock import MagicMock, patch
+from tests.support.mock import AsyncMock, MagicMock, patch
 
 
 async def test_async_pub_channel_connect_cb(minion_opts):
     """
     Validate connect_callback closes the request channel it creates.
+
+    ``connect_callback`` uses ``async with AsyncReqChannel.factory(...)``
+    (switched from sync ``with`` as part of the VCOPS-90587 close-async
+    audit), so the mock has to expose ``__aenter__`` / ``__aexit__``,
+    not ``__enter__`` / ``__exit__``.
     """
     minion_opts["master_uri"] = "tcp://127.0.0.1:4506"
     minion_opts["master_ip"] = "127.0.0.1"
@@ -17,9 +22,10 @@ async def send_id(*args):
         channel._reconnected = True
 
         mock = MagicMock(salt.channel.client.AsyncReqChannel)
-        mock.__enter__ = lambda self: mock
+        mock.__aenter__ = AsyncMock(return_value=mock)
+        mock.__aexit__ = AsyncMock(return_value=None)
 
         with patch("salt.channel.client.AsyncReqChannel.factory", return_value=mock):
             await channel.connect_callback(None)
             mock.send.assert_called_once()
-            mock.__exit__.assert_called_once()
+            mock.__aexit__.assert_called_once()