diff --git a/changelog/69816.fixed.md b/changelog/69816.fixed.md new file mode 100644 index 000000000000..d1f6f91249fd --- /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 84046009b70e..88d8daed6da3 100644 --- a/salt/transport/zeromq.py +++ b/salt/transport/zeromq.py @@ -2154,17 +2154,20 @@ 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. + # 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 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(): diff --git a/tests/pytests/unit/transport/test_zeromq_concurrency.py b/tests/pytests/unit/transport/test_zeromq_concurrency.py index 3a37d5264b57..f79943e2578e 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."