"""Соединение с NATS: core NATS без JetStream (design D-6; spec recognition-job-exchange, «Соединение с NATS по шаблону, без JetStream»; docs/context/03-architecture.md, «Контракт обмена через NATS»). - Аутентификация как в шаблоне: заданы `NATS_SEED` и `NATS_JWT` — nkeys/JWT; оба пусты — без учётных данных. - `publish(subject, payload, msg_id)` — публикация с заголовком `Nats-Msg-Id` и `flush`: ошибка соединения проявляется сразу исключением `NatsPublishError` (outbox оставляет задание в `queued`). - `subscribe_results(handler)` — подписка на `files.tabular_parsed` в queue group `arm-core`: из нескольких инстансов ядра сообщение получает ровно один. Исключения обработчика логируются, подписка не падает. - Stream'ов и consumer'ов JetStream нет. Переподключение после разрыва — встроенное в nats-py (`max_reconnect_attempts=-1`, `reconnect_time_wait=2`), подписки nats-py восстанавливает сам. - Первое подключение выполняет фоновый supervisor (`start()`): старт приложения не ждёт NATS. nats-py 2.14 с `max_reconnect_attempts=-1` сам повторяет и первое подключение (раз в `reconnect_time_wait`), поэтому `connect` возвращается только при успехе; если он всё же завершился исключением, supervisor повторяет попытку с экспоненциальной паузой 1 → 2 → 4 → … ≤ 30 с. Ошибки подключения логируются без шума: первая ошибка серии — warning с причиной, дальше не чаще раза в минуту; серия сбрасывается при (пере)подключении. """ import asyncio import base64 import time from collections.abc import Awaitable, Callable from contextlib import suppress from typing import Any 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, RESULT_SUBJECT from src.core.config import NATSSettings RESULT_QUEUE_GROUP = "arm-core" CLIENT_NAME = "arm-core" MAX_RECONNECT_ATTEMPTS = -1 # переподключаться бесконечно RECONNECT_TIME_WAIT = 2 # секунд между попытками переподключения nats-py CONNECT_TIMEOUT = 2 # секунд на одну попытку TCP-подключения FLUSH_TIMEOUT = 5 # секунд на подтверждение публикации сервером (PING/PONG) BACKOFF_INITIAL = 1.0 BACKOFF_MAX = 30.0 ERROR_LOG_INTERVAL = 60.0 # не чаще раза в минуту повторяем warning об ошибках одной серии 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): """Публикация не подтверждена: соединения нет или сервер не ответил на flush.""" 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.get_secret_value() jwt = settings.NATS_JWT.get_secret_value() 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: фоновое первое подключение, публикация с `Nats-Msg-Id`, подписка на результаты. `on_connected(conn)` вызывается один раз после первого успешного подключения (подписки); при исключении повторяется с той же экспоненциальной паузой. `result_subject` и `queue` параметризуются для тестов — по умолчанию `files.tabular_parsed` и `arm-core`. """ def __init__( # noqa: PLR0913 - все параметры, кроме settings, именованные: подмены для тестов и тайминги self, settings: NATSSettings, *, on_connected: Callable[["NatsConnection"], Awaitable[None]] | None = None, result_subject: str = RESULT_SUBJECT, queue: str = RESULT_QUEUE_GROUP, client_factory: Callable[[], Any] = NATS, sleep: SleepFn = asyncio.sleep, backoff_initial: float = BACKOFF_INITIAL, backoff_max: float = BACKOFF_MAX, error_log_interval: float = ERROR_LOG_INTERVAL, flush_timeout: int = FLUSH_TIMEOUT, ) -> None: self._settings = settings self._on_connected = on_connected self.result_subject = result_subject self.queue = queue self._client_factory = client_factory self._sleep = sleep self._backoff_initial = backoff_initial self._backoff_max = backoff_max self._error_log_interval = error_log_interval self._flush_timeout = flush_timeout self._client: Any = None self._supervisor: asyncio.Task[None] | None = None self._ready = asyncio.Event() self._stopping = False # Серия ошибок подключения: сколько подавлено и когда последний раз писали warning self._error_streak = 0 self._last_error_log = 0.0 # --- состояние ------------------------------------------------------------------------------ @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 if self._supervisor is not None: self._supervisor.cancel() with suppress(asyncio.CancelledError, Exception): await self._supervisor self._supervisor = None await self.close() async def close(self) -> None: """Подключённый клиент — `drain` (дообработать полученное), иначе `close`; без исключений наружу.""" 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, **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: 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._sleep(delay) delay = min(delay * 2, self._backoff_max) self._ready.set() # --- публикация и подписка ------------------------------------------------------------------ async def publish(self, subject: str, payload: bytes, msg_id: str) -> None: """Публикует сообщение с заголовком `Nats-Msg-Id: ` и ждёт подтверждения сервера (`flush`).""" if not self.is_connected: raise NatsPublishError(f"Нет соединения с NATS: сообщение {msg_id} в {subject} не опубликовано") 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"Публикация в NATS не подтверждена: {subject}, {msg_id} ({type(exc).__name__})" ) from exc async def subscribe_results(self, handler: MessageHandler) -> Any: """Подписка на `result_subject` в queue group `queue`; исключения `handler` логируются, подписка живёт.""" if not self.is_connected: raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {self.result_subject} невозможна") async def safe_handler(msg: Msg) -> None: try: await handler(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(self.result_subject, queue=self.queue, cb=safe_handler) logger.info( "Подписка NATS активна. subject={subject}, queue={queue}", subject=self.result_subject, queue=self.queue ) return subscription # --- колбэки 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 >= self._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: self._report_error(exc) async def _on_disconnected(self) -> None: if self._stopping: return logger.warning("Соединение с NATS потеряно, nats-py переподключается") async def _on_reconnected(self) -> None: self._reset_error_streak() logger.info("Соединение с NATS восстановлено") async def _on_closed(self) -> None: logger.info("Соединение с NATS закрыто")