`ConnectionSource` (and `AsyncConnectionPool` it uses) is not multi-event-loop safe

[aivarsk]

When multiple Subscribers/PoolingServices are being run with IsolationMode.DEDICATED_THREAD, the shared ConnectionSource (and AsyncConnectionPool it uses) may get this error:

Traceback (most recent call last):
 File "/srv/.poetry/xxx/lib/python3.13/site-packages/logicblocks/event/processing/broker/strategies/distributed/observer.py", line 63, in observe
 await self.synchronise()
 File "/srv/.poetry/xxx/lib/python3.13/site-packages/logicblocks/event/processing/broker/strategies/distributed/observer.py", line 80, in synchronise
 updated = await self._subscription_state_store.list()
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 File "/srv/.poetry/xxx/distributed/subscriptions/stores/state/postgres.py", line 181, in list
 async with self.connection_pool.connection() as connection:
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^
 File "/usr/local/lib/python3.13/contextlib.py", line 214, in __aenter__
 return await anext(self.gen)
 ^^^^^^^^^^^^^^^^^^^^^
 File "/srv/.poetry/xxx/lib/python3.13/site-packages/psycopg_pool/pool_async.py", line 195, in connection
 conn = await self.getconn(timeout=timeout)
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 File "/srv/.poetry/xxx/lib/python3.13/site-packages/psycopg_pool/pool_async.py", line 225, in getconn
 return await self._getconn_with_check_loop(deadline)
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 File "/srv/.poetry/xxx/lib/python3.13/site-packages/psycopg_pool/pool_async.py", line 238, in _getconn_with_check_loop
 conn = await self._getconn_unchecked(deadline - monotonic())
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 File "/srv/.poetry/xxx/lib/python3.13/site-packages/psycopg_pool/pool_async.py", line 263, in _getconn_unchecked
 async with self._lock:
 ^^^^^^^^^^
 File "/usr/local/lib/python3.13/asyncio/locks.py", line 14, in __aenter__
 await self.acquire()
 File "/usr/local/lib/python3.13/asyncio/locks.py", line 105, in acquire
 fut = self._get_loop().create_future()
 ~~~~~~~~~~~~~~^^
 File "/usr/local/lib/python3.13/asyncio/mixins.py", line 20, in _get_loop
 raise RuntimeError(f'{self!r} is bound to a different event loop')
RuntimeError: <asyncio.locks.Lock object at 0x7fe656974590 [unlocked, waiters:1]> is bound to a different event loop

After that, it gets stuck with the

task run <ShrinkPool 'pool-1' at 0x7fe656974c20> failed: RuntimeError: <asyncio.locks.Lock object at 0x7fe656974590 [unlocked, waiters:1]> is bound to a different event loop

and never recovers while the application itself is up and running.

[tobyclemson]

Hi @aivarsk,

Thanks for raising this issue. Could you share how you are creating your AsyncConnectionPool?

Thanks,
Toby

[aivarsk]

class PostgresConnectionSource(AsyncConnectionPool[AsyncConnection]):
    def __init__(self, settings: Settings, *args, **kwargs):
        super().__init__(settings.DB_URI, *args, **kwargs)
        self.settings = settings

connection_source = PostgresConnectionSource(
    settings,
    open=False,
    min_size=10,
    max_size=10,
)

[tobyclemson]

This appears to be an undocumented limitation of psycopg_pool: psycopg/psycopg#737.

In order to resolve this, I think you'd need to either switch to using a shared thread for all subscribers or inject a different connection pool for use by each subscriber that is on a dedicated event loop.

The latter could be done by switching away from make_subscriber and wiring up the EventSubscriptionConsumer yourself, as is done here:

event.store/src/logicblocks/event/processing/consumers/subscription.py

Line 59 in 7445745

def make_subscriber[E: Event](