From bf2f60b3b589455abb11f09a3e11ac4c9fdf01c8 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 4 Oct 2026 03:46:33 +0400 Subject: [PATCH] =?UTF-8?q?fix:=20NATS=20=E2=80=94=20=D0=BE=D1=82=D0=BA?= =?UTF-8?q?=D0=B0=D0=B7=20=D0=B2=20=D0=BF=D1=80=D0=B0=D0=B2=D0=B0=D1=85=20?= =?UTF-8?q?=D0=BB=D0=BE=D0=B3=D0=B8=D1=80=D1=83=D0=B5=D1=82=D1=81=D1=8F=20?= =?UTF-8?q?=D0=BE=D1=82=D0=B4=D0=B5=D0=BB=D1=8C=D0=BD=D0=BE;=20drain=20?= =?UTF-8?q?=D0=BE=D0=B3=D1=80=D0=B0=D0=BD=D0=B8=D1=87=D0=B5=D0=BD=2010=20?= =?UTF-8?q?=D1=81=20(I-3,=20M-7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit nats-py 2.14 при Permissions Violation не закрывает соединение и не роняет subscribe/flush/publish, а только вызывает error_cb. Раньше это логировалось как «Нет связи с NATS, повторяем подключение» с лимитом раз в минуту. Теперь _on_error распознаёт «permissions violation» и пишет error error_category=nats_permission с subject, каждый раз и без серии ошибок подключения. Исправлены ложные утверждения docstring («ошибки прав проявляются сразу»). connect получает drain_timeout=DRAIN_TIMEOUT (10 с вместо 30 с nats-py): STOP_TIMEOUT воркера + drain укладываются в 30 с платформы. Co-Authored-By: Claude Fable 5.1 --- api/src/app.py | 3 ++- api/src/core/nats.py | 26 +++++++++++++++++++++++--- api/tests/unit/test_nats.py | 35 +++++++++++++++++++++++++++++++++++ 3 files changed, 60 insertions(+), 4 deletions(-) diff --git a/api/src/app.py b/api/src/app.py index 1b4ca99..2956145 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -6,7 +6,8 @@ (`nats = ok | down | off`, R-17), обработчики исключений. Сервис без БД и фронта (DECISION-4). При `NATS_ENABLED=false` ни NATS, ни воркер не запускаются. Остановка: воркер перестаёт принимать запросы и -дорабатывает очередь (NATS ещё нужен для публикации результатов), затем закрывается NATS (drain), затем сессия S3. +дорабатывает очередь не дольше `STOP_TIMEOUT` (NATS ещё нужен для публикации результатов), затем закрывается NATS +(drain не дольше `DRAIN_TIMEOUT`), затем сессия S3 — в сумме в пределах 30 с, которые платформа даёт на завершение. """ from contextlib import asynccontextmanager diff --git a/api/src/core/nats.py b/api/src/core/nats.py index b2c1f6a..872baa5 100644 --- a/api/src/core/nats.py +++ b/api/src/core/nats.py @@ -11,17 +11,22 @@ вызывается `on_connected(conn)` (подписки воркера); если он упал, созданные в этой попытке подписки снимаются и вызов повторяется с той же паузой. Подписки после разрыва nats-py восстанавливает сам. - `subscribe_requests(cb)` — подписка на `recognition.tabular.request` в queue group `tabular-parser`: из нескольких - инстанций сообщение получает одна. После `subscribe` — `flush`: подписка считается активной, когда сервер принял SUB - (ошибки прав проявляются сразу). Исключения обработчика логируются, подписка живёт. + инстанций сообщение получает одна. После `subscribe` — `flush`: сервер обработал SUB, ошибки соединения проявляются + здесь. Отказ в правах (`Permissions Violation`) nats-py 2.14 так не проявляет: соединение остаётся открытым, + `subscribe`/`flush`/`publish` проходят, сервер лишь присылает `-ERR` в `error_cb` — он логируется отдельно уровнем + error (`error_category=nats_permission`, subject). Исключения обработчика логируются, подписка живёт. - `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, дальше не чаще раза в минуту. +- Остановка: `drain` ограничен `DRAIN_TIMEOUT` (у nats-py по умолчанию 30 с), чтобы вместе со `STOP_TIMEOUT` воркера + уложиться в 30 с, которые платформа даёт на завершение. """ import asyncio import base64 +import re import time from collections.abc import Awaitable, Callable from contextlib import suppress @@ -41,11 +46,14 @@ MAX_RECONNECT_ATTEMPTS = -1 # переподключаться бесконеч RECONNECT_TIME_WAIT = 2 # секунд между попытками переподключения nats-py CONNECT_TIMEOUT = 2 # секунд на одну попытку TCP-подключения FLUSH_TIMEOUT = 5 # секунд на подтверждение публикации сервером (PING/PONG) +DRAIN_TIMEOUT = 10 # секунд на drain при остановке (nats-py: 30); STOP_TIMEOUT воркера + drain ≤ 30 с платформы PUBLISH_RETRIES = 3 # повторов публикации после первой попытки (settings.PUBLISH_RETRIES, R-16) PUBLISH_RETRY_PAUSE = 2.0 # секунд между попытками публикации (D-5) BACKOFF_INITIAL = 1.0 BACKOFF_MAX = 30.0 ERROR_LOG_INTERVAL = 60.0 # не чаще раза в минуту повторяем warning об ошибках одной серии +PERMISSIONS_VIOLATION = "permissions violation" # текст `-ERR` сервера при отказе в правах (nats-py: в нижнем регистре) +_SUBJECT_IN_ERROR_RE = re.compile(r'(?:to|for)\s+"([^"]+)"') MessageHandler = Callable[[Msg], Awaitable[None]] SleepFn = Callable[[float], Awaitable[None]] @@ -213,6 +221,7 @@ class NatsConnection: max_reconnect_attempts=MAX_RECONNECT_ATTEMPTS, reconnect_time_wait=RECONNECT_TIME_WAIT, connect_timeout=CONNECT_TIMEOUT, + drain_timeout=DRAIN_TIMEOUT, **auth, ) except BaseException: @@ -287,7 +296,7 @@ class NatsConnection: subscription = await self._client.subscribe(subject, queue=queue, cb=safe_cb) self._subscriptions.append(subscription) - # Подписка активна, когда сервер обработал SUB: ошибки прав/соединения проявляются здесь, а не позже + # Сервер обработал SUB: ошибки соединения проявляются здесь; отказ в правах приходит только в error_cb await self._client.flush(timeout=self._flush_timeout) logger.info("Подписка NATS активна. subject={subject}, queue={queue}", subject=subject, queue=queue) return subscription @@ -344,6 +353,17 @@ class NatsConnection: self._last_error_log = 0.0 async def _on_error(self, exc: Exception) -> None: + text = str(exc) + if PERMISSIONS_VIOLATION in text.lower(): + # Соединение живо, повтор подключения не поможет: нужны права на subject (каждый отказ — в лог) + match = _SUBJECT_IN_ERROR_RE.search(text) + logger.error( + "NATS отказал в правах: подписка или публикация не работает. error_category=nats_permission, " + "subject={subject}, error={error}", + subject=match.group(1) if match else "-", + error=text, + ) + return self._report_error(exc) async def _on_disconnected(self) -> None: diff --git a/api/tests/unit/test_nats.py b/api/tests/unit/test_nats.py index 42bf391..8aa0b36 100644 --- a/api/tests/unit/test_nats.py +++ b/api/tests/unit/test_nats.py @@ -18,6 +18,7 @@ from loguru import logger from src.contract.subjects import NATS_MSG_ID_HEADER, RESULT_SUBJECT, TABULAR_REQUEST_SUBJECT from src.core.config import NATSSettings from src.core.nats import ( + DRAIN_TIMEOUT, REQUEST_QUEUE_GROUP, NatsConfigError, NatsConnection, @@ -26,8 +27,10 @@ from src.core.nats import ( Publisher, build_nats_auth, ) +from src.core.worker import STOP_TIMEOUT JWT = "header.payload.signature" +PLATFORM_STOP_GRACE = 30.0 # секунд, которые платформа даёт процессу на завершение def _nats_settings(seed: str = "", jwt: str = "") -> NATSSettings: @@ -188,6 +191,9 @@ async def test_connect_without_credentials_infinite_reconnect_and_callbacks(): assert options["servers"] == "nats://localhost:4222" assert options["max_reconnect_attempts"] == -1 assert options["name"] == "tabular-parser" + # drain при остановке ограничен, чтобы остановка уложилась в 30 с платформы (STOP_TIMEOUT воркера + drain) + assert options["drain_timeout"] == DRAIN_TIMEOUT + assert STOP_TIMEOUT + DRAIN_TIMEOUT <= PLATFORM_STOP_GRACE for callback in ("error_cb", "disconnected_cb", "reconnected_cb", "closed_cb"): assert callable(options[callback]) assert "user_jwt_cb" not in options @@ -300,6 +306,35 @@ async def test_on_closed_logs_error_when_not_stopping_and_info_on_stop(log_recor # --- подписка ----------------------------------------------------------------------------------- +async def test_permission_violation_is_logged_as_permission_error_not_reconnect(log_records: list[Any]): + """nats-py 2.14 при Permissions Violation не рвёт соединение, только вызывает error_cb (текст в нижнем регистре).""" + conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld(), sleep=_RecordingSleep()) + error = Exception('nats: permissions violation for subscription to "recognition.tabular.request"') + + repeats = 2 # повтор не глушится лимитом ошибок подключения + for _ in range(repeats): + await conn._on_error(error) + + permission = [r for r in log_records if "error_category=nats_permission" in r["message"]] + assert len(permission) == repeats + assert all(r["level"].name == "ERROR" for r in permission) + assert all("subject=recognition.tabular.request" in r["message"] for r in permission) + assert not any("повторяем подключение" in r["message"] for r in log_records) + assert conn._error_streak == 0 # серия ошибок подключения не затронута + + +async def test_connection_error_in_error_cb_keeps_rate_limited_reconnect_warning(log_records: list[Any]): + conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld(), sleep=_RecordingSleep()) + + await conn._on_error(OSError("Connection refused")) + await conn._on_error(OSError("Connection refused")) + + warnings = [r for r in log_records if "повторяем подключение" in r["message"]] + assert len(warnings) == 1 # вторая ошибка серии — не чаще раза в минуту + assert warnings[0]["level"].name == "WARNING" + assert not any("nats_permission" in r["message"] for r in log_records) + + async def test_subscribe_requests_uses_tabular_subject_and_queue_group(): world = _FakeWorld() conn = await _connected(world, _RecordingSleep())