From 23d7a7a29fb8b2f8d7c3c04e687511950b3d793d Mon Sep 17 00:00:00 2001 From: huniu20 Date: Fri, 23 Oct 2026 15:33:22 +0800 Subject: [PATCH] [fix] Bound controller RPC send/recv so a lost reply cannot hang the caller ## Problem The client's controller RPC pool is created without a timeout, so every controller request (`get_meta`, `kv_retrieve_meta`, `kv_list`, ...) waits on `recv` forever. A DEALER silently drops a reply it can no longer route, so one lost response blocks the caller indefinitely with no error. Storage RPCs are already bounded by `TQ_SIMPLE_STORAGE_SEND_RECV_TIMEOUT`. ## Change - Add `TQ_CONTROLLER_SEND_RECV_TIMEOUT` (seconds, default 200, matching `TQ_SIMPLE_STORAGE_SEND_RECV_TIMEOUT`) and pass it to the client's `controller_rpc_pool`. - On timeout the call raises (`RuntimeError` wrapping `zmq.Again`); the pool already discards a socket whose lease raised, so a late reply cannot be read by the next request. ## Validation - New `test_controller_rpc_raises_when_controller_never_replies`: a ROUTER that never answers makes `get_partition_list` raise within the timeout. - `python -m pytest -q tests --ignore=tests/e2e -k "not yuanrong and not mooncake"`: 612 passed, 3 skipped (baseline 611 passed; the extra one is the new test). - `python -m compileall -q transfer_queue tutorial tests` and ruff pass. Signed-off-by: huniu20 --- tests/test_client.py | 21 +++++++++++++++++++++ transfer_queue/client.py | 4 ++++ 2 files changed, 25 insertions(+) diff --git a/tests/test_client.py b/tests/test_client.py index 0fd98761..e2750613 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1355,6 +1355,27 @@ def test_controller_rpc_reuses_one_socket(client_setup): assert len(mock_controller.caller_identities) == 1, "each request opened its own socket" +def test_controller_rpc_raises_when_controller_never_replies(): + """A lost controller reply must surface as an error, not block the caller forever.""" + context = zmq.Context() + silent_controller = context.socket(zmq.ROUTER) + port = silent_controller.bind_to_random_port("tcp://127.0.0.1") + controller_info = ZMQServerInfo( + role=Role.CONTROLLER, id="silent_controller", ip="127.0.0.1", ports={"request_handle_socket": port} + ) + with patch("transfer_queue.client.TQ_CONTROLLER_SEND_RECV_TIMEOUT", 1): + client = TransferQueueClient(client_id="client_timeout", controller_info=controller_info) + try: + start = time.monotonic() + with pytest.raises(RuntimeError, match="Resource temporarily unavailable"): + client.get_partition_list() + assert time.monotonic() - start < 10 + finally: + client.close() + silent_controller.close(linger=0) + context.term() + + # ===================================================== # Checkpoint Interface Tests # ===================================================== diff --git a/transfer_queue/client.py b/transfer_queue/client.py index 73f458f4..93367243 100644 --- a/transfer_queue/client.py +++ b/transfer_queue/client.py @@ -48,6 +48,9 @@ # Raising it also needs enough file descriptors (``ulimit -n``). TQ_CLIENT_ZMQ_MAX_SOCKETS = os.environ.get("TQ_CLIENT_ZMQ_MAX_SOCKETS") or None DEFAULT_CLIENT_ZMQ_MAX_SOCKETS = 8192 +# A DEALER drops a reply it can no longer route, so without a bound a lost response blocks +# the caller forever. Generous: a busy controller is normal, an unanswered request is not. +TQ_CONTROLLER_SEND_RECV_TIMEOUT = int(os.environ.get("TQ_CONTROLLER_SEND_RECV_TIMEOUT", 200)) # seconds # Pre-bound decorator for controller socket operations. with_controller_socket = with_zmq_socket( @@ -156,6 +159,7 @@ def __init__( self.zmq_context, client_id, "request_handle_socket", + timeout=TQ_CONTROLLER_SEND_RECV_TIMEOUT, ) # Backstop for a client that is never closed, so the context and its I/O threads do