Drop abandoned requests when draining the ZeroMQ send queue

`AsyncReqMessageClient.send()` enqueues `(future, message)` on `self._queue`
and arms `_timeout_message()` to complete the future when the caller's timeout
expires. When that fires the caller is released, but the queue entry is not
removed -- it keeps pinning its serialized payload until the drain loop
happens to reach it.

Before 145a06e in-flight requests were tracked in `self._send_future_map`, a
dict, so a timed-out entry could be removed by key. `Queue` has no
removal-from-middle and `self._queue` has no maxsize, so an abandoned entry
becomes unreachable garbage that still occupies the queue.

A REQ socket permits one request/reply in flight, each drain iteration polls
for up to 300ms, and `_send_recv()` restarts from scratch on every reconnect.
Under sustained load the enqueue rate outruns the drain rate and the queue
grows without bound. Observed on a 120-minion master: one process at ~53 GiB,
OOM-killed roughly fortnightly.

Skip requests whose future is already done: the reply can no longer be
delivered to anyone, so there is no reason to spend a round trip on it.

Two existing tests queued a single already-done future and relied on it
reaching the send path to terminate the loop. With the entry now dropped they
fall through to the idle poll branch, which calls `.result()` on the
AsyncMock's coroutine. Both now queue a shutdown sentinel so the loop exits.
Note this makes the #68506 repro structurally unreachable on this path -- a
timed-out future never reaches `socket.send` -- so that test now asserts only
its invariant, that the original timeout exception survives untouched.

Fixes #68660

Author: bretep

changelog/68660.fixed.md

@@ -0,0 +1,4 @@
+Drop abandoned requests when draining the ZeroMQ send queue in
+``AsyncReqMessageClient``. A request whose caller had already timed out stayed
+in ``self._queue`` holding its serialized payload until the drain loop reached
+it, which under sustained load it never did, growing the queue without bound.

salt/transport/zeromq.py

@@ -1001,6 +1001,26 @@ def _send_recv(self, socket, _TimeoutError=salt.ext.tornado.gen.TimeoutError):
                     send_recv_running = False
                     break
 
+                if future.done():
+                    # The caller already abandoned this request: send() arms
+                    # _timeout_message(), which completes the future when the
+                    # caller's timeout expires. The reply can no longer be
+                    # delivered to anyone, so drop the request instead of
+                    # spending a round trip on it.
+                    #
+                    # Without this, self._queue grows without bound. Before
+                    # 145a06e in-flight requests were tracked in
+                    # self._send_future_map, a dict, so a timed-out entry was
+                    # removed by key. A Queue has no removal-from-middle and
+                    # self._queue has no maxsize, so an abandoned entry stays
+                    # queued -- pinning its serialized payload -- until the
+                    # drain loop reaches it. A REQ socket permits one
+                    # request/reply in flight and _send_recv() restarts on
+                    # every reconnect, so under sustained load the enqueue rate
+                    # outruns the drain rate and it never does.
+                    log.trace("Dropping request whose caller already timed out")
+                    continue
+
                 try:
                     yield socket.send(message)
                 except zmq.eventloop.future.CancelledError as exc:

tests/pytests/unit/transport/test_zeromq.py

@@ -1825,6 +1825,16 @@ async def test_client_send_recv_on_cancelled_error(minion_opts):
         client.socket = AsyncMock()
         client.socket.poll.side_effect = zmq.eventloop.future.CancelledError
         client._queue.put_nowait((mock_future, {"meh": "bah"}))
+        # The future is already done, so _send_recv drops it (see #68660) and
+        # returns to the queue. Queue a shutdown sentinel so the loop exits
+        # rather than falling through to the idle poll branch, which would
+        # call .result() on the AsyncMock's coroutine.
+        client._queue.put_nowait(
+            (
+                salt.ext.tornado.concurrent.Future(),
+                salt.transport.zeromq._REQ_QUEUE_SHUTDOWN,
+            )
+        )
         await client._send_recv(client.socket)
         mock_future.set_exception.assert_not_called()
     finally:
@@ -1856,6 +1866,17 @@ async def test_client_send_recv_no_double_set_exception_after_timeout(minion_opt
         client.socket = AsyncMock()
         client.socket.send.side_effect = zmq.ZMQError(zmq.ETERM)
         client._queue.put_nowait((future, {"meh": "bah"}))
+        # Since #68660 a request whose future is already done is dropped before
+        # it reaches socket.send, so this repro no longer exercises the send
+        # failure path -- the double-set it guarded against is now structurally
+        # unreachable here. The invariant still asserted below is that the
+        # original timeout exception survives untouched.
+        client._queue.put_nowait(
+            (
+                salt.ext.tornado.concurrent.Future(),
+                salt.transport.zeromq._REQ_QUEUE_SHUTDOWN,
+            )
+        )
         # Before the fix this raises TypeError from tornado's _set_done.
         await client._send_recv(client.socket)
         # The timeout exception must be preserved, not overwritten.
@@ -1864,6 +1885,53 @@ async def test_client_send_recv_no_double_set_exception_after_timeout(minion_opt
         client.close()
 
 
+async def test_client_send_recv_drops_abandoned_request(minion_opts):
+    """
+    Regression test for #68660.
+
+    ``send()`` enqueues ``(future, message)`` and arms ``_timeout_message``.
+    When the caller's timeout fires the future is completed, but its queue
+    entry remains and keeps pinning the serialized payload until the drain
+    loop reaches it. ``self._queue`` has no maxsize and a REQ socket permits
+    one request/reply in flight, so under sustained load the enqueue rate
+    outruns the drain rate and the queue grows without bound.
+
+    ``_send_recv`` must drop a request whose future is already done rather
+    than spend a round trip on a reply nobody can receive.
+    """
+    client = salt.transport.zeromq.AsyncReqMessageClient(
+        minion_opts, "tcp://127.0.0.1:4506"
+    )
+
+    abandoned = salt.ext.tornado.concurrent.Future()
+    # Exactly what _timeout_message does when the caller's timeout expires.
+    client._timeout_message(abandoned)
+    assert abandoned.done()
+
+    # Keep our own reference: without the fix ``_send_recv`` takes its error
+    # path and ``_reconnect()`` swaps ``client.socket`` for a real socket, so
+    # asserting against ``client.socket`` afterwards would inspect the wrong
+    # object and fail for the wrong reason.
+    sock = AsyncMock()
+    try:
+        client.socket = sock
+        client._queue.put_nowait((abandoned, {"meh": "bah"}))
+        # Sentinel stops the drain loop after the abandoned entry is handled.
+        client._queue.put_nowait(
+            (
+                salt.ext.tornado.concurrent.Future(),
+                salt.transport.zeromq._REQ_QUEUE_SHUTDOWN,
+            )
+        )
+        await client._send_recv(sock)
+        # The abandoned payload must never reach the wire.
+        sock.send.assert_not_called()
+        # And its timeout exception must be left intact.
+        assert isinstance(abandoned.exception(), salt.exceptions.SaltReqTimeoutError)
+    finally:
+        client.close()
+
+
 def test_async_req_message_client_close_never_connected(minion_opts):
     """
     close() must not hang when connect() was never called (#68637).