Retry ETTN creation on Nova Poshta rate limit instead of failing

Checkbox relays Nova Poshta's "To many requests" (20000401501) as a 4xx
third_party.generic error, not a 429, so bursts of receipt requests ended
up as failed receipts. Such responses are now CheckboxRateLimitedError:
the receipt stays pending and the worker job is retried with arq.Retry
after the "Try again after N seconds" delay plus backoff. NP timeouts
relayed the same way are treated as unavailable (unknown outcome).

The HTTP client also sends Checkbox requests one at a time with a
CHECKBOX_MIN_REQUEST_INTERVAL_MS pause and signs the cashier in once for
concurrent jobs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-09-24 21:25:59 +03:00
co-authored by Claude Opus 5.5
parent f88b5cccc0
commit 528a869657
9 changed files with 220 additions and 23 deletions
+2
View File
@@ -67,6 +67,8 @@ NOVA_POSHTA_API_KEY=change-me-novaposhta-apikey
CHECKBOX_BASE_URL=https://api.checkbox.ua
CHECKBOX_CLIENT_NAME=lux_fiscal
CHECKBOX_CLIENT_VERSION=0.1.0
# Пауза между запросами к Checkbox (мс): Новая Почта за Checkbox ограничивает частоту.
CHECKBOX_MIN_REQUEST_INTERVAL_MS=1000
# ЕТТН-чеки на тестовой кассе Checkbox не работают: локально весь цикл
# прогоняется через стаб (чек «фискализируется» на втором опросе). В production запрещено.
CHECKBOX_USE_STUB=false
+1
View File
@@ -126,4 +126,5 @@ Stages 1–4 (scaffolding, auth/audit, CRM order queue, Nova Poshta tracking) ar
- Two-phase create: `POST /receipts` writes `Receipt(pending)` + audit and commits, then enqueues `create_ettn_receipt`; the worker calls Checkbox. A timeout leaves the row `pending` with `error` set — the retry first looks the TTN up via `find_ettn` instead of blindly re-posting (would create a second receipt). Keep this.
- State machine and the "one live receipt per order" partial unique index live in `app/db/models/receipt.py`. `orders.receipt_created_at` is set on request and reset to NULL when a receipt ends `failed`/`cancelled` (order goes back to the queue).
- Once Checkbox accepts the receipt, the order is moved to `PACKED` in the CRM (`services/receipts.sync_crm_statuses`, marked by `receipts.crm_status_set_at`; runs right after creation and is retried by cron). The live exoCRM `SetStatus` differs from its docs: params must be `{"Orders": [id], "Status": ...}` (the documented `{"ID": ...}` returns "Undefined order list."), and the reply has no `status: OK` — success is `{"<id>": {"Status": "Success"}}`.
- Nova Poshta's rate limit comes back through Checkbox as a 4xx with `code=third_party.generic` and «To many requests» / `20000401501`, not as a 429. `http_client._transient_error` maps it to `CheckboxRateLimitedError`: the receipt stays `pending` and the worker retries with `arq.Retry`. The client also sends requests one at a time with a `CHECKBOX_MIN_REQUEST_INTERVAL_MS` pause, so don't parallelize Checkbox calls in the worker.
- ETTN does **not** work on a Checkbox test cash register. Locally use `CHECKBOX_USE_STUB=true`; client selection is only in `services/checkbox/client.get_checkbox_client()`.
+3
View File
@@ -69,6 +69,9 @@ class Settings(BaseSettings):
checkbox_base_url: str = "https://api.checkbox.ua"
checkbox_client_name: str = "lux_fiscal"
checkbox_client_version: str = "0.1.0"
# Пауза между запросами к Checkbox из одного процесса: при создании ЕТТН
# Checkbox ходит в API Новой Почты, а та отвечает «To many requests» на частые вызовы.
checkbox_min_request_interval_ms: int = 1000
# Стаб вместо реального Checkbox: ЕТТН работает только на боевой кассе,
# поэтому локально весь цикл прогоняется через стаб. В проде запрещено.
checkbox_use_stub: bool = False
+12
View File
@@ -23,6 +23,18 @@ class CheckboxUnavailableError(CheckboxError):
"""Сеть, таймаут или 5xx — исход запроса неизвестен, можно повторить."""
class CheckboxRateLimitedError(CheckboxUnavailableError):
"""Лимит частоты запросов (Checkbox 429 или «To many requests» от Новой Почты).
В отличие от прочих `CheckboxUnavailableError` исход известен — запрос отклонён,
повторять можно не раньше чем через `retry_after` секунд.
"""
def __init__(self, message: str, retry_after: float) -> None:
super().__init__(message)
self.retry_after = retry_after
@dataclass(frozen=True)
class CheckboxCredentials:
"""Расшифрованные доступы одной кассы. Живут только в памяти."""
+74 -12
View File
@@ -2,10 +2,17 @@
Авторизация — токен кассира по PIN-коду (`/api/v1/cashier/signinPinCode`).
Токен кэшируется в памяти процесса по кассе; на 401 — один повторный вход.
Запросы из одного процесса идут строго по одному с паузой
`CHECKBOX_MIN_REQUEST_INTERVAL_MS`: при создании ЕТТН Checkbox синхронно ходит
в API Новой Почты, а та на частые вызовы отвечает «To many requests».
"""
from __future__ import annotations
import asyncio
import re
import time
import uuid
from datetime import UTC, datetime, timedelta
from typing import Any
@@ -17,6 +24,7 @@ from app.schemas.checkbox import EttnOut, EttnStatus
from app.services.checkbox.client import (
CheckboxCredentials,
CheckboxError,
CheckboxRateLimitedError,
CheckboxUnavailableError,
)
@@ -27,6 +35,15 @@ _TIMEOUT = httpx.Timeout(30, connect=10)
_FIND_PAGES = 3
_FIND_PAGE_SIZE = 50
# Ошибки Новой Почты, которые Checkbox отдаёт 4xx с `code=third_party.*`:
# {"message": "To many requests", "info": ["Try again after 1 seconds"],
# "errorCodes": ["20000401501"]} — лимит частоты НП;
# "cURL error 28: SSL connection timeout" — Checkbox не дождался НП.
_NP_RATE_LIMIT_RE = re.compile(r"20000401501|too? many requests", re.IGNORECASE)
_NP_RETRY_AFTER_RE = re.compile(r"try again after (\d+) second", re.IGNORECASE)
_NP_TIMEOUT_RE = re.compile(r"curl error|timed? ?out", re.IGNORECASE)
_MIN_RETRY_AFTER = 1.0
def _error_message(response: httpx.Response) -> str:
try:
@@ -55,6 +72,29 @@ def _error_message(response: httpx.Response) -> str:
return f"HTTP {response.status_code}"
def _retry_after(response: httpx.Response) -> float:
seconds: float = 0
if match := _NP_RETRY_AFTER_RE.search(response.text):
seconds = float(match.group(1))
elif (header := response.headers.get("Retry-After", "")).isdigit():
seconds = float(header)
return max(seconds, _MIN_RETRY_AFTER)
def _transient_error(response: httpx.Response) -> CheckboxUnavailableError | None:
"""Ответ, после которого запрос можно повторить, или None, если ошибка окончательная."""
if response.status_code == 429:
return CheckboxRateLimitedError(_error_message(response), _retry_after(response))
if response.status_code >= 500:
return CheckboxUnavailableError(_error_message(response))
if response.status_code >= 400 and '"third_party.' in response.text:
if _NP_RATE_LIMIT_RE.search(response.text):
return CheckboxRateLimitedError(_error_message(response), _retry_after(response))
if _NP_TIMEOUT_RE.search(response.text):
return CheckboxUnavailableError(_error_message(response))
return None
class HttpCheckboxClient:
def __init__(self, settings: Settings) -> None:
self._base_url = settings.checkbox_base_url.rstrip("/")
@@ -63,6 +103,10 @@ class HttpCheckboxClient:
"X-Client-Version": settings.checkbox_client_version,
}
self._tokens: dict[uuid.UUID, str] = {}
self._min_interval = settings.checkbox_min_request_interval_ms / 1000
self._send_lock = asyncio.Lock()
self._sign_in_lock = asyncio.Lock()
self._last_sent_at = float("-inf")
async def _send(
self,
@@ -73,18 +117,36 @@ class HttpCheckboxClient:
json: Any = None,
params: dict[str, Any] | None = None,
) -> httpx.Response:
try:
async with httpx.AsyncClient(base_url=self._base_url, timeout=_TIMEOUT) as client:
response = await client.request(
method, path, headers=headers, json=json, params=params
)
except httpx.HTTPError as exc:
raise CheckboxUnavailableError(f"Checkbox недоступен: {exc!r}") from exc
# 429 — «Занадто часто виконуються запити»: исход не ошибка данных, повторяемо.
if response.status_code >= 500 or response.status_code == 429:
raise CheckboxUnavailableError(_error_message(response))
# Строго по одному запросу с паузой — параллельные задачи worker'а
# иначе пачкой упираются в лимит Новой Почты.
async with self._send_lock:
delay = self._last_sent_at + self._min_interval - time.monotonic()
if delay > 0:
await asyncio.sleep(delay)
self._last_sent_at = time.monotonic()
try:
async with httpx.AsyncClient(base_url=self._base_url, timeout=_TIMEOUT) as client:
response = await client.request(
method, path, headers=headers, json=json, params=params
)
except httpx.HTTPError as exc:
raise CheckboxUnavailableError(f"Checkbox недоступен: {exc!r}") from exc
if error := _transient_error(response):
raise error
return response
async def _token(self, creds: CheckboxCredentials, *, stale: str | None = None) -> str:
"""Токен кассира; `stale` — отвергнутый Checkbox'ом (401), его не переиспользуем.
Под блокировкой: параллельные задачи без токена входят один раз,
остальные берут токен, полученный первой.
"""
async with self._sign_in_lock:
token = self._tokens.get(creds.cash_register_id)
if token is not None and token != stale:
return token
return await self._sign_in(creds)
async def _sign_in(self, creds: CheckboxCredentials) -> str:
response = await self._send(
"POST",
@@ -107,7 +169,7 @@ class HttpCheckboxClient:
json: Any = None,
params: dict[str, Any] | None = None,
) -> httpx.Response:
token = self._tokens.get(creds.cash_register_id) or await self._sign_in(creds)
token = await self._token(creds)
for attempt in range(2):
response = await self._send(
method,
@@ -121,7 +183,7 @@ class HttpCheckboxClient:
params=params,
)
if response.status_code == 401 and attempt == 0:
token = await self._sign_in(creds)
token = await self._token(creds, stale=token)
continue
break
if response.status_code >= 400:
+15 -2
View File
@@ -38,6 +38,7 @@ from app.services.checkbox.client import (
CheckboxClient,
CheckboxCredentials,
CheckboxError,
CheckboxRateLimitedError,
CheckboxUnavailableError,
)
from app.services.crm.client import CrmClient, CrmError
@@ -387,7 +388,10 @@ def _apply_ettn(receipt: Receipt, ettn: EttnOut) -> None:
async def create_ettn_for_receipt(
session: AsyncSession, client: CheckboxClient, receipt_id: uuid.UUID
) -> Receipt | None:
"""Отправляет `pending`-чек в Checkbox. Идемпотентна — безопасно вызывать повторно."""
"""Отправляет `pending`-чек в Checkbox. Идемпотентна — безопасно вызывать повторно.
На лимит частоты пробрасывает `CheckboxRateLimitedError`, чек остаётся `pending`.
"""
receipt = await session.get(Receipt, receipt_id, with_for_update=True)
if receipt is None or receipt.status != ReceiptStatus.PENDING:
return receipt
@@ -403,6 +407,12 @@ async def create_ettn_for_receipt(
await client.find_ettn(creds, receipt.waybill_number) if receipt.error else None
)
ettn = existing or await client.create_ettn(creds, receipt.request_body)
except CheckboxRateLimitedError as exc:
# Запрос отклонён, чек точно не создан: остаётся `pending` без пометки
# «исход неизвестен». Повтор с задержкой — забота вызывающего (worker).
await session.commit() # снять блокировку строки
log.info("ettn_rate_limited", receipt_id=str(receipt_id), retry_after=exc.retry_after)
raise
except CheckboxUnavailableError as exc:
receipt.error = str(exc)
await session.commit()
@@ -435,7 +445,10 @@ async def retry_pending_receipts(session: AsyncSession, client: CheckboxClient)
)
)
for receipt_id in receipt_ids:
await create_ettn_for_receipt(session, client, receipt_id)
try:
await create_ettn_for_receipt(session, client, receipt_id)
except CheckboxRateLimitedError:
break # остальные — в следующем проходе cron'а
# --- Отмена и опрос ----------------------------------------------------------
+15 -6
View File
@@ -12,20 +12,24 @@ from __future__ import annotations
import uuid
from typing import Any
from arq import cron
from arq import Retry, cron, func
from arq.connections import RedisSettings
from app.core.config import settings
from app.core.logging import configure_logging, get_logger
from app.db.session import SessionFactory
from app.services import receipts as receipts_service
from app.services.checkbox.client import get_checkbox_client
from app.services.checkbox.client import CheckboxRateLimitedError, get_checkbox_client
from app.services.crm.exo_client import ExoCrmClient
from app.services.nova_poshta.np_client import NpTrackingClient
from app.services.orders import sync_np_statuses
log = get_logger(__name__)
# Повторы `create_ettn_receipt` при лимите частоты: пауза retry_after + 5 с × номер попытки.
_RATE_LIMIT_BACKOFF = 5
_CREATE_ETTN_MAX_TRIES = 6
async def startup(ctx: dict[str, Any]) -> None:
configure_logging()
@@ -43,9 +47,14 @@ async def poll_np_statuses(ctx: dict[str, Any]) -> None:
async def create_ettn_receipt(ctx: dict[str, Any], receipt_id: str) -> None:
async with SessionFactory() as session:
await receipts_service.create_ettn_for_receipt(
session, ctx["checkbox_client"], uuid.UUID(receipt_id)
)
try:
await receipts_service.create_ettn_for_receipt(
session, ctx["checkbox_client"], uuid.UUID(receipt_id)
)
except CheckboxRateLimitedError as exc:
# Лимит частоты Новой Почты: повтор с нарастающей паузой. Когда попытки
# кончатся, `pending`-чек подберёт cron `poll_receipts`.
raise Retry(defer=exc.retry_after + _RATE_LIMIT_BACKOFF * ctx["job_try"]) from exc
# Чек принят Checkbox — сразу переводим заказ в CRM в PACKED.
await receipts_service.sync_crm_statuses(session, ctx["crm_client"])
@@ -61,7 +70,7 @@ async def poll_receipts(ctx: dict[str, Any]) -> None:
class WorkerSettings:
redis_settings = RedisSettings.from_dsn(settings.redis_url)
on_startup = startup
functions = [create_ettn_receipt]
functions = [func(create_ettn_receipt, max_tries=_CREATE_ETTN_MAX_TRIES)]
cron_jobs = [
cron(poll_np_statuses, minute=set(range(60)), run_at_startup=True),
cron(poll_receipts, minute=set(range(60)), second=30),
+79 -2
View File
@@ -2,7 +2,9 @@
from __future__ import annotations
import asyncio
import json
import time
import uuid
import httpx
@@ -14,6 +16,7 @@ from app.core.config import Settings
from app.services.checkbox.client import (
CheckboxCredentials,
CheckboxError,
CheckboxRateLimitedError,
CheckboxUnavailableError,
)
from app.services.checkbox.http_client import HttpCheckboxClient
@@ -35,11 +38,12 @@ ETTN = {
}
def _client() -> HttpCheckboxClient:
def _client(interval_ms: int = 0) -> HttpCheckboxClient:
settings = Settings(
secret_key="test-secret-key",
encryption_key="dGVzdC1lbmNyeXB0aW9uLWtleS0zMi1ieXRlcyEh",
checkbox_base_url=BASE,
checkbox_min_request_interval_ms=interval_ms,
) # type: ignore[arg-type]
return HttpCheckboxClient(settings)
@@ -178,5 +182,78 @@ async def test_rate_limit_is_retryable() -> None:
respx.post(f"{BASE}/api/v1/ettn").mock(
return_value=Response(429, json={"message": "Занадто часто виконуються запити"})
)
with pytest.raises(CheckboxUnavailableError):
with pytest.raises(CheckboxRateLimitedError):
await _client().create_ettn(CREDS, {})
def _np_error(message: str, info: list[str], codes: list[str]) -> Response:
"""Ошибка Новой Почты в том виде, в каком её отдаёт боевой Checkbox."""
inner = {
"type": "ettn",
"message": message,
"code": 0,
"context": {"success": False, "errors": [message], "info": info, "errorCodes": codes},
}
return Response(400, json={"code": "third_party.generic", "message": json.dumps(inner)})
@respx.mock
async def test_np_too_many_requests_is_rate_limited() -> None:
_signin()
respx.post(f"{BASE}/api/v1/ettn").mock(
return_value=_np_error("To many requests", ["Try again after 3 seconds"], ["20000401501"])
)
with pytest.raises(CheckboxRateLimitedError) as exc_info:
await _client().create_ettn(CREDS, {})
assert exc_info.value.retry_after == 3
@respx.mock
async def test_np_zero_retry_after_waits_at_least_a_second() -> None:
_signin()
respx.post(f"{BASE}/api/v1/ettn").mock(
return_value=_np_error("To many requests", ["Try again after 0 seconds"], ["20000401501"])
)
with pytest.raises(CheckboxRateLimitedError) as exc_info:
await _client().create_ettn(CREDS, {})
assert exc_info.value.retry_after == 1
@respx.mock
async def test_np_timeout_is_unavailable_not_rate_limited() -> None:
_signin()
respx.post(f"{BASE}/api/v1/ettn").mock(
return_value=_np_error("cURL error 28: SSL connection timeout", [], [])
)
with pytest.raises(CheckboxUnavailableError) as exc_info:
await _client().create_ettn(CREDS, {})
assert not isinstance(exc_info.value, CheckboxRateLimitedError)
@respx.mock
async def test_concurrent_calls_sign_in_once() -> None:
signin = _signin()
respx.get(f"{BASE}/api/v1/ettn/e-1").mock(return_value=Response(200, json=ETTN))
client = _client()
await asyncio.gather(*(client.get_ettn(CREDS, "e-1") for _ in range(5)))
assert signin.call_count == 1
@respx.mock
async def test_requests_are_spaced_by_min_interval() -> None:
_signin()
sent_at: list[float] = []
def record(_: httpx.Request) -> Response:
sent_at.append(time.monotonic())
return Response(200, json=ETTN)
respx.get(f"{BASE}/api/v1/ettn/e-1").mock(side_effect=record)
client = _client(interval_ms=50)
await asyncio.gather(*(client.get_ettn(CREDS, "e-1") for _ in range(3)))
gaps = [b - a for a, b in zip(sent_at, sent_at[1:], strict=False)]
assert len(gaps) == 2 and all(gap >= 0.045 for gap in gaps)
+19 -1
View File
@@ -18,7 +18,7 @@ from app.db.models.order import Order
from app.db.models.receipt import Receipt, ReceiptStatus
from app.schemas.checkbox import EttnStatus
from app.services import receipts as svc
from app.services.checkbox.client import CheckboxUnavailableError
from app.services.checkbox.client import CheckboxRateLimitedError, CheckboxUnavailableError
from app.services.checkbox.stub_client import StubCheckboxClient
@@ -266,6 +266,24 @@ class TestCreateEttn:
assert receipt.status == ReceiptStatus.CREATED
assert len(client.orders) == 1 # второго чека на ту же ТТН нет
async def test_rate_limit_keeps_pending_without_unknown_outcome(self) -> None:
order, register = _order(receipt_created_at=datetime.now(UTC)), _register()
receipt = _pending(order, register)
client = StubCheckboxClient()
session = FakeSession(order, register, receipt)
async def rate_limited(creds: Any, body: dict[str, Any]) -> Any:
raise CheckboxRateLimitedError("To many requests", retry_after=1)
client.create_ettn = rate_limited # type: ignore[method-assign]
with pytest.raises(CheckboxRateLimitedError):
await svc.create_ettn_for_receipt(session, client, receipt.id)
assert receipt.status == ReceiptStatus.PENDING
assert receipt.error is None # повтор не пойдёт через find_ettn
assert order.receipt_created_at is not None # заказ не вернулся в очередь
assert session.commits == 1 # блокировка строки снята
class TestApplyEttn:
@pytest.mark.parametrize(