transport/zeromq + channel + minion: async close for req channel reconnect

salt/channel/client.py

@@ -408,6 +408,36 @@ def close(self):
         self._closing = True
         self.transport.close()
 
+    async def close_async(self):
+        """Async-aware close.
+
+        When the caller is running on the ioloop that owns the underlying
+        transport, this awaits the transport's ``close_async`` (which in
+        turn awaits its ``_send_recv`` task's exit future) before
+        returning.  That ordering matters: sync ``close()``'s
+        same-thread + loop-running fallback runs teardown immediately,
+        which can race the still-running ``_send_recv`` task and leave
+        the ``zmq.Context`` alive on refs the task holds -- the Context
+        is then finalized from a later ioloop callback and can wedge in
+        ``zmq_ctx_term()``.  Awaiting ``close_async`` lets the send/recv
+        task drain the shutdown sentinel and release its socket
+        reference first, so the subsequent teardown is a clean
+        release-of-last-references rather than a race.
+
+        Falls back to sync ``close()`` for transports (e.g. TCP) that
+        do not implement ``close_async``.  Ioloop-owning callers such
+        as ``salt.minion``'s reconnect path should prefer this method.
+        """
+        if self._closing:
+            return
+        log.debug("Async-closing %s instance", self.__class__.__name__)
+        self._closing = True
+        close_async = getattr(self.transport, "close_async", None)
+        if close_async is None:
+            self.transport.close()
+        else:
+            await close_async()
+
     def __enter__(self):
         return self
 
@@ -419,7 +449,7 @@ async def __aenter__(self):
         return self
 
     async def __aexit__(self, *_):
-        self.close()
+        await self.close_async()
 
 
 class AsyncPubChannel:

salt/minion.py

@@ -2018,7 +2018,20 @@ async def connect_master(self, failed=False):
             if hasattr(self.pub_channel, "close"):
                 self.pub_channel.close()
         if hasattr(self, "req_channel") and self.req_channel:
-            self.req_channel.close()
+            # Wait for the underlying transport's ``_send_recv`` task to
+            # drain its shutdown sentinel and release the socket
+            # reference before we drop our reference to the channel.
+            # Otherwise the Context stays alive on the task's coroutine
+            # locals and is finalized later from a plain ioloop
+            # callback, where pyzmq's ``Context.__del__`` can wedge in
+            # ``zmq_ctx_term()``.  ``close_async`` is available on
+            # ``AsyncReqChannel``; guard so this still works if a
+            # third-party channel subclass only exposes sync ``close``.
+            close_async = getattr(self.req_channel, "close_async", None)
+            if close_async is not None:
+                await close_async()
+            else:
+                self.req_channel.close()
             self.req_channel = None
 
         # Consider refactoring so that eval_master does not have a subtle side-effect on the contents of the opts array
@@ -4655,7 +4668,18 @@ async def handle_event(self, package):
                     if hasattr(self.pub_channel, "close"):
                         self.pub_channel.close()
                 if hasattr(self, "req_channel") and self.req_channel:
-                    self.req_channel.close()
+                    # See ``connect_master`` for why ``close_async`` is
+                    # preferred here over the sync ``close``:  the
+                    # transport's send/recv task must drain before we
+                    # drop our reference to the channel, or the
+                    # underlying ``zmq.Context`` gets finalized from a
+                    # later ioloop callback and can wedge in
+                    # ``zmq_ctx_term()``.
+                    close_async = getattr(self.req_channel, "close_async", None)
+                    if close_async is not None:
+                        await close_async()
+                    else:
+                        self.req_channel.close()
                     self.req_channel = None
 
                 # if eval_master finds a new master for us, self.connected

salt/transport/zeromq.py

@@ -2322,6 +2322,66 @@ async def _reconnect(self):
             self.socket = None
         await self.connect()
 
+    async def close_async(self):
+        """Preferred close for ioloop-owning callers.
+
+        The sync ``close()``'s same-thread + loop-running branch cannot
+        await the running ``_send_recv`` task (blocking would deadlock
+        the loop it is on), so it runs ``_sync_teardown`` immediately.
+        ``_send_recv`` may still be holding a reference to ``socket``
+        in its coroutine locals when that teardown runs, which leaves
+        the ``Context`` refcount above zero.  When ``_send_recv``
+        eventually exits and drops its socket reference, the
+        ``Context`` is finalized -- from an ioloop callback, on the
+        loop's own thread -- and pyzmq's ``Context.__del__`` can wedge
+        the loop in ``zmq_ctx_term()``.
+
+        ``close_async`` fixes the race by actually awaiting
+        ``_send_recv_exit_future`` before the teardown runs.  Once
+        ``_send_recv`` has drained the shutdown sentinel and returned,
+        the socket / context are the only remaining references; the
+        explicit ``socket.close()`` + ``context.destroy(linger=1000)``
+        below then release them deterministically, with the bounded
+        linger acting as a safety net so the destroy never blocks the
+        caller indefinitely.
+
+        Callers on the ioloop thread that are about to drop the
+        underlying client reference (e.g. ``salt.minion`` on its
+        reconnect path) should use this instead of the sync
+        ``close()``.
+        """
+        if self._closing:
+            return
+        self._closing = True
+        if hasattr(self, "_queue") and self._queue is not None:
+            try:
+                self._queue.put_nowait((None, None))
+            except Exception:  # pylint: disable=broad-except
+                pass
+        socket = self.socket
+        context = self.context
+        exit_future = self._send_recv_exit_future
+        self.socket = None
+        self.context = None
+        self._send_recv_exit_future = None
+        if exit_future is not None:
+            try:
+                await asyncio.wait_for(asyncio.shield(exit_future), timeout=5)
+            except (asyncio.TimeoutError, asyncio.CancelledError):
+                pass
+            except Exception:  # pylint: disable=broad-except
+                log.debug("RequestClient graceful drain failed", exc_info=True)
+        if socket is not None:
+            try:
+                socket.close()
+            except Exception:  # pylint: disable=broad-except
+                pass
+        if context is not None and not context.closed:
+            try:
+                context.destroy(linger=1000)
+            except Exception:  # pylint: disable=broad-except
+                pass
+
     async def send(self, load, timeout=60):
         """
         Return a future which will be completed when the message has a response

tests/pytests/scenarios/reauth/test_reauth.py

@@ -106,6 +106,13 @@ def test_reauth(salt_cli, salt_minion, salt_master, timeout, event_listener):
 
 
 def test_presence_events(salt_cli, salt_minion, salt_master, event_listener):
+    # ``test_reauth`` above tears the minion down at end (``stop_event.set()``
+    # -> ``minion_func`` finally -> ``salt_minion.terminate()``).  The
+    # package-scoped ``salt_minion`` fixture does not auto-restart, so
+    # this second test in the file runs against a dead minion.  Bring it
+    # back up before pinging.
+    if not salt_minion.is_running():
+        salt_minion.start()
     # On slow runners (FIPS/Arm64 in particular) the first ping after
     # master+minion startup can exceed the factory-level 30 s CLI timeout
     # while the minion finishes auth/reauth and the presence machinery