OPS-5449: One async client per event loop — fix cross-loop 'bound to a different event loop' (3.8.7)

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: <asyncio.locks.Event ...> 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.

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
+        # "<asyncio.locks.Event ...> 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"""

pygqlc/__version__.py

@@ -1 +1 @@
-__version__ = "3.8.6"
+__version__ = "3.8.7"

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,22 +125,22 @@ 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
 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

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: <asyncio.locks.Event object at 0x...> 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()