From 994fe3607f2064d206d9df8d0eae35df567cf613 Mon Sep 17 00:00:00 2001 From: "Sam.Richards@taurich.org" Date: Thu, 27 Aug 2026 17:52:21 +0100 Subject: [PATCH] Return a response that arrived instead of reporting a timeout _dequeue_messages files every response it dequeues into self.responses, whatever request it was watching for - only its break is specific to watch_for. response() then let TimeoutError propagate without consulting self.responses, so when any other consumer of the connection's queue took our response first, the answer was stored correctly and the caller was told the request timed out. This is not a rare race. Connection(background_processing=True) starts exactly such a consumer via process_events_forever, and with one running, 20 of 20 bounded reads raised TimeoutError with the answer already recorded. It also fires with no threads at all: a request issued from inside a broadcast callback re-enters the same pump on the calling thread, so an outer frame can file the answer while the inner call concludes it timed out. It matters beyond a spurious exception, because callers treat TimeoutError as evidence that an actor is unresponsive and act on it - dropping cached handles, re-acquiring, marking a peer unhealthy. Re-check self.responses before propagating. Kept at the response() level rather than in _dequeue_messages, whose break condition is correct: the loop's job is to pump until it sees what it was told to watch for. What was wrong is concluding "no answer" from "I did not dequeue it myself". A genuine timeout, where no response was ever recorded, still raises. Signed-off-by: Sam.Richards@taurich.org --- python/src/xstudio/connection/__init__.py | 14 +++++++++- python/test/test_connection.py | 33 +++++++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) create mode 100644 python/test/test_connection.py diff --git a/python/src/xstudio/connection/__init__.py b/python/src/xstudio/connection/__init__.py index 8ebc9af7a..d3d269b3e 100644 --- a/python/src/xstudio/connection/__init__.py +++ b/python/src/xstudio/connection/__init__.py @@ -401,7 +401,19 @@ def request(self, *args): def response(self, req_id, timeout_milli=None): """""" if timeout_milli is not None: - self._dequeue_messages(timeout_milli, req_id) + try: + self._dequeue_messages(timeout_milli, req_id) + except TimeoutError: + # _dequeue_messages files every response it dequeues into + # self.responses, whatever request it was watching for - only + # its break is specific to watch_for. So another consumer of + # this connection's queue may have taken our response and + # stored it correctly while we were still waiting. Not having + # dequeued it ourselves is not the same as no answer arriving, + # and callers treat a timeout as evidence an actor is + # unresponsive. + if self.responses.get(req_id) is None: + raise return self._response(req_id) diff --git a/python/test/test_connection.py b/python/test/test_connection.py new file mode 100644 index 000000000..c8107c40c --- /dev/null +++ b/python/test/test_connection.py @@ -0,0 +1,33 @@ +# SPDX-License-Identifier: Apache-2.0 +import pytest +from xstudio.core import version_atom + + +def test_response_already_received_is_not_a_timeout(spawn): + """A response dequeued by another consumer must be returned, not reported + as a timeout. + + _dequeue_messages files every response it dequeues into self.responses, + whatever request it was watching for - only its break is specific to + watch_for. So any other consumer of the queue can pull our response and + store it correctly while we are still waiting for it. + + dequeue_messages() here is that other consumer. Calling it on this thread + rather than another removes the race without changing the mechanism: it + pumps with watch_for unset, so it files the response and carries on instead + of breaking on it. + """ + req_id = spawn.request(spawn.remote(), version_atom()) + spawn.dequeue_messages(300) + + assert spawn.responses[req_id] is not None, "precondition: the answer was recorded" + assert spawn.response(req_id, 300) is not None + + +def test_genuine_timeout_still_raises(spawn): + """No response was ever recorded for this id, so it must still raise.""" + unused_req_id = 0x7FFFFFF0 + assert unused_req_id not in spawn.responses + + with pytest.raises(TimeoutError): + spawn.response(unused_req_id, 300)