OAuth-токен истёк у восьми запросов: как выполнить одно обновление
Разбираем single-flight для OAuth: один запрос обновляет access token и refresh token, остальные ждут результат, а ошибка token endpoint не портит сохранённую пару.
В этой статье
Если восемь параллельных запросов одновременно увидели истёкший OAuth access token, к token endpoint должен пойти один запрос обновления. Остальные ждут его результат, затем повторно проверяют общую пару токенов и используют уже новый access token. Для одного процесса это даёт связка «быстрая проверка → блокировка → повторная проверка → обновление состояния после успешного ответа».
Ниже — безопасная учебная модель для одного процесса на CPython 3.12.14. Она не обращается к сети, не пишет секреты в файлы и не изображает контракт конкретного провайдера или модуля 1С-Битрикс. В рабочей интеграции параметры запроса, срок жизни, способ хранения и правила ротации нужно брать из документации авторизационного сервера.
Откуда берётся шторм обновления
Представим восемь задач, которые почти одновременно отправляют запросы к внешнему API. Все читают одну и ту же пару A1/R1, замечают истёкший A1 и решают обновить токен. Без координации восемь задач отправят восемь запросов с одним R1.
В OAuth refresh token используется клиентом для получения нового access token. Авторизационный сервер может выдать вместе с новым access token и новый refresh token; тогда клиент должен заменить старый refresh token новым. Это поведение описано в разделе 6 RFC 6749.
Почему ротация refresh token повышает цену гонки
При ротации предыдущий refresh token инвалидируется после успешного обновления. RFC 9700 рекомендует для публичных клиентов либо привязанные к отправителю refresh token, либо ротацию с обнаружением повторного использования. Два параллельных refresh — уже не просто лишняя нагрузка: один запрос может успеть заменить R1 на R2, а следующий получить ошибку повторного использования и спровоцировать отзыв активной цепочки.
Компонент | Правило учебной модели | Что предотвращает |
|---|---|---|
Запас по сроку | Обновлять за 5 секунд до expires_at | Пограничные запросы со старым A1 |
Блокировка | В один момент обновляет одна задача | Параллельное использование R1 |
Повторная проверка | После получения mutex задача снова читает expires_at | Последовательную очередь лишних refresh |
Замена состояния | Сохранять A2/R2 только после полного успешного ответа | Пару A2/R1 или потерю рабочего R1 |
Ошибка | Оставлять текущую пару без изменений | Частичную запись при сбое token endpoint |
Single-flight: mutex и повторная проверка
Тут есть подвох: использование mutex само по себе ещё не гарантирует один refresh. Если каждая ожидавшая задача без повторной проверки сразу обновит токен, восемь параллельных refresh превратятся в восемь последовательных. Поэтому условие срока действия проверяется дважды: до ожидания и внутри критической секции.
В примере используется asyncio.Lock. Документация Python описывает его как средство исключительного доступа для задач asyncio и отдельно предупреждает: этот объект не предназначен для синхронизации потоков ОС. Тем более он не координирует разные процессы или серверы.
Воспроизводимая модель
Команда ниже создаёт весь стенд в памяти и запускает три проверки. Начальное состояние — истёкший A1 и refresh token R1. Восемь задач одновременно запрашивают токен; затем имитируется отказ token endpoint и повтор после отказа. Все значения синтетические.
python3 - <<'PY'
import asyncio
from dataclasses import dataclass
@dataclass
class Tokens:
access_token: str
refresh_token: str
expires_at: int
class FakeIssuer:
def __init__(self):
self.calls = 0
self.used_refresh_tokens = set()
self.fail_next = False
async def refresh(self, refresh_token: str, now: int) -> Tokens:
await asyncio.sleep(0)
self.calls += 1
if self.fail_next:
self.fail_next = False
raise RuntimeError("token_endpoint_unavailable")
if refresh_token in self.used_refresh_tokens:
raise RuntimeError("refresh_token_reused")
self.used_refresh_tokens.add(refresh_token)
n = self.calls + 1
return Tokens(f"A{n}", f"R{n}", now + 60)
class TokenManager:
def __init__(self, tokens: Tokens, issuer: FakeIssuer):
self.tokens = tokens
self.issuer = issuer
self.lock = asyncio.Lock()
async def get_access_token(self, now: int) -> str:
if self.tokens.expires_at > now + 5:
return self.tokens.access_token
async with self.lock:
if self.tokens.expires_at > now + 5:
return self.tokens.access_token
updated = await self.issuer.refresh(self.tokens.refresh_token, now)
self.tokens = updated
return updated.access_token
async def main():
issuer = FakeIssuer()
manager = TokenManager(Tokens("A1", "R1", 0), issuer)
tokens = await asyncio.gather(
*(manager.get_access_token(now=100) for _ in range(8))
)
assert tokens == ["A2"] * 8
assert issuer.calls == 1
assert manager.tokens == Tokens("A2", "R2", 160)
print("parallel_8: one refresh, all received A2")
manager.tokens.expires_at = 100
issuer.fail_next = True
try:
await manager.get_access_token(now=100)
except RuntimeError as error:
assert str(error) == "token_endpoint_unavailable"
else:
raise AssertionError("failure was not propagated")
assert manager.tokens.refresh_token == "R2"
print("failed_refresh: state kept R2")
token = await manager.get_access_token(now=100)
assert token == "A4"
assert manager.tokens == Tokens("A4", "R4", 160)
print("retry_after_failure: received A4 with R4")
asyncio.run(main())
PY
parallel_8: one refresh, all received A2
failed_refresh: state kept R2
retry_after_failure: received A4 with R4
Проверка | Ожидаемый результат | Наблюдение |
|---|---|---|
Восемь параллельных задач | Один refresh; всем A2 | one refresh, all received A2 |
Сбой token endpoint | Сохранить R2 | state kept R2 |
Повтор после сбоя | Получить новую пару | received A4 with R4 |
Сценарий завершился с кодом 0. Он подтверждает только координацию восьми задач в одном процессе, повторную проверку под блокировкой и отсутствие частичной записи при синтетической ошибке. Реальный token endpoint, сетевые тайм-ауты, несколько процессов, база данных и конкретный провайдер не тестировались.
Что происходит при сбое token endpoint
Метод refresh() сначала формирует новый объект Tokens и только затем присваивает его общему состоянию. Если запрос завершается исключением, старый объект остаётся без изменений. Это важно для ротации: нельзя успеть записать A2, но оставить R1, или затереть R2 пустым значением из неполного ответа.
Не записывать новую пару до проверки обязательных полей ответа.
Не очищать рабочий refresh token из-за сетевого тайм-аута: сервер мог принять запрос, но ответ потерялся.
Не запускать бесконечные повторы; после неоднозначного результата действовать по контракту провайдера.
Не выводить access token и refresh token в журнал; достаточно результата, времени, идентификатора подключения и безопасного кода ошибки.
Как встроить механизм в клиент API
Хранить access token, refresh token и expires_at как одну версионируемую запись.
Сначала читать запись без блокировки; если access token ещё годен с запасом, сразу возвращать его.
Если пора обновлять, входить в критическую секцию для конкретного подключения.
Внутри секции снова читать актуальную запись: другая задача уже могла выполнить refresh.
Отправлять один запрос к token endpoint и атомарно заменять всю пару только после валидации ответа.
После выхода из секции отдавать ожидавшим задачам новый access token; ошибку передавать по контролируемой политике повторов.
Ответ 401 от ресурсного API не стоит превращать в безусловный refresh для каждого запроса. Сначала проверьте, действительно ли access token истёк и разрешает ли контракт безопасный повтор исходной операции. Для изменения заказа повтор может быть опаснее, чем краткая ошибка авторизации.
Граница одного процесса
asyncio.Lock решает задачу только внутри одного event loop. Если интеграция работает в нескольких процессах или на нескольких серверах, нужен общий координатор: например, блокировка по идентификатору подключения плюс сравнение версии записи перед заменой. Конкретную технологию выбирают вместе с хранилищем; локальный lock нельзя выдавать за распределённую гарантию.
Для распределённого варианта полезно хранить номер версии. Процесс читает версию 12, получает новую пару и пытается заменить запись только при условии, что версия всё ещё 12. Если уже записана версия 13, ответ устарел и не должен перезаписывать более новое состояние. Это проектный ориентир, а не испытанный в этой статье стенд.
Критерий готовности
При одновременном истечении токена token endpoint получает один refresh для одного подключения.
Все ожидавшие задачи получают одну и ту же новую пару.
После успешной ротации старый refresh token больше не используется.
Сбой до успешного ответа не меняет сохранённую пару.
Несколько процессов координируются общим механизмом, а не локальным asyncio.Lock.
Журналы содержат результат и идентификаторы, но не значения токенов.
Один успешно обновлённый токен проверяет только счастливый путь. Приёмка начинается там, где одновременно приходят восемь запросов, token endpoint один раз падает, а сохранённая пара остаётся согласованной.
If eight parallel requests simultaneously encounter an expired OAuth access token, only one request should go to the token endpoint for renewal. The others wait for its result, then recheck the shared token pair and use the new access token. For a single process, this creates a sequence of fast check → lock → recheck → update state after successful response.
Below is a safe educational model for a single process on CPython 3.12.14. It does not access the network, does not write secrets to files, and does not represent the contract of a specific provider or 1C-Bitrix module. In a production integration, request parameters, lifetime, storage method, and rotation rules must be taken from the authorization server documentation.
Where the Update Storm Comes From
Imagine eight tasks that almost simultaneously send requests to an external API. All read the same A1/R1 pair, notice that A1 has expired, and decide to update the token. Without coordination, eight tasks will send eight requests with the same R1.
In OAuth, a client uses a refresh token to obtain a new access token. The authorization server may issue a new refresh token along with the new access token; in that case, the client must replace the old refresh token with the new one. This behavior is described in section 6 of RFC 6749.
Why Refresh Token Rotation Increases the Cost of a Race
With rotation, the previous refresh token is invalidated after a successful update. RFC 9700 recommends either sender-bound refresh tokens or rotation with reuse detection for public clients. Two parallel refreshes are no longer just extra load: one request may replace R1 with R2, while the next triggers a reuse error and provokes revocation of the active chain.
Component | Training model rule | What it prevents |
|---|---|---|
Validity buffer | Update 5 seconds before expires_at | Edge requests with old A1 |
Blocking | One task updates at a time | Parallel use of R1 |
Re-validation | After acquiring the mutex, the task reads expires_at again | Sequential queue of unnecessary refreshes |
State replacement | Save A2/R2 only after a complete successful response | A pair of A2/R1 or a lost working R1 |
Error | Leave the current pair unchanged | Partial write on token endpoint failure |
Single-flight: mutex and double-check
There is a catch: using a mutex alone does not guarantee a single refresh. If every waiting task updates the token immediately without a double-check, eight parallel refreshes will become eight sequential ones. Therefore, the expiration condition is checked twice: before waiting and inside the critical section.
The example uses asyncio.Lock. Python documentation describes it as a mechanism for exclusive access in asyncio tasks and separately warns: this object is not intended for synchronizing OS threads. It certainly does not coordinate different processes or servers.
Reproducible model
The command below creates the entire environment in memory and runs three checks. The initial state is an expired A1 and refresh token R1. Eight tasks simultaneously request a token; then a token endpoint failure is simulated followed by a retry after failure. All values are synthetic.
python3 - <<'PY'
import asyncio
from dataclasses import dataclass
@dataclass
class Tokens:
access_token: str
refresh_token: str
expires_at: int
class FakeIssuer:
def __init__(self):
self.calls = 0
self.used_refresh_tokens = set()
self.fail_next = False
async def refresh(self, refresh_token: str, now: int) -> Tokens:
await asyncio.sleep(0)
self.calls += 1
if self.fail_next:
self.fail_next = False
raise RuntimeError("token_endpoint_unavailable")
if refresh_token in self.used_refresh_tokens:
raise RuntimeError("refresh_token_reused")
self.used_refresh_tokens.add(refresh_token)
n = self.calls + 1
return Tokens(f"A{n}", f"R{n}", now + 60)
class TokenManager:
def __init__(self, tokens: Tokens, issuer: FakeIssuer):
self.tokens = tokens
self.issuer = issuer
self.lock = asyncio.Lock()
async def get_access_token(self, now: int) -> str:
if self.tokens.expires_at > now + 5:
return self.tokens.access_token
async with self.lock:
if self.tokens.expires_at > now + 5:
return self.tokens.access_token
updated = await self.issuer.refresh(self.tokens.refresh_token, now)
self.tokens = updated
return updated.access_token
async def main():
issuer = FakeIssuer()
manager = TokenManager(Tokens("A1", "R1", 0), issuer)
tokens = await asyncio.gather(
*(manager.get_access_token(now=100) for _ in range(8))
)
assert tokens == ["A2"] * 8
assert issuer.calls == 1
assert manager.tokens == Tokens("A2", "R2", 160)
print("parallel_8: one refresh, all received A2")
manager.tokens.expires_at = 100
issuer.fail_next = True
try:
await manager.get_access_token(now=100)
except RuntimeError as error:
assert str(error) == "token_endpoint_unavailable"
else:
raise AssertionError("failure was not propagated")
assert manager.tokens.refresh_token == "R2"
print("failed_refresh: state kept R2")
token = await manager.get_access_token(now=100)
assert token == "A4"
assert manager.tokens == Tokens("A4", "R4", 160)
print("retry_after_failure: received A4 with R4")
asyncio.run(main())
PY
parallel_8: one refresh, all received A2
failed_refresh: state kept R2
retry_after_failure: received A4 with R4
Check | Expected result | Observation |
|---|---|---|
Eight parallel tasks | One refresh; all A2 | one refresh, all received A2 |
Token endpoint failure | Keep R2 | state kept R2 |
Retry after failure | Get a new pair | received A4 with R4 |
The scenario completed with code 0. It confirms only the coordination of eight tasks in a single process, the double-check under the lock, and the absence of partial writes during a synthetic error. The real token endpoint, network timeouts, multiple processes, database, and specific provider were not tested.
What happens when the token endpoint fails
The refresh() method first constructs a new Tokens object and only then assigns it to the shared state. If the request terminates with an exception, the old object remains unchanged. This is critical for rotation: one must not succeed in writing A2 while leaving R1, nor overwrite R2 with an empty value from an incomplete response.
Do not write a new pair until the mandatory response fields are verified.
Do not clear the working refresh token due to a network timeout: the server may have accepted the request, but the response was lost.
Do not launch infinite retries; after an ambiguous result, act according to the provider contract.
Do not output the access token and refresh token to the log; the result, time, connection identifier, and a safe error code are sufficient.
How to embed the mechanism in an API client
Store the access token, refresh token, and expires_at as a single versioned record.
First, read the record without locking; if the access token is still valid with a margin, return it immediately.
If an update is needed, enter the critical section for that specific connection.
Inside the section, read the current record again: another task may have already performed the refresh.
Send a single request to the token endpoint and atomically replace the entire pair only after validating the response.
After exiting the section, provide the waiting tasks with the new access token; pass errors according to a controlled retry policy.
A 401 response from a resource API should not be converted into an unconditional refresh for every request. First, verify whether the access token has actually expired and whether the contract permits safely retrying the original operation. For order modifications, a retry can be more dangerous than a brief authorization error.
Boundary of a single process
asyncio.Lock solves the problem only within a single event loop. If the integration runs across multiple processes or servers, a shared coordinator is required: for example, locking by connection ID plus comparing the record version before replacement. The specific technology is chosen together with the storage; a local lock cannot be presented as a distributed guarantee.
For a distributed variant, it is useful to store the version number. The process reads version 12, receives a new pair, and attempts to replace the record only if the version is still 12. If version 13 has already been written, the response is stale and must not overwrite a newer state. This is a design guideline, not a tested setup in this article.
Readiness criteria
When a token expires simultaneously, the token endpoint receives one refresh request for one connection.
All waiting tasks receive the same new pair.
After a successful rotation, the old refresh token is no longer used.
A failure before a successful response does not change the saved pair.
Several processes are coordinated by a shared mechanism, not a local asyncio.Lock.
Logs contain the result and identifiers, but not token values.
A single successfully updated token validates only the happy path. Acceptance testing begins when eight requests arrive simultaneously, the token endpoint fails once, and the saved pair remains consistent.
Обсуждение 0
Делись опытом и задавай вопросы. Комментарии без ссылок появляются после проверки редактором.
Пока никто не написал. Начни обсуждение.