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