oauth: OAuthRefreshTokenSource caches its asyncio.Lock across event loops, breaking reuse in a new loop

[pcbeingused333]

Description

OAuthRefreshTokenSource.resolve_async creates its asyncio.Lock lazily and then keeps it for the lifetime of the source:

# integrations/oauth/src/haystack_integrations/utils/oauth/sources.py
if self._async_lock is None:
    self._async_lock = asyncio.Lock()
async with self._async_lock:

An asyncio.Lock binds itself to the loop that first awaits it under contention and raises RuntimeError when awaited from any other one. A source is a plain, serializable object that routinely outlives the loop that first used it — one asyncio.run(...) per request is a common deployment, and it is also how a lot of test code is written — so the second loop's first contended refresh fails.

Worth stressing that contention is not an edge case here. Collapsing a burst of concurrent callers into a single network refresh is the only reason the lock exists, and the class docstring says so.

The lazy creation itself is not a data race within one loop: there is no await between the is None check and the assignment, so a single event loop cannot interleave there. Two threads each driving their own loop can, but the loop-binding failure below is the one that reproduces deterministically.

Reproduction

import asyncio
from unittest.mock import patch

import httpx
from haystack.utils import Secret
from haystack_integrations.utils.oauth import OAuthRefreshTokenSource

source = OAuthRefreshTokenSource(
    token_url="https://idp.example.com/oauth2/token",
    client_id="client-1",
    refresh_token=Secret.from_token("rt"),
)


async def slow_post(*_args, **_kwargs):
    # Yield to the loop so the second caller reaches the lock while the first holds it.
    await asyncio.sleep(0)
    return httpx.Response(200, json={"access_token": "acc", "expires_in": 0})


async def two_concurrent_callers():
    with patch("httpx.AsyncClient.post", side_effect=slow_post):
        return await asyncio.gather(source.resolve_async(), source.resolve_async())


print(asyncio.run(two_concurrent_callers()))  # ['acc', 'acc']
print(asyncio.run(two_concurrent_callers()))  # RuntimeError

RuntimeError: <asyncio.locks.Lock object at 0x...> is bound to a different event loop

Reproduced on Python 3.12.3, oauth-haystack at main.

Note that dropping the asyncio.sleep(0) makes the script pass: an uncontended Lock.acquire() returns before it ever calls _get_loop(), so the loop is never bound. That is why the existing test_resolve_async_success_and_caches does not catch this — it awaits one caller at a time.

Expected Behavior

A source reused from a new event loop should refresh normally. The lock is per-loop runtime state, so it should be rebuilt when the running loop changes rather than cached for the lifetime of the object.

Additional Context

OAuthTokenExchangeSource is not affected: it uses a threading.Lock held only around cache access and never across the network call, which its own comment calls out.

I have a fix and a regression test ready and will open a PR referencing this issue.

---

Disclosure: I used AI assistance while investigating and writing this up. I ran the reproduction and confirmed the failure and the root cause myself.