InventoryAllocation

Rotating OAuth2 Tokens Without Dropping In-Flight Requests

When an access token expires in the middle of a burst of concurrent channel-manager calls, the naive fix — every coroutine that sees a 401 refreshes the token — produces a thundering herd of refresh calls, and the last one to write wins while every other in-flight request carries a token that was just invalidated. This page shows how to rotate the token exactly once, let the other coroutines wait for that single refresh, and keep both the old and new token accepted for a short grace window so nothing already on the wire is dropped. It is the concurrency-hardened companion to the broader OAuth2 token refresh strategies that govern how a sync worker stays authenticated.

The audience is the engineer who runs a httpx.AsyncClient with a pool of coroutines pushing rate and availability updates to booking_com and expedia at once. A single expiry event should cost one refresh round-trip and zero failed pushes — not a cascade of invalid_token responses that a revenue manager later sees as a gap in a channel’s rate calendar.

Prerequisites & environment

The pattern below lives in the auth layer that sits under every outbound request, so pin these so the lock semantics and clock math behave identically across every worker:

The one hard requirement the provider must document is the grace window: how long, if at all, a rotated token stays valid after a refresh. Many OTAs keep the previous token live for 30–120 seconds; that overlap is what makes zero-drop rotation possible, and Step 4 explains what to do when it is zero.

Single-flight token refresh under an asyncio lock with a dual-token grace window Three concurrent coroutines A, B and C all observe an expired access token at nearly the same moment. Coroutine A wins the asyncio lock and performs the one network refresh; coroutines B and C await the same lock and, on entry, re-read the now-fresh cached token instead of refreshing again. During the refresh, requests already dispatched with the old token still succeed because the provider keeps the previous token valid for a grace window, so no in-flight request is dropped. After the window closes only the new token is accepted. Single-flight refresh: one round-trip, no dropped requests Three coroutines see the same expiry; only the lock holder refreshes, the rest await and re-read. coroutine A coroutine B coroutine C token expires asyncio.Lock A holds lock POST /oauth/token B awaits lock C awaits lock cached token new access_token + expires_at writes re-read, no refresh Grace window — old and new token both accepted, so requests already on the wire still succeed. old token still valid (30–120s overlap) only new token accepted
Figure 1: One coroutine refreshes under the lock; the others await and re-read the fresh token, while the provider's grace window keeps in-flight requests alive.

Step-by-step implementation

Four moving parts: a validated token envelope with a monotonic expiry check, a single-flight refresh guarded by an asyncio.Lock, an auth hook that reads the cache on every request, and a retry-once-on-401 wrapper that survives the narrow race the grace window cannot cover.

Step 1 — Model the token with a skew-aware expiry

Store the token as a validated envelope and compute a stale flag against time.monotonic(), not wall-clock time, so an NTP correction or a clock jump mid-sync can never make a live token look expired or an expired one look live.

python
import time
from pydantic import BaseModel, Field

class TokenEnvelope(BaseModel):
    access_token: str
    refresh_token: str | None = None
    # monotonic deadline computed at fetch time, immune to wall-clock jumps
    expires_at_monotonic: float
    skew_seconds: float = Field(default=30.0, ge=0)

    @classmethod
    def from_response(cls, body: dict, skew: float = 30.0) -> "TokenEnvelope":
        return cls(
            access_token=body["access_token"],
            refresh_token=body.get("refresh_token"),
            expires_at_monotonic=time.monotonic() + float(body["expires_in"]),
            skew_seconds=skew,
        )

    def is_stale(self) -> bool:
        # refresh 'skew' seconds early so we rotate before the OTA rejects us
        return time.monotonic() >= (self.expires_at_monotonic - self.skew_seconds)

Using time.monotonic() rather than datetime.now() matters because these workers are long-lived: a wall-clock adjustment during a multi-hour sync would otherwise silently shift every expiry deadline, and the whole point of proactive rotation is that the deadline stays trustworthy.

Step 2 — Refresh exactly once under an asyncio lock

Wrap the refresh in a TokenProvider that holds an asyncio.Lock. The critical detail is the double-checked read: after acquiring the lock, re-inspect the cached token before refreshing, because by the time a waiting coroutine gets the lock, the winner has usually already written a fresh token.

python
import asyncio
import httpx
import structlog

log = structlog.get_logger()

class TokenProvider:
    def __init__(self, client: httpx.AsyncClient, token_url: str, creds: dict):
        self._client = client
        self._token_url = token_url
        self._creds = creds
        self._token: TokenEnvelope | None = None
        self._lock = asyncio.Lock()

    async def get_token(self) -> str:
        if self._token is not None and not self._token.is_stale():
            return self._token.access_token  # fast path: no lock, no await
        async with self._lock:
            # double-check: another coroutine may have refreshed while we waited
            if self._token is not None and not self._token.is_stale():
                return self._token.access_token
            resp = await self._client.post(self._token_url, data=self._creds)
            resp.raise_for_status()
            self._token = TokenEnvelope.from_response(resp.json())
            log.info("oauth_token_rotated", channel="booking_com",
                     expires_in=self._creds.get("expires_in"))
            return self._token.access_token

The double-checked read inside the lock is what collapses N concurrent expiries into one HTTP call: without it, every coroutine that queued on the lock would perform its own redundant refresh the instant it acquired the lock, recreating the thundering herd the lock was meant to prevent.

Step 3 — Inject the token through an auth hook, not per call site

Bind the provider to the client with an httpx.Auth subclass so every request reads the current token at send time. This guarantees a request dispatched a millisecond after a rotation picks up the new token, and no call site can accidentally cache a stale bearer string in a local variable.

python
class RotatingBearerAuth(httpx.Auth):
    requires_request_body = False

    def __init__(self, provider: TokenProvider):
        self._provider = provider

    async def async_auth_flow(self, request):
        token = await self._provider.get_token()
        request.headers["Authorization"] = f"Bearer {token}"
        yield request  # httpx sends it; response handled by the caller/retry wrapper

Reading the token inside async_auth_flow rather than passing a string into the client means the freshest cached token is stamped onto the request at the last possible moment, closing the window where a request built before a rotation would otherwise ship the old bearer.

Step 4 — Retry once on 401 to cover the grace-window edge

The grace window covers requests already on the wire, but there is still a hairline race: a request whose header was stamped microseconds before a rotation, on a provider with a zero grace window, can land a 401. Handle it by forcing one refresh and replaying the request exactly once — never in a loop, or a genuinely revoked credential becomes an infinite retry storm.

python
async def send_with_rotation(client: httpx.AsyncClient, provider: TokenProvider,
                             request: httpx.Request) -> httpx.Response:
    resp = await client.send(request)
    if resp.status_code == 401:
        provider._token = None          # force the next get_token() to refresh
        fresh = await provider.get_token()
        request.headers["Authorization"] = f"Bearer {fresh}"
        log.warning("oauth_401_retry", channel="expedia",
                    property_id=request.headers.get("X-Property-Id"))
        resp = await client.send(request)   # replay once, then surface the result
    return resp

Capping the replay at one attempt is deliberate: a 401 after a fresh forced refresh is no longer an expiry problem but a revoked or misscoped credential, and retrying that further would only hammer the OTA and delay the real signal to error categorization and retry logic.

Gotchas & production notes

Verification snippet

The one behaviour worth proving is single-flight: many concurrent callers hitting an expired cache must trigger exactly one network refresh. Stub the HTTP post with a counter and fan out.

python
import asyncio

async def test_single_flight_refresh() -> None:
    calls = {"n": 0}

    class FakeProvider(TokenProvider):
        async def _do_refresh(self) -> None:
            calls["n"] += 1
            await asyncio.sleep(0.05)   # simulate network latency
            self._token = TokenEnvelope(
                access_token=f"tok-{calls['n']}", refresh_token="r",
                expires_at_monotonic=time.monotonic() + 3600)

        async def get_token(self) -> str:
            if self._token is not None and not self._token.is_stale():
                return self._token.access_token
            async with self._lock:
                if self._token is not None and not self._token.is_stale():
                    return self._token.access_token
                await self._do_refresh()
                return self._token.access_token

    prov = FakeProvider.__new__(FakeProvider)
    prov._token = None
    prov._lock = asyncio.Lock()
    tokens = await asyncio.gather(*(prov.get_token() for _ in range(50)))
    assert calls["n"] == 1              # 50 callers, one refresh
    assert len(set(tokens)) == 1        # everyone got the same token

asyncio.run(test_single_flight_refresh())

A green run proves the two guarantees that matter under load: fifty coroutines that all observed an empty cache produced exactly one refresh, and every one of them walked away with the identical token — so a live expiry event costs one round-trip, and no concurrent push is stranded on a token that was rotated out from under it.

← Back to OAuth2 Token Refresh Strategies