From e2d8fbb758779c1f5dc07948626bef39a354ed60 Mon Sep 17 00:00:00 2001 From: ncttjz Date: Thu, 16 Jul 2026 14:28:17 +0700 Subject: [PATCH 1/2] fix: poll POLLOUT in a loop instead of aborting after single 300ms miss In RequestClient._send_recv, a single 300ms socket.poll(POLLOUT) miss was treated as an immediate SaltReqTimeoutError, ignoring the caller's configured request timeout (return_retry_timer / request_channel_timeout). Mirror the POLLIN loop pattern already used later in the same method: poll in a loop, only abort when the future completes (timeout fired). Fixes #69802 --- salt/transport/zeromq.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/salt/transport/zeromq.py b/salt/transport/zeromq.py index 84046009b70..3599256c9dc 100644 --- a/salt/transport/zeromq.py +++ b/salt/transport/zeromq.py @@ -2154,8 +2154,17 @@ async def _send_recv( break try: - # Wait for socket to be ready for sending - if not await socket.poll(300, zmq.POLLOUT): + # Wait for socket to be ready for sending. + # Poll in a loop so a single 300ms POLLOUT miss does not + # abort the send before the configured request timeout fires. + sent = False + while not future.done(): + if await socket.poll(300, zmq.POLLOUT): + await socket.send(message) + sent = True + break + if not sent: + # The future was completed by _timeout_message or cancel if not future.done(): future.set_exception( SaltReqTimeoutError("Socket not ready for sending") @@ -2163,8 +2172,6 @@ async def _send_recv( if not self._closing: await self._reconnect() break - - await socket.send(message) except (zmq.eventloop.future.CancelledError, asyncio.CancelledError) as exc: send_recv_running = False if not future.done(): From 0c42191f8fbbb0663e4d709731d93bd80ffe3536 Mon Sep 17 00:00:00 2001 From: ncttjz Date: Fri, 17 Jul 2026 11:33:58 +0700 Subject: [PATCH 2/2] Address review: remove unreachable code, add changelog, add POLLOUT retry test - Remove unreachable lines 2168-2171 (future already done per review) - Add changelog/69816.fixed.md - Add test_request_client_pollout_retries_on_transient_failure verifying that consecutive POLLOUT misses are retried --- changelog/69816.fixed.md | 1 + salt/transport/zeromq.py | 4 -- .../unit/transport/test_zeromq_concurrency.py | 67 +++++++++++++++++++ 3 files changed, 68 insertions(+), 4 deletions(-) create mode 100644 changelog/69816.fixed.md diff --git a/changelog/69816.fixed.md b/changelog/69816.fixed.md new file mode 100644 index 00000000000..d1f6f91249f --- /dev/null +++ b/changelog/69816.fixed.md @@ -0,0 +1 @@ +Fix ZMQ POLLOUT retry loop in `salt/transport/zeromq.py` so that a single 300ms poll miss does not abort the send before the configured request timeout fires. diff --git a/salt/transport/zeromq.py b/salt/transport/zeromq.py index 3599256c9dc..88d8daed6da 100644 --- a/salt/transport/zeromq.py +++ b/salt/transport/zeromq.py @@ -2165,10 +2165,6 @@ async def _send_recv( break if not sent: # The future was completed by _timeout_message or cancel - if not future.done(): - future.set_exception( - SaltReqTimeoutError("Socket not ready for sending") - ) if not self._closing: await self._reconnect() break diff --git a/tests/pytests/unit/transport/test_zeromq_concurrency.py b/tests/pytests/unit/transport/test_zeromq_concurrency.py index 3a37d5264b5..f79943e2578 100644 --- a/tests/pytests/unit/transport/test_zeromq_concurrency.py +++ b/tests/pytests/unit/transport/test_zeromq_concurrency.py @@ -177,3 +177,70 @@ async def test_request_client_reconnect_task_safety(minion_opts, io_loop): # The old task should have exited cleanly. client.close() + + +async def test_request_client_pollout_retries_on_transient_failure( + minion_opts, io_loop +): + """ + Regression test for the POLLOUT retry loop in _send_recv. + + A single 300ms POLLOUT miss should not abort the send. Instead, the + polling loop re-tries ``socket.poll`` so long as the request future is + still pending. If the socket eventually becomes writable within the + request timeout, the message goes through. + + This test sets up a mock socket that returns ``False`` from ``poll`` + for the first N calls (simulating transient ZMQ POLLOUT misses) and + then returns ``True``, confirming that the retry loop keeps trying + rather than falling through to the timeout/reconnect path. + """ + client = salt.transport.zeromq.RequestClient(minion_opts, io_loop) + + poll_attempts = [] + + async def mocked_poll(timeout, flag=None, **kwargs): + poll_attempts.append(True) + # Return False (not ready) for the first 3 calls, then True + return len(poll_attempts) > 3 + + sent_payloads = [] + + async def mocked_send(msg, **kwargs): + sent_payloads.append(msg) + + async def mocked_recv(**kwargs): + return salt.payload.dumps({"ret": "ok"}) + + mock_socket = AsyncMock() + mock_socket.poll = mocked_poll + mock_socket.send = mocked_send + mock_socket.recv = mocked_recv + + await client.connect() + + if client.socket: + client.socket.close() + client.socket = mock_socket + if client.send_recv_task: + client.send_recv_task.cancel() + + client.send_recv_task = asyncio.create_task( + client._send_recv(mock_socket, client._queue, task_id=client.send_recv_task_id) + ) + + # Send a message with a generous timeout + result = await client.send({"cmd": "test"}, timeout=30) + + client._closing = True + if client.send_recv_task and not client.send_recv_task.done(): + client.send_recv_task.cancel() + + # We should have polled at least 4 times (3 failures + 1 success) + assert len(poll_attempts) >= 4, ( + f"Expected at least 4 POLLOUT attempts, got {len(poll_attempts)}. " + "The retry loop did not re-poll on transient failures." + ) + # The message should have been sent on the wire + assert len(sent_payloads) == 1, "Message was never sent despite socket becoming ready." + assert result == {"ret": "ok"}, "Unexpected response payload."