Broadcast Traffic Scheduling Automation

Refreshing Avstar OAuth Tokens Automatically

An Avstar bearer token lives for an hour, maybe less, but a nightly traffic ingestion run lasts longer — and the moment the token expires mid-batch, every upload comes back 401 Unauthorized and the run stalls with half the logs posted. This guide solves one exact task: detecting token expiry before it bites, refreshing the credential with the OAuth refresh-token flow, and doing it in a way that is safe when dozens of async workers hit the expiry boundary at once — a single-flight refresh guarded by a lock, plus a one-shot retry on a 401 that slipped through. It is the credential-lifecycle procedure beneath Avstar API Authentication and Rate Limits, part of Avion & Avstar Ingestion Pipelines. It matters for audit because a run that dies mid-batch leaves an incomplete as-run record, and an uncoordinated refresh storm — every worker refreshing at once — can invalidate the refresh token and lock the whole station group out of the API.

The core requirement is single-flight refresh: when the token is stale, exactly one worker refreshes while the rest wait for that result, rather than each firing its own refresh and racing to overwrite the shared credential. That is a concurrency-correctness concern distinct from, but adjacent to, the session-timeout handling that recovers a dropped connection — this page keeps the credential itself valid so the connection never gets a 401 to begin with.

Prerequisites

  • Python 3.11+ — required for asyncio.Lock, X | None union syntax, and timezone-aware datetime.now(timezone.utc).
  • httpx — pin httpx==0.27.0 for its async client; the refresh call and the protected requests share one client and connection pool.
  • Pydantic v2 — pin pydantic==2.7.1 to model the token response, so a malformed refresh reply is caught before it poisons the cached credential.
  • A registered OAuth client — client id, client secret, the token endpoint URL, and a long-lived refresh token, all held as secrets, never in source.
  • A clock-skew margin — a refresh-ahead window (for example 60 seconds) so the token is renewed before it expires, absorbing round-trip latency and small clock differences between the app and the auth server.
  • A shared token store — a single in-process cache guarded by a lock, or an external store, so every worker reads one credential rather than each holding its own.

Step-by-Step Implementation

The token manager holds the current access token and its expiry, hands it out to callers, and refreshes it single-flight when it is within the skew margin of expiring. A protected-request wrapper retries exactly once on a 401, forcing a refresh in case the token was revoked server-side before its stated expiry.

Single-flight token refresh state machine A worker requests a valid access token from the token manager. The manager checks whether the cached token is still fresh, meaning its expiry is more than the skew margin away. If fresh, it returns the cached token immediately. If stale, the worker acquires the refresh lock. The first worker to acquire the lock re-checks freshness under the lock, then performs the OAuth refresh-token grant against the auth server, receives a new access token and expiry, and stores them. Workers that arrive while the refresh is in progress wait on the lock and, once it releases, find the token already fresh and return it without issuing a second refresh. Separately, a protected request that still receives a 401 forces one refresh and retries once before failing. Worker get_token() fresh? exp − skew Return cached token Bearer … Acquire lock re-check under lock others wait Refresh grant POST /oauth/token grant=refresh_token Auth server new access token + expiry yes stale store token + expiry → cache fresh

Figure — Single-flight refresh: a fresh token is returned from cache; a stale one is refreshed by exactly one worker under a lock while the rest wait and then read the newly cached credential, so the auth server sees one refresh, not a storm.

Step 1 — Structured logging and the token model

Goal: emit greppable audit lines in the timestamp | level | module | spot_id shape and model the token response so a malformed refresh reply never poisons the cache. The credential lifecycle is a compliance surface — every refresh must be reconstructable from the log.

python 
from __future__ import annotations

import asyncio
import logging
from datetime import datetime, timedelta, timezone

import httpx
from pydantic import BaseModel, Field

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("traffic.auth.avstar_token_manager")


class TokenResponse(BaseModel):
    access_token: str = Field(min_length=1)
    refresh_token: str | None = None       # some servers rotate the refresh token
    expires_in: int = Field(gt=0)          # seconds until the access token expires
    token_type: str = "Bearer"


class CachedToken(BaseModel):
    access_token: str
    expires_at: datetime                   # absolute UTC expiry, not a duration

    def is_fresh(self, skew: timedelta) -> bool:
        # Fresh means expiry is more than the skew margin into the future.
        return datetime.now(timezone.utc) + skew < self.expires_at

Step 2 — The single-flight refresh

Goal: perform the refresh-token grant under a lock so concurrent workers trigger exactly one network refresh. The double-check inside the lock is the crux — a worker that waited on the lock must re-test freshness, because the worker ahead of it has probably already refreshed.

python 
class AvstarTokenManager:
    def __init__(self, client: httpx.AsyncClient, token_url: str,
                 client_id: str, client_secret: str, refresh_token: str,
                 skew_seconds: int = 60) -> None:
        self._client = client
        self._token_url = token_url
        self._client_id = client_id
        self._client_secret = client_secret
        self._refresh_token = refresh_token
        self._skew = timedelta(seconds=skew_seconds)
        self._cached: CachedToken | None = None
        self._lock = asyncio.Lock()        # serializes refresh, not reads

    async def _do_refresh(self) -> CachedToken:
        # OAuth refresh-token grant: exchange the refresh token for a new access token.
        resp = await self._client.post(
            self._token_url,
            data={"grant_type": "refresh_token",
                  "refresh_token": self._refresh_token,
                  "client_id": self._client_id,
                  "client_secret": self._client_secret},
            timeout=10.0,
        )
        resp.raise_for_status()
        parsed = TokenResponse.model_validate(resp.json())
        if parsed.refresh_token:           # honour refresh-token rotation
            self._refresh_token = parsed.refresh_token
        expires_at = datetime.now(timezone.utc) + timedelta(seconds=parsed.expires_in)
        logger.info("token refreshed | expires_at=%s", expires_at.isoformat())
        return CachedToken(access_token=parsed.access_token, expires_at=expires_at)

    async def get_token(self) -> str:
        # Fast path: a fresh cached token needs no lock.
        if self._cached is not None and self._cached.is_fresh(self._skew):
            return self._cached.access_token
        async with self._lock:
            # Double-check: a worker ahead of us may have already refreshed.
            if self._cached is not None and self._cached.is_fresh(self._skew):
                logger.debug("token already refreshed by concurrent worker")
                return self._cached.access_token
            self._cached = await self._do_refresh()
            return self._cached.access_token

Step 3 — A protected request that retries once on 401

Goal: handle the case where a token was revoked server-side before its stated expiry. A 401 forces one refresh and one retry; a second 401 is a real auth failure, not a stale-token blip, and is raised.

python 
class AvstarTokenManager:  # continued — request() sits alongside get_token()/_do_refresh() from above
    async def request(self, method: str, url: str, **kwargs: object) -> httpx.Response:
        token = await self.get_token()
        headers = {**kwargs.pop("headers", {}), "Authorization": f"Bearer {token}"}
        resp = await self._client.request(method, url, headers=headers, **kwargs)
        if resp.status_code != 401:
            return resp
        # 401 despite a "fresh" token => revoked early. Force one refresh + retry.
        logger.warning("401 on protected request | forcing token refresh")
        async with self._lock:
            self._cached = await self._do_refresh()
            token = self._cached.access_token
        headers["Authorization"] = f"Bearer {token}"
        retry = await self._client.request(method, url, headers=headers, **kwargs)
        if retry.status_code == 401:
            logger.error("401 after refresh | credential invalid, not stale")
        return retry

Step 4 — Wire it into an upload worker

Goal: show the manager in a realistic call site — an ingestion worker posting a spot log — so the token lifecycle is invisible to the upload logic. The worker just calls manager.request; freshness and refresh are handled beneath it.

python 
async def upload_spot_log(manager: AvstarTokenManager, spot_id: str,
                          payload: dict[str, str]) -> bool:
    resp = await manager.request(
        "POST", "https://avstar.example/v1/logs", json=payload,
        headers={"Idempotency-Key": spot_id},
    )
    ok = resp.status_code < 300
    logger.info("%s | upload status=%d ok=%s", spot_id, resp.status_code, ok)
    return ok

A representative log line reads 2026-07-17T00:59:41+00:00 | INFO | traffic.auth.avstar_token_manager | token refreshed | expires_at=2026-07-17T01:59:41+00:00.

Verification & Testing

Correctness rests on two properties: freshness detection (a token within the skew margin is treated as stale) and single-flight refresh (concurrent callers trigger one refresh, not many). Both are assertable with a counting stub transport.

python 
import httpx

calls = {"refresh": 0}

def handler(req: httpx.Request) -> httpx.Response:
    if req.url.path.endswith("/oauth/token"):
        calls["refresh"] += 1
        return httpx.Response(200, json={"access_token": f"tok-{calls['refresh']}",
                                         "expires_in": 3600})
    return httpx.Response(200, json={"ok": True})

async def _test() -> None:
    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        mgr = AvstarTokenManager(client, "https://auth.example/oauth/token",
                                 "cid", "secret", "refresh-tok", skew_seconds=60)
        # 10 workers hit an empty cache at once: exactly one refresh must occur.
        tokens = await asyncio.gather(*(mgr.get_token() for _ in range(10)))
    assert calls["refresh"] == 1                 # single-flight held
    assert len(set(tokens)) == 1                 # all workers got the same token

asyncio.run(_test())

# Freshness: a token expiring inside the skew window is stale.
soon = CachedToken(access_token="x",
                   expires_at=datetime.now(timezone.utc) + timedelta(seconds=30))
assert soon.is_fresh(timedelta(seconds=60)) is False

Run this in CI so a regression that drops the lock or the double-check — turning single-flight into a refresh storm — fails the build. Assert on the refresh count, not just that a token came back, because a correct-looking token can still hide N redundant refreshes.

Edge Cases & Failure Handling

  • Refresh-token rotation. Some auth servers issue a new refresh token on every grant and invalidate the old one. _do_refresh captures parsed.refresh_token and replaces the stored value; skip that and the next refresh presents a dead token and every worker is locked out. Persist the rotated token to the shared store if the manager outlives the process.
  • Clock skew and the refresh-ahead margin. If the app clock runs behind the auth server, a token the app thinks is fresh may already be rejected. The skew_seconds margin refreshes ahead of expiry to absorb this; widen it if you see sporadic 401s that a retry clears, and keep it in sync with the session-timeout handling so both layers agree on when a credential is spent.
  • Refresh failure mid-batch. If the refresh grant itself returns 4xx — a revoked refresh token, a clock far out of range — raise_for_status propagates and the batch must stop rather than loop. Surface it to the backpressure and retry layer’s circuit breaker so a dead credential halts the run instead of hammering the auth server, and alert the traffic desk to re-authorize.

FAQ

Why refresh ahead of expiry instead of waiting for a 401?

Because a token that expires mid-request costs a wasted round trip, a 401, and a retry for every worker that hit the boundary — under load that is a burst of failures right when the batch is busiest. Refreshing when the token is within the skew margin of expiring renews it proactively, so requests almost never see a 401. The one-shot retry in Step 3 is a safety net for early revocation, not the primary mechanism, which keeps load off the rate-limited API.

Why does the lock double-check freshness after acquiring?

Because between deciding the token was stale and acquiring the lock, another worker may have already refreshed it. Without the re-check, every worker that queued on the lock would refresh in turn — exactly the storm the lock is meant to prevent. Re-testing is_fresh inside the lock means only the first worker refreshes and the rest return the freshly cached token. It is the same double-checked pattern that makes async batch processing safe under concurrency.

Is an asyncio.Lock enough, or do I need a threading lock too?

Within a single event loop, asyncio.Lock is sufficient and correct — coroutines yield cooperatively, so there is no preemption to guard against beyond the await points. If you run the manager across OS threads, or across processes, you need a threading lock or an external mutex around the shared store instead. Pick the primitive that matches your concurrency model; mixing them, or using threading.Lock inside async code, will block the loop.

How does token refresh relate to session-timeout handling?

They are two layers of the same resilience story. Token refresh keeps the credential valid so a request is authorized; session-timeout handling recovers the connection when the transport drops. A run needs both: a fresh token sent over a dead socket still fails, and a live socket carrying an expired token still returns 401. Keep their timeout margins aligned so neither masks the other.

Related content

Up one levelAvstar API Authentication and Rate Limits RelatedHandling Avstar Session Timeouts in Python