Подписка активна, а API возвращает 429: как отличить квоту от ограничения частоты запросов

У приложения есть обработчик: если сервер ответил 429, подождать и повторить запрос. На нагрузочном всплеске это выглядит разумно. Но тот же обработчик будет терпеливо повторять запросы, когда закончились API-кредиты или сработал лимит расходов проекта. Пользователь увидит долгую загрузку, хотя ни одна дополнительная секунда ожидания не устраняет причину.

Задача этого разбора узкая: по одному неуспешному HTTP-ответу выбрать следующее действие. Продолжать с меньшей интенсивностью, проверять биллинг или останавливать автоматику до выяснения причины? Результат должен определяться проверяемыми признаками, а не наличием оплаченной подписки.

Ниже рассматривается прямое обращение к OpenAI API, а не сторонний шлюз с совместимым SDK. Документация проверена 31 октября 2026 года. Примеры ответов синтетические; код проверяется локально и не обращается к моделям. Это не рассказ о клиентском инциденте и не нагрузочный тест.

Почему активная подписка ничего не говорит об этом запросе

Начать стоит с уточнения, что именно оплачено и к какой организации относится запрос. Условия подписки ChatGPT и состояние API-биллинга нельзя свести к одному флажку «оплачено».

При этом привычное объяснение «подписка никогда не включает API» уже слишком категорично. В актуальной документации описана возможность получить API-кредиты по подходящим подпискам Pro и Business. Льготу нужно активировать и связать с API-организацией; кредиты действуют именно в ней, имеют срок действия и не отменяют rate limits. Условия применимости проверяются отдельно.

Представим учебную конфигурацию: организация personal получила кредиты, а приложение использует ключ проекта из work. Скриншот положительного баланса personal ничего не доказывает о запросе приложения. Сначала нужно сопоставить окружение приложения, используемый ключ, организацию и проект. Перевыпускать ключ наугад бессмысленно: новый секрет сам по себе не объяснит, куда списывается потребление.

В API есть явные заголовки выбора организации и проекта для соответствующих сценариев авторизации. Но их не следует добавлять вслепую ко всем запросам. Для диагностики важен фактически выбранный контекст, а не организация, которая последней открывалась в браузере.

Сначала сохранить свидетельства

Запись «API error 429» слишком бедна для разбора. До повторной попытки полезно сохранить время с часовым поясом, HTTP-статус, error.code, error.type, модель, идентификатор проекта из конфигурации и x-request-id, если он пришёл. Отдельно нужны доступные Retry-After и заголовки x-ratelimit-*. Идентификатор запроса позволяет обсуждать с поддержкой конкретный ответ.

Я бы собирал такую запись по разрешённому списку полей, а не сохранял целиком HTTP-объект. В нём легко оставить Authorization, cookie или пользовательский документ. Даже текст ошибки перед помещением в общий лог стоит проверить на чувствительные данные. Для локальной диагностики обычно достаточно технических признаков; публиковать их вместе с секретом не требуется.

Дальше читается тело ответа. У ошибок биллинга общий error.type может оставаться insufficient_quota, тогда как более точная причина находится в error.code. Поэтому сравнение только с одним словом quota теряет полезную информацию.

Коды и их смысл приведены по официальному справочнику ошибок. Последняя строка — консервативное правило предлагаемого обработчика, а не утверждение о любом 429.

У бюджета есть режим работы

Слово «лимит» в интерфейсе тоже недостаточно. Spend alert уведомляет о расходах, но не прекращает API-трафик. Hard spend limit способен его остановить. Причём организационный лимит распространяется на проекты организации, а проектный действует в пределах одного проекта. Применение ограничения не мгновенно: учтённая сумма может немного превысить установленное значение.

Отсюда практическая проверка для учебного инцидента: если один проект перестал работать, а другой продолжает, нужно сравнить их настройки и коды ответов. Это основание для проверки области ограничения, но не готовый диагноз. У проектов могут различаться ключи, модели и нагрузка.

Повышение лимита расходов — отдельное решение владельца бюджета. Обработчик ошибки не должен автоматически снимать ограничение ради зелёного статуса мониторинга. Его задача — указать причину и прекратить бесполезные повторы.

Когда ожидание действительно помогает

Для временного ограничения важны единицы измерения. Запросы в минуту и токены в минуту ограничивают разные вещи. Проверяются также область действия лимита и выбранная модель: часть лимитов относится к организации или проекту, некоторые модели делят общий лимит. Код slow_down описывает слишком резкое наращивание трафика и возможен даже без исчерпания RPM и TPM.

Если подтверждена временная ошибка, валидный Retry-After задаёт минимальную паузу. Без серверной подсказки используют увеличивающуюся задержку со случайной добавкой, ограничивая число попыток и общее время. Длинную серверную паузу нельзя обрезать до удобных пяти секунд: если операция столько ждать не может, её следует отложить.

У приложения должен быть один понятный бюджет повторов. Например, в учебной политике ниже разрешены две дополнительные попытки, а на очередной сетевой вызов резервируются три секунды. Эти числа выбраны для демонстрации, а не рекомендованы поставщиком. Вызывающий код обязан уменьшать оставшееся время после каждого ожидания и сетевого вызова.

Нужно учесть и SDK. Официальная Python-библиотека автоматически повторяет определённые ошибки дважды по умолчанию; max_retries=0 отключает этот механизм. Внешний цикл поверх SDK без общего учёта способен дать больше запросов, чем видно из числа итераций приложения. Поведение следует сверять с установленной версией.

Проверка обработчика без расходов на API

Чтобы проверить выбор действия, не нужен настоящий исчерпанный баланс. Достаточно функции, которая получает статус, разобранное JSON-тело и заголовки, а возвращает план. Она ничего не отправляет и не спит, поэтому тест проверяет именно ветвление.

В примере есть небольшой список известных временных кодов. rate_limit_exceeded оставлен для соответствующих ответов API; это не обещание одинакового формата у всех методов. Незнакомый код отправляется на разбор. Для 503 разрешена только явно указанная перегрузка server_is_overloaded.

Воспроизводимый файл retry_plan.py использует только стандартную библиотеку Python. Полный исполняемый пример:

"""Synthetic, offline retry-policy example. No HTTP calls or credentials."""

import math
import random

BILLING_CODES = {
    "credit_balance_exhausted",
    "project_spend_limit_exceeded",
    "organization_spend_limit_exceeded",
    "organization_usage_limit_exceeded",
}


def plan(status, body, headers, retry_index=0, remaining_s=20.0,
         attempt_timeout_s=3.0, rng=None):
    """Return (action, delay_s); retry_index counts retries already used."""
    if retry_index < 0 or remaining_s < 0 or attempt_timeout_s <= 0:
        raise ValueError("Invalid retry budget")
    if not math.isfinite(remaining_s + attempt_timeout_s):
        raise ValueError("Budget must be finite")
    error = body.get("error") if isinstance(body, dict) else None
    if not isinstance(error, dict):
        return "inspect", None
    code, kind = error.get("code"), error.get("type")
    if not isinstance(code, (str, type(None))):
        return "inspect", None
    if status == 429 and (
        code in BILLING_CODES or kind == "insufficient_quota"
    ):
        return "fix_billing_or_limit", None
    transient = (
        status == 429 and code in {"rate_limit_exceeded", "slow_down"}
    ) or (status == 503 and code == "server_is_overloaded")
    if not transient:
        return "inspect", None
    if retry_index >= 2:
        return "retry_budget_exhausted", None
    normalized = {str(k).lower(): v for k, v in headers.items()}
    raw = normalized.get("retry-after")
    if raw is None:
        delay = 2.0 ** retry_index
    else:
        try:
            delay = float(raw)
        except (TypeError, ValueError, OverflowError):
            return "inspect_retry_after", None
        if not math.isfinite(delay) or delay < 0:
            return "inspect_retry_after", None
    rng = rng if rng is not None else random
    delay += rng.uniform(0.0, 0.25)
    if delay + attempt_timeout_s > remaining_s:
        return "defer", None
    return "retry_after", delay


def demo():
    fixtures = [
        ("traffic", 429, "slow_down", "rate_limit_error", "2"),
        ("credits", 429, "credit_balance_exhausted",
         "insufficient_quota", "2"),
        ("long_wait", 429, "slow_down", "rate_limit_error", "120"),
        ("unknown", 429, "new_code", "new_type", "2"),
    ]
    for name, status, code, kind, hint in fixtures:
        body = {"error": {"code": code, "type": kind}}
        action, delay = plan(status, body, {"Retry-After": hint},
                             rng=random.Random(0))
        print(name, action, "-" if delay is None else f"{delay:.3f}")


if __name__ == "__main__":
    demo()

Здесь retry_index — число уже использованных повторов, а remaining_s — оставшееся время всей операции. При каждом новом ответе вызывающий код передаёт актуальные значения. Возвращённое defer не ставит задачу в очередь автоматически: приложение само решает, можно ли продолжить её позже или нужно сообщить об отказе.

Поддержка заголовка намеренно ограничена числом секунд. Если приходит дата либо непонятное значение, пример прекращает автоматическое планирование. Для рабочего клиента нужен полноценный разбор поддерживаемых форматов; превращать любое непонятное значение в нулевую задержку нельзя.

Теперь подадим четыре вымышленных ответа. У первых двух одинаковые HTTP-статус 429 и Retry-After: 2, но разные коды. В третьем сервер просит ждать 120 секунд при оставшемся бюджете 20 секунд. В четвёртом код неизвестен. С фиксированным генератором случайных чисел запуск python retry_plan.py печатает:

traffic retry_after 2.211
credits fix_billing_or_limit -
long_wait defer -
unknown inspect -

Самый полезный контрпример — второй. Даже наличие Retry-After не заставляет функцию повторять запрос при исчерпанных кредитах. В третьем случае она также не подменяет 120 секунд более коротким ожиданием. Это свойства нашей политики, проверенные на заданных входных данных.

Отдельный локальный набор из 18 тестов прошёл успешно. Он проверяет все четыре кода биллинга, общий тип insufficient_quota, неизвестные и повреждённые тела, регистр заголовка, предел повторов и временной бюджет. Команда запуска: python -m unittest -v test_retry_plan.py. Прохождение этих тестов не доказывает доступность API и не измеряет производительность модели.

Что ещё нужно до рабочего внедрения

Такой планировщик удобен как тестируемый фрагмент, но сетевой слой остаётся за его пределами. Понадобятся общий deadline, отмена, ограничения параллелизма и аккуратное завершение задачи. Если несколько воркеров одновременно упираются в один лимит, независимые паузы каждого из них не заменяют управление общей очередью.

Отдельно обрабатываются неоднозначные тайм-ауты и уже начавшийся поток. Этот пример относится к полученному HTTP-отказу до выдачи результата. Он не разрешает повторять операцию после частичного ответа или выполненного внешнего действия: сначала нужно установить, что уже произошло. Для потоковых запросов документация прямо отделяет HTTP-ошибку до старта от ошибки внутри потока.

После исправления полезно повторить один контролируемый запрос в прежнем проекте и проверить новый ответ. Если одновременно сменить ключ, модель, организацию и параметры нагрузки, причина успешного результата останется неизвестной.

В итоге разбор заканчивается конкретным действием: исправить биллинг в установленной области, снизить нагрузку с ограниченными повторами либо собрать недостающие свидетельства. Активная подписка остаётся контекстом. Решение о следующем запросе принимает обработчик, который различает причины отказа.

import random
import unittest

from retry_plan import BILLING_CODES, plan


def body(code="slow_down", kind="rate_limit_error"):
    return {"error": {"code": code, "type": kind}}


class RetryPolicyTests(unittest.TestCase):
    def call(self, payload=None, headers=None, **kw):
        return plan(429, body() if payload is None else payload,
                    {} if headers is None else headers,
                    rng=random.Random(0), **kw)

    def test_billing_codes_override_retry_header(self):
        for code in BILLING_CODES:
            with self.subTest(code=code):
                self.assertEqual(self.call(body(code), {"Retry-After": "2"}),
                                 ("fix_billing_or_limit", None))

    def test_broad_quota_type(self):
        self.assertEqual(self.call(body(None, "insufficient_quota"))[0],
                         "fix_billing_or_limit")

    def test_quota_type_wins_in_conflicting_fixture(self):
        self.assertEqual(self.call(body("slow_down", "insufficient_quota"))[0],
                         "fix_billing_or_limit")

    def test_numeric_hint_is_minimum(self):
        action, delay = self.call(headers={"Retry-After": "2"})
        self.assertEqual(action, "retry_after")
        self.assertGreaterEqual(delay, 2)
        self.assertLessEqual(delay, 2.25)

    def test_header_names_case_insensitive(self):
        self.assertEqual(self.call(headers={"Retry-After": "2"}),
                         self.call(headers={"RETRY-AFTER": "2"}))

    def test_fallback_increases(self):
        self.assertAlmostEqual(self.call(retry_index=1)[1] -
                               self.call(retry_index=0)[1], 1.0)

    def test_retry_count_is_finite(self):
        self.assertEqual(self.call(retry_index=2),
                         ("retry_budget_exhausted", None))

    def test_server_delay_never_truncated(self):
        self.assertEqual(self.call(headers={"Retry-After": "120"}),
                         ("defer", None))

    def test_timeout_reserved(self):
        self.assertEqual(self.call(headers={"Retry-After": "18"}),
                         ("defer", None))

    def test_unknown_code_not_blindly_retried(self):
        self.assertEqual(self.call(body("unknown")), ("inspect", None))

    def test_missing_code_not_blindly_retried(self):
        self.assertEqual(self.call(body(None)), ("inspect", None))

    def test_legacy_rate_code(self):
        self.assertEqual(self.call(body("rate_limit_exceeded"))[0],
                         "retry_after")

    def test_other_status_not_matched(self):
        self.assertEqual(plan(401, body(), {}), ("inspect", None))

    def test_overload_code_only_for_503(self):
        payload = body("server_is_overloaded", "service_unavailable_error")
        self.assertEqual(plan(503, payload, {}, rng=random.Random(0))[0],
                         "retry_after")
        self.assertEqual(self.call(payload), ("inspect", None))

    def test_malformed_body(self):
        for payload in [[], "HTML", {}, {"error": []}, {"error": None}]:
            with self.subTest(payload=payload):
                self.assertEqual(self.call(payload), ("inspect", None))

    def test_malformed_code(self):
        self.assertEqual(self.call(body([])), ("inspect", None))

    def test_unsupported_or_invalid_hint(self):
        for hint in ["NaN", "inf", "-1", "later", {},
                     "Sat, 31 Oct 2026 13:00:00 GMT"]:
            with self.subTest(hint=hint):
                self.assertEqual(self.call(headers={"Retry-After": hint}),
                                 ("inspect_retry_after", None))

    def test_invalid_budget(self):
        for kw in [{"remaining_s": -1}, {"attempt_timeout_s": 0},
                   {"retry_index": -1}, {"remaining_s": float("inf")},
                   {"remaining_s": float("nan")}]:
            with self.subTest(kw=kw):
                with self.assertRaises(ValueError):
                    self.call(**kw)


if __name__ == "__main__":
    unittest.main(verbosity=2)

Официальные источники

• Коды ошибок API

• Лимиты расходов

• API-кредиты по подписке

Если проверка показала, что нужна именно оплата зарубежного сервиса, условия можно уточнить в RaketaPay. Оплата не устраняет ограничение частоты запросов и не назначает права проекту.