From 45fb003ebf11f8f27ef821fff2caf0d15556da5f Mon Sep 17 00:00:00 2001 From: Ricardo Corral Date: Thu, 30 Jul 2026 19:26:31 -0600 Subject: [PATCH] fix: one async client per event loop, never shared across loops (3.8.7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 3.8.6 cached a single httpx.AsyncClient per GraphQLClient instance. The client's connection-pool primitives bind to the event loop that first awaits them, so consumers running coroutines on more than one loop — a Temporal worker's main loop plus a subscription callback thread using asyncio.run per event — failed with 'RuntimeError: is bound to a different event loop', and the message matched no retry predicate so it propagated. Intermittent: a fresh loop only trips when the pool still holds another loop's connections. The cache is now keyed by the running loop (weak keys — a GC'd loop drops its client). Per-loop reuse keeps 3.8.6's no-churn goal; _drop_async_client and async_cleanup act on the running loop's client; _close() schedules aclose() on each client's own loop; and 'is bound to a different event loop' joins the retryable-on-fresh-connection predicate as a defensive self-heal. Regression tests pin the per-loop contract end-to-end through async_execute. --- CHANGELOG.md | 4 + pygqlc/GraphQLClient.py | 68 ++++++++++----- pygqlc/__version__.py | 2 +- .../gql_client/test_async_client_lifecycle.py | 49 +++++++---- .../gql_client/test_async_client_per_loop.py | 86 +++++++++++++++++++ 5 files changed, 170 insertions(+), 39 deletions(-) create mode 100644 tests/pygqlc/gql_client/test_async_client_per_loop.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c5c7308..5f138f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # CHANGELOG +## [3.8.7] - 2026-07-30 + +- [Fixed] The cached async client is now PER EVENT LOOP. 3.8.6's shared client bound its httpx connection-pool primitives to whichever loop first used it; any consumer running coroutines on more than one loop — e.g. a Temporal worker's main loop plus a subscription callback thread using `asyncio.run` per event (valuechainos-queues' `trigger_by_subscription`) — then failed with `RuntimeError: is bound to a different event loop` (observed live as `Error processing workflow QUEUE_REPLENISHMENT_FOR_CSV_REPORT`). Each loop now gets (and reuses) its own client; per-loop reuse keeps 3.8.6's no-churn goal, `_close()` schedules `aclose()` on each client's own loop, and "is bound to a different event loop" joined the retryable-on-fresh-connection predicate as a defensive self-heal. + ## [3.8.6] - 2026-06-26 - [Fixed] `_get_async_client` reuses the shared client instead of recreating it every call. Its probe awaited a non-existent `get_timeout()`, so every call closed + rebuilt the client — under concurrency that closed it mid-use elsewhere (`Cannot send a request, as the client has been closed.`). Now rebuilt only when missing or `is_closed`, and that error is retryable. diff --git a/pygqlc/GraphQLClient.py b/pygqlc/GraphQLClient.py index 3450b6f..d705533 100644 --- a/pygqlc/GraphQLClient.py +++ b/pygqlc/GraphQLClient.py @@ -9,6 +9,7 @@ import asyncio import traceback +import weakref import time import threading from functools import lru_cache @@ -293,7 +294,12 @@ def __init__(self): # Reuse HTTP client for better performance self._http_client = None self._thread_local = threading.local() - self._async_client = None + # One async client PER EVENT LOOP: httpx.AsyncClient's pool primitives bind to + # the loop that first uses them, so a client shared across loops fails with + # " is bound to a different event loop" (e.g. a worker's + # main loop plus a subscription thread running asyncio.run per callback). + # Weak keys: a GC'd loop drops its client reference with it. + self._async_clients = weakref.WeakKeyDictionary() # Configure sleep time for polling loops self.poll_interval = 0.005 # reduced from 0.01 for faster response @@ -1121,24 +1127,34 @@ def execute(self, query: str, variables: dict | None = None) -> dict: # * ASYNC METHODS ---------------------------------- async def _get_async_client(self): - """Return the shared async client, rebuilding only when missing or closed. + """Return the RUNNING loop's async client, rebuilding only when missing or closed. - No per-call liveness probe — a dead event loop is recovered lazily by - async_execute's retry. + One client per event loop (never shared): the client's connection-pool + primitives bind to the loop that first awaits them, so reuse from another + loop raises "is bound to a different event loop". Within a loop the client + is reused as before. No per-call liveness probe — a dead event loop is + recovered lazily by async_execute's retry. """ - if self._async_client is None or self._async_client.is_closed: - self._async_client = httpx.AsyncClient(**self.async_client_params) - return self._async_client + loop = asyncio.get_running_loop() + client = self._async_clients.get(loop) + if client is None or client.is_closed: + client = httpx.AsyncClient(**self.async_client_params) + self._async_clients[loop] = client + return client async def _drop_async_client(self): - """Best-effort aclose() of the current async client before dropping it. + """Best-effort aclose() of the RUNNING loop's async client before dropping it. Closing may fail when the client's original event loop is gone; transports are then unavoidably left to GC, but every avoidable path closes promptly so socket finalizers don't pile up for the cyclic GC (TMPRL1101 fallout in Temporal workers — see valiot/python-tooling#151). """ - client, self._async_client = self._async_client, None + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return + client = self._async_clients.pop(loop, None) if client is None: return try: @@ -1152,7 +1168,13 @@ def _should_retry_on_fresh_connection(error: Exception) -> bool: """True when the connection is unusable but a fresh one should work: a closed/closed-down client or a transient transport error.""" msg = str(error) - if "Event loop is closed" in msg or "client has been closed" in msg: + if ( + "Event loop is closed" in msg + or "client has been closed" in msg + # Defensive: cannot arise with per-loop clients, but if stale cross-loop + # state ever surfaces, drop-and-rebuild self-heals instead of failing. + or "is bound to a different event loop" in msg + ): return True return isinstance(error, TRANSIENT_TRANSPORT_ERRORS) @@ -1326,17 +1348,21 @@ def _close(self): except Exception: # pylint: disable=broad-except pass - # For the async client we can't await here. If this thread has a running - # event loop (e.g. __del__ triggered by GC inside a loop thread), schedule - # aclose() on it; otherwise the transports are left to GC, as before. - # __del__ can run on any thread, hence call_soon_threadsafe. - if hasattr(self, "_async_client") and self._async_client is not None: - client, self._async_client = self._async_client, None - try: - loop = asyncio.get_running_loop() - loop.call_soon_threadsafe(lambda: loop.create_task(client.aclose())) - except Exception: # pylint: disable=broad-except - pass # no usable loop — GC fallback, as before + # For the async clients we can't await here. Schedule aclose() on each + # client's OWN loop when that loop is still open; a closed/gone loop's + # transports are left to GC, as before. __del__ can run on any thread, + # hence call_soon_threadsafe. + if hasattr(self, "_async_clients"): + clients = list(self._async_clients.items()) + self._async_clients.clear() + for loop, client in clients: + try: + if not loop.is_closed(): + loop.call_soon_threadsafe( + lambda l=loop, c=client: l.create_task(c.aclose()) + ) + except Exception: # pylint: disable=broad-except + pass # no usable loop — GC fallback, as before def __del__(self): """Cleanup resources when the instance is being destroyed""" diff --git a/pygqlc/__version__.py b/pygqlc/__version__.py index f64508e..08f7211 100644 --- a/pygqlc/__version__.py +++ b/pygqlc/__version__.py @@ -1 +1 @@ -__version__ = "3.8.6" +__version__ = "3.8.7" diff --git a/tests/pygqlc/gql_client/test_async_client_lifecycle.py b/tests/pygqlc/gql_client/test_async_client_lifecycle.py index 4f8849f..05e5d48 100644 --- a/tests/pygqlc/gql_client/test_async_client_lifecycle.py +++ b/tests/pygqlc/gql_client/test_async_client_lifecycle.py @@ -3,6 +3,9 @@ during cyclic-GC sweeps on arbitrary threads, which contributed to false TMPRL1101 deadlocks in Temporal workers (see valiot/python-tooling#151). +Clients are registered PER EVENT LOOP (see test_async_client_per_loop.py for +the cross-loop contract); these tests seed the running loop's slot directly. + Hermetic: no real sockets, no sleeps.""" import asyncio @@ -21,16 +24,25 @@ def client(): gql = GraphQLClient() gql.addEnvironment("lifecycle-test", url="http://ex", default=True) yield gql - gql._async_client = None # never let teardown touch a mock + gql._async_clients.clear() # never let teardown touch a mock Singleton._instances.pop(GraphQLClient, None) +def _seed(gql, mock_client): + """Register a mock client for the RUNNING loop (what _get_async_client keys by).""" + gql._async_clients[asyncio.get_running_loop()] = mock_client + + +def _current(gql): + return gql._async_clients.get(asyncio.get_running_loop()) + + @pytest.mark.asyncio async def test_get_async_client_reuses_live_client(client): """A live client is reused on every call — never closed and recreated.""" live = AsyncMock() live.is_closed = False - client._async_client = live + _seed(client, live) with patch("pygqlc.GraphQLClient.httpx.AsyncClient") as new_client: first = await client._get_async_client() @@ -46,7 +58,7 @@ async def test_get_async_client_replaces_closed_client(client): """A client that has been closed is replaced with a fresh one.""" closed = AsyncMock() closed.is_closed = True - client._async_client = closed + _seed(client, closed) fresh = AsyncMock() with patch("pygqlc.GraphQLClient.httpx.AsyncClient", return_value=fresh): @@ -62,7 +74,7 @@ async def test_async_execute_retry_closes_stale_client(client): stale = AsyncMock() stale.is_closed = False # live, so _get_async_client returns it (then .post fails) stale.post.side_effect = RuntimeError("Event loop is closed") - client._async_client = stale + _seed(client, stale) response = MagicMock(status_code=200, content=b'{"data": {"ok": true}}') fresh = AsyncMock() @@ -77,14 +89,14 @@ async def test_async_execute_retry_closes_stale_client(client): @pytest.mark.asyncio -async def test_close_schedules_aclose_on_running_loop(client): - """_close() (sync, called by __del__) must schedule aclose() on the running - loop instead of dropping the client to GC.""" +async def test_close_schedules_aclose_on_client_loop(client): + """_close() (sync, called by __del__) must schedule aclose() on the client's + own loop instead of dropping the client to GC.""" stale = AsyncMock() - client._async_client = stale + _seed(client, stale) client._close() - assert client._async_client is None + assert _current(client) is None # Let the call_soon_threadsafe callback and the task it creates run. await asyncio.sleep(0) @@ -93,14 +105,17 @@ async def test_close_schedules_aclose_on_running_loop(client): stale.aclose.assert_awaited_once() -def test_close_without_running_loop_falls_back_to_gc(client): - """_close() outside any event loop must not raise — GC fallback as before.""" +def test_close_with_closed_origin_loop_falls_back_to_gc(client): + """_close() must not raise when a client's origin loop is already closed — + its transports are left to GC, as before.""" + dead_loop = asyncio.new_event_loop() + dead_loop.close() stale = AsyncMock() - client._async_client = stale + client._async_clients[dead_loop] = stale client._close() - assert client._async_client is None + assert client._async_clients.get(dead_loop) is None stale.aclose.assert_not_awaited() @@ -110,12 +125,12 @@ async def test_drop_async_client_swallows_aclose_errors(client): still dropped.""" stale = AsyncMock() stale.aclose.side_effect = RuntimeError("Event loop is closed") - client._async_client = stale + _seed(client, stale) await client._drop_async_client() stale.aclose.assert_awaited_once() - assert client._async_client is None + assert _current(client) is None @pytest.mark.asyncio @@ -123,9 +138,9 @@ async def test_async_cleanup_closes_client(client): """async_cleanup must actually aclose() a live client (previously the broken get_timeout probe sent every client down the 'let GC handle it' branch).""" stale = AsyncMock() - client._async_client = stale + _seed(client, stale) await client.async_cleanup() stale.aclose.assert_awaited_once() - assert client._async_client is None + assert _current(client) is None diff --git a/tests/pygqlc/gql_client/test_async_client_per_loop.py b/tests/pygqlc/gql_client/test_async_client_per_loop.py new file mode 100644 index 0000000..6b3097c --- /dev/null +++ b/tests/pygqlc/gql_client/test_async_client_per_loop.py @@ -0,0 +1,86 @@ +"""Regression tests: the async client must be per-event-loop, never shared. + +httpx.AsyncClient's connection-pool primitives (asyncio Events/Locks inside +httpcore) bind to the event loop that first awaits them. 3.8.6 cached ONE +client per GraphQLClient instance, so a consumer running coroutines on more +than one loop — e.g. a Temporal worker's main loop plus a pygqlc subscription +thread calling ``asyncio.run(...)`` per callback (valuechainos-queues' +``trigger_by_subscription``) — failed with:: + + RuntimeError: is bound to a different event loop + +observed live as ``Error processing workflow QUEUE_REPLENISHMENT_FOR_CSV_REPORT``. + +Hermetic: no real sockets, no sleeps.""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from pygqlc import GraphQLClient +from pygqlc.helper_modules.Singleton import Singleton + + +@pytest.fixture +def client(): + """A fresh GraphQLClient (bypassing the process-wide singleton cache).""" + Singleton._instances.pop(GraphQLClient, None) + gql = GraphQLClient() + gql.addEnvironment("per-loop-test", url="http://ex", default=True) + yield gql + gql._async_clients.clear() + Singleton._instances.pop(GraphQLClient, None) + + +def test_each_loop_gets_its_own_client(client): + """Two sequential asyncio.run loops must NOT share an httpx.AsyncClient.""" + seen = [] + + async def grab(): + seen.append(await client._get_async_client()) + + with patch("pygqlc.GraphQLClient.httpx.AsyncClient", side_effect=lambda **_: AsyncMock(is_closed=False)): + asyncio.run(grab()) + asyncio.run(grab()) + + assert len(seen) == 2 + assert seen[0] is not seen[1], "client leaked across event loops" + + +def test_same_loop_reuses_its_client(client): + """Within one loop the client is still cached — no per-call churn (the 3.8.6 goal).""" + seen = [] + + async def grab_twice(): + seen.append(await client._get_async_client()) + seen.append(await client._get_async_client()) + + with patch( + "pygqlc.GraphQLClient.httpx.AsyncClient", side_effect=lambda **_: AsyncMock(is_closed=False) + ) as ctor: + asyncio.run(grab_twice()) + + assert seen[0] is seen[1] + assert ctor.call_count == 1 + + +def test_async_execute_across_loops_posts_on_each_loops_client(client): + """The field failure, pinned end-to-end: async_execute from two different + loops must post on two different clients — never the first loop's.""" + response = MagicMock(status_code=200, content=b'{"data": {"ok": true}}') + created = [] + + def make_client(**_): + mock = AsyncMock(is_closed=False) + mock.post.return_value = response + created.append(mock) + return mock + + with patch("pygqlc.GraphQLClient.httpx.AsyncClient", side_effect=make_client) as ctor: + asyncio.run(client.async_execute("query { ok }")) + asyncio.run(client.async_execute("query { ok }")) + + assert ctor.call_count == 2, "second loop must build its own client" + created[0].post.assert_awaited_once() + created[1].post.assert_awaited_once()