From 702b5baf2fcbc759c4a2fe59be04330902db38b7 Mon Sep 17 00:00:00 2001 From: "Daniel A. Wozniak" Date: Tue, 14 Jul 2026 14:22:48 -0700 Subject: [PATCH] Fix POLLOUT single-shot timeout in AsyncReqMessageClient._send_recv (#69802) RequestClient._send_recv (and AsyncReqMessageClient._send_recv, both in salt/transport/zeromq.py) treated a single 300 ms zmq.POLLOUT miss as an immediate SaltReqTimeoutError and reconnected, ignoring the caller's own send timeout registered via io_loop.call_later(timeout, _timeout_message, future). A slow master or brief socket congestion therefore aborted the request within 300 ms even when the caller had requested a much longer timeout. Loop the POLLOUT poll the same way the POLLIN branch below already does: continue polling until either the socket becomes writable or future.done() is True (set by the caller's own timeout callback). On future.done() without a ready socket, reconnect for the next send and break without overwriting the caller's exception. Add regression tests covering both the slow-but-eventually-ready path (request succeeds) and the never-ready-with-caller-timeout path (caller's exception preserved, socket.send never called, reconnect triggered). Fixes #69802 --- changelog/69802.fixed.md | 1 + salt/transport/zeromq.py | 23 +++- tests/pytests/unit/transport/test_zeromq.py | 125 ++++++++++++++++++++ 3 files changed, 143 insertions(+), 6 deletions(-) create mode 100644 changelog/69802.fixed.md diff --git a/changelog/69802.fixed.md b/changelog/69802.fixed.md new file mode 100644 index 000000000000..1bf909c7df1c --- /dev/null +++ b/changelog/69802.fixed.md @@ -0,0 +1 @@ +Fix ``AsyncReqMessageClient._send_recv`` treating a single 300ms POLLOUT miss as a hard timeout instead of looping until the caller's timeout fires. diff --git a/salt/transport/zeromq.py b/salt/transport/zeromq.py index 84046009b70e..a30c583b9248 100644 --- a/salt/transport/zeromq.py +++ b/salt/transport/zeromq.py @@ -2154,12 +2154,23 @@ async def _send_recv( break try: - # Wait for socket to be ready for sending - if not await socket.poll(300, zmq.POLLOUT): - if not future.done(): - future.set_exception( - SaltReqTimeoutError("Socket not ready for sending") - ) + # Wait for socket to be ready for sending. Loop until either + # POLLOUT reports the socket is ready or the caller's own + # timeout (scheduled via ``io_loop.call_later`` in ``send``) + # marks ``future`` as done with ``SaltReqTimeoutError``. + # A single 300 ms poll miss is not a hard failure; the caller + # decides how long to wait via ``send(..., timeout=...)``. + ready = False + while True: + ready = await socket.poll(300, zmq.POLLOUT) + if ready: + break + if future.done(): + break + if not ready: + # The caller's timeout already errored ``future``. Reconnect + # so the next send starts from a clean socket, but do not + # overwrite the exception the caller sees. if not self._closing: await self._reconnect() break diff --git a/tests/pytests/unit/transport/test_zeromq.py b/tests/pytests/unit/transport/test_zeromq.py index 29e1c0aed21f..39dbcf53ba07 100644 --- a/tests/pytests/unit/transport/test_zeromq.py +++ b/tests/pytests/unit/transport/test_zeromq.py @@ -17,6 +17,7 @@ from pytestshellutils.utils import ports import salt.config +import salt.payload import salt.transport.base import salt.transport.zeromq import salt.utils.platform @@ -1904,6 +1905,130 @@ async def test_client_send_recv_no_double_set_exception_after_timeout(minion_opt client.close() +async def test_request_client_send_recv_slow_pollout_does_not_timeout( + minion_opts, io_loop +): + """ + Regression test for #69802. + + ``RequestClient._send_recv`` must NOT treat a single 300 ms ``zmq.POLLOUT`` + miss as an immediate ``SaltReqTimeoutError``. It should loop the + POLLOUT poll (mirroring the POLLIN loop below it) until either the + socket becomes writable or the caller's own timeout (registered via + ``io_loop.call_later`` in ``send``) marks the future done. + + Simulates a slow socket by returning 0 (not ready) from + ``socket.poll`` on the first two calls and then returning + ``zmq.POLLOUT`` on the third call. The message should be sent and + the response should be delivered normally. + """ + client = salt.transport.zeromq.RequestClient(minion_opts, io_loop) + + future = tornado.concurrent.Future() + + poll_calls = {"n": 0} + + async def slow_pollout(timeout, flag): + # First two POLLOUT polls miss; third is ready. POLLIN afterwards + # is ready immediately. + if flag == zmq.POLLOUT: + poll_calls["n"] += 1 + if poll_calls["n"] <= 2: + return 0 + return zmq.POLLOUT + return zmq.POLLIN + + async def fake_send(msg): + return None + + async def fake_recv(): + return salt.payload.dumps({"ok": True}) + + try: + client.socket = AsyncMock() + client.socket.poll.side_effect = slow_pollout + client.socket.send.side_effect = fake_send + client.socket.recv.side_effect = fake_recv + + client._queue.put_nowait((future, b"payload")) + # Sentinel to break the outer loop once our message is processed. + client._queue.put_nowait((None, None)) + + await client._send_recv(client.socket, client._queue) + + assert future.done() + assert future.exception() is None + assert future.result() == {"ok": True} + # Two misses + one hit. + assert poll_calls["n"] == 3 + finally: + client.close() + + +async def test_request_client_send_recv_pollout_respects_caller_timeout( + minion_opts, io_loop +): + """ + Negative companion to + ``test_request_client_send_recv_slow_pollout_does_not_timeout``. + + When POLLOUT is never ready AND the caller's timeout has fired + (setting ``SaltReqTimeoutError`` on the future), ``_send_recv`` must + exit its POLLOUT loop and preserve the caller's exception without + overwriting it or raising a different error. + """ + client = salt.transport.zeromq.RequestClient(minion_opts, io_loop) + + future = tornado.concurrent.Future() + + poll_calls = {"n": 0} + + async def never_ready(timeout, flag): + poll_calls["n"] += 1 + # Return 0 (POLLOUT miss) on the first poll. The caller's + # ``_timeout_message`` then fires between iterations, so on the + # second poll the loop must observe ``future.done()`` and exit + # without overwriting the caller's exception. + if poll_calls["n"] >= 2: + # Simulate the caller's timeout callback having fired + # between polls. + if not future.done(): + future.set_exception( + salt.exceptions.SaltReqTimeoutError("Message timed out") + ) + return 0 + + try: + client.socket = AsyncMock() + client.socket.poll.side_effect = never_ready + + client._queue.put_nowait((future, b"payload")) + client._queue.put_nowait((None, None)) + + # Prevent _reconnect from actually rebuilding sockets during test. + with patch.object( + client, "_reconnect", new=AsyncMock(return_value=None) + ) as reconnect_mock: + await client._send_recv(client.socket, client._queue) + + assert future.done() + exc = future.exception() + assert isinstance(exc, salt.exceptions.SaltReqTimeoutError) + # The caller's "Message timed out" exception must NOT be + # overwritten by the "Socket not ready for sending" exception + # that the pre-fix code raised on the very first POLLOUT miss. + assert "Message timed out" in str(exc) + assert "Socket not ready for sending" not in str(exc) + # send() must never have been attempted because POLLOUT never + # reported ready. + client.socket.send.assert_not_called() + # We should reconnect after abandoning the send so the next + # request starts from a clean socket. + reconnect_mock.assert_awaited() + finally: + client.close() + + def test_async_req_message_client_close_never_connected(minion_opts): """ close() must not hang when connect() was never called (#68637).