RSHB/arm-tabular-parser
9
0
Fork 0
Code Issues 9 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: NATS — PUBLISH_RETRIES как повторы после первой попытки (R-16), flush после подписки, снятие подписок при сбое on_connected, проброс отмены в stop(), error-лог самопроизвольного закрытия

Browse Source 
Находки ревью Task 8: T8.2–T8.6.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-04 02:57:23 +04:00
parent a8dc5cc314
commit 6c8d7b6e9b
5 changed files with 150 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -37,7 +37,7 @@ Stateless-сервис распознавания табличных файло
5. Результат (схема `recognition-result`) записывается в S3 под ключом `results/{file_id}/{job_id}/result.json`.
6. В `recognition.result` публикуется конверт результата (схема `recognition-result-message`; статусы
`processing | done | needs_check | error`) с заголовком `Nats-Msg-Id = result:<job_id>:<attempt>`. Публикация
повторяется до `PUBLISH_RETRIES` раз.
при неудаче повторяется до `PUBLISH_RETRIES` раз (после первой попытки, с паузой 2 с).
Ошибки транспорта процесс не роняют, а превращаются в конверт `status = error`: нечитаемый или несовместимый запрос —
`SCHEMA_MISMATCH` (ack при этом не публикуется), объект S3 не читается — `FILE_READ_ERROR`, исключение парсера —
@ -133,7 +133,7 @@ curl -s localhost:8000/healthcheck
| `ENV_FILE` | `.env` | путь к файлу окружения (от `api/`) |
| `NATS_ENABLED` | `true` | `false` — не подключаться к NATS, воркер не стартует |
| `WORKER_CONCURRENCY` | `1` | сколько заданий обрабатывается одновременно |
| `PUBLISH_RETRIES` | `3` | попытки публикации ack и результата в NATS |
| `PUBLISH_RETRIES` | `3` | повторы публикации ack и результата в NATS после первой попытки (пауза 2 с) |
| `DONE_CACHE_SIZE` | `1000` | размер LRU обработанных `job_id` (дедупликация внутри процесса) |
| `SERVER__HOST` | `0.0.0.0` | адрес привязки HTTP-сервера |
| `SERVER__PORT` | `8000` | порт HTTP-сервера |

2
api/.env.example
View File

@ -31,7 +31,7 @@ SERVICE_NAME=tabular-parser
NATS_ENABLED=true
# Число заданий, обрабатываемых одновременно
WORKER_CONCURRENCY=1
# Попытки публикации ack и результата в NATS
# Повторы публикации ack и результата в NATS после первой попытки (3 — всего до 4 попыток, пауза 2 с)
PUBLISH_RETRIES=3
# Размер process-local LRU обработанных job_id (защита от дублей доставки)
DONE_CACHE_SIZE=1000

2
api/src/core/config.py
View File

@ -44,7 +44,7 @@ class Settings(BaseSettings):
SERVICE_NAME: str = "tabular-parser" # `service.name` конвертов контракта и имя queue group NATS
NATS_ENABLED: bool = True # false — приложение не подключается к NATS (воркер не стартует)
WORKER_CONCURRENCY: PositiveInt = 1 # число заданий, обрабатываемых одновременно
PUBLISH_RETRIES: PositiveInt = 3 # попытки публикации ack/результата в NATS
PUBLISH_RETRIES: PositiveInt = 3 # повторы публикации ack/результата в NATS после первой попытки (до 4 попыток)
DONE_CACHE_SIZE: PositiveInt = 1000 # размер process-local LRU обработанных job_id
model_config = SettingsConfigDict(

66
api/src/core/nats.py
View File

@ -8,12 +8,15 @@
`max_reconnect_attempts=-1` сам повторяет и первое подключение (параметра `retry_on_failed_connect` в 2.14 нет:
`connect` просто не возвращается, пока сервер недоступен), поэтому ждать его в lifespan нельзя. Если `connect` всё же
завершился исключением, supervisor повторяет попытку с паузой 1 → 2 → 4 → … ≤ 30 с. После подключения один раз
вызывается `on_connected(conn)` (подписки воркера); подписки после разрыва nats-py восстанавливает сам.
вызывается `on_connected(conn)` (подписки воркера); если он упал, созданные в этой попытке подписки снимаются и
вызов повторяется с той же паузой. Подписки после разрыва nats-py восстанавливает сам.
- `subscribe_requests(cb)` — подписка на `recognition.tabular.request` в queue group `tabular-parser`: из нескольких
инстанций сообщение получает одна. Исключения обработчика логируются, подписка живёт.
- `publish(subject, payload, msg_id)` — заголовок `Nats-Msg-Id` и `flush` (подтверждение сервером); до
`publish_retries` попыток с паузой `publish_retry_pause` между ними (D-5: 3 попытки, 2 с), затем `NatsPublishError` —
воркер логирует ошибку, задание дойдёт до `timeout` в ядре.
инстанций сообщение получает одна. После `subscribe` — `flush`: подписка считается активной, когда сервер принял SUB
(ошибки прав проявляются сразу). Исключения обработчика логируются, подписка живёт.
- `publish(subject, payload, msg_id)` — заголовок `Nats-Msg-Id` и `flush` (подтверждение сервером); первая попытка и
до `publish_retries` повторов с тем же `msg_id` и паузой `publish_retry_pause` между попытками (D-5, R-16: 3 повтора,
2 с — всего до 4 попыток), затем `NatsPublishError` — воркер логирует ошибку, задание дойдёт до `timeout` в ядре.
Нет соединения — попытка неудачна сразу (без ожидания), пауза и повтор как обычно.
- Ошибки подключения логируются без шума: первая ошибка серии — warning, дальше не чаще раза в минуту.
"""
@ -38,7 +41,7 @@ MAX_RECONNECT_ATTEMPTS = -1 # переподключаться бесконеч
RECONNECT_TIME_WAIT = 2 # секунд между попытками переподключения nats-py
CONNECT_TIMEOUT = 2 # секунд на одну попытку TCP-подключения
FLUSH_TIMEOUT = 5 # секунд на подтверждение публикации сервером (PING/PONG)
PUBLISH_RETRIES = 3 # попыток публикации (settings.PUBLISH_RETRIES)
PUBLISH_RETRIES = 3 # повторов публикации после первой попытки (settings.PUBLISH_RETRIES, R-16)
PUBLISH_RETRY_PAUSE = 2.0 # секунд между попытками публикации (D-5)
BACKOFF_INITIAL = 1.0
BACKOFF_MAX = 30.0
@ -61,7 +64,7 @@ class NatsNotConnectedError(NatsError):
class NatsPublishError(NatsError):
"""Публикация не подтверждена за все попытки."""
"""Публикация не подтверждена ни первой попыткой, ни повторами."""
@runtime_checkable
@ -121,8 +124,8 @@ class NatsConnection:
backoff_max: float = BACKOFF_MAX,
flush_timeout: int = FLUSH_TIMEOUT,
) -> None:
if publish_retries < 1:
raise ValueError("publish_retries должен быть ≥ 1")
if publish_retries < 0:
raise ValueError("publish_retries должен быть ≥ 0")
self._settings = settings
self._on_connected = on_connected
self._publish_retries = publish_retries
@ -138,6 +141,7 @@ class NatsConnection:
self._stopping = False
self._error_streak = 0
self._last_error_log = 0.0
self._subscriptions: list[Any] = [] # подписки subscribe_requests (снимаются при сбое on_connected)
# --- состояние ------------------------------------------------------------------------------
@property
@ -162,10 +166,18 @@ class NatsConnection:
async def stop(self) -> None:
"""Останавливает supervisor и закрывает соединение; исключения наружу не выходят."""
self._stopping = True
if self._supervisor is not None:
self._supervisor.cancel()
with suppress(asyncio.CancelledError, Exception):
await self._supervisor
supervisor = self._supervisor
if supervisor is not None:
supervisor.cancel()
try:
await supervisor
except asyncio.CancelledError:
# Подавляем только отмену самого supervisor'а; отмену вызывающей задачи пробрасываем
current = asyncio.current_task()
if current is not None and current.cancelling():
raise
except Exception:
logger.exception("Supervisor NATS завершился с ошибкой. error_category=nats_error")
self._supervisor = None
client = self._client
if client is None or client.is_closed:
@ -230,6 +242,7 @@ class NatsConnection:
delay = self._backoff_initial
while self._on_connected is not None:
created_before = len(self._subscriptions)
try:
await self._on_connected(self)
break
@ -240,10 +253,18 @@ class NatsConnection:
error_type=type(exc).__name__,
error=repr(exc),
)
await self._drop_subscriptions(created_before)
await self._sleep(delay)
delay = min(delay * 2, self._backoff_max)
self._ready.set()
async def _drop_subscriptions(self, keep: int) -> None:
"""Снимает подписки сверх первых `keep` (неудачная попытка `on_connected`): повтор не дублирует их."""
stale, self._subscriptions = self._subscriptions[keep:], self._subscriptions[:keep]
for subscription in stale:
with suppress(Exception):
await subscription.unsubscribe()
# --- подписка и публикация ------------------------------------------------------------------
async def subscribe_requests(
self, cb: MessageHandler, *, queue: str = REQUEST_QUEUE_GROUP, subject: str = TABULAR_REQUEST_SUBJECT
@ -265,13 +286,17 @@ class NatsConnection:
)
subscription = await self._client.subscribe(subject, queue=queue, cb=safe_cb)
self._subscriptions.append(subscription)
# Подписка активна, когда сервер обработал SUB: ошибки прав/соединения проявляются здесь, а не позже
await self._client.flush(timeout=self._flush_timeout)
logger.info("Подписка NATS активна. subject={subject}, queue={queue}", subject=subject, queue=queue)
return subscription
async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
"""Публикует с `Nats-Msg-Id: <msg_id>`; до `publish_retries` попыток с паузой, затем `NatsPublishError`."""
"""`Nats-Msg-Id: <msg_id>`; попытка + до `publish_retries` повторов с паузой, затем `NatsPublishError`."""
attempts = self._publish_retries + 1
last_error: str = ""
for attempt in range(1, self._publish_retries + 1):
for attempt in range(1, attempts + 1):
if attempt > 1:
await self._sleep(self._publish_retry_pause)
try:
@ -285,12 +310,10 @@ class NatsConnection:
subject=subject,
msg_id=msg_id,
attempt=attempt,
attempts=self._publish_retries,
attempts=attempts,
error=last_error,
)
raise NatsPublishError(
f"Публикация {msg_id} в {subject} не удалась за {self._publish_retries} попыток: {last_error}"
)
raise NatsPublishError(f"Публикация {msg_id} в {subject} не удалась за {attempts} попыток: {last_error}")
async def _publish_once(self, subject: str, payload: bytes, msg_id: str) -> None:
if not self.is_connected:
@ -332,4 +355,7 @@ class NatsConnection:
logger.info("Соединение с NATS восстановлено")
async def _on_closed(self) -> None:
logger.info("Соединение с NATS закрыто")
if self._stopping:
logger.info("Соединение с NATS закрыто")
else:
logger.error("Соединение с NATS закрыто без остановки сервиса. error_category=nats_error")

108
api/tests/unit/test_nats.py
View File

@ -76,6 +76,19 @@ def test_build_nats_auth_valid_credentials_give_callbacks():
# --- поддельный клиент nats-py ------------------------------------------------------------------
class _FakeSubscription:
def __init__(self, client: "_FakeClient", subject: str, queue: str, cb: Any) -> None:
self.client = client
self.subject = subject
self.queue = queue
self.cb = cb
self.active = True
async def unsubscribe(self) -> None:
self.active = False
self.client.calls.append("unsubscribe")
class _FakeClient:
"""Минимальный двойник `nats.aio.client.Client`."""
@ -87,6 +100,7 @@ class _FakeClient:
self.subscriptions: list[dict[str, Any]] = []
self.published: list[dict[str, Any]] = []
self.drained = False
self.calls: list[str] = [] # порядок subscribe/flush/unsubscribe
async def connect(self, servers: str, **options: Any) -> None:
self.world.connect_calls += 1
@ -97,14 +111,17 @@ class _FakeClient:
raise OSError("Connection refused")
self.is_connected = True
async def subscribe(self, subject: str, queue: str = "", cb: Any = None) -> SimpleNamespace:
self.subscriptions.append({"subject": subject, "queue": queue, "cb": cb})
return SimpleNamespace(subject=subject, queue=queue)
async def subscribe(self, subject: str, queue: str = "", cb: Any = None) -> _FakeSubscription:
self.calls.append("subscribe")
subscription = _FakeSubscription(self, subject, queue, cb)
self.subscriptions.append({"subject": subject, "queue": queue, "cb": cb, "sub": subscription})
return subscription
async def publish(self, subject: str, payload: bytes = b"", headers: dict[str, str] | None = None) -> None:
self.published.append({"subject": subject, "payload": payload, "headers": headers})
async def flush(self, timeout: int = 10) -> None:
self.calls.append("flush")
if self.world.flush_failures > 0:
self.world.flush_failures -= 1
raise TimeoutError("flush timeout")
@ -114,6 +131,8 @@ class _FakeClient:
await self.close()
async def close(self) -> None:
if self.world.hang_close:
await asyncio.Event().wait() # медленная очистка соединения
self.is_connected = False
self.is_closed = True
@ -121,10 +140,13 @@ class _FakeClient:
class _FakeWorld:
"""Фабрика клиентов: первые `fail_times` подключений падают; `flush_failures` первых flush — таймаут."""
def __init__(self, fail_times: int = 0, flush_failures: int = 0, hang_connect: bool = False) -> None:
def __init__(
self, fail_times: int = 0, flush_failures: int = 0, hang_connect: bool = False, hang_close: bool = False
) -> None:
self.fail_times = fail_times
self.flush_failures = flush_failures
self.hang_connect = hang_connect
self.hang_close = hang_close
self.connect_calls = 0
self.clients: list[_FakeClient] = []
@ -246,6 +268,37 @@ async def test_stop_drains_connected_client():
assert not conn.is_connected
async def test_stop_propagates_cancellation_of_its_caller():
"""T8.5: отмена задачи, вызвавшей stop(), не проглатывается (подавляется только отмена самого supervisor'а)."""
world = _FakeWorld(hang_connect=True, hang_close=True)
conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep())
await conn.start()
await asyncio.sleep(0)
stop_task = asyncio.create_task(conn.stop())
for _ in range(5):
await asyncio.sleep(0) # supervisor отменён и висит в медленном close()
assert not stop_task.done()
stop_task.cancel()
# asyncio.wait, а не await: без исправления stop() проглатывает отмену и виснет в close()
await asyncio.wait({stop_task}, timeout=1)
assert stop_task.cancelled()
async def test_on_closed_logs_error_when_not_stopping_and_info_on_stop(log_records: list[Any]):
"""T8.6: самопроизвольное закрытие клиента — error; закрытие в stop() — info."""
world = _FakeWorld()
conn = await _connected(world, _RecordingSleep())
await conn._on_closed()
await conn.stop()
await conn._on_closed()
closed = [r for r in log_records if "Соединение с NATS закрыто" in r["message"]]
assert [r["level"].name for r in closed] == ["ERROR", "INFO"]
# --- подписка -----------------------------------------------------------------------------------
async def test_subscribe_requests_uses_tabular_subject_and_queue_group():
world = _FakeWorld()
@ -259,6 +312,42 @@ async def test_subscribe_requests_uses_tabular_subject_and_queue_group():
await conn.stop()
async def test_subscribe_requests_flushes_after_subscribe():
"""T8.3: подписка считается активной, когда сервер обработал SUB (flush после subscribe)."""
world = _FakeWorld()
conn = await _connected(world, _RecordingSleep())
await conn.subscribe_requests(_noop_handler)
assert world.clients[-1].calls == ["subscribe", "flush"]
await conn.stop()
async def test_on_connected_failure_is_retried_without_duplicate_subscriptions(log_records: list[Any]):
"""T8.4: on_connected упал после подписки — подписка снимается, повтор с паузой, в итоге одна активная подписка."""
world = _FakeWorld()
sleep = _RecordingSleep()
calls = 0
async def on_connected(conn: NatsConnection) -> None:
nonlocal calls
calls += 1
await conn.subscribe_requests(_noop_handler)
if calls == 1:
raise RuntimeError("сбой после подписки")
conn = NatsConnection(_nats_settings(), on_connected=on_connected, client_factory=world, sleep=sleep)
await conn.start()
assert await conn.wait_ready(timeout=2)
assert calls == 2 # noqa: PLR2004
assert sleep.delays == [1]
subscriptions = [s["sub"] for s in world.clients[-1].subscriptions]
assert [s.active for s in subscriptions] == [False, True]
assert any("Не удалось выполнить подписки NATS" in r["message"] for r in log_records)
await conn.stop()
async def test_subscribe_requests_handler_errors_are_logged_not_raised(log_records: list[Any]):
world = _FakeWorld()
conn = await _connected(world, _RecordingSleep())
@ -327,10 +416,13 @@ async def test_publish_retries_exhausted_raises_publish_error(log_records: list[
with pytest.raises(NatsPublishError, match="result:j-1:1"):
await conn.publish(RESULT_SUBJECT, b"{}", "result:j-1:1")
assert len(world.clients[-1].published) == 3 # noqa: PLR2004
assert sleep.delays == [0.5, 0.5] # пауза только между попытками
# PUBLISH_RETRIES=3 — три повтора после первой попытки (R-16): 4 попытки, 3 паузы, один и тот же Nats-Msg-Id
published = world.clients[-1].published
assert len(published) == 4 # noqa: PLR2004
assert {p["headers"][NATS_MSG_ID_HEADER] for p in published} == {"result:j-1:1"}
assert sleep.delays == [0.5, 0.5, 0.5] # пауза только между попытками
warnings = [r for r in log_records if r["level"].name == "WARNING" and "result:j-1:1" in r["message"]]
assert len(warnings) == 3 # noqa: PLR2004
assert len(warnings) == 4 # noqa: PLR2004
await conn.stop()
@ -341,7 +433,7 @@ async def test_publish_without_connection_retries_then_raises():
with pytest.raises(NatsPublishError, match="Нет соединения"):
await conn.publish(RESULT_SUBJECT, b"{}", "ack:j-1:1")
assert sleep.delays == [2.0]
assert sleep.delays == [2.0, 2.0] # 1 попытка + 2 повтора
async def test_nats_connection_satisfies_publisher_protocol():