diff --git a/changelog/68660.fixed.md b/changelog/68660.fixed.md new file mode 100644 index 000000000000..fc1b85c2e438 --- /dev/null +++ b/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. diff --git a/salt/transport/zeromq.py b/salt/transport/zeromq.py index a01687548433..76df541fd50c 100644 --- a/salt/transport/zeromq.py +++ b/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: diff --git a/tests/pytests/unit/transport/test_zeromq.py b/tests/pytests/unit/transport/test_zeromq.py index 73afcfee541c..83b2e3db89b8 100644 --- a/tests/pytests/unit/transport/test_zeromq.py +++ b/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).