InventoryAllocation

OAuth2 Token Refresh Strategies for Hotel Rate Parity Automation

Uninterrupted API connectivity between property management systems and channel managers directly dictates rate parity compliance and inventory accuracy. When an OAuth2 access token expires mid-sync, rate pushes fail silently, parity violations cascade across online travel agencies, and revenue managers absorb the financial impact days later when a competitor undercuts a rate the property never intended to publish. Token refresh is not an authentication footnote — it is the operational backbone of reliable API sync and data ingestion workflows, and the failure mode it prevents (a half-completed rate push abandoned at a 401) is one of the most common root causes of drift that ops teams cannot explain after the fact.

This page specifies a deterministic refresh design for Python automation engineers: validate before you dispatch, rotate atomically under a lock, coordinate with OTA rate limits so the token endpoint does not become the thing that gets you throttled, and prove correctness with log assertions rather than hope. The audience is the engineer who owns the sync worker and the revenue manager who has to trust its output.

Architecture and prerequisites

The refresh subsystem sits between your credential store and every outbound mutation. Treat authentication as a pre-flight gate, never an inline fallback: no rate or availability write leaves the process without a credential whose remaining lifetime clears a safety margin. This mirrors the boundary design in security and authentication boundaries and the initial grant flow covered in implementing OAuth2 for PMS API access.

Atomic OAuth2 refresh under a distributed lock with two concurrent workers A sequence diagram across four lifelines: Sync worker A, Sync worker B, Redis (credential cache plus lock), and the provider token endpoint. Worker A's pre-flight check finds the access token inside its 15% refresh window and issues SET lock NX to Redis, which succeeds and is acquired. Worker B also needs a token and issues SET lock NX, but Redis denies it with refresh_in_progress because A holds the lock. While the lock is held, worker A POSTs the refresh_token grant to the token endpoint, receives a new access token and a rotated refresh token, HSETs the rotated pair into the shared Redis cache, then DELetes the lock to release it. Only after release does worker B re-read the cache and obtain the freshly rotated token without performing a second refresh. Both workers then dispatch the rate push with the same valid credential, and exactly one refresh occurred. Sync worker A TTL < 15% → refresh Sync worker B concurrent, needs token Redis credential cache + lock Token endpoint refresh_token grant 1 · SET lock NX → acquired 2 · SET lock NX denied: refresh_in_progress 3 · POST /oauth/token (refresh_token grant) access_token + rotated refresh_token 4 · HSET creds — persist rotated pair 5 · DEL lock — release 6 · re-read cache fresh token — no second refresh While A holds the lock, B's SET NX is denied — it waits, then re-reads the freshly cached token. Both workers dispatch the rate push with one shared credential — exactly one refresh occurred.

Inputs, outputs and assumptions for the reference implementation:

The one non-negotiable prerequisite is a shared, atomic credential store. Multiple sync workers will race to refresh the same expiring token; without a lock they submit the same one-use refresh token twice and the second call returns invalid_grant, permanently invalidating the chain until a human re-authorizes the property.

Implementation

Step 1 — Decide whether a refresh is even needed

Inspect the exp claim before every dispatch. If the remaining validity falls below a configurable slice of total lifespan — 15% is a sane default — refresh proactively rather than discovering expiry mid-request.

python
import time
import jwt  # PyJWT
import structlog

log = structlog.get_logger()

def should_refresh(access_token: str, property_id: str, threshold_pct: float = 0.15) -> bool:
    """Return True when the token is inside its refresh window (or unreadable)."""
    try:
        # Signature is verified by the resource server, not us — we only read exp/iat.
        claims = jwt.decode(access_token, options={"verify_signature": False})
    except jwt.InvalidTokenError:
        log.warning("token_unreadable", property_id=property_id)
        return True  # fail toward refreshing rather than dispatching a bad token

    exp = claims.get("exp")
    if exp is None:
        return True  # opaque token with no exp: treat as always-stale, see FAQ
    remaining = exp - time.time()
    lifespan = exp - claims.get("iat", exp - 300)
    threshold_met = remaining < (lifespan * threshold_pct)
    log.info("token_pre_flight_check", property_id=property_id,
             remaining_ttl=round(remaining, 1), threshold_met=threshold_met)
    return threshold_met

Reading exp/iat without signature verification is deliberate: the resource server validates the signature on every call, so re-verifying locally only adds a key-fetch dependency without improving safety.

Step 2 — Refresh atomically under a distributed lock

Wrap the network call and the credential write in a single critical section so concurrent workers cannot double-spend a single-use refresh token.

python
import redis.asyncio as redis
from contextlib import asynccontextmanager

@asynccontextmanager
async def refresh_lock(rds: redis.Redis, property_id: str, channel: str, ttl: int = 30):
    """Hold a per-property/channel lock for the duration of a refresh."""
    key = f"oauth:refresh_lock:{property_id}:{channel}"
    token_value = str(time.time())  # unique holder id for safe release
    acquired = await rds.set(key, token_value, nx=True, ex=ttl)
    if not acquired:
        raise RuntimeError("refresh_in_progress")  # caller re-reads cache instead
    try:
        yield
    finally:
        # Only delete the lock if we still own it (avoid clobbering a re-acquire).
        if await rds.get(key) == token_value:
            await rds.delete(key)

The holder-id check on release prevents a slow worker whose lock has already expired from deleting a second worker’s freshly acquired lock — the classic lock-stealing bug that reintroduces the race you were trying to close.

Step 3 — Exchange the refresh token and persist the rotated pair

On a RuntimeError("refresh_in_progress") the caller should not refresh — it should wait briefly and re-read the cache, because another worker is already rotating the credential. When it does hold the lock, it exchanges the grant, backing off on transient throttling.

python
import asyncio, random
from httpx import AsyncClient, HTTPStatusError

async def refresh_access_token(client: AsyncClient, rds: redis.Redis,
                               property_id: str, channel: str,
                               refresh_token: str, max_retries: int = 4) -> dict:
    base_delay = 1.0
    for attempt in range(max_retries):
        try:
            resp = await client.post("/oauth/token", data={
                "grant_type": "refresh_token",
                "refresh_token": refresh_token,
            })
            resp.raise_for_status()
        except HTTPStatusError as exc:
            code = exc.response.status_code
            if code in (429, 503):  # transient: back off with jitter, then retry
                delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
                log.warning("token_endpoint_throttled", property_id=property_id,
                            status=code, attempt=attempt, delay=round(delay, 2))
                await asyncio.sleep(delay)
                continue
            raise  # 400/invalid_grant is terminal — do not retry (see Error handling)
        body = resp.json()
        # Persist the ROTATED refresh token before returning, per RFC 6749 §6.
        await rds.hset(f"oauth:creds:{property_id}:{channel}", mapping={
            "access_token": body["access_token"],
            "refresh_token": body.get("refresh_token", refresh_token),
            "exp": int(time.time()) + body["expires_in"],
        })
        log.info("token_rotation_complete", property_id=property_id, channel=channel,
                 new_exp=int(time.time()) + body["expires_in"])
        return body
    raise TimeoutError("token_refresh_exhausted_retry_budget")

Writing the rotated refresh_token to the store before the function returns is the load-bearing line: if the process crashes after a successful exchange but before persisting, the old one-use token is already dead and the property is locked out.

Step 4 — The read-through accessor every worker calls

Steps 1–3 are building blocks; no rate push should call them directly. Instead, every worker asks a single accessor for a valid credential, and that accessor owns the entire decision: read the shared hash, decide whether a refresh is due, contend for the lock, and — critically — do the right thing when the lock is already held by a peer.

python
async def get_valid_token(client: AsyncClient, rds: redis.Redis,
                          property_id: str, channel: str,
                          wait_backoff: float = 0.25, wait_attempts: int = 8) -> str:
    """Return a live access token, refreshing once across all workers if due."""
    key = f"oauth:creds:{property_id}:{channel}"
    cred = await rds.hgetall(key)
    if cred and not should_refresh(cred["access_token"], property_id):
        return cred["access_token"]  # hot path: cached token still comfortably valid

    try:
        async with refresh_lock(rds, property_id, channel):
            # Re-read inside the lock: a peer may have rotated while we blocked.
            cred = await rds.hgetall(key)
            if cred and not should_refresh(cred["access_token"], property_id):
                return cred["access_token"]
            body = await refresh_access_token(client, rds, property_id, channel,
                                              cred["refresh_token"])
            return body["access_token"]
    except RuntimeError:  # refresh_in_progress: a peer holds the lock, so wait it out
        for _ in range(wait_attempts):
            await asyncio.sleep(wait_backoff)
            cred = await rds.hgetall(key)
            if cred and not should_refresh(cred["access_token"], property_id):
                return cred["access_token"]
        raise TimeoutError("peer_refresh_did_not_land")

The double-check of should_refresh after acquiring the lock is what collapses a refresh storm into a single network call: the worker that loses the race for the lock never reaches the token endpoint, and the worker that wins re-reads before spending the grant so it does not refresh a token a peer just rotated microseconds earlier.

Decision flow inside get_valid_token: hot path, refresh path, and contended path A flowchart of the single read-through accessor every worker calls. It starts at get_valid_token, which reads the shared credential hash. The first decision asks whether the cached token is still comfortably valid. If yes, the hot path returns the cached access token with no lock and no network call. If no, the worker contends for the refresh_lock via SET NX and a second decision asks whether it won the lock. If it won (the refresh path, in violet), it re-checks should_refresh inside the lock; if a peer already rotated the credential while it blocked, it skips the network call, otherwise it POSTs the refresh grant and persists the rotated pair — either way it returns the new access token. If it lost the lock (the contended path, in pink), it polls the cache every 0.25 seconds up to eight times; when the peer's token lands it returns that freshly rotated token, and only if the peer's refresh never lands does it raise a TimeoutError. Net effect: exactly one refresh occurs across all workers. get_valid_token(property_id, channel) read the shared credential hash — hgetall Cached token still comfortably valid? yes HOT PATH return cached access_token no lock · no network no Won the refresh_lock (SET lock NX)? CONTENDED · lock loser no Poll cache every 0.25s up to 8 attempts Peer's token landed yet? yes return peer's rotated token — no network call no raise TimeoutError peer_refresh_did_not_land yes REFRESH PATH · lock winner re-check should_refresh inside lock Peer rotated while we blocked? no POST refresh_token grant → persist rotated pair, then return new access_token yes skip network — peer already rotated hot / peer-served — no network lock winner — one refresh lock loser — poll & wait Every branch converges on one valid token — exactly one refresh occurs across all workers.

Schema and data contracts

Model the credential payload with Pydantic v2 so a malformed token response is rejected at the boundary instead of corrupting the cache. This is the canonical shape every worker reads and writes.

python
from pydantic import BaseModel, Field, field_validator

class OAuthCredential(BaseModel):
    property_id: str = Field(pattern=r"^PROP_\d+$")
    channel: str  # OTA slug, e.g. "booking_com", "expedia"
    access_token: str
    refresh_token: str
    expires_in: int = Field(gt=0, description="Access-token lifespan in seconds")
    token_type: str = "Bearer"

    @field_validator("channel")
    @classmethod
    def known_channel(cls, v: str) -> str:
        allowed = {"booking_com", "expedia", "agoda", "hostelworld"}
        if v not in allowed:
            raise ValueError(f"unknown channel slug: {v}")
        return v

# Persisted form: model_dump() gives a plain dict for the Redis hash.
cred = OAuthCredential(property_id="PROP_8842", channel="booking_com",
                       access_token="eyJ...", refresh_token="rt_9f...",
                       expires_in=180)
record = cred.model_dump()

field_validator on channel catches a mistyped OTA slug at parse time, which matters because a credential filed under the wrong channel silently authenticates rate pushes against the wrong distribution partner.

Error handling and retry strategy

Token failures split cleanly into retryable and terminal — conflating them is what turns a transient hiccup into an account suspension. Align this taxonomy with the broader treatment in error categorization and retry logic.

The dedicated child page automating channel manager token renewal packages this taxonomy into a scheduled renewal service you can deploy per property.

Verification and testing

Prove the refresh path works before it runs against production credentials. Assert on structured-log events and cache state rather than eyeballing HTTP output.

python
import pytest

@pytest.mark.asyncio
async def test_refresh_rotates_and_persists(fake_token_server, rds):
    channel, prop = "booking_com", "PROP_8842"
    old = "rt_expiring"
    body = await refresh_access_token(fake_token_server, rds, prop, channel, old)

    stored = await rds.hgetall(f"oauth:creds:{prop}:{channel}")
    assert stored["refresh_token"] != old            # rotation happened
    assert int(stored["exp"]) > int(time.time())     # future expiry cached
    assert body["access_token"] == stored["access_token"]  # cache == response

The three assertions map to the three ways this subsystem silently breaks: no rotation (single-use provider will reject the next call), a stale/past exp (every dispatch needlessly refreshes), and a cache that disagrees with the wire (workers read a token the server never issued).

Every refresh must emit these events so an operator can distinguish provider outages from client-side drift on a dashboard: token_pre_flight_check (remaining_ttl, threshold_met), token_refresh_initiated (worker_id, lock_status), token_rotation_complete (new_exp), and token_refresh_failed (error_code, retry_count, fallback_action).

Troubleshooting

Symptom Root cause Fix
Sporadic invalid_grant under load Two workers refreshed the same single-use token concurrently Wrap refresh in the refresh_lock; on refresh_in_progress, wait and re-read the cache instead of refreshing
Property locked out after a deploy Process persisted the new token to memory but not the shared store before restart Write the rotated refresh_token to Redis/secrets inside the exchange function, before returning
Refresh works but rate pushes still 401 Cache and resource server disagree; workers read a stale access token Use a single canonical hash key per property_id:channel and always read-through it, never a per-worker copy
Token endpoint returns 429 during peak Refresh calls share a throttling bucket with rate pushes Route both through one rate limiter; back off refreshes with jitter and honour Retry-After
Constant refreshing every request Provider issues opaque tokens with no exp Cache a local TTL at 80% of the documented lifespan (see FAQ) rather than decoding a claim that isn’t there

FAQ

How large should the refresh threshold be?

Start at 15% of the total lifespan. For a 180-second token that refreshes with roughly 27 seconds of headroom — enough to cover a multi-step rate push plus one backoff-retry without expiring mid-flight. Shorten it only if your longest sync transaction plus worst-case retry delay exceeds that margin; lengthen it if you are refreshing more often than necessary and burning token-endpoint quota.

The channel manager returns opaque tokens with no exp claim. Now what?

You cannot introspect an opaque token locally, so should_refresh will always return True. Instead, cache a local TTL derived from the provider’s documented lifespan and refresh at 80% of it. Store the computed expiry alongside the token in the same hash the OAuthCredential model persists, and drive should_refresh off that stored value rather than a decoded claim.

Should each sync worker manage its own token?

No. Per-worker tokens multiply refresh calls and, with single-use refresh tokens, guarantee invalid_grant races. Share one credential per property_id + channel through Redis, gate refreshes with the distributed lock, and have every worker read-through the same cache key. This keeps refresh volume proportional to properties, not to worker count.

How does token refresh interact with inventory polling?

Token validity windows bound how long a poll loop can run before it must re-check credentials. When a refresh completes, resume any queued inventory mutations and trigger a targeted reconciliation to confirm rate changes propagated. The polling cadence and webhook-fallback thresholds are covered in async polling for inventory updates.

What happens to in-flight requests when a refresh fires mid-batch?

Pause new dispatches at the pre-flight gate, let outstanding requests finish or fail into the retry queue, then refresh once under the lock and replay the paused writes. Because each mutation carries an idempotency key, replays the OTA already accepted are deduplicated rather than double-applied — no overbooking or duplicated restriction results.

← Back to API Sync & Data Ingestion Workflows