diff --git a/README.md b/README.md index 38b931f..793221d 100644 --- a/README.md +++ b/README.md @@ -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::`. Публикация - повторяется до `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-сервера | diff --git a/api/.env.example b/api/.env.example index ef59d5c..17adc3a 100644 --- a/api/.env.example +++ b/api/.env.example @@ -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 diff --git a/api/src/core/config.py b/api/src/core/config.py index 9e4fd8c..191c48e 100644 --- a/api/src/core/config.py +++ b/api/src/core/config.py @@ -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( diff --git a/api/src/core/nats.py b/api/src/core/nats.py index 7692a67..b2c1f6a 100644 --- a/api/src/core/nats.py +++ b/api/src/core/nats.py @@ -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: `; до `publish_retries` попыток с паузой, затем `NatsPublishError`.""" + """`Nats-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") diff --git a/api/tests/unit/test_nats.py b/api/tests/unit/test_nats.py index d064662..42bf391 100644 --- a/api/tests/unit/test_nats.py +++ b/api/tests/unit/test_nats.py @@ -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():