estuary-cdk: serialize token fetches

Concurrent tasks can race to refresh an expiring token and can result in
401 responses depending on the source system's behavior. Each task
independently calls the OAuth2 token endpoint to obtain a new token
without any coordination between them. Some providers revoke the
previous token upon issuing a new one, so a task that received the first
new token finds it already revoked by the time it makes an API request
and can result in a `401` response that crashes the connector.

Fix by adding an asyncio.Lock to TokenSource.fetch_token(). The lock
serializes access to the check-and-refresh logic in the new _fetch_token
method. In the common case (cached token still valid), no await occurs
while the lock is held, so there is no actual contention. This _does_
serialize access for connectors that have non-expiring tokens too, but
those code paths contain no await points, so they were already
effectively atomic and the lock adds no meaningful overhead.

Author: Alex-Bair

estuary-cdk/estuary_cdk/http.py

@@ -3,7 +3,7 @@
 import base64
 import json
 import time
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from logging import Logger
 from typing import Any, AsyncGenerator, Awaitable, Callable, Protocol, TypeVar
 
@@ -277,8 +277,18 @@ class AccessTokenResponse(BaseModel):
     google_spec: GoogleServiceAccountSpec | None = None
     _access_token: AccessTokenResponse | GoogleServiceAccountCredentials | None = None
     _fetched_at: int = 0
+    _lock: asyncio.Lock = field(default_factory=asyncio.Lock)
 
     async def fetch_token(self, log: Logger, session: HTTPSession) -> tuple[str, str]:
+        # Serialize token fetches so that concurrent tasks don't race to refresh
+        # an expiring token. Without this, multiple tasks can each request a new
+        # token from the OAuth2 provider. Some providers revoke the previous token
+        # when a new one is issued, causing in-flight requests that used the old
+        # token to fail.
+        async with self._lock:
+            return await self._fetch_token(log, session)
+
+    async def _fetch_token(self, log: Logger, session: HTTPSession) -> tuple[str, str]:
         if isinstance(
             self.credentials,
             (