diff --git a/README.md b/README.md index 99037f8..7db9d47 100644 --- a/README.md +++ b/README.md @@ -211,8 +211,8 @@ docker compose -f ../docs/dev/docker-compose.yml down -v # полный сб | `OPENBAO__SMB__SMB_PATH` | — (обязательно) | заглушка SMB | | `OPENBAO__SMB__SMB_LOGIN` | — (обязательно) | заглушка SMB | | `OPENBAO__SMB__SMB_PASSWORD` | — (обязательно) | заглушка SMB | -| `NATS_ENABLED` | `false` | подключение к NATS и обмен через `recognition_job`; без него `/healthcheck` показывает `nats=off` | -| `OUTBOX_POLL_SECONDS` | `5` | период опроса outbox | +| `NATS_ENABLED` | `false` | подключение к NATS и обмен через `recognition_job`: фоновые outbox-публикатор и вахтёр сроков ответа (период 30 с, `processing`/`published` с прошедшим `reply_deadline` → `timeout`); без него фоновых задач нет, `/healthcheck` показывает `nats=off` | +| `OUTBOX_POLL_SECONDS` | `5` | период опроса outbox (`queued` → публикация → `published`) | | `ACK_TIMEOUT_SECONDS` | `30` | ожидание ack сервиса распознавания до повторной публикации | | `OUTBOX_MAX_REPUBLISH` | `3` | число повторных публикаций до перевода в `dlq` | | `AUTH_DEV_ISSUER` | `false` | dev-эмитент токенов: работает **только** при `AUTH_DEV_ISSUER=true` **и** `DEBUG=true` | diff --git a/api/src/app.py b/api/src/app.py index ba70527..980c82f 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -5,7 +5,8 @@ NATS без JetStream, подключение в фоне при NATS_ENABLED=tr group `arm-core`), SMB (горячие папки ИС Сканирование), middleware (gzip, proxy headers, CORS, РАИСА при DEBUG=False), healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает -состояние. Outbox-паблишер и вахтёр сроков — задача 23 (docs/context/03-architecture.md). +состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор и +вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) — и останавливает их до закрытия NATS. """ import os @@ -28,7 +29,8 @@ from src.core.clients import S3Storage from src.core.config import get_settings from src.core.exception_handlers import register_exception_handlers from src.core.nats import NatsConnection -from src.services.recognition_exchange import handle_message +from src.db.db import async_session_maker +from src.services.recognition_exchange import ExchangeTasks, handle_message settings = get_settings() @@ -102,10 +104,16 @@ def _create_nats() -> NatsConnection | None: return NatsConnection(settings.OPENBAO.NATS, on_connected=_subscribe_results) +def _start_exchange(nats: NatsConnection) -> ExchangeTasks: + """Фоновые задачи обмена: публикатор outbox и вахтёр сроков (только вместе с NATS, т.е. при NATS_ENABLED=true).""" + return ExchangeTasks.start(async_session_maker, nats, settings) + + @asynccontextmanager async def lifespan(app: FastAPI): app.state.storage = None app.state.nats = None + app.state.exchange = None try: smbclient.register_session( @@ -119,12 +127,15 @@ async def lifespan(app: FastAPI): app.state.nats = _create_nats() if app.state.nats is not None: await app.state.nats.start() # подключение в фоне: старт приложения не ждёт NATS + app.state.exchange = _start_exchange(app.state.nats) # публикатор сам ждёт соединения logger.info("Приложение запущено") yield finally: logger.info("Приложение закрывается") smbclient.reset_connection_cache() + if app.state.exchange is not None: # до закрытия NATS: публикатор не должен писать в закрытое соединение + await app.state.exchange.stop() if app.state.nats is not None: await app.state.nats.stop() if app.state.storage: diff --git a/api/src/services/recognition_exchange.py b/api/src/services/recognition_exchange.py index 3ddb740..59372e7 100644 --- a/api/src/services/recognition_exchange.py +++ b/api/src/services/recognition_exchange.py @@ -1,12 +1,436 @@ -"""Обмен с сервисами распознавания через `recognition_job` (design D-6; docs/context/03-architecture.md). +"""Обмен с сервисами распознавания через `recognition_job` (design D-6; spec recognition-job-exchange; +docs/context/03-architecture.md, «Контракт обмена через NATS»). -Пока здесь только заглушка обработчика `recognition.result`: подписка в `src/core/nats.py` уже вызывает её. -Полноценная обработка (ack `processing`, результат, дедуп по `result_msg_id`) — задача 24 (рулинг R-12, design #23); -`publish_pending()` и `expire_deadlines()` — задача 23. +Outbox (задача 23): +- `publish_pending()` — один проход публикатора: `queued` → первая публикация → `published`; `published` без ack дольше + `ACK_TIMEOUT_SECONDS` → повторная публикация с тем же `Nats-Msg-Id` и тем же `attempt` (счётчик — + `metrics.republish_count`); после `OUTBOX_MAX_REPUBLISH` повторов → `dlq`, `error_code = NO_ACK`. `attempt` не растёт: + обработка не начиналась. Replay (`attempt + 1`) — вне Среза 0. +- `expire_deadlines()` — вахтёр: `processing`/`published` с `reply_deadline < now` → `timeout`, `error_code = TIMEOUT`. +- `run_publisher_loop()` / `run_watchdog_loop()` — фоновые циклы; `ExchangeTasks` запускает их в lifespan при + `NATS_ENABLED=true` и останавливает до закрытия NATS. + +Гарантия доставки — at-least-once за счёт повторов без ack, а не за счёт `flush`: успешный `NatsConnection.publish` +означает лишь, что сервер ответил на PING после отправки, но не что принял именно это сообщение (ошибка прав, например, +приходит асинхронно в `error_cb`). Сбой commit после успешной публикации тоже даёт повтор того же сообщения. Поэтому +сервисы распознавания идемпотентны по `job_id` + `attempt` (`Nats-Msg-Id`). + +Модель транзакций публикатора: одно задание — одна короткая транзакция (`SELECT … LIMIT 1 FOR UPDATE OF recognition_job +SKIP LOCKED` → публикация → `UPDATE` → `COMMIT`). Блокировка строки держится на время одной публикации, а не пачки; +параллельный проход (второй инстанс ядра) пропускает заблокированную строку и не публикует её повторно, а после commit +строка уже не подходит под условие выборки. + +Обмен не пишет `audit_event` (рулинг R-7, design #20). + +Inbox (`handle_message`) — пока заглушка: подписка в `src/core/nats.py` уже вызывает её; полноценная обработка +(ack `processing`, результат, дедуп по `result_msg_id`) — задача 24 (рулинг R-12, design #23). """ +import asyncio +import json +import uuid +from collections.abc import Awaitable, Callable, Sequence +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import Any, Protocol + from loguru import logger from nats.aio.msg import Msg +from pydantic import ValidationError +from sqlalchemy import CursorResult, and_, case, func, literal, or_, select, update +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.contract import CONTRACT_VERSION, RecognitionRequest, S3Ref, request_msg_id, request_subject +from src.core.config import Settings +from src.core.nats import NatsPublishError +from src.models.enums import RecognitionPipeline, RecognitionStatus +from src.models.files import RecognitionJob, S3Object, StoredFile + +WATCHDOG_PERIOD_SECONDS = 30 # период вахтёра сроков (spec «Таймаут ожидания результата») +PUBLISH_BATCH_SIZE = 50 # заданий за один проход публикатора (design D-6, LIMIT 50) +# Срок ответа сервиса, если задание создано без `reply_deadline`: OCR — 5 мин (DBML note), парсер — так же +DEFAULT_REPLY_TIMEOUT: dict[RecognitionPipeline, timedelta] = { + RecognitionPipeline.tabular: timedelta(minutes=5), + RecognitionPipeline.image: timedelta(minutes=5), +} +REPUBLISH_COUNT_KEY = "republish_count" # счётчик повторов в `recognition_job.metrics` (design #6) + +ERROR_NO_ACK = "NO_ACK" +ERROR_TIMEOUT = "TIMEOUT" +ERROR_INVALID_REQUEST = "INVALID_REQUEST" +ERROR_MESSAGE_MAX_LENGTH = 2000 + +SleepFn = Callable[[float], Awaitable[None]] +Clock = Callable[[], datetime] + + +class RequestPublisher(Protocol): + """То, что публикатору нужно от соединения NATS (`src.core.nats.NatsConnection`).""" + + @property + def is_connected(self) -> bool: ... + + async def publish(self, subject: str, payload: bytes, msg_id: str) -> None: ... + + +@dataclass +class PublishReport: + """Итог одного прохода публикатора.""" + + published: int = 0 # первая публикация: queued → published + republished: int = 0 # повторная публикация без ack + dlq: int = 0 # повторы исчерпаны: published → dlq + invalid: int = 0 # запрос не собирается по контракту: → error / INVALID_REQUEST + failed: int = 0 # ошибка публикации (проход прерван) + skipped: bool = False # нет соединения с NATS: проход не выполнялся + + @property + def total(self) -> int: + """Сколько заданий изменено за проход.""" + return self.published + self.republished + self.dlq + self.invalid + + +def _utcnow() -> datetime: + return datetime.now(UTC) + + +def _republish_count(metrics: Any) -> int: + """`metrics.republish_count` (нет поля / `metrics` NULL / не число — 0).""" + if isinstance(metrics, dict): + value = metrics.get(REPUBLISH_COUNT_KEY) + if isinstance(value, int) and not isinstance(value, bool): + return value + return 0 + + +def _candidate_stmt(now: datetime, ack_cutoff: datetime) -> Any: + """Следующее задание прохода: `queued` или `published` без ack дольше таймаута (срок ответа ещё не прошёл — + иначе задание за вахтёром). Строка задания блокируется, занятые другим проходом пропускаются.""" + job = RecognitionJob + awaiting_ack = and_( + job.status == RecognitionStatus.published, + or_(job.published_at.is_(None), job.published_at < ack_cutoff), + or_(job.reply_deadline.is_(None), job.reply_deadline >= now), + ) + return ( + select( + job.id, + job.status, + job.pipeline, + job.attempt, + job.reply_deadline, + job.requested_at, + job.metrics, + job.request_msg_id, + StoredFile.id.label("file_id"), + StoredFile.file_type, + StoredFile.original_name, + S3Object.bucket, + S3Object.key, + ) + .join(StoredFile, StoredFile.id == job.file_id) + .join(S3Object, S3Object.id == StoredFile.s3_object_id) + .where(or_(job.status == RecognitionStatus.queued, awaiting_ack)) + .order_by(job.requested_at.asc().nulls_first(), job.id) + .limit(1) + .with_for_update(of=job, skip_locked=True) + ) + + +def _job_update(job_id: uuid.UUID, expected: RecognitionStatus) -> Any: + """`UPDATE recognition_job` только из ожидаемого статуса (страховка от гонки с обработчиком результата).""" + return ( + update(RecognitionJob) + .where(RecognitionJob.id == job_id, RecognitionJob.status == expected) + .execution_options(synchronize_session=False) + ) + + +def _build_request(row: Any, reply_deadline: datetime) -> RecognitionRequest: + """Запрос `recognition-request.schema.json` из `recognition_job ⋈ stored_file ⋈ s3_object`.""" + return RecognitionRequest( + schema_version=CONTRACT_VERSION, + job_id=row.id, + file_id=row.file_id, + pipeline=row.pipeline.value, + file_type=row.file_type.value, + original_name=row.original_name, + s3=S3Ref(bucket=row.bucket, key=row.key), + reply_deadline=reply_deadline, + attempt=row.attempt, + ) + + +def _short(text: str) -> str: + return text if len(text) <= ERROR_MESSAGE_MAX_LENGTH else text[: ERROR_MESSAGE_MAX_LENGTH - 1] + "…" + + +async def _process_one( # noqa: PLR0913 - один шаг прохода: все параметры прохода + session: AsyncSession, + nats: RequestPublisher, + report: PublishReport, + *, + clock: Clock, + ack_cutoff: datetime, + pass_started: datetime, + max_republish: int, +) -> bool: + """Обработать одно задание в транзакции `session`; `False` — подходящих заданий больше нет.""" + row = (await session.execute(_candidate_stmt(pass_started, ack_cutoff))).one_or_none() + if row is None: + return False + now = clock() + first = row.status is RecognitionStatus.queued + + if not first and _republish_count(row.metrics) >= max_republish: + await session.execute( + _job_update(row.id, RecognitionStatus.published).values( + status=RecognitionStatus.dlq, + error_code=ERROR_NO_ACK, + error_message=( + f"Сервис распознавания не подтвердил получение запроса (ack) после {max_republish} " + "повторных публикаций" + ), + finished_at=now, + ) + ) + report.dlq += 1 + logger.warning( + "Запрос распознавания без ack: задание переведено в dlq. error_category=no_ack, job_id={job_id}, " + "republish_count={count}", + job_id=row.id, + count=_republish_count(row.metrics), + ) + return True + + reply_deadline = row.reply_deadline or now + DEFAULT_REPLY_TIMEOUT[row.pipeline] + try: + request = _build_request(row, reply_deadline) + subject = request_subject(row.pipeline.value) + except (ValidationError, ValueError) as exc: + await session.execute( + _job_update(row.id, row.status).values( + status=RecognitionStatus.error, + error_code=ERROR_INVALID_REQUEST, + error_message=_short(f"Запрос распознавания не соответствует контракту: {exc}"), + finished_at=now, + ) + ) + report.invalid += 1 + logger.error( + "Запрос распознавания не собран по контракту, задание переведено в error. " + "error_category=invalid_request, job_id={job_id}, error={error}", + job_id=row.id, + error=str(exc), + ) + return True + + # первая публикация — id из текущего attempt; повтор — сохранённый id (тот же Nats-Msg-Id) + msg_id = request_msg_id(row.id, row.attempt) if first else row.request_msg_id or request_msg_id(row.id, row.attempt) + payload = json.dumps(request.to_wire(), ensure_ascii=False).encode("utf-8") + await nats.publish(subject, payload, msg_id) # NatsPublishError → откат транзакции у вызывающего + + if first: + await session.execute( + _job_update(row.id, RecognitionStatus.queued).values( + status=RecognitionStatus.published, + published_at=now, + request_msg_id=msg_id, + request_subject=subject, + reply_deadline=reply_deadline, + requested_at=row.requested_at or now, + ) + ) + report.published += 1 + logger.info( + "Запрос распознавания опубликован. job_id={job_id}, subject={subject}, msg_id={msg_id}", + job_id=row.id, + subject=subject, + msg_id=msg_id, + ) + else: + count = _republish_count(row.metrics) + 1 + # jsonb `||`: прочие ключи metrics сохраняются; не объект (SQL NULL или JSON null — так ORM пишет None) → {} + is_object = func.jsonb_typeof(RecognitionJob.metrics) == "object" + base = case((is_object, RecognitionJob.metrics), else_=literal({}, JSONB)) + metrics = base.op("||", return_type=JSONB)(literal({REPUBLISH_COUNT_KEY: count}, JSONB)) + await session.execute( + _job_update(row.id, RecognitionStatus.published).values( + published_at=now, reply_deadline=reply_deadline, metrics=metrics + ) + ) + report.republished += 1 + logger.warning( + "Нет ack от сервиса распознавания, запрос опубликован повторно. job_id={job_id}, subject={subject}, " + "msg_id={msg_id}, republish_count={count}", + job_id=row.id, + subject=subject, + msg_id=msg_id, + count=count, + ) + return True + + +async def publish_pending( # noqa: PLR0913 - параметры прохода именованные: настройки и подмены для тестов + session_factory: async_sessionmaker[AsyncSession], + nats: RequestPublisher, + *, + ack_timeout: timedelta, + max_republish: int, + batch_size: int = PUBLISH_BATCH_SIZE, + now: Clock | None = None, +) -> PublishReport: + """Один проход outbox-публикатора (spec «Outbox публикует задания из `queued`», «Запрос без ack публикуется + повторно, затем — `dlq`»; design D-6). + + До `batch_size` заданий, каждое — в своей транзакции (см. docstring модуля). Нет соединения с NATS — проход + пропускается (задания остаются как есть). Первая `NatsPublishError` откатывает транзакцию задания (оно остаётся + `queued` / `published` без изменений) и прерывает проход: следующий проход повторит публикацию. + """ + clock = now or _utcnow + report = PublishReport() + if not nats.is_connected: + report.skipped = True + return report + pass_started = clock() + ack_cutoff = pass_started - ack_timeout + for _ in range(batch_size): + try: + async with session_factory() as session, session.begin(): + found = await _process_one( + session, + nats, + report, + clock=clock, + ack_cutoff=ack_cutoff, + pass_started=pass_started, + max_republish=max_republish, + ) + except NatsPublishError as exc: + report.failed += 1 + logger.warning( + "Публикация запроса распознавания не удалась, повтор в следующем проходе. error_category=nats_error, " + "error={error}", + error=str(exc), + ) + break + if not found: + break + return report + + +async def expire_deadlines(session_factory: async_sessionmaker[AsyncSession], *, now: Clock | None = None) -> int: + """Вахтёр сроков (spec «Таймаут ожидания результата»): `processing`/`published` с `reply_deadline < now` → + `timeout`, `error_code = TIMEOUT`, `finished_at = now`. Один атомарный `UPDATE` с условием по статусу: результат, + успевший перевести задание в финальный статус, не перетирается. Возвращает число заданий.""" + moment = (now or _utcnow)() + stmt = ( + update(RecognitionJob) + .where( + RecognitionJob.status.in_([RecognitionStatus.processing, RecognitionStatus.published]), + RecognitionJob.reply_deadline < moment, + ) + .values( + status=RecognitionStatus.timeout, + error_code=ERROR_TIMEOUT, + error_message="Сервис распознавания не прислал результат до срока ответа (reply_deadline)", + finished_at=moment, + ) + .execution_options(synchronize_session=False) + ) + async with session_factory() as session, session.begin(): + result = await session.execute(stmt) + count = result.rowcount if isinstance(result, CursorResult) else 0 + if count: + logger.warning("Задания распознавания переведены в timeout по сроку ответа. count={count}", count=count) + return count + + +async def run_publisher_loop( # noqa: PLR0913 - настройки цикла именованные + session_factory: async_sessionmaker[AsyncSession], + nats: RequestPublisher, + *, + poll_seconds: float, + ack_timeout: timedelta, + max_republish: int, + sleep: SleepFn = asyncio.sleep, +) -> None: + """Бесконечный цикл публикатора с периодом `poll_seconds`; ошибка прохода логируется, цикл продолжается.""" + while True: + try: + await publish_pending(session_factory, nats, ack_timeout=ack_timeout, max_republish=max_republish) + except Exception as exc: + logger.exception( + "Ошибка прохода outbox-публикатора. error_category=outbox_error, error_type={error_type}, " + "error={error}", + error_type=type(exc).__name__, + error=repr(exc), + ) + await sleep(poll_seconds) + + +async def run_watchdog_loop( + session_factory: async_sessionmaker[AsyncSession], + *, + period_seconds: float = WATCHDOG_PERIOD_SECONDS, + sleep: SleepFn = asyncio.sleep, +) -> None: + """Бесконечный цикл вахтёра сроков с периодом `period_seconds`; ошибка прохода логируется, цикл продолжается.""" + while True: + try: + await expire_deadlines(session_factory) + except Exception as exc: + logger.exception( + "Ошибка прохода вахтёра сроков. error_category=watchdog_error, error_type={error_type}, error={error}", + error_type=type(exc).__name__, + error=repr(exc), + ) + await sleep(period_seconds) + + +class ExchangeTasks: + """Фоновые задачи обмена в lifespan (design D-6): публикатор и вахтёр. Запускаются только при `NATS_ENABLED=true` + (вызывающий код создаёт их вместе с `NatsConnection`) и останавливаются до закрытия NATS.""" + + def __init__(self, tasks: Sequence[asyncio.Task[None]]) -> None: + self.tasks = tuple(tasks) + + @classmethod + def start( + cls, session_factory: async_sessionmaker[AsyncSession], nats: RequestPublisher, settings: Settings + ) -> "ExchangeTasks": + publisher = asyncio.create_task( + run_publisher_loop( + session_factory, + nats, + poll_seconds=settings.OUTBOX_POLL_SECONDS, + ack_timeout=timedelta(seconds=settings.ACK_TIMEOUT_SECONDS), + max_republish=settings.OUTBOX_MAX_REPUBLISH, + ), + name="recognition-outbox-publisher", + ) + watchdog = asyncio.create_task( + run_watchdog_loop(session_factory, period_seconds=WATCHDOG_PERIOD_SECONDS), + name="recognition-deadline-watchdog", + ) + logger.info( + "Фоновые задачи обмена запущены: публикатор (период {poll} с), вахтёр сроков (период {watchdog} с)", + poll=settings.OUTBOX_POLL_SECONDS, + watchdog=WATCHDOG_PERIOD_SECONDS, + ) + return cls([publisher, watchdog]) + + async def stop(self) -> None: + """Отменяет задачи и дожидается их завершения; исключения наружу не выходят.""" + for task in self.tasks: + task.cancel() + results = await asyncio.gather(*self.tasks, return_exceptions=True) + for task, result in zip(self.tasks, results, strict=True): + if isinstance(result, Exception): # CancelledError — штатная остановка, не Exception + logger.opt(exception=result).error( + "Фоновая задача обмена завершилась с ошибкой. task={name}", name=task.get_name() + ) async def handle_message(msg: Msg) -> None: diff --git a/api/tests/factories.py b/api/tests/factories.py index 150a585..e338b4f 100644 --- a/api/tests/factories.py +++ b/api/tests/factories.py @@ -1,11 +1,18 @@ -"""Фабрики тестовых данных для БД-тестов: минимальные цепочки FK поверх сессии (`flush`, без `commit`).""" +"""Фабрики тестовых данных для БД-тестов. + +- `make_osv_entry` — минимальная цепочка FK поверх сессии (`flush`, без `commit`): откатывается фикстурой `db_session`. +- `CommittedJobs` — задания распознавания, ЗАФИКСИРОВАННЫЕ в БД (`commit`): для фоновых процессов обмена, которые + работают в собственных транзакциях и не видят чужую незафиксированную транзакцию. Всё созданное удаляется + `cleanup()` в порядке FK. +""" import datetime import decimal import uuid from typing import Any -from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import delete +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.models.directories import LegalEntity from src.models.enums import ( @@ -20,6 +27,8 @@ from src.models.enums import ( from src.models.files import Document, RecognitionJob, S3Object, StoredFile from src.models.osv import OsvEntry +DEFAULT_FILE_TYPES = {RecognitionPipeline.tabular: FileType.XLSX, RecognitionPipeline.image: FileType.PDF} + async def make_osv_entry(session: AsyncSession, **overrides: Any) -> OsvEntry: """Строка ОСВ с минимальной цепочкой FK (ЮЛ, s3_object, stored_file, recognition_job, document); `flush`. @@ -69,3 +78,106 @@ async def make_osv_entry(session: AsyncSession, **overrides: Any) -> OsvEntry: session.add(entry) await session.flush() return entry + + +class CommittedJobs: + """Задания `recognition_job` с цепочкой `legal_entity` + `s3_object` + `stored_file`, зафиксированные в БД. + + Каждый `make_queued_job` — отдельная транзакция с `commit`; созданные id запоминаются, `cleanup()` удаляет их + в порядке FK (задание → файл → объект S3 и ЮЛ). Вызывать `cleanup()` в teardown фикстуры — даже при падении теста. + """ + + def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: + self._session_factory = session_factory + self.job_ids: list[uuid.UUID] = [] + self._file_ids: list[uuid.UUID] = [] + self._s3_ids: list[uuid.UUID] = [] + self._entity_ids: list[uuid.UUID] = [] + + async def make_queued_job( # noqa: PLR0913 - каждое поле задания задаётся тестом явно + self, + *, + pipeline: RecognitionPipeline = RecognitionPipeline.tabular, + file_type: FileType | None = None, + original_name: str | None = "osv.xlsx", + status: RecognitionStatus = RecognitionStatus.queued, + reply_deadline: datetime.datetime | None = None, + published_at: datetime.datetime | None = None, + request_msg_id: str | None = None, + metrics: Any = None, + attempt: int = 1, + ) -> uuid.UUID: + """Задание (по умолчанию `queued`, tabular, XLSX) со своей цепочкой FK; возвращает `recognition_job.id`.""" + now = datetime.datetime.now(datetime.UTC) + async with self._session_factory() as session, session.begin(): + entity = LegalEntity(inn=str(uuid.uuid4().int)[:12], name="ООО Тест обмена", valid_from=now) + s3 = S3Object( + bucket_role=S3BucketRole.uploads, + bucket="arm-ka", + key=f"uploads/{uuid.uuid4()}/{original_name or 'file'}", + created_at=now, + ) + session.add_all([entity, s3]) + await session.flush() + self._entity_ids.append(entity.id) + self._s3_ids.append(s3.id) + file = StoredFile( + legal_entity_id=entity.id, + file_type=file_type or DEFAULT_FILE_TYPES[pipeline], + original_name=original_name, + s3_object_id=s3.id, + status=FileStatus.UPLOADED, + ) + session.add(file) + await session.flush() + self._file_ids.append(file.id) + job = RecognitionJob( + file_id=file.id, + pipeline=pipeline, + request_subject=f"recognition.{pipeline.value}.request", + schema_version="1.0", + status=status, + attempt=attempt, + reply_deadline=reply_deadline, + published_at=published_at, + request_msg_id=request_msg_id, + metrics=metrics, + ) + session.add(job) + await session.flush() + self.job_ids.append(job.id) + return job.id + + async def s3_key(self, job_id: uuid.UUID) -> str: + """Ключ S3 исходного файла задания.""" + async with self._session_factory() as session: + job = await session.get(RecognitionJob, job_id) + assert job is not None + file = await session.get(StoredFile, job.file_id) + assert file is not None + s3 = await session.get(S3Object, file.s3_object_id) + assert s3 is not None + return s3.key + + async def get(self, job_id: uuid.UUID) -> RecognitionJob: + """Свежее состояние задания из БД (новая сессия).""" + async with self._session_factory() as session: + job = await session.get(RecognitionJob, job_id) + assert job is not None + return job + + async def cleanup(self) -> None: + """Удаляет всё созданное в порядке FK.""" + async with self._session_factory() as session, session.begin(): + if self.job_ids: + await session.execute(delete(RecognitionJob).where(RecognitionJob.id.in_(self.job_ids))) + if self._file_ids: + await session.execute(delete(StoredFile).where(StoredFile.id.in_(self._file_ids))) + if self._s3_ids: + await session.execute(delete(S3Object).where(S3Object.id.in_(self._s3_ids))) + if self._entity_ids: + await session.execute(delete(LegalEntity).where(LegalEntity.id.in_(self._entity_ids))) + self.job_ids.clear() + self._file_ids.clear() + self._s3_ids.clear() + self._entity_ids.clear() diff --git a/api/tests/test_app.py b/api/tests/test_app.py index 5e5ff58..1755540 100644 --- a/api/tests/test_app.py +++ b/api/tests/test_app.py @@ -14,6 +14,7 @@ import pytest from src.api.router import router from src.core.nats import NatsConnection +from src.services import recognition_exchange # src/__init__.py реэкспортирует объект app под именем модуля, поэтому модуль берём через importlib app_module = importlib.import_module("src.app") @@ -93,6 +94,24 @@ def _no_smb(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(app_module.smbclient, "reset_connection_cache", lambda: None) +@pytest.fixture(autouse=True) +def exchange_loops(monkeypatch: pytest.MonkeyPatch) -> list[str]: + """Циклы публикатора и вахтёра подменены ожиданием: быстрые тесты lifespan не ходят в БД.""" + started: list[str] = [] + + async def publisher(session_factory: Any, nats: Any, **kwargs: Any) -> None: + started.append("publisher") + await asyncio.Event().wait() + + async def watchdog(session_factory: Any, **kwargs: Any) -> None: + started.append("watchdog") + await asyncio.Event().wait() + + monkeypatch.setattr(recognition_exchange, "run_publisher_loop", publisher) + monkeypatch.setattr(recognition_exchange, "run_watchdog_loop", watchdog) + return started + + @pytest.mark.usefixtures("_no_smb") async def test_lifespan_starts_when_s3_and_nats_unavailable(monkeypatch: pytest.MonkeyPatch): async def unavailable() -> None: @@ -198,3 +217,61 @@ async def test_connect_s3_passes_secure_flag(monkeypatch: pytest.MonkeyPatch, se assert storage is not None assert created["secure"] is secure + + +class _RecordingNats(_FakeNats): + """Запоминает, остановлены ли фоновые задачи обмена к моменту `stop()` соединения NATS.""" + + def __init__(self) -> None: + super().__init__() + self.exchange_done_at_stop: bool | None = None + + async def stop(self) -> None: + exchange = app_module.app.state.exchange + self.exchange_done_at_stop = all(task.done() for task in exchange.tasks) + await super().stop() + + +@pytest.mark.usefixtures("_no_smb") +async def test_lifespan_starts_exchange_tasks_and_stops_them_before_nats( + monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str] +): + """NATS_ENABLED=true: публикатор и вахтёр — две фоновые задачи; остановка — до `nats.stop()`.""" + + async def unavailable() -> None: + return None + + fake_nats = _RecordingNats() + monkeypatch.setattr(app_module, "_connect_s3", unavailable) + monkeypatch.setattr(app_module, "_create_nats", lambda: fake_nats) + + async with app_module.lifespan(app_module.app): + exchange = app_module.app.state.exchange + assert exchange is not None + assert len(exchange.tasks) == 2 # noqa: PLR2004 - публикатор и вахтёр + await asyncio.sleep(0) + assert sorted(exchange_loops) == ["publisher", "watchdog"] + assert not any(task.done() for task in exchange.tasks) + + assert fake_nats.exchange_done_at_stop is True + assert fake_nats.stopped + + +@pytest.mark.usefixtures("_no_smb") +async def test_lifespan_without_nats_starts_no_exchange_tasks( + monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str] +): + """NATS_ENABLED=false (`_create_nats` → None): фоновые задачи обмена не запускаются.""" + + async def unavailable() -> None: + return None + + monkeypatch.setattr(app_module, "_connect_s3", unavailable) + monkeypatch.setattr(app_module.settings, "NATS_ENABLED", False) + tasks_before = asyncio.all_tasks() + + async with app_module.lifespan(app_module.app): + assert app_module.app.state.exchange is None + await asyncio.sleep(0) + assert exchange_loops == [] + assert asyncio.all_tasks() - tasks_before == set() diff --git a/api/tests/test_recognition_exchange.py b/api/tests/test_recognition_exchange.py new file mode 100644 index 0000000..48430fd --- /dev/null +++ b/api/tests/test_recognition_exchange.py @@ -0,0 +1,698 @@ +"""Outbox-публикатор и вахтёр сроков `recognition_job` (spec recognition-job-exchange: «Outbox публикует задания из +`queued`», «Запрос без ack публикуется повторно, затем — `dlq`», «Таймаут ожидания результата»; design D-6). + +Интеграционные тесты работают против реального PostgreSQL (БД сессии pytest) и NATS из `.env.test`. Публикатор и +вахтёр ведут собственные транзакции, поэтому данные фиксируются реально (`CommittedJobs`) и удаляются в teardown. +Subject'ы запросов общие (`recognition.*.request`): тестовый подписчик фильтрует сообщения по `job_id` своих заданий. +Время в тестах повторов — инжектируемые часы `now`, без реальных ожиданий таймаутов. +""" + +import asyncio +import json +import time +import uuid +from collections.abc import AsyncIterator, Callable +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any, get_args + +import pytest +import pytest_asyncio +from jsonschema import Draft202012Validator, FormatChecker +from nats.aio.client import Client as NATS +from nats.aio.msg import Msg +from sqlalchemy import update +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker + +from src.contract import NATS_MSG_ID_HEADER, RecognitionRequest +from src.contract.request import FileType as ContractFileType +from src.core.config import get_settings +from src.core.nats import NatsConnection, NatsPublishError +from src.models.enums import FileType, RecognitionPipeline, RecognitionStatus +from src.models.files import RecognitionJob +from src.services import recognition_exchange as rx +from src.services.recognition_exchange import ( + WATCHDOG_PERIOD_SECONDS, + ExchangeTasks, + expire_deadlines, + publish_pending, + run_publisher_loop, + run_watchdog_loop, +) +from tests.factories import CommittedJobs + +CONTRACTS_DIR = Path(__file__).resolve().parents[1] / "contracts" +REQUEST_SCHEMA = json.loads((CONTRACTS_DIR / "recognition-request.schema.json").read_text(encoding="utf-8")) +REQUEST_VALIDATOR = Draft202012Validator(REQUEST_SCHEMA, format_checker=FormatChecker()) +REQUEST_SUBJECTS = ("recognition.tabular.request", "recognition.image.request") + +ACK = timedelta(seconds=30) +MAX_REPUBLISH = 3 +READY_TIMEOUT = 5.0 +CONCURRENT_JOBS = 20 +MIN_LOOP_PASSES = 3 # цикл пережил несколько ошибок прохода подряд +EXCHANGE_TASKS = 2 # публикатор и вахтёр + + +# --- инфраструктура --------------------------------------------------------------------------------------------- + + +class Clock: + """Инжектируемые часы: `now` для публикатора и вахтёра, `advance` — сдвиг без реального ожидания.""" + + def __init__(self) -> None: + self.current = datetime.now(UTC) + + def __call__(self) -> datetime: + return self.current + + def advance(self, delta: timedelta) -> None: + self.current += delta + + +class RequestSubscriber: + """Тестовый подписчик на subject'ы запросов: «голый» клиент nats-py, сообщения фильтруются по `job_id`.""" + + def __init__(self) -> None: + self.messages: list[Msg] = [] + self._nc = NATS() + + async def start(self) -> None: + url = get_settings().OPENBAO.NATS.NATS_URL + try: + await self._nc.connect(url, connect_timeout=2, allow_reconnect=False) + except Exception as exc: + pytest.fail(f"NATS недоступен ({url}, .env.test): {type(exc).__name__}", pytrace=False) + + async def collect(msg: Msg) -> None: + self.messages.append(msg) + + for subject in REQUEST_SUBJECTS: + await self._nc.subscribe(subject, cb=collect) + await self._nc.flush() + + async def stop(self) -> None: + await self._nc.close() + + def for_job(self, job_id: uuid.UUID) -> list[Msg]: + return [m for m in self.messages if json.loads(m.data).get("job_id") == str(job_id)] + + async def settle(self) -> None: + """Всё, что сервер принял до этого момента, доставлено подписчику (PING/PONG после сообщений) и обработано.""" + await self._nc.flush() + await asyncio.sleep(0.1) + + async def wait_for(self, job_id: uuid.UUID, count: int, timeout: float = 10.0) -> list[Msg]: + deadline = time.monotonic() + timeout + while len(self.for_job(job_id)) < count: + if time.monotonic() > deadline: + raise AssertionError(f"за {timeout} с получено {len(self.for_job(job_id))} из {count} сообщений") + await asyncio.sleep(0.02) + return self.for_job(job_id) + + +class FakeNats: + """Подставной NATS: состояние соединения и счётчик публикаций; первые `fail_first` публикаций — ошибка.""" + + def __init__(self, *, connected: bool = True, fail_first: int = 0, delegate: Any = None, delay: float = 0) -> None: + self.is_connected_value = connected + self.fail_first = fail_first + self.delegate = delegate + self.delay = delay + self.calls: list[tuple[str, str]] = [] + + @property + def is_connected(self) -> bool: + return self.is_connected_value + + async def publish(self, subject: str, payload: bytes, msg_id: str) -> None: + self.calls.append((subject, msg_id)) + if len(self.calls) <= self.fail_first: + raise NatsPublishError(f"подставная ошибка публикации {msg_id}") + if self.delay: + await asyncio.sleep(self.delay) + if self.delegate is not None: + await self.delegate.publish(subject, payload, msg_id) + + +@pytest.fixture +def session_factory(db_engine: AsyncEngine) -> async_sessionmaker[AsyncSession]: + return async_sessionmaker(db_engine, expire_on_commit=False) + + +@pytest_asyncio.fixture +async def jobs(session_factory: async_sessionmaker[AsyncSession]) -> AsyncIterator[CommittedJobs]: + committed = CommittedJobs(session_factory) + try: + yield committed + finally: + await committed.cleanup() + + +@pytest_asyncio.fixture +async def nats() -> AsyncIterator[NatsConnection]: + conn = NatsConnection(get_settings().OPENBAO.NATS) + await conn.start() + try: + if not await conn.wait_ready(timeout=READY_TIMEOUT): + pytest.fail(f"NATS недоступен ({get_settings().OPENBAO.NATS.NATS_URL}, .env.test)", pytrace=False) + yield conn + finally: + await conn.stop() + + +@pytest_asyncio.fixture +async def subscriber() -> AsyncIterator[RequestSubscriber]: + sub = RequestSubscriber() + await sub.start() + try: + yield sub + finally: + await sub.stop() + + +async def _set_status( + session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID, status: RecognitionStatus +) -> None: + async with session_factory() as session, session.begin(): + await session.execute(update(RecognitionJob).where(RecognitionJob.id == job_id).values(status=status)) + + +def _publish( + session_factory: async_sessionmaker[AsyncSession], nats: Any, now: Callable[[], datetime] | None = None +) -> Any: + return publish_pending(session_factory, nats, ack_timeout=ACK, max_republish=MAX_REPUBLISH, now=now) + + +# --- юнит: соответствие enum, константы ------------------------------------------------------------------------ + + +def test_orm_file_type_matches_contract_enum(): + """`stored_file.file_type` (DBML) и `file_type` схемы запроса — одно множество значений.""" + assert {m.value for m in FileType} == set(get_args(ContractFileType)) + assert set(REQUEST_SCHEMA["properties"]["file_type"]["enum"]) == {m.value for m in FileType} + + +def test_orm_pipeline_matches_request_subjects(): + assert {m.value for m in RecognitionPipeline} == {"tabular", "image"} + + +def test_watchdog_period_is_30_seconds(): + assert WATCHDOG_PERIOD_SECONDS == 30 # noqa: PLR2004 - значение из spec + + +# --- публикация ------------------------------------------------------------------------------------------------- + + +@pytest.mark.integration +@pytest.mark.parametrize( + ("pipeline", "file_type", "subject"), + [ + (RecognitionPipeline.tabular, FileType.XLSX, "recognition.tabular.request"), + (RecognitionPipeline.image, FileType.PDF, "recognition.image.request"), + ], +) +async def test_publish_request_to_service( # noqa: PLR0913,PLR0917 - фикстуры и параметры + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + nats: NatsConnection, + subscriber: RequestSubscriber, + pipeline: RecognitionPipeline, + file_type: FileType, + subject: str, +): + """Сценарий «Публикация запроса парсеру»: сообщение по схеме, `attempt = 1`, строка `published`.""" + job_id = await jobs.make_queued_job(pipeline=pipeline, file_type=file_type, original_name="ОСВ 60.xlsx") + before = datetime.now(UTC) + + report = await _publish(session_factory, nats) + + assert report.published == 1 + (msg,) = await subscriber.wait_for(job_id, 1) + body = json.loads(msg.data) + errors = [e.message for e in REQUEST_VALIDATOR.iter_errors(body)] + assert errors == [] + request = RecognitionRequest.model_validate_json(msg.data) + job = await jobs.get(job_id) + assert msg.subject == subject + assert msg.headers == {NATS_MSG_ID_HEADER: f"{job_id}:1"} + assert request.job_id == job_id + assert request.file_id == job.file_id + assert request.pipeline == pipeline.value + assert request.file_type == file_type.value + assert request.original_name == "ОСВ 60.xlsx" + assert request.s3.bucket == "arm-ka" + assert request.s3.key == await jobs.s3_key(job_id) + assert request.attempt == 1 + assert request.schema_version == "1.0" + # reply_deadline не был задан: вычислен при первой публикации (5 мин) и сохранён в задание + assert job.reply_deadline is not None + assert request.reply_deadline == job.reply_deadline + assert timedelta(minutes=4) < job.reply_deadline - before < timedelta(minutes=6) + + assert job.status is RecognitionStatus.published + assert job.request_msg_id == f"{job_id}:1" + assert job.request_subject == subject + assert job.published_at is not None and job.published_at >= before + assert job.requested_at is not None + assert job.attempt == 1 + + +@pytest.mark.integration +async def test_publish_keeps_existing_reply_deadline( + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + nats: NatsConnection, + subscriber: RequestSubscriber, +): + deadline = datetime.now(UTC).replace(microsecond=0) + timedelta(hours=2) + job_id = await jobs.make_queued_job(reply_deadline=deadline) + + await _publish(session_factory, nats) + + (msg,) = await subscriber.wait_for(job_id, 1) + assert RecognitionRequest.model_validate_json(msg.data).reply_deadline == deadline + assert (await jobs.get(job_id)).reply_deadline == deadline + + +@pytest.mark.integration +async def test_first_publication_msg_id_follows_attempt( + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + nats: NatsConnection, + subscriber: RequestSubscriber, +): + """`queued` с `attempt = 2` и устаревшим `request_msg_id` (будущий replay): id и тело — по текущему attempt.""" + job_id = await jobs.make_queued_job(attempt=2, request_msg_id=f"stale-{uuid.uuid4()}") + + await _publish(session_factory, nats) + + (msg,) = await subscriber.wait_for(job_id, 1) + assert msg.headers == {NATS_MSG_ID_HEADER: f"{job_id}:2"} + assert RecognitionRequest.model_validate_json(msg.data).attempt == 2 # noqa: PLR2004 - attempt задания + job = await jobs.get(job_id) + assert (job.request_msg_id, job.attempt) == (f"{job_id}:2", 2) + + +@pytest.mark.integration +async def test_invalid_request_marks_job_error_and_pass_continues( + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + nats: NatsConnection, + subscriber: RequestSubscriber, +): + """Запрос не собирается по схеме (нет `original_name`) — задание `error/INVALID_REQUEST`, проход идёт дальше.""" + broken = await jobs.make_queued_job(original_name=None) + good = await jobs.make_queued_job() + + report = await _publish(session_factory, nats) + + assert (report.invalid, report.published) == (1, 1) + await subscriber.wait_for(good, 1) + await subscriber.settle() + assert subscriber.for_job(broken) == [] + job = await jobs.get(broken) + assert job.status is RecognitionStatus.error + assert job.error_code == "INVALID_REQUEST" + assert job.error_message and "original_name" in job.error_message + assert job.finished_at is not None + assert job.request_msg_id is None + + +# --- повторы без ack и dlq --------------------------------------------------------------------------------------- + + +@pytest.mark.integration +async def test_republish_without_ack_then_dlq( + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + nats: NatsConnection, + subscriber: RequestSubscriber, +): + """Сценарий «Сервис недоступен»: 1 + 3 публикации с одним `Nats-Msg-Id`, затем `dlq / NO_ACK`.""" + clock = Clock() + job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1), metrics={"note": "сохранить"}) + + assert (await _publish(session_factory, nats, clock)).published == 1 + first = await jobs.get(job_id) + # раньше ACK_TIMEOUT повтора нет + clock.advance(ACK - timedelta(seconds=1)) + assert (await _publish(session_factory, nats, clock)).republished == 0 + + for n in range(1, MAX_REPUBLISH + 1): + clock.advance(timedelta(seconds=2)) # с последней публикации прошло больше ACK_TIMEOUT + report = await _publish(session_factory, nats, clock) + assert report.republished == 1 + job = await jobs.get(job_id) + assert job.metrics == {"note": "сохранить", "republish_count": n} + assert job.published_at == clock() + assert job.status is RecognitionStatus.published + clock.advance(ACK - timedelta(seconds=1)) + + messages = await subscriber.wait_for(job_id, 1 + MAX_REPUBLISH) + assert {m.headers[NATS_MSG_ID_HEADER] for m in messages} == {f"{job_id}:1"} + assert len({m.data for m in messages}) == 1 # то же тело: тот же attempt и reply_deadline + + clock.advance(timedelta(seconds=2)) + report = await _publish(session_factory, nats, clock) + assert (report.dlq, report.republished, report.published) == (1, 0, 0) + job = await jobs.get(job_id) + assert job.status is RecognitionStatus.dlq + assert job.error_code == "NO_ACK" + assert job.error_message + assert job.finished_at == clock() + assert job.request_msg_id == first.request_msg_id == f"{job_id}:1" + assert job.attempt == 1 + assert job.metrics == {"note": "сохранить", "republish_count": MAX_REPUBLISH} + + clock.advance(ACK * 2) + assert (await _publish(session_factory, nats, clock)).total == 0 + await subscriber.settle() + assert len(subscriber.for_job(job_id)) == 1 + MAX_REPUBLISH + + +@pytest.mark.integration +async def test_ack_after_first_republish_stops_republishing( + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + nats: NatsConnection, + subscriber: RequestSubscriber, +): + """Сценарий «Сервис появился после первого повтора»: после ack (`processing`) повторов нет.""" + clock = Clock() + job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1)) + await _publish(session_factory, nats, clock) + clock.advance(ACK + timedelta(seconds=1)) + assert (await _publish(session_factory, nats, clock)).republished == 1 + await subscriber.wait_for(job_id, 2) + + await _set_status(session_factory, job_id, RecognitionStatus.processing) # имитация ack + for _ in range(MAX_REPUBLISH + 1): + clock.advance(ACK + timedelta(seconds=1)) + assert (await _publish(session_factory, nats, clock)).total == 0 + + await subscriber.settle() + assert len(subscriber.for_job(job_id)) == 2 # noqa: PLR2004 - первая публикация и один повтор + job = await jobs.get(job_id) + assert job.status is RecognitionStatus.processing + assert job.metrics == {"republish_count": 1} + + +@pytest.mark.integration +async def test_publisher_skips_published_job_with_passed_deadline( + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + nats: NatsConnection, + subscriber: RequestSubscriber, +): + """`published` с прошедшим `reply_deadline` не републикуется и не уходит в dlq — его забирает вахтёр.""" + now = datetime.now(UTC) + job_id = await jobs.make_queued_job( + status=RecognitionStatus.published, + published_at=now - timedelta(hours=1), + reply_deadline=now - timedelta(minutes=1), + request_msg_id=f"{uuid.uuid4()}:1", + metrics={"republish_count": MAX_REPUBLISH}, + ) + + assert (await _publish(session_factory, nats)).total == 0 + await subscriber.settle() + assert subscriber.for_job(job_id) == [] + assert (await jobs.get(job_id)).status is RecognitionStatus.published + + assert await expire_deadlines(session_factory) == 1 + job = await jobs.get(job_id) + assert (job.status, job.error_code) == (RecognitionStatus.timeout, "TIMEOUT") + + +# --- NATS недоступен --------------------------------------------------------------------------------------------- + + +@pytest.mark.integration +async def test_pass_skipped_when_nats_disconnected( + session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs +): + job_id = await jobs.make_queued_job() + fake = FakeNats(connected=False) + + report = await _publish(session_factory, fake) + + assert report.skipped + assert fake.calls == [] + assert (await jobs.get(job_id)).status is RecognitionStatus.queued + + +@pytest.mark.integration +async def test_publish_error_keeps_queued_and_stops_pass( + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + nats: NatsConnection, + subscriber: RequestSubscriber, +): + """Сценарий «NATS недоступен»: задание остаётся `queued`, проход прерван; следующий проход публикует.""" + first = await jobs.make_queued_job() + second = await jobs.make_queued_job() + fake = FakeNats(fail_first=1) + + report = await _publish(session_factory, fake) + + assert (report.failed, report.published) == (1, 0) + assert len(fake.calls) == 1 # второе задание в этом проходе не пытались публиковать + for job_id in (first, second): + job = await jobs.get(job_id) + assert job.status is RecognitionStatus.queued + assert job.request_msg_id is None + assert job.published_at is None + assert job.reply_deadline is None # откат: срок не сохранён + + report = await _publish(session_factory, nats) + + assert report.published == 2 # noqa: PLR2004 - оба задания + for job_id in (first, second): + await subscriber.wait_for(job_id, 1) + assert (await jobs.get(job_id)).status is RecognitionStatus.published + + +@pytest.mark.integration +async def test_republish_error_keeps_published_state( + session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection +): + clock = Clock() + job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1)) + await _publish(session_factory, nats, clock) + published_at = (await jobs.get(job_id)).published_at + clock.advance(ACK + timedelta(seconds=1)) + + report = await _publish(session_factory, FakeNats(fail_first=1), clock) + + assert report.failed == 1 + job = await jobs.get(job_id) + assert job.status is RecognitionStatus.published + assert job.published_at == published_at + assert job.metrics is None + + +# --- конкурентность ----------------------------------------------------------------------------------------------- + + +@pytest.mark.integration +async def test_concurrent_publishers_publish_each_job_once( + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + nats: NatsConnection, + subscriber: RequestSubscriber, +): + """Два одновременных прохода (SKIP LOCKED): каждое задание опубликовано ровно один раз.""" + job_ids = [await jobs.make_queued_job() for _ in range(CONCURRENT_JOBS)] + # задержка внутри публикации держит блокировку строки, пока второй проход выбирает задания + first, second = FakeNats(delegate=nats, delay=0.02), FakeNats(delegate=nats, delay=0.02) + + reports = await asyncio.gather(_publish(session_factory, first), _publish(session_factory, second)) + + assert sum(r.published for r in reports) == CONCURRENT_JOBS + assert all(r.published > 0 for r in reports) # оба прохода реально работали параллельно + published_ids = [msg_id for fake in (first, second) for _, msg_id in fake.calls] + assert sorted(published_ids) == sorted(f"{job_id}:1" for job_id in job_ids) + await subscriber.settle() + for job_id in job_ids: + assert len(await subscriber.wait_for(job_id, 1)) == 1 + assert (await jobs.get(job_id)).status is RecognitionStatus.published + + +# --- вахтёр ---------------------------------------------------------------------------------------------------------- + + +@pytest.mark.integration +@pytest.mark.parametrize( + ("status", "deadline_offset", "expected"), + [ + (RecognitionStatus.processing, timedelta(minutes=-1), RecognitionStatus.timeout), + (RecognitionStatus.published, timedelta(minutes=-1), RecognitionStatus.timeout), + (RecognitionStatus.processing, timedelta(minutes=5), RecognitionStatus.processing), + (RecognitionStatus.published, timedelta(minutes=5), RecognitionStatus.published), + (RecognitionStatus.processing, None, RecognitionStatus.processing), + (RecognitionStatus.queued, timedelta(minutes=-1), RecognitionStatus.queued), + (RecognitionStatus.done, timedelta(minutes=-1), RecognitionStatus.done), + (RecognitionStatus.dlq, timedelta(minutes=-1), RecognitionStatus.dlq), + ], +) +async def test_expire_deadlines( + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + status: RecognitionStatus, + deadline_offset: timedelta | None, + expected: RecognitionStatus, +): + """Сценарий «Сервис взял, но не ответил»: `processing`/`published` с прошедшим сроком → `timeout`.""" + clock = Clock() + deadline = None if deadline_offset is None else clock() + deadline_offset + job_id = await jobs.make_queued_job(status=status, reply_deadline=deadline) + + expired = await expire_deadlines(session_factory, now=clock) + + job = await jobs.get(job_id) + assert job.status is expected + if expected is RecognitionStatus.timeout: + assert expired == 1 + assert job.error_code == "TIMEOUT" + assert job.error_message + assert job.finished_at == clock() + assert job.attempt == 1 + else: + assert expired == 0 + assert job.error_code is None + assert job.finished_at is None + + +# --- фоновые циклы ----------------------------------------------------------------------------------------------- + + +async def _wait_until(predicate: Callable[[], bool], timeout: float = 2.0) -> None: + deadline = time.monotonic() + timeout + while not predicate(): + if time.monotonic() > deadline: + raise AssertionError("условие не выполнено за отведённое время") + await asyncio.sleep(0.01) + + +async def _cancel(task: asyncio.Task[Any]) -> None: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +@pytest.mark.integration +async def test_publisher_loop_publishes_and_stops_on_cancel( + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + nats: NatsConnection, + subscriber: RequestSubscriber, +): + job_id = await jobs.make_queued_job() + task = asyncio.create_task( + run_publisher_loop(session_factory, nats, poll_seconds=0.1, ack_timeout=ACK, max_republish=MAX_REPUBLISH) + ) + try: + await subscriber.wait_for(job_id, 1, timeout=2.0) + finally: + await _cancel(task) + assert (await jobs.get(job_id)).status is RecognitionStatus.published + + +@pytest.mark.integration +async def test_watchdog_loop_expires_and_stops_on_cancel( + session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs +): + job_id = await jobs.make_queued_job( + status=RecognitionStatus.processing, reply_deadline=datetime.now(UTC) - timedelta(seconds=1) + ) + task = asyncio.create_task(run_watchdog_loop(session_factory, period_seconds=0.1)) + try: + deadline = time.monotonic() + 2.0 + while (await jobs.get(job_id)).status is not RecognitionStatus.timeout: + assert time.monotonic() < deadline, "вахтёр не перевёл задание в timeout за 2 с" + await asyncio.sleep(0.05) + finally: + await _cancel(task) + + +class RecordingSleep: + """Подмена `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 test_publisher_loop_survives_pass_errors(monkeypatch: pytest.MonkeyPatch): + calls: list[int] = [] + + async def failing_pass(*args: Any, **kwargs: Any) -> Any: + calls.append(1) + raise RuntimeError("БД недоступна") + + monkeypatch.setattr(rx, "publish_pending", failing_pass) + sleep = RecordingSleep() + task = asyncio.create_task( + run_publisher_loop( + object(), # type: ignore[arg-type] # проход подменён: фабрика сессий не используется + FakeNats(), + poll_seconds=5, + ack_timeout=ACK, + max_republish=MAX_REPUBLISH, + sleep=sleep, + ) + ) + try: + await _wait_until(lambda: len(calls) >= MIN_LOOP_PASSES) + finally: + await _cancel(task) + assert set(sleep.delays) == {5} + + +async def test_watchdog_loop_survives_pass_errors(monkeypatch: pytest.MonkeyPatch): + calls: list[int] = [] + + async def failing_pass(*args: Any, **kwargs: Any) -> int: + calls.append(1) + raise RuntimeError("БД недоступна") + + monkeypatch.setattr(rx, "expire_deadlines", failing_pass) + sleep = RecordingSleep() + task = asyncio.create_task(run_watchdog_loop(object(), sleep=sleep)) # type: ignore[arg-type] + try: + await _wait_until(lambda: len(calls) >= MIN_LOOP_PASSES) + finally: + await _cancel(task) + assert set(sleep.delays) == {WATCHDOG_PERIOD_SECONDS} + + +async def test_exchange_tasks_start_two_loops_and_stop(monkeypatch: pytest.MonkeyPatch): + started: dict[str, dict[str, Any]] = {} + + async def fake_publisher(session_factory: Any, nats: Any, **kwargs: Any) -> None: + started["publisher"] = kwargs + await asyncio.Event().wait() + + async def fake_watchdog(session_factory: Any, **kwargs: Any) -> None: + started["watchdog"] = kwargs + await asyncio.Event().wait() + + monkeypatch.setattr(rx, "run_publisher_loop", fake_publisher) + monkeypatch.setattr(rx, "run_watchdog_loop", fake_watchdog) + settings = get_settings().model_copy( + update={"OUTBOX_POLL_SECONDS": 7, "ACK_TIMEOUT_SECONDS": 11, "OUTBOX_MAX_REPUBLISH": 2} + ) + + exchange = ExchangeTasks.start(object(), FakeNats(), settings) # type: ignore[arg-type] + await _wait_until(lambda: len(started) == EXCHANGE_TASKS) + + assert len(exchange.tasks) == EXCHANGE_TASKS + assert started["publisher"] == {"poll_seconds": 7, "ack_timeout": timedelta(seconds=11), "max_republish": 2} + assert started["watchdog"] == {"period_seconds": WATCHDOG_PERIOD_SECONDS} + await exchange.stop() + assert all(task.done() for task in exchange.tasks) + await exchange.stop() # повторная остановка безопасна