Skip to content

[fix] Bound controller RPC send/recv so a lost reply cannot hang the caller #203

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
0oshowero0 merged 1 commit into from
Oct 23, 2026
Merged

[fix] Bound controller RPC send/recv so a lost reply cannot hang the caller #203

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions tests/test_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
# =====================================================
Expand Down
4 changes: 4 additions & 0 deletions transfer_queue/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -156,6 +159,7 @@ def __init__(
self.zmq_context,
client_id,
"request_handle_socket",
timeout=TQ_CONTROLLER_SEND_RECV_TIMEOUT,

@chatgpt-codex-connector chatgpt-codex-connector Bot Oct 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Translate controller timeouts in the shared request helper

When a controller is silent long enough to hit this new timeout, pyzmq raises zmq.error.Again from _request_controller; public methods such as async_get_meta and async_set_custom_meta call that helper without a catch, so their synchronous and asynchronous forms now leak Again instead of the documented RuntimeError. The new test only exercises get_partition_list, which has its own blanket wrapper, so translate Again once in _request_controller to give every controller RPC the same public failure contract.

Useful? React with 👍 / 👎.

)

# Backstop for a client that is never closed, so the context and its I/O threads do
Expand Down
Loading