From ff0890558073c542a23600e5f5d8c1de9f5c56bb Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 4 Oct 2026 02:47:17 +0400 Subject: [PATCH] =?UTF-8?q?feat:=20=D1=81=D0=BB=D0=BE=D0=B9=20NATS=20src/c?= =?UTF-8?q?ore/nats.py=20=E2=80=94=20=D1=84=D0=BE=D0=BD=D0=BE=D0=B2=D0=BE?= =?UTF-8?q?=D0=B5=20=D0=BF=D0=BE=D0=B4=D0=BA=D0=BB=D1=8E=D1=87=D0=B5=D0=BD?= =?UTF-8?q?=D0=B8=D0=B5=20(nkeys/=D0=B1=D0=B5=D0=B7),=20=D0=BF=D0=BE=D0=B4?= =?UTF-8?q?=D0=BF=D0=B8=D1=81=D0=BA=D0=B0=20=D0=BD=D0=B0=20=D0=B7=D0=B0?= =?UTF-8?q?=D0=BF=D1=80=D0=BE=D1=81=D1=8B=20=D0=B2=20queue=20group,=20?= =?UTF-8?q?=D0=BF=D1=83=D0=B1=D0=BB=D0=B8=D0=BA=D0=B0=D1=86=D0=B8=D1=8F=20?= =?UTF-8?q?=D1=81=20=D0=BF=D0=BE=D0=B2=D1=82=D0=BE=D1=80=D0=B0=D0=BC=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5.1 --- api/src/core/nats.py | 335 ++++++++++++++++++++++++++++++++++ api/tests/unit/test_nats.py | 349 ++++++++++++++++++++++++++++++++++++ 2 files changed, 684 insertions(+) create mode 100644 api/src/core/nats.py create mode 100644 api/tests/unit/test_nats.py diff --git a/api/src/core/nats.py b/api/src/core/nats.py new file mode 100644 index 0000000..7692a67 --- /dev/null +++ b/api/src/core/nats.py @@ -0,0 +1,335 @@ +"""Соединение сервиса с 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`: из нескольких + инстанций сообщение получает одна. Исключения обработчика логируются, подписка живёт. +- `publish(subject, payload, msg_id)` — заголовок `Nats-Msg-Id` и `flush` (подтверждение сервером); до + `publish_retries` попыток с паузой `publish_retry_pause` между ними (D-5: 3 попытки, 2 с), затем `NatsPublishError` — + воркер логирует ошибку, задание дойдёт до `timeout` в ядре. +- Ошибки подключения логируются без шума: первая ошибка серии — warning, дальше не чаще раза в минуту. +""" + +import asyncio +import base64 +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) +PUBLISH_RETRIES = 3 # попыток публикации (settings.PUBLISH_RETRIES) +PUBLISH_RETRY_PAUSE = 2.0 # секунд между попытками публикации (D-5) +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): + """Публикация не подтверждена за все попытки.""" + + +@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 < 1: + raise ValueError("publish_retries должен быть ≥ 1") + 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 + + # --- состояние ------------------------------------------------------------------------------ + @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 + 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 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) + 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`.""" + last_error: str = "" + for attempt in range(1, self._publish_retries + 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=self._publish_retries, + error=last_error, + ) + raise NatsPublishError( + f"Публикация {msg_id} в {subject} не удалась за {self._publish_retries} попыток: {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: + 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: + logger.info("Соединение с NATS закрыто") diff --git a/api/tests/unit/test_nats.py b/api/tests/unit/test_nats.py new file mode 100644 index 0000000..d064662 --- /dev/null +++ b/api/tests/unit/test_nats.py @@ -0,0 +1,349 @@ +"""Unit-тесты `src/core/nats.py` без сервера NATS: аутентификация, фоновое подключение, подписка, публикация. + +Клиент nats-py подменяется фабрикой (`client_factory`), паузы — инжектируемым `sleep` (design D-5, D-6; spec +«Соединение с NATS по правилам шаблона, без JetStream», «Результат по контракту» — повтор публикации). +""" + +import asyncio +import base64 +import os +from collections.abc import Iterator +from types import SimpleNamespace +from typing import Any + +import nkeys +import pytest +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 ( + REQUEST_QUEUE_GROUP, + NatsConfigError, + NatsConnection, + NatsNotConnectedError, + NatsPublishError, + Publisher, + build_nats_auth, +) + +JWT = "header.payload.signature" + + +def _nats_settings(seed: str = "", jwt: str = "") -> NATSSettings: + return NATSSettings(NATS_URL="nats://localhost:4222", NATS_SEED=seed, NATS_JWT=jwt) + + +def _valid_seed() -> str: + return nkeys.encode_seed(os.urandom(32), nkeys.PREFIX_BYTE_USER).decode("utf-8") + + +@pytest.fixture +def log_records() -> Iterator[list[Any]]: + """Записи loguru, выпущенные за время теста.""" + records: list[Any] = [] + handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG") + yield records + logger.remove(handler_id) + + +# --- build_nats_auth ---------------------------------------------------------------------------- +def test_build_nats_auth_empty_credentials_gives_no_kwargs(): + assert build_nats_auth(_nats_settings()) == {} + + +@pytest.mark.parametrize(("seed", "jwt"), [("SU-only-seed", ""), ("", "only-jwt")]) +def test_build_nats_auth_partial_credentials_is_config_error(seed: str, jwt: str): + with pytest.raises(NatsConfigError, match="NATS_SEED"): + build_nats_auth(_nats_settings(seed, jwt)) + + +def test_build_nats_auth_invalid_seed_is_config_error(): + with pytest.raises(NatsConfigError, match="NATS_SEED"): + build_nats_auth(_nats_settings("not-a-seed", JWT)) + + +def test_build_nats_auth_valid_credentials_give_callbacks(): + seed = _valid_seed() + auth = build_nats_auth(_nats_settings(seed, JWT)) + + assert set(auth) == {"user_jwt_cb", "signature_cb"} + assert auth["user_jwt_cb"]() == JWT.encode() + signature = auth["signature_cb"]("nonce") + # Подпись проверяется публичным ключом той же пары + keypair = nkeys.from_seed(seed.encode()) + assert keypair.verify(b"nonce", base64.b64decode(signature)) + + +# --- поддельный клиент nats-py ------------------------------------------------------------------ +class _FakeClient: + """Минимальный двойник `nats.aio.client.Client`.""" + + def __init__(self, world: "_FakeWorld") -> None: + self.world = world + self.is_connected = False + self.is_closed = False + self.connect_options: dict[str, Any] = {} + self.subscriptions: list[dict[str, Any]] = [] + self.published: list[dict[str, Any]] = [] + self.drained = False + + async def connect(self, servers: str, **options: Any) -> None: + self.world.connect_calls += 1 + self.connect_options = {"servers": servers, **options} + if self.world.hang_connect: + await asyncio.Event().wait() # NATS недоступен: nats-py 2.14 повторяет подключение бесконечно + if self.world.connect_calls <= self.world.fail_times: + 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 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: + if self.world.flush_failures > 0: + self.world.flush_failures -= 1 + raise TimeoutError("flush timeout") + + async def drain(self) -> None: + self.drained = True + await self.close() + + async def close(self) -> None: + self.is_connected = False + self.is_closed = True + + +class _FakeWorld: + """Фабрика клиентов: первые `fail_times` подключений падают; `flush_failures` первых flush — таймаут.""" + + def __init__(self, fail_times: int = 0, flush_failures: int = 0, hang_connect: bool = False) -> None: + self.fail_times = fail_times + self.flush_failures = flush_failures + self.hang_connect = hang_connect + self.connect_calls = 0 + self.clients: list[_FakeClient] = [] + + def __call__(self) -> _FakeClient: + client = _FakeClient(self) + self.clients.append(client) + return client + + +class _RecordingSleep: + """Подмена `asyncio.sleep`: запоминает паузы и не ждёт.""" + + def __init__(self) -> None: + self.delays: list[float] = [] + + async def __call__(self, delay: float) -> None: + self.delays.append(delay) + await asyncio.sleep(0) + + +async def _noop_handler(msg: Any) -> None: + return None + + +async def _connected(world: _FakeWorld, sleep: _RecordingSleep, **kwargs: Any) -> NatsConnection: + conn = NatsConnection(_nats_settings(), client_factory=world, sleep=sleep, **kwargs) + await conn.start() + assert await conn.wait_ready(timeout=2) + return conn + + +# --- подключение ------------------------------------------------------------------------------ +async def test_connect_without_credentials_infinite_reconnect_and_callbacks(): + world = _FakeWorld() + conn = await _connected(world, _RecordingSleep()) + + assert conn.is_connected + options = world.clients[-1].connect_options + assert options["servers"] == "nats://localhost:4222" + assert options["max_reconnect_attempts"] == -1 + assert options["name"] == "tabular-parser" + for callback in ("error_cb", "disconnected_cb", "reconnected_cb", "closed_cb"): + assert callable(options[callback]) + assert "user_jwt_cb" not in options + assert "signature_cb" not in options + await conn.stop() + + +async def test_connect_with_credentials_passes_nkeys_callbacks(): + world = _FakeWorld() + conn = NatsConnection(_nats_settings(_valid_seed(), JWT), client_factory=world, sleep=_RecordingSleep()) + await conn.start() + assert await conn.wait_ready(timeout=2) + + options = world.clients[-1].connect_options + assert options["user_jwt_cb"]() == JWT.encode() + assert callable(options["signature_cb"]) + await conn.stop() + + +async def test_supervisor_retries_with_backoff_then_connects_and_runs_on_connected_once(log_records: list[Any]): + world = _FakeWorld(fail_times=7) + sleep = _RecordingSleep() + called: list[NatsConnection] = [] + + async def on_connected(conn: NatsConnection) -> None: + called.append(conn) + await conn.subscribe_requests(_noop_handler) + + conn = NatsConnection(_nats_settings(), on_connected=on_connected, client_factory=world, sleep=sleep) + assert not conn.is_connected + await conn.start() + assert await conn.wait_ready(timeout=2) + + assert world.connect_calls == 8 # noqa: PLR2004 + assert sleep.delays == [1, 2, 4, 8, 16, 30, 30] + assert called == [conn] + # Первая ошибка серии — warning с причиной, остальные в пределах минуты подавлены + warnings = [r for r in log_records if r["level"].name == "WARNING"] + assert len(warnings) == 1 + assert "error_category=nats_error" in warnings[0]["message"] + await conn.stop() + + +async def test_start_does_not_wait_for_unavailable_nats_and_stop_is_clean(): + """Регрессия O-3: NATS недоступен при старте — start() возвращается сразу, stop() отменяет подключение.""" + world = _FakeWorld(hang_connect=True) + conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep()) + + await asyncio.wait_for(conn.start(), timeout=0.5) + await asyncio.sleep(0) + assert not conn.is_connected + assert not await conn.wait_ready(timeout=0.05) + + await asyncio.wait_for(conn.stop(), timeout=1) + assert world.clients[-1].is_closed + + +async def test_partial_credentials_stop_supervisor_with_config_error(log_records: list[Any]): + world = _FakeWorld() + conn = NatsConnection(_nats_settings(seed="", jwt=JWT), client_factory=world, sleep=_RecordingSleep()) + + await conn.start() + assert not await conn.wait_ready(timeout=0.1) + + assert world.connect_calls == 0 + errors = [r for r in log_records if r["level"].name == "ERROR"] + assert any("error_category=config_error" in r["message"] for r in errors) + await conn.stop() + + +async def test_stop_drains_connected_client(): + world = _FakeWorld() + conn = await _connected(world, _RecordingSleep()) + + await conn.stop() + + assert world.clients[-1].drained + assert not conn.is_connected + + +# --- подписка ----------------------------------------------------------------------------------- +async def test_subscribe_requests_uses_tabular_subject_and_queue_group(): + world = _FakeWorld() + conn = await _connected(world, _RecordingSleep()) + + await conn.subscribe_requests(_noop_handler) + + assert REQUEST_QUEUE_GROUP == "tabular-parser" + subscription = world.clients[-1].subscriptions[0] + assert (subscription["subject"], subscription["queue"]) == (TABULAR_REQUEST_SUBJECT, "tabular-parser") + 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()) + received: list[bytes] = [] + + async def handler(msg: Any) -> None: + if msg.data == b"boom": + raise ValueError("сломанное сообщение") + received.append(msg.data) + + await conn.subscribe_requests(handler) + safe_cb = world.clients[-1].subscriptions[0]["cb"] + await safe_cb(SimpleNamespace(subject=TABULAR_REQUEST_SUBJECT, data=b"boom")) + await safe_cb(SimpleNamespace(subject=TABULAR_REQUEST_SUBJECT, data=b"ok")) + + assert received == [b"ok"] + assert any("error_category=handler_error" in r["message"] for r in log_records) + await conn.stop() + + +async def test_subscribe_requests_without_connection_raises(): + conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld(), sleep=_RecordingSleep()) + + with pytest.raises(NatsNotConnectedError): + await conn.subscribe_requests(_noop_handler) + + +# --- публикация --------------------------------------------------------------------------------- +async def test_publish_adds_nats_msg_id_header_and_flushes(): + world = _FakeWorld() + sleep = _RecordingSleep() + conn = await _connected(world, sleep) + + await conn.publish(RESULT_SUBJECT, b'{"status": "processing"}', "ack:j-1:1") + + assert world.clients[-1].published == [ + { + "subject": RESULT_SUBJECT, + "payload": b'{"status": "processing"}', + "headers": {NATS_MSG_ID_HEADER: "ack:j-1:1"}, + } + ] + assert sleep.delays == [] + await conn.stop() + + +async def test_publish_retries_with_pause_then_succeeds_with_same_msg_id(): + world = _FakeWorld(flush_failures=1) + sleep = _RecordingSleep() + conn = await _connected(world, sleep) + + await conn.publish(RESULT_SUBJECT, b"{}", "result:j-1:1") + + published = world.clients[-1].published + assert len(published) == 2 # noqa: PLR2004 - первая попытка не подтверждена flush, вторая успешна + assert {p["headers"][NATS_MSG_ID_HEADER] for p in published} == {"result:j-1:1"} + assert sleep.delays == [2.0] + await conn.stop() + + +async def test_publish_retries_exhausted_raises_publish_error(log_records: list[Any]): + world = _FakeWorld(flush_failures=10) + sleep = _RecordingSleep() + conn = await _connected(world, sleep, publish_retries=3, publish_retry_pause=0.5) + + 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] # пауза только между попытками + 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 + await conn.stop() + + +async def test_publish_without_connection_retries_then_raises(): + sleep = _RecordingSleep() + conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld(), sleep=sleep, publish_retries=2) + + with pytest.raises(NatsPublishError, match="Нет соединения"): + await conn.publish(RESULT_SUBJECT, b"{}", "ack:j-1:1") + + assert sleep.delays == [2.0] + + +async def test_nats_connection_satisfies_publisher_protocol(): + publisher: Publisher = NatsConnection(_nats_settings(), client_factory=_FakeWorld()) + assert isinstance(publisher, Publisher)