REQ/ZMQ/TCP cleanup: teardown helpers, minion cruft removal, balancer race guard

Drop unused timestamp assignments left in SaltClientError / generic reconnect
handlers in MinionManager._connect_minion and the matching Syndic path.
Refactor functional zeromq RequestClient tests to prefer close_future-backed
cleanup with IOLoop-friendly polling and a synchronous fallback
(_close_zmq_only / _mark_teardown_finished) when graceful shutdown stalls.
Always set RequestClient._closing after the fallback so Transport.__del__ does
not emit Unclosed transport warnings after tests stop the IOLoop.
Harden TCP LoadBalancerServer accept loop: if close() clears the socket while
accept() fails, exit the loop instead of calling getsockname() on None
(fixes PytestUnhandledThreadExceptionWarning in test_tcp_load_balancer_server).

Author: twangboy

salt/minion.py

@@ -1177,7 +1177,6 @@ def _connect_minion(self, minion):
                     minion.opts["master"],
                     exc,
                 )
-                last = time.time()
                 if auth_wait < self.max_auth_wait:
                     auth_wait += self.auth_wait
                 yield salt.ext.tornado.gen.sleep(auth_wait)  # TODO: log?
@@ -1202,7 +1201,6 @@ def _connect_minion(self, minion):
                 # Match SaltClientError path: without a delay, connect_master can be
                 # retried in a tight loop and create zmq contexts faster than they are
                 # torn down (libzMQ pthread / EMFILE failures on some hosts).
-                last = time.time()
                 if auth_wait < self.max_auth_wait:
                     auth_wait += self.auth_wait
                 yield salt.ext.tornado.gen.sleep(auth_wait)
@@ -4557,7 +4555,6 @@ def _connect_syndic(self, opts):
                     "master at %s responding?",
                     opts["master"],
                 )
-                last = time.time()
                 if auth_wait < self.max_auth_wait:
                     auth_wait += self.auth_wait
                 yield salt.ext.tornado.gen.sleep(auth_wait)  # TODO: log?

salt/transport/tcp.py

@@ -183,7 +183,10 @@ def run(self):
                 # ECONNABORTED indicates that there was a connection
                 # but it was closed while still in the accept queue.
                 # (observed on FreeBSD).
-                name = self._socket.getsockname()
+                sock = self._socket
+                if sock is None:
+                    break
+                name = sock.getsockname()
                 if isinstance(name, tuple):
                     name = name[0]
                 if salt.ext.tornado.util.errno_from_exception(e) == errno.ECONNABORTED:

tests/pytests/functional/transport/zeromq/test_request_client.py

@@ -12,47 +12,109 @@
 
 log = logging.getLogger(__name__)
 
-
-_REQ_DRAIN_POLLS = 300
-_REQ_DRAIN_SLEEP_S = 0.01
 _REQ_DRAIN_TIMEOUT_S = 10
-
+_REQ_POLL_S = 0.01
 
 pytestmark = [
     pytest.mark.windows_whitelisted,
 ]
 
 
-def _sync_finalize_req_client(cli, io_loop):
+def _blocking_teardown_req_message_client(cli):
+    """
+    When ``close_future`` cannot finish, release ZMQ resources.
+
+    ``AsyncReqMessageClient.close()`` is a no-op if ``close_future()`` already
+    started teardown but ``finalize()`` never ran (stuck loop, half-finished
+    ``_send_recv``, etc.). ``_close_zmq_only`` matches the resource release in
+    ``finalize`` without re-entering ``_initiate_async_req_close``.
+    """
+    try:
+        mc = cli.message_client
+        if getattr(mc, "socket", None) is not None:
+            try:
+                mc.close()
+            except Exception:  # pylint: disable=broad-except
+                log.debug(
+                    "REQ MessageClient synchronous close failed during cleanup",
+                    exc_info=True,
+                )
+        if getattr(mc, "socket", None) is not None:
+            mc._close_zmq_only()
+        mc._mark_teardown_finished()
+    except Exception:  # pylint: disable=broad-except
+        log.debug(
+            "REQ MessageClient synchronous teardown fallback failed during cleanup",
+            exc_info=True,
+        )
+    finally:
+        # ``Transport.__del__`` warns unless ``_closing`` is set; when the owning
+        # ``IOLoop`` is stopped we never yield ``cli.close_future()`` (#68637).
+        setattr(cli, "_closing", True)
+
+
+def _sync_yield_req_client_close_future(cli, io_loop):
     """
-    Explicit REQ close plus bounded wait for AsyncReqMessageClient teardown.
+    Block until zeromq RequestClient asynchronous teardown completes (#68637).
 
-    When close() is deferred onto the loop (#68637), dropping the REQ client immediately
-    can leave Zeromq sockets until GC and trigger noisy ``Socket.__del__`` tracebacks at
-    process exit unless we drain ``message_client.socket is None``.
+    Uses ``RequestClient.close_future()``—same completion contract as
+    ``AsyncReqChannel.close_async`` / ``Minion.destroy_async``—rather than polling
+    ``message_client.socket is None``. If the loop is stopped or ``run_sync`` fails,
+    fall back to synchronous ``AsyncReqMessageClient.close()`` (covers tests that stop
+    the ``IOLoop`` before teardown).
     """
 
     @salt.ext.tornado.gen.coroutine
-    def _runner():
-        cli.close()
-        for _ in range(_REQ_DRAIN_POLLS):
-            if cli.message_client.socket is None:
-                break
-            yield salt.ext.tornado.gen.sleep(_REQ_DRAIN_SLEEP_S)
+    def _wait():
+        fut = cli.close_future()
+        yield fut
 
+    ok = False
     try:
-        io_loop.run_sync(_runner, timeout=_REQ_DRAIN_TIMEOUT_S)
+        if getattr(io_loop, "_running", False):
+            io_loop.run_sync(_wait, timeout=_REQ_DRAIN_TIMEOUT_S)
+            ok = True
     except Exception:  # pylint: disable=broad-except
-        log.debug("REQ client teardown drain aborted during cleanup", exc_info=True)
+        log.debug(
+            "REQ client close_future waiter aborted during cleanup", exc_info=True
+        )
+    if not ok or getattr(cli.message_client, "socket", None) is not None:
+        _blocking_teardown_req_message_client(cli)
+
+
+def _sync_finalize_req_client(cli, io_loop):
+    """Explicit teardown for fixtures (wait on ``close_future``)."""
+    _sync_yield_req_client_close_future(cli, io_loop)
 
 
 async def async_finalize_req_client(cli):
-    """Async equivalent of :func:`_sync_finalize_req_client`."""
-    cli.close()
-    for _ in range(_REQ_DRAIN_POLLS):
-        if cli.message_client.socket is None:
-            break
-        await salt.ext.tornado.gen.sleep(_REQ_DRAIN_SLEEP_S)
+    """
+    Async cleanup: spin the I/O loop via ``gen.sleep`` until ``close_future`` completes.
+
+    Plain ``await`` on a Tornado ``Future`` does not drive ``cli.io_loop`` in this
+    test harness (#68637).
+    """
+    await _async_wait_close_future(cli, "fixture REQ close_future did not finish")
+
+
+async def _await_req_teardown_after_close(cli):
+    """After ``RequestClient.close()`` in-test, wait for deferred teardown."""
+    await _async_wait_close_future(
+        cli, "REQ message client did not finish teardown after RequestClient.close()"
+    )
+
+
+async def _async_wait_close_future(cli, fail_msg):
+    fut = cli.close_future()
+    n = max(1, int(_REQ_DRAIN_TIMEOUT_S / _REQ_POLL_S))
+    for _ in range(n):
+        if fut.done():
+            fut.result()
+            return
+        await salt.ext.tornado.gen.sleep(_REQ_POLL_S)
+    _blocking_teardown_req_message_client(cli)
+    if getattr(cli.message_client, "socket", None) is not None:
+        pytest.fail(fail_msg)
 
 
 def _zmq_teardown_rep(stream=None, rep_socket=None, ctx=None):
@@ -96,7 +158,7 @@ async def test_request_channel_issue_64627(io_loop, request_client, minion_opts,
     Validate socket is preserved until request channel is explicitly closed.
 
     When ``AsyncReqMessageClient.close()`` runs on an active ``IOLoop``, teardown is
-    scheduled on the loop (#68637); yield until ``socket`` is cleared before asserting.
+    scheduled on the loop (#68637); yield ``close_future`` before asserting the socket is gone.
     """
     minion_opts["master_uri"] = f"tcp://127.0.0.1:{port}"
 
@@ -117,15 +179,7 @@ def req_handler(stream, msg):
         rep = await request_client.send(b"foo")
         assert req_socket is request_client.message_client.socket
         request_client.close()
-        for _ in range(300):
-            if request_client.message_client.socket is None:
-                break
-            await salt.ext.tornado.gen.sleep(0.01)
-        else:
-            pytest.fail(
-                "REQ message client socket not cleared after RequestClient.close() "
-                "(deferred teardown on running IOLoop; see #68637)"
-            )
+        await _await_req_teardown_after_close(request_client)
         assert request_client.message_client.socket is None
 
     finally:
@@ -221,15 +275,7 @@ async def test_request_client_send_recv_socket_closed(
 
         with caplog.at_level(logging.TRACE):
             request_client.close()
-            for _ in range(300):
-                if request_client.message_client.socket is None:
-                    break
-                await salt.ext.tornado.gen.sleep(0.01)
-            else:
-                pytest.fail(
-                    "REQ message client socket not cleared after RequestClient.close() "
-                    "(deferred teardown on running IOLoop; see #68637)"
-                )
+            await _await_req_teardown_after_close(request_client)
 
             assert any(
                 "Send and receive coroutine ending" in msg and "closed" in msg