"""Соединение сервиса с NATS: core NATS без JetStream (design D-2, D-5, D-6; spec «Соединение с NATS по правилам шаблона, без JetStream», «Подтверждение запроса до обработки», «Результат по контракту»). Сторона сервиса распознавания; образец — `arm-core/api/src/core/nats.py` (сторона ядра). - Аутентификация как в шаблоне: заданы `NATS_SEED` и `NATS_JWT` — nkeys/JWT; оба пусты — без учётных данных; задан только один — `NatsConfigError` (подключение не выполняется, ошибка в логе). - Первое подключение выполняет фоновый supervisor (`start()`): старт приложения не ждёт NATS. nats-py 2.14 с `max_reconnect_attempts=-1` сам повторяет и первое подключение (параметра `retry_on_failed_connect` в 2.14 нет: `connect` просто не возвращается, пока сервер недоступен), поэтому ждать его в lifespan нельзя. Если `connect` всё же завершился исключением, supervisor повторяет попытку с паузой 1 → 2 → 4 → … ≤ 30 с. После подключения один раз вызывается `on_connected(conn)` (подписки воркера); если он упал, созданные в этой попытке подписки снимаются и вызов повторяется с той же паузой. Подписки после разрыва nats-py восстанавливает сам. - `subscribe_requests(cb)` — подписка на `recognition.tabular.request` в queue group `tabular-parser`: из нескольких инстанций сообщение получает одна. После `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 from typing import Any, Protocol, runtime_checkable import nkeys from loguru import logger from nats.aio.client import Client as NATS from nats.aio.msg import Msg from src.contract.subjects import NATS_MSG_ID_HEADER, TABULAR_REQUEST_SUBJECT from src.core.config import NATSSettings REQUEST_QUEUE_GROUP = "tabular-parser" CLIENT_NAME = "tabular-parser" 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]] class NatsError(Exception): """Базовая ошибка слоя NATS сервиса.""" class NatsConfigError(NatsError, RuntimeError): """Ошибка конфигурации учётных данных NATS: повтор подключения бесполезен.""" class NatsNotConnectedError(NatsError): """Операция требует установленного соединения с NATS.""" class NatsPublishError(NatsError): """Публикация не подтверждена ни первой попыткой, ни повторами.""" @runtime_checkable class Publisher(Protocol): """Публикация в NATS, нужная воркеру (design D-7): исключение `NatsPublishError` — все попытки исчерпаны.""" async def publish(self, subject: str, payload: bytes, msg_id: str) -> None: ... def build_nats_auth(settings: NATSSettings) -> dict[str, Any]: """Параметры аутентификации для `Client.connect`. Пустые `NATS_SEED` и `NATS_JWT` — без учётных данных (docker compose, тесты): `{}`. Заданы оба — колбэки nkeys/JWT, как в шаблоне. Задан только один или seed не читается — `NatsConfigError`. """ seed = settings.NATS_SEED jwt = settings.NATS_JWT if not seed and not jwt: return {} if not seed or not jwt: raise NatsConfigError( "Неполные учётные данные NATS: NATS_SEED и NATS_JWT задаются вместе или не задаются вовсе" ) try: keypair = nkeys.from_seed(seed.encode("utf-8")) except Exception as exc: raise NatsConfigError(f"Некорректный NATS_SEED: {type(exc).__name__}") from None def user_jwt_cb() -> bytes: return jwt.encode("utf-8") def signature_cb(nonce: str | bytes) -> bytes: if isinstance(nonce, str): nonce = nonce.encode("utf-8") return base64.b64encode(keypair.sign(nonce)) return {"user_jwt_cb": user_jwt_cb, "signature_cb": signature_cb} class NatsConnection: """Соединение сервиса с NATS: фоновое подключение, подписка на запросы, публикация с повторами. Реализует `Publisher`. `client_factory` и `sleep` подменяются в тестах; `sleep` используется и для пауз переподключения supervisor'а, и для пауз между попытками публикации. """ def __init__( # noqa: PLR0913 - все параметры, кроме settings, именованные: подмены для тестов и тайминги self, settings: NATSSettings, *, on_connected: Callable[["NatsConnection"], Awaitable[None]] | None = None, publish_retries: int = PUBLISH_RETRIES, publish_retry_pause: float = PUBLISH_RETRY_PAUSE, client_factory: Callable[[], Any] = NATS, sleep: SleepFn = asyncio.sleep, backoff_initial: float = BACKOFF_INITIAL, backoff_max: float = BACKOFF_MAX, flush_timeout: int = FLUSH_TIMEOUT, ) -> None: if publish_retries < 0: raise ValueError("publish_retries должен быть ≥ 0") self._settings = settings self._on_connected = on_connected self._publish_retries = publish_retries self._publish_retry_pause = publish_retry_pause self._client_factory = client_factory self._sleep = sleep self._backoff_initial = backoff_initial self._backoff_max = backoff_max self._flush_timeout = flush_timeout self._client: Any = None self._supervisor: asyncio.Task[None] | None = None self._ready = asyncio.Event() self._stopping = False self._error_streak = 0 self._last_error_log = 0.0 self._subscriptions: list[Any] = [] # подписки subscribe_requests (снимаются при сбое on_connected) # --- состояние ------------------------------------------------------------------------------ @property def is_connected(self) -> bool: """Фактическое состояние соединения (False до первого подключения, во время переподключения и после stop).""" return self._client is not None and bool(self._client.is_connected) async def wait_ready(self, timeout: float) -> bool: """Ждёт подключения и выполнения `on_connected`; `False`, если не дождались за `timeout` секунд.""" try: await asyncio.wait_for(self._ready.wait(), timeout) except TimeoutError: return False return True # --- жизненный цикл ------------------------------------------------------------------------- async def start(self) -> None: """Запускает фоновый supervisor подключения; NATS не ждёт.""" if self._supervisor is None: self._supervisor = asyncio.create_task(self._supervise(), name="nats-supervisor") async def stop(self) -> None: """Останавливает supervisor и закрывает соединение; исключения наружу не выходят.""" self._stopping = True 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: return try: if client.is_connected: await client.drain() # дообработать полученное else: await client.close() except Exception as exc: logger.warning( "Ошибка при закрытии соединения NATS. error_category=nats_error, error_type={error_type}, " "error={error}", error_type=type(exc).__name__, error=repr(exc), ) with suppress(Exception): await client.close() async def _connect(self) -> None: """Одна попытка подключения новым клиентом (nats-py с `max_reconnect_attempts=-1` повторяет её сам).""" auth = build_nats_auth(self._settings) client = self._client_factory() self._client = client try: await client.connect( self._settings.NATS_URL, name=CLIENT_NAME, error_cb=self._on_error, disconnected_cb=self._on_disconnected, reconnected_cb=self._on_reconnected, closed_cb=self._on_closed, max_reconnect_attempts=MAX_RECONNECT_ATTEMPTS, reconnect_time_wait=RECONNECT_TIME_WAIT, connect_timeout=CONNECT_TIMEOUT, drain_timeout=DRAIN_TIMEOUT, **auth, ) except BaseException: with suppress(Exception): await client.close() raise async def _supervise(self) -> None: delay = self._backoff_initial while True: try: await self._connect() break except NatsConfigError as exc: logger.error( "Конфигурация NATS некорректна, подключение не выполняется. error_category=config_error, " "error={error}", error=str(exc), ) return except Exception as exc: self._report_error(exc) await self._sleep(delay) delay = min(delay * 2, self._backoff_max) self._reset_error_streak() logger.info("Подключение к NATS установлено. url={url}", url=self._settings.NATS_URL) delay = self._backoff_initial while self._on_connected is not None: created_before = len(self._subscriptions) try: await self._on_connected(self) break except Exception as exc: logger.error( "Не удалось выполнить подписки NATS. error_category=nats_error, error_type={error_type}, " "error={error}", 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 ) -> Any: """Подписка на запросы (`recognition.tabular.request`) в queue group; исключения `cb` логируются.""" if not self.is_connected: raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {subject} невозможна") async def safe_cb(msg: Msg) -> None: try: await cb(msg) except Exception as exc: logger.exception( "Ошибка обработки сообщения NATS. error_category=handler_error, subject={subject}, " "error_type={error_type}, error={error}", subject=msg.subject, error_type=type(exc).__name__, error=repr(exc), ) subscription = await self._client.subscribe(subject, queue=queue, cb=safe_cb) self._subscriptions.append(subscription) # Сервер обработал SUB: ошибки соединения проявляются здесь; отказ в правах приходит только в error_cb 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: `; попытка + до `publish_retries` повторов с паузой, затем `NatsPublishError`.""" attempts = self._publish_retries + 1 last_error: str = "" for attempt in range(1, attempts + 1): if attempt > 1: await self._sleep(self._publish_retry_pause) try: await self._publish_once(subject, payload, msg_id) return except NatsPublishError as exc: last_error = str(exc) logger.warning( "Публикация в NATS не удалась. error_category=nats_error, subject={subject}, msg_id={msg_id}, " "attempt={attempt}, attempts={attempts}, error={error}", subject=subject, msg_id=msg_id, attempt=attempt, attempts=attempts, error=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: raise NatsPublishError("Нет соединения с NATS") try: await self._client.publish(subject, payload, headers={NATS_MSG_ID_HEADER: msg_id}) await self._client.flush(timeout=self._flush_timeout) except Exception as exc: raise NatsPublishError(f"публикация не подтверждена ({type(exc).__name__})") from exc # --- колбэки nats-py и логирование без шума ------------------------------------------------- def _report_error(self, exc: BaseException) -> None: now = time.monotonic() self._error_streak += 1 if self._error_streak == 1 or now - self._last_error_log >= ERROR_LOG_INTERVAL: logger.warning( "Нет связи с NATS, повторяем подключение. error_category=nats_error, url={url}, " "error_type={error_type}, error={error}, attempts={attempts}", url=self._settings.NATS_URL, error_type=type(exc).__name__, error=repr(exc), attempts=self._error_streak, ) self._last_error_log = now def _reset_error_streak(self) -> None: self._error_streak = 0 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: if not self._stopping: logger.warning("Соединение с NATS потеряно, nats-py переподключается") async def _on_reconnected(self) -> None: self._reset_error_streak() logger.info("Соединение с NATS восстановлено") async def _on_closed(self) -> None: if self._stopping: logger.info("Соединение с NATS закрыто") else: logger.error("Соединение с NATS закрыто без остановки сервиса. error_category=nats_error")