diff --git a/api/src/app.py b/api/src/app.py index 83fac09..3d59e9f 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -10,7 +10,7 @@ middleware не подключается), healthcheck (uptime, version, nats: o состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор и вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) — и останавливает их до закрытия NATS. Сообщения `recognition.result` (ack и результаты сервисов) обрабатывает inbox `handle_message` того же модуля с фабрикой сессий -приложения и `NoopMapper` (маппинг результата в доменные таблицы — Срез 2). +приложения (inbox меняет задание и статус файла; маппинг результата в доменные таблицы — отдельный цикл, Срез 2c D-1). Остановка: фоновые задачи обмена → NATS → S3 → SMB → пул соединений БД; каждый шаг независим — ошибка одного логируется и не отменяет остальные. """ @@ -42,7 +42,7 @@ from src.core.config import Settings, get_settings from src.core.exception_handlers import register_exception_handlers from src.core.nats import MessageHandler, NatsConnection from src.db.db import async_session_maker, engine, session_transaction -from src.services.recognition_exchange import ExchangeTasks, NoopMapper, handle_message +from src.services.recognition_exchange import ExchangeTasks, handle_message from src.services.session_audit import SessionAudit settings = get_settings() @@ -102,8 +102,9 @@ async def _connect_s3() -> S3Storage | None: def _result_handler() -> MessageHandler: - """Inbox `recognition.result`: `handle_message` с фабрикой сессий приложения и `NoopMapper` (Срез 0).""" - return partial(handle_message, session_factory=async_session_maker, mapper=NoopMapper()) + """Inbox `recognition.result`: `handle_message` с фабрикой сессий приложения (маппинг результата — отдельный цикл, + Срез 2c design D-1).""" + return partial(handle_message, session_factory=async_session_maker) async def _subscribe_results(nats: NatsConnection) -> None: diff --git a/api/src/services/file_status.py b/api/src/services/file_status.py new file mode 100644 index 0000000..c9b70b4 --- /dev/null +++ b/api/src/services/file_status.py @@ -0,0 +1,85 @@ +"""Статус `stored_file` по заданию распознавания (spec primary-files «Жизненный цикл статуса файла»; design D-5; +рулинги R-4, R-5; docs/context/05-domain-rules.md §9.2). + +Единственное правило перехода: статус файла меняет только его ТЕКУЩЕЕ задание +(`stored_file.current_recognition_job_id = job.id`). Результат, ack, таймаут или dlq старого задания (повторная загрузка +после `SUPERSEDED`, будущий replay) файл не трогает. `PROCESSED` и `SUPERSEDED` — итог маппинга и автозамены — обменом +не перезаписываются. + +Функции выполняют `UPDATE` в транзакции вызывающего (inbox, вахтёр, публикатор) — после изменения строки задания: +порядок блокировок «задание → файл» одинаков у всех участников, взаимоблокировок нет. `audit_event` не пишется: +обмен не аудируется (ASSUMPTION(OQ-55)). +""" + +import uuid +from collections.abc import Collection, Sequence +from datetime import datetime +from typing import Any, Protocol + +from sqlalchemy import CursorResult, update +from sqlalchemy.ext.asyncio import AsyncSession + +from src.models.enums import FileStatus +from src.models.files import StoredFile + +# Итоги маппинга и автозамены: события обмена их не перезаписывают (design D-5) +PROTECTED_FILE_STATUSES: frozenset[FileStatus] = frozenset({FileStatus.PROCESSED, FileStatus.SUPERSEDED}) + + +class JobRef(Protocol): + """Что нужно от задания: `recognition_job.id` и `file_id` (ORM `RecognitionJob` или строка выборки).""" + + @property + def id(self) -> uuid.UUID: ... + + @property + def file_id(self) -> uuid.UUID: ... + + +def _rowcount(result: Any) -> int: + return result.rowcount if isinstance(result, CursorResult) else 0 + + +async def set_file_status_for_job( + session: AsyncSession, + job: JobRef, + status: FileStatus, + *, + now: datetime, + only_from: Collection[FileStatus] | None = None, +) -> bool: + """Перевести файл задания `job` в `status` (`updated_at = now`), если задание — текущее для файла и файл не в + `PROCESSED`/`SUPERSEDED`; `only_from` — допустимые исходные статусы (R-5: ack и результат без ack — только из + `UPLOADED`, запоздавший ack не воскрешает `ERROR`). Возвращает `True`, если строка файла изменена.""" + stmt = ( + update(StoredFile) + .where( + StoredFile.id == job.file_id, + StoredFile.current_recognition_job_id == job.id, + StoredFile.status.not_in(PROTECTED_FILE_STATUSES), + ) + .values(status=status, updated_at=now) + .execution_options(synchronize_session=False) + ) + if only_from is not None: + stmt = stmt.where(StoredFile.status.in_(list(only_from))) + return _rowcount(await session.execute(stmt)) == 1 + + +async def set_file_status_for_jobs( + session: AsyncSession, job_ids: Sequence[uuid.UUID], status: FileStatus, *, now: datetime +) -> int: + """Пакетный вариант для вахтёра: файлы, у которых текущее задание из `job_ids` и статус не `PROCESSED`/ + `SUPERSEDED`, → `status`. Возвращает число изменённых файлов.""" + if not job_ids: + return 0 + stmt = ( + update(StoredFile) + .where( + StoredFile.current_recognition_job_id.in_(list(job_ids)), + StoredFile.status.not_in(PROTECTED_FILE_STATUSES), + ) + .values(status=status, updated_at=now) + .execution_options(synchronize_session=False) + ) + return _rowcount(await session.execute(stmt)) diff --git a/api/src/services/recognition_exchange.py b/api/src/services/recognition_exchange.py index 8ca7f2e..19934ce 100644 --- a/api/src/services/recognition_exchange.py +++ b/api/src/services/recognition_exchange.py @@ -4,9 +4,13 @@ docs/context/03-architecture.md, «Контракт обмена через NATS 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`. + `metrics.republish_count`); после `OUTBOX_MAX_REPUBLISH` повторов → `dlq`, `error_code = NO_ACK`, файл текущего + задания → `ERROR`. `attempt` не растёт: обработка не начиналась. Replay (`attempt + 1`) — вне Среза 2c. + +Статусы `stored_file` (Срез 2c; spec primary-files «Жизненный цикл статуса файла», design D-5, рулинги R-4, R-5) — +только через `services/file_status.py` и только для текущего задания файла (`current_recognition_job_id = job.id`). +- `expire_deadlines()` — вахтёр: `processing`/`published` с `reply_deadline < now` → `timeout`, `error_code = TIMEOUT`; + файлы этих заданий (по `current_recognition_job_id`) → `ERROR` в той же транзакции. - `run_publisher_loop()` / `run_watchdog_loop()` — фоновые циклы; `ExchangeTasks` запускает их в lifespan при `NATS_ENABLED=true` и останавливает до закрытия NATS. @@ -20,22 +24,25 @@ SKIP LOCKED` → публикация → `UPDATE` → `COMMIT`). Блокиро параллельный проход (второй инстанс ядра) пропускает заблокированную строку и не публикует её повторно, а после commit строка уже не подходит под условие выборки. -Обмен не пишет `audit_event` (рулинг R-7, design #20). +Обмен не пишет `audit_event` (рулинг R-7, design #20; ASSUMPTION(OQ-55)). Inbox (задача 24, рулинги R-12, R-22): `handle_message()` — обработчик подписки `recognition.result` (`src/core/nats.py`, queue group `arm-core`). Одно сообщение — одна короткая транзакция без сетевого IO: `SELECT … FOR UPDATE` строки задания (ЖДЁТ блокировку: ack, пришедший, пока публикатор держит строку до своего commit, применяется после него, а не теряется) → проверки статуса → изменения → `COMMIT`. -- ack (`status = processing`): `published` → `processing`, `started_at`, `service_*`; повторный ack для `processing` — - тихо игнорируется; ack для другого статуса — warning, задание не меняется. +- ack (`status = processing`): `published` → `processing`, `started_at`, `service_*`, файл `UPLOADED → PROCESSING` + (R-5: только из `UPLOADED`); повторный ack для `processing` — тихо игнорируется; ack для другого статуса — warning, + задание и файл не меняются. - результат (`done | needs_check | error`): дубликат определяется ТОЛЬКО по `recognition_job.result_msg_id == Nats-Msg-Id` (at-least-once) — без изменений; задание в финальном статусе (`FINAL_STATUSES`) — warning, без изменений; иначе (`queued | published | processing`: потерянный ack не блокирует результат) — регистрация файла результата в - `s3_object` (`bucket_role = results`), обновление задания, `ResultMapper.map_result` (Срез 0 — `NoopMapper`; Срез 2 - заменит маппингом результата в `document` / `osv_entry` / `financial_statement_*`) и commit. + `s3_object` (`bucket_role = results`), обновление задания, статус файла и commit. `error` → файл `ERROR`; + `done`/`needs_check` → файл `UPLOADED → PROCESSING` (R-4: ack мог потеряться — core NATS без JetStream), из + `PROCESSING` файл не меняется и ждёт цикла маппинга (design D-1). Inbox НЕ читает файл результата и не создаёт + документы: `ResultMapper` удалён (OQ-67), маппинг — отдельный цикл `services/result_mapping.py`. - Гонка с вахтёром (`expire_deadlines` — один `UPDATE` по статусу) решается блокировкой строки: кто первым взял строку, тот и задаёт финальный статус; второй видит финальный статус и ничего не меняет. -- Невалидное сообщение, неизвестный `job_id`, ошибка БД или mapper'а — лог без повторов; исключения наружу не выходят. +- Невалидное сообщение, неизвестный `job_id`, ошибка БД — лог без повторов; исключения наружу не выходят. Задание без применённого результата дойдёт до `timeout` (осознанный компромисс D-6). """ @@ -49,7 +56,7 @@ from typing import Any, Protocol from loguru import logger from pydantic import ValidationError -from sqlalchemy import CursorResult, and_, case, func, literal, or_, select, update +from sqlalchemy import and_, case, func, literal, or_, select, update from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker @@ -69,8 +76,9 @@ from src.contract import ( ) from src.core.config import Settings from src.core.nats import NatsPublishError -from src.models.enums import RecognitionPipeline, RecognitionStatus, S3BucketRole +from src.models.enums import FileStatus, RecognitionPipeline, RecognitionStatus, S3BucketRole from src.models.files import RecognitionJob, S3Object, StoredFile +from src.services.file_status import set_file_status_for_job, set_file_status_for_jobs WATCHDOG_PERIOD_SECONDS = 30 # период вахтёра сроков (spec «Таймаут ожидания результата») PUBLISH_BATCH_SIZE = 50 # заданий за один проход публикатора (design D-6, LIMIT 50) @@ -220,6 +228,7 @@ async def _process_one( # noqa: PLR0913 - один шаг прохода: вс finished_at=now, ) ) + await set_file_status_for_job(session, row, FileStatus.ERROR, now=now) report.dlq += 1 logger.warning( "Запрос распознавания без ack: задание переведено в dlq. error_category=no_ack, job_id={job_id}, " @@ -348,8 +357,9 @@ async def publish_pending( # noqa: PLR0913 - параметры прохода 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` с условием по статусу: результат, - успевший перевести задание в финальный статус, не перетирается. Возвращает число заданий.""" + `timeout`, `error_code = TIMEOUT`, `finished_at = now`. Один атомарный `UPDATE … RETURNING id` с условием по + статусу: результат, успевший перевести задание в финальный статус, не перетирается. В той же транзакции файлы этих + заданий (только где задание — текущее, `current_recognition_job_id`) → `ERROR`. Возвращает число заданий.""" moment = (now or _utcnow)() stmt = ( update(RecognitionJob) @@ -363,13 +373,19 @@ async def expire_deadlines(session_factory: async_sessionmaker[AsyncSession], *, error_message="Сервис распознавания не прислал результат до срока ответа (reply_deadline)", finished_at=moment, ) + .returning(RecognitionJob.id) .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 + job_ids = list((await session.execute(stmt)).scalars()) + files = await set_file_status_for_jobs(session, job_ids, FileStatus.ERROR, now=moment) + count = len(job_ids) if count: - logger.warning("Задания распознавания переведены в timeout по сроку ответа. count={count}", count=count) + logger.warning( + "Задания распознавания переведены в timeout по сроку ответа. count={count}, files_error={files}", + count=count, + files=files, + ) return count @@ -495,31 +511,6 @@ class InboundMessage(Protocol): def subject(self) -> str: ... -class ResultMapper(Protocol): - """Точка расширения inbox (design D-6): маппинг принятого результата в доменные таблицы. - - Вызывается в транзакции обработки сообщения, после обновления задания (`job` уже содержит статус и ссылку на - файл результата и может быть уточнено mapper'ом). Исключение откатывает всю транзакцию: задание не меняется и - дойдёт до `timeout`. Сетевой IO внутри — на совести реализации: транзакция держит блокировку строки задания. - Срез 0 — `NoopMapper`; Срез 2 заменит чтением файла результата и записью `document` / `osv_entry` / … - """ - - async def map_result( - self, session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage - ) -> None: ... - - -class NoopMapper: - """Mapper Среза 0: результат только регистрируется (`s3_object`, задание), доменные таблицы не заполняются.""" - - async def map_result(self, session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage) -> None: - logger.debug("Маппинг результата не выполняется (NoopMapper, Срез 0). job_id={job_id}", job_id=message.job_id) - - -class _MapperError(Exception): - """Исключение `ResultMapper.map_result` — отличаем от ошибок БД в логе.""" - - def header_msg_id(headers: Mapping[str, str] | None) -> str | None: """Значение заголовка `Nats-Msg-Id` без учёта регистра имени; нет заголовка или пустое значение — `None`.""" if not headers: @@ -588,19 +579,26 @@ async def _lock_job(session: AsyncSession, job_id: uuid.UUID) -> RecognitionJob return (await session.execute(stmt)).scalar_one_or_none() -def _apply_ack(job: RecognitionJob, message: RecognitionResultMessage, msg_id: str, now: datetime) -> None: - """Ack (`status = processing`): `published` → `processing`; повтор для `processing` — тихо; иначе — warning.""" +async def _apply_ack( + session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage, msg_id: str, now: datetime +) -> None: + """Ack (`status = processing`): `published` → `processing` и файл `UPLOADED → PROCESSING` (R-5); повтор для + `processing` — тихо; иначе — warning (запоздавший ack после `timeout`/`dlq` файл не трогает).""" if job.status is RecognitionStatus.published: job.status = RecognitionStatus.processing job.started_at = now job.service_name = message.service.name.value job.service_version = message.service.version + file_changed = await set_file_status_for_job( + session, job, FileStatus.PROCESSING, now=now, only_from={FileStatus.UPLOADED} + ) logger.info( "Сервис подтвердил получение запроса (ack): задание в processing. job_id={job_id}, service={service}, " - "msg_id={msg_id}", + "msg_id={msg_id}, file_processing={file_changed}", job_id=job.id, service=message.service.name.value, msg_id=msg_id, + file_changed=file_changed, ) elif job.status is RecognitionStatus.processing: logger.debug("Повторный ack для задания в processing, игнорируется. job_id={job_id}", job_id=job.id) @@ -653,7 +651,7 @@ async def _apply_result( session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage, msg_id: str, now: datetime ) -> bool: """Результат (`done | needs_check | error`): дубликат и финальный статус — без изменений (`False`); иначе — - регистрация файла результата и обновление задания (`True`, дальше — mapper и commit).""" + регистрация файла результата, обновление задания и статуса файла (`True`, дальше — commit).""" if job.result_msg_id == msg_id: logger.debug( "Повторная доставка результата (дубликат по result_msg_id), изменений нет. job_id={job_id}, " @@ -689,26 +687,26 @@ async def _apply_result( job.result_schema_version = message.result_schema_version job.result_s3_object_id = result_s3_object_id job.result_msg_id = msg_id - return True - - -async def _map_result( - session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage, mapper: ResultMapper -) -> None: - """`mapper.map_result` в транзакции сообщения; его исключение — `_MapperError` (откат всей транзакции).""" - try: - await mapper.map_result(session, job, message) - except Exception as exc: - raise _MapperError(f"ResultMapper.map_result: {type(exc).__name__}") from exc + file_changed = await _apply_result_file_status(session, job, now) logger.info( "Результат распознавания принят. job_id={job_id}, status={status}, service={service}, msg_id={msg_id}, " - "error_code={error_code}", + "error_code={error_code}, file_changed={file_changed}", job_id=job.id, status=job.status.value, service=message.service.name.value, - msg_id=job.result_msg_id, + msg_id=msg_id, error_code=job.error_code, + file_changed=file_changed, ) + return True + + +async def _apply_result_file_status(session: AsyncSession, job: RecognitionJob, now: datetime) -> bool: + """Статус файла по принятому результату (design D-5): `error` → `ERROR`; `done`/`needs_check` → `PROCESSING` только + из `UPLOADED` (R-4: ack мог потеряться), иначе файл ждёт цикла маппинга как есть. Только текущее задание файла.""" + if job.status is RecognitionStatus.error: + return await set_file_status_for_job(session, job, FileStatus.ERROR, now=now) + return await set_file_status_for_job(session, job, FileStatus.PROCESSING, now=now, only_from={FileStatus.UPLOADED}) def _metrics_json(metrics: Metrics) -> dict[str, Any]: @@ -716,7 +714,7 @@ def _metrics_json(metrics: Metrics) -> dict[str, Any]: async def _apply_message( - session: AsyncSession, message: RecognitionResultMessage, header_id: str | None, mapper: ResultMapper, clock: Clock + session: AsyncSession, message: RecognitionResultMessage, header_id: str | None, clock: Clock ) -> None: job = await _lock_job(session, message.job_id) if job is None: @@ -748,36 +746,30 @@ async def _apply_message( ) now = clock() if is_ack: - _apply_ack(job, message, msg_id, now) - elif await _apply_result(session, job, message, msg_id, now): - await _map_result(session, job, message, mapper) + await _apply_ack(session, job, message, msg_id, now) + else: + await _apply_result(session, job, message, msg_id, now) async def handle_message( msg: InboundMessage, *, session_factory: async_sessionmaker[AsyncSession], - mapper: ResultMapper | None = None, now: Clock | None = None, ) -> None: """Обработчик `recognition.result` (spec «Ack сервиса переводит задание в `processing`», «Inbox принимает результат идемпотентно»; design D-6). Модель транзакции и правила — в docstring модуля. - Исключения наружу не выходят: невалидное сообщение — warning; ошибка БД или mapper'а — откат транзакции, - `logger.exception`, без повторов (задание дойдёт до `timeout`). `audit_event` не пишется (R-7). + Исключения наружу не выходят: невалидное сообщение — warning; ошибка БД — откат транзакции (задание и файл), + `logger.exception`, без повторов (задание дойдёт до `timeout`). `audit_event` не пишется (R-7, + ASSUMPTION(OQ-55)). Файл результата не читается: маппинг — отдельный цикл (design D-1). """ message = _parse_message(msg) if message is None: return try: async with session_factory() as session, session.begin(): - await _apply_message(session, message, header_msg_id(msg.headers), mapper or NoopMapper(), now or _utcnow) - except _MapperError: - logger.exception( - "Ошибка маппинга результата распознавания, транзакция откачена; задание дойдёт до timeout. " - "error_category=mapper_error, job_id={job_id}", - job_id=message.job_id, - ) + await _apply_message(session, message, header_msg_id(msg.headers), now or _utcnow) except Exception as exc: logger.exception( "Ошибка обработки сообщения {subject}, транзакция откачена (без повторов). error_category=inbox_error, " diff --git a/api/tests/factories.py b/api/tests/factories.py index 5af4174..551dcc2 100644 --- a/api/tests/factories.py +++ b/api/tests/factories.py @@ -8,7 +8,8 @@ `FakeStorage` для тестов загрузки и маппинга (Срез 2c; `flush`, без `commit`). `make_entries_from_result` создаёт Task 9 (R-2). - `CommittedJobs` — задания распознавания, ЗАФИКСИРОВАННЫЕ в БД (`commit`): для фоновых процессов обмена, которые - работают в собственных транзакциях и не видят чужую незафиксированную транзакцию. Всё созданное удаляется + работают в собственных транзакциях и не видят чужую незафиксированную транзакцию. `make_file_job` / `add_job` — файл + с текущим заданием и второе задание того же файла (статусы `stored_file` в обмене, Срез 2c). Всё созданное удаляется `cleanup()` в порядке FK, включая строки `s3_object` файлов результата, зарегистрированные inbox'ом. """ @@ -400,6 +401,71 @@ class CommittedJobs: self.job_ids.append(job.id) return job.id + async def make_file_job( + self, + *, + file_status: FileStatus = FileStatus.UPLOADED, + job_status: RecognitionStatus = RecognitionStatus.queued, + **job_fields: Any, + ) -> tuple[uuid.UUID, uuid.UUID]: + """Файл с текущим заданием (`make_stored_file`: `current_recognition_job_id` = задание), зафиксированный в БД. + + `job_fields` — поля `recognition_job` поверх фабрики (`reply_deadline`, `published_at`, `metrics`, …). + Возвращает `(stored_file.id, recognition_job.id)`. + """ + async with self._session_factory() as session, session.begin(): + entity = await make_legal_entity(session, name="ООО Тест статусов файла") + file, job = await make_stored_file(session, entity=entity, status=file_status, job_status=job_status) + for name, value in job_fields.items(): + setattr(job, name, value) + await session.flush() + self._entity_ids.append(entity.id) + self._s3_ids.append(file.s3_object_id) + self._file_ids.append(file.id) + self.job_ids.append(job.id) + return file.id, job.id + + async def add_job( + self, + file_id: uuid.UUID, + *, + status: RecognitionStatus = RecognitionStatus.queued, + current: bool = True, + **job_fields: Any, + ) -> uuid.UUID: + """Ещё одно задание файла `file_id` (как replay: `attempt` = максимум по файлу + 1); `current=True` — оно + становится текущим заданием файла.""" + async with self._session_factory() as session, session.begin(): + file = await session.get(StoredFile, file_id) + assert file is not None + last = ( + await session.execute(select(func.max(RecognitionJob.attempt)).where(RecognitionJob.file_id == file_id)) + ).scalar_one() + job_fields.setdefault("attempt", (last or 0) + 1) + tabular = file.file_type in TABULAR_FILE_TYPES + pipeline = RecognitionPipeline.tabular if tabular else RecognitionPipeline.image + job = RecognitionJob( + file_id=file_id, + pipeline=pipeline, + request_subject=f"recognition.{pipeline.value}.request", + schema_version="1.0", + status=status, + **job_fields, + ) + session.add(job) + await session.flush() + if current: + file.current_recognition_job_id = job.id + self.job_ids.append(job.id) + return job.id + + async def file(self, file_id: uuid.UUID) -> StoredFile: + """Свежее состояние файла из БД (новая сессия).""" + async with self._session_factory() as session: + file = await session.get(StoredFile, file_id) + assert file is not None + return file + async def s3_key(self, job_id: uuid.UUID) -> str: """Ключ S3 исходного файла задания.""" async with self._session_factory() as session: diff --git a/api/tests/test_app.py b/api/tests/test_app.py index 842dbcc..63bb893 100644 --- a/api/tests/test_app.py +++ b/api/tests/test_app.py @@ -226,7 +226,8 @@ def test_create_nats_enabled_returns_connection_for_results(monkeypatch: pytest. async def test_result_subscription_uses_inbox_handler_with_dependencies(): - """Подписка `recognition.result` получает `handle_message` с фабрикой сессий приложения и `NoopMapper`.""" + """Подписка `recognition.result` получает `handle_message` с фабрикой сессий приложения (без `mapper`: маппинг + результата — отдельный цикл, Срез 2c D-1).""" subscribed: list[Any] = [] class _FakeConnection: @@ -238,9 +239,8 @@ async def test_result_subscription_uses_inbox_handler_with_dependencies(): (handler,) = subscribed assert handler.func is recognition_exchange.handle_message assert handler.args == () - assert set(handler.keywords) == {"session_factory", "mapper"} + assert set(handler.keywords) == {"session_factory"} assert handler.keywords["session_factory"] is app_module.async_session_maker - assert isinstance(handler.keywords["mapper"], recognition_exchange.NoopMapper) def test_create_nats_subscribes_results_on_connect(monkeypatch: pytest.MonkeyPatch): diff --git a/api/tests/test_exchange_lifespan_nats.py b/api/tests/test_exchange_lifespan_nats.py index 133a365..af7e739 100644 --- a/api/tests/test_exchange_lifespan_nats.py +++ b/api/tests/test_exchange_lifespan_nats.py @@ -32,7 +32,7 @@ from src.core.config import get_settings from src.core.nats import RESULT_QUEUE_GROUP, NatsConnection from src.models.enums import RecognitionStatus, S3BucketRole from src.models.files import RecognitionJob, S3Object -from src.services.recognition_exchange import NoopMapper, handle_message +from src.services.recognition_exchange import handle_message from tests.factories import CommittedJobs from tests.test_recognition_inbox import MESSAGE_VALIDATOR, envelope @@ -277,10 +277,7 @@ async def test_two_core_instances_handle_result_once( """Два `NatsConnection` в queue group `arm-core` на одном subject: результат обрабатывает ровно один инстанс, задание меняется один раз; повторная доставка того же сообщения ничего не меняет.""" settings = get_settings().OPENBAO.NATS - handlers = [ - _CountingHandler(partial(handle_message, session_factory=session_factory, mapper=NoopMapper())) - for _ in range(2) - ] + handlers = [_CountingHandler(partial(handle_message, session_factory=session_factory)) for _ in range(2)] def subscriber(handler: _CountingHandler) -> Callable[[NatsConnection], Awaitable[None]]: async def subscribe(conn: NatsConnection) -> None: diff --git a/api/tests/test_recognition_exchange.py b/api/tests/test_recognition_exchange.py index 1deb48e..8af29a9 100644 --- a/api/tests/test_recognition_exchange.py +++ b/api/tests/test_recognition_exchange.py @@ -1,5 +1,7 @@ """Outbox-публикатор и вахтёр сроков `recognition_job` (spec recognition-job-exchange: «Outbox публикует задания из -`queued`», «Запрос без ack публикуется повторно, затем — `dlq`», «Таймаут ожидания результата»; design D-6). +`queued`», «Запрос без ack публикуется повторно, затем — `dlq`», «Таймаут ожидания результата»; design D-6). Срез 2c: +`timeout` вахтёра и `dlq` публикатора переводят файл текущего задания в `ERROR` (spec primary-files «Жизненный цикл +статуса файла», design D-5). Интеграционные тесты работают против реального PostgreSQL (БД сессии pytest) и NATS из `.env.test`. Публикатор и вахтёр ведут собственные транзакции, поэтому данные фиксируются реально (`CommittedJobs`) и удаляются в teardown. @@ -28,7 +30,7 @@ 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.enums import FileStatus, FileType, RecognitionPipeline, RecognitionStatus from src.models.files import RecognitionJob from src.services import recognition_exchange as rx from src.services.recognition_exchange import ( @@ -591,6 +593,93 @@ async def test_expire_deadlines( assert job.finished_at is None +@pytest.mark.integration +@pytest.mark.parametrize("file_status", [FileStatus.PROCESSING, FileStatus.UPLOADED]) +@pytest.mark.parametrize("job_status", [RecognitionStatus.processing, RecognitionStatus.published]) +async def test_expire_deadlines_moves_current_file_to_error( + session_factory: async_sessionmaker[AsyncSession], + jobs: CommittedJobs, + file_status: FileStatus, + job_status: RecognitionStatus, +): + """Сценарий «Сервис не ответил» (primary-files): задание — `timeout`, файл — `ERROR` в той же транзакции.""" + clock = Clock() + file_id, job_id = await jobs.make_file_job( + file_status=file_status, job_status=job_status, reply_deadline=clock() - timedelta(minutes=1) + ) + + assert await expire_deadlines(session_factory, now=clock) == 1 + + job, file = await jobs.get(job_id), await jobs.file(file_id) + assert (job.status, job.error_code) == (RecognitionStatus.timeout, "TIMEOUT") + assert (file.status, file.updated_at) == (FileStatus.ERROR, clock()) + + +@pytest.mark.integration +@pytest.mark.parametrize("file_status", [FileStatus.PROCESSED, FileStatus.SUPERSEDED]) +async def test_expire_deadlines_keeps_processed_and_superseded_file( + session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, file_status: FileStatus +): + clock = Clock() + file_id, job_id = await jobs.make_file_job( + file_status=file_status, job_status=RecognitionStatus.processing, reply_deadline=clock() - timedelta(minutes=1) + ) + + assert await expire_deadlines(session_factory, now=clock) == 1 + + assert (await jobs.get(job_id)).status is RecognitionStatus.timeout + file = await jobs.file(file_id) + assert (file.status, file.updated_at) == (file_status, None) + + +@pytest.mark.integration +async def test_expire_deadlines_of_non_current_job_keeps_file( + session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs +): + """Review Focus 1: таймаут старого задания файла, у которого уже новое текущее задание, файл не трогает.""" + clock = Clock() + file_id, old_job = await jobs.make_file_job( + file_status=FileStatus.UPLOADED, + job_status=RecognitionStatus.processing, + reply_deadline=clock() - timedelta(minutes=1), + ) + new_job = await jobs.add_job(file_id, status=RecognitionStatus.queued, current=True) + + assert await expire_deadlines(session_factory, now=clock) == 1 + + assert (await jobs.get(old_job)).status is RecognitionStatus.timeout + file = await jobs.file(file_id) + assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None) + + +@pytest.mark.integration +@pytest.mark.parametrize("current", [True, False], ids=["current", "non_current"]) +async def test_dlq_moves_current_file_to_error( + session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, current: bool +): + """Публикатор переводит задание в `dlq` — файл текущего задания `ERROR`; файл чужого задания не меняется.""" + clock = Clock() + file_id, job_id = await jobs.make_file_job( + job_status=RecognitionStatus.published, + published_at=clock() - ACK * 2, + request_msg_id="dlq-test:1", + reply_deadline=clock() + timedelta(hours=1), + metrics={"republish_count": MAX_REPUBLISH}, + ) + # новое задание — `processing`, чтобы публикатор его не трогал (в проходе только задание старого) + new_job = None if current else await jobs.add_job(file_id, status=RecognitionStatus.processing, current=True) + + report = await _publish(session_factory, FakeNats(), clock) + + assert report.dlq == 1 + assert (await jobs.get(job_id)).status is RecognitionStatus.dlq + file = await jobs.file(file_id) + if current: + assert (file.status, file.updated_at) == (FileStatus.ERROR, clock()) + else: + assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None) + + # --- фоновые циклы ----------------------------------------------------------------------------------------------- diff --git a/api/tests/test_recognition_inbox.py b/api/tests/test_recognition_inbox.py index e3d69c3..5a1bec3 100644 --- a/api/tests/test_recognition_inbox.py +++ b/api/tests/test_recognition_inbox.py @@ -1,5 +1,7 @@ """Inbox обмена `recognition_job`: `handle_message()` (spec recognition-job-exchange: «Ack сервиса переводит задание в -`processing`», «Inbox принимает результат идемпотентно»; design D-6; рулинги R-7, R-12, R-22). +`processing`», «Inbox принимает результат идемпотентно»; design D-6; рулинги R-7, R-12, R-22). Срез 2c: статусы +`stored_file` по ack / результату (spec primary-files «Жизненный цикл статуса файла», design D-5, рулинги R-4, R-5); +`ResultMapper` удалён — inbox не читает S3 и не создаёт документы (маппинг — отдельный цикл, D-1). Быстрые тесты (без маркера) — разбор заголовка `Nats-Msg-Id`, слияние метрик, отказ невалидным сообщениям без обращения к БД, ошибка БД не выходит наружу, шаблонный `nats_callback` удалён. @@ -10,6 +12,7 @@ import asyncio import importlib.util +import inspect import json import time import uuid @@ -29,14 +32,14 @@ from nats.aio.msg import Msg from sqlalchemy import JSON, func, select from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine -from src.contract import NATS_MSG_ID_HEADER, RecognitionResultMessage, ack_msg_id, result_msg_id +from src.contract import NATS_MSG_ID_HEADER, ack_msg_id, result_msg_id from src.core.config import get_settings from src.core.nats import NatsConnection -from src.models.enums import RecognitionStatus, S3BucketRole -from src.models.files import RecognitionJob, S3Object +from src.models.enums import FileStatus, RecognitionStatus, S3BucketRole +from src.models.files import RecognitionJob, S3Object, StoredFile +from src.services import recognition_exchange as rx from src.services.recognition_exchange import ( FINAL_STATUSES, - NoopMapper, expire_deadlines, handle_message, header_msg_id, @@ -54,6 +57,7 @@ BUCKET = "arm-ka" READY_TIMEOUT = 5.0 RACE_JOBS = 10 LOCK_HOLD_SECONDS = 0.5 +_ERROR = {"code": "UNSUPPORTED_FORMAT", "message": "формат файла не поддерживается"} # --- инфраструктура --------------------------------------------------------------------------------------------- @@ -103,21 +107,20 @@ def fake_msg(body: Any, msg_id: str | None, *, header: str = NATS_MSG_ID_HEADER, return SimpleNamespace(data=data, headers=headers, subject="recognition.result") -class RecordingMapper: - """Подставной mapper: запоминает вызовы и видимое состояние задания; `fail=True` — исключение; `delay` — пауза - внутри транзакции (строка задания остаётся заблокированной).""" +class SlowFileStatus: + """Подмена `set_file_status_for_job` в inbox: запоминает вызовы (id задания, статус) и держит паузу `delay` + ВНУТРИ транзакции сообщения (строка задания остаётся заблокированной), затем вызывает настоящую функцию.""" - def __init__(self, *, fail: bool = False, delay: float = 0) -> None: - self.fail = fail + def __init__(self, monkeypatch: pytest.MonkeyPatch, *, delay: float) -> None: self.delay = delay - self.calls: list[tuple[uuid.UUID, RecognitionStatus, RecognitionResultMessage]] = [] + self.calls: list[tuple[uuid.UUID, FileStatus]] = [] + self._original = rx.set_file_status_for_job + monkeypatch.setattr(rx, "set_file_status_for_job", self) - async def map_result(self, session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage) -> None: - self.calls.append((job.id, job.status, message)) - if self.delay: - await asyncio.sleep(self.delay) - if self.fail: - raise RuntimeError("подставная ошибка маппинга") + async def __call__(self, session: AsyncSession, job: RecognitionJob, status: FileStatus, **kwargs: Any) -> bool: + self.calls.append((job.id, status)) + await asyncio.sleep(self.delay) + return await self._original(session, job, status, **kwargs) @pytest.fixture @@ -152,6 +155,13 @@ async def _s3_total(session_factory: async_sessionmaker[AsyncSession]) -> int: return int((await session.execute(select(func.count()).select_from(S3Object))).scalar_one()) +async def _file_row(session_factory: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> dict[str, Any]: + """Снимок всех колонок строки `stored_file`.""" + async with session_factory() as session: + row = (await session.execute(select(StoredFile.__table__).where(StoredFile.id == file_id))).one() + return dict(row._mapping) + + async def _file_id(jobs: CommittedJobs, job_id: uuid.UUID) -> uuid.UUID: return (await jobs.get(job_id)).file_id @@ -264,9 +274,11 @@ async def test_database_error_is_logged_and_not_raised(logs: list[Any]): assert record["exception"] is not None -async def test_noop_mapper_does_nothing(): - message = RecognitionResultMessage.model_validate(envelope(_JOB, _FILE, "done")) - assert await NoopMapper().map_result(None, None, message) is None # type: ignore[arg-type] +def test_inbox_has_no_result_mapper(): + """OQ-67 (Срез 2c, D-1): точки расширения `ResultMapper` / `NoopMapper` нет, `handle_message` без `mapper`.""" + assert not hasattr(rx, "ResultMapper") + assert not hasattr(rx, "NoopMapper") + assert "mapper" not in inspect.signature(handle_message).parameters def test_final_statuses(): @@ -422,22 +434,23 @@ async def test_result_metrics_merge_over_null_metrics( @pytest.mark.integration async def test_redelivery_changes_nothing(session_factory, jobs: CommittedJobs, logs: list[Any]): - """Сценарий «Повторная доставка»: ни recognition_job, ни s3_object не меняются; mapper не вызывается.""" - job_id = await jobs.make_queued_job(status=RecognitionStatus.processing) - file_id = await _file_id(jobs, job_id) + """Сценарий «Повторная доставка»: ни recognition_job, ни s3_object, ни stored_file не меняются.""" + file_id, job_id = await jobs.make_file_job( + file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing + ) msg = fake_msg(envelope(job_id, file_id, "done", metrics={"rows": 3}), result_msg_id(job_id, 1)) - mapper = RecordingMapper() - await _handle(session_factory, msg, mapper=mapper) + await _handle(session_factory, msg) job_before = await _row(session_factory, job_id) + file_before = await _file_row(session_factory, file_id) s3_before = await _s3_rows(session_factory, _result_key(file_id, job_id)) total_before = await _s3_total(session_factory) - await _handle(session_factory, msg, mapper=mapper, now=lambda: datetime.now(UTC) + timedelta(hours=1)) + await _handle(session_factory, msg, now=lambda: datetime.now(UTC) + timedelta(hours=1)) assert await _row(session_factory, job_id) == job_before assert await _s3_rows(session_factory, _result_key(file_id, job_id)) == s3_before assert await _s3_total(session_factory) == total_before - assert len(mapper.calls) == 1 + assert await _file_row(session_factory, file_id) == file_before assert _logged(logs, "дубликат", "DEBUG") assert not _logged(logs, "финальном статусе") @@ -482,13 +495,11 @@ async def test_result_for_final_status_is_logged_and_ignored( file_id = await _file_id(jobs, job_id) before = await _row(session_factory, job_id) total_before = await _s3_total(session_factory) - mapper = RecordingMapper() - await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)), mapper=mapper) + await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))) assert await _row(session_factory, job_id) == before assert await _s3_total(session_factory) == total_before - assert mapper.calls == [] (record,) = _logged(logs, "финальном статусе", "WARNING") assert status.value in record["message"] @@ -587,39 +598,163 @@ async def test_existing_s3_object_is_reused(session_factory, jobs: CommittedJobs assert len(await _s3_rows(session_factory, _result_key(file_id, job_id))) == 1 -# --- mapper ------------------------------------------------------------------------------------------------------ +# --- статусы stored_file (Срез 2c: spec primary-files «Жизненный цикл статуса файла», D-5, R-4, R-5) ------------- @pytest.mark.integration -async def test_mapper_sees_job_and_message(session_factory, jobs: CommittedJobs): - job_id = await jobs.make_queued_job(status=RecognitionStatus.processing) - file_id = await _file_id(jobs, job_id) - mapper = RecordingMapper() +async def test_ack_moves_uploaded_file_to_processing(session_factory, jobs: CommittedJobs): + """Сценарий «Сервис подтвердил»: файл текущего задания — `PROCESSING`, `updated_at` — момент ack.""" + file_id, job_id = await jobs.make_file_job(job_status=RecognitionStatus.published) + moment = datetime(2026, 10, 5, 9, 0, tzinfo=UTC) - await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)), mapper=mapper) + await _handle( + session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)), now=lambda: moment + ) - ((seen_job_id, seen_status, message),) = mapper.calls - assert seen_job_id == job_id - assert seen_status is RecognitionStatus.done # mapper видит уже обновлённое задание (может его уточнить) - assert message.job_id == job_id - assert message.result_s3 is not None - assert message.result_s3.key == _result_key(file_id, job_id) + file = await _file_row(session_factory, file_id) + assert file["status"] is FileStatus.PROCESSING + assert file["updated_at"] == moment + assert file["current_recognition_job_id"] == job_id + assert (await _row(session_factory, job_id))["status"] is RecognitionStatus.processing @pytest.mark.integration -async def test_mapper_error_rolls_back_whole_transaction(session_factory, jobs: CommittedJobs, logs: list[Any]): - """Исключение mapper'а — откат всего: задание не изменено, s3_object не создан (дойдёт до timeout, D-6).""" - job_id = await jobs.make_queued_job(status=RecognitionStatus.processing) - file_id = await _file_id(jobs, job_id) - before = await _row(session_factory, job_id) - total_before = await _s3_total(session_factory) +@pytest.mark.parametrize( + "file_status", [FileStatus.ERROR, FileStatus.PROCESSED, FileStatus.SUPERSEDED, FileStatus.PROCESSING] +) +async def test_ack_moves_file_only_from_uploaded(session_factory, jobs: CommittedJobs, file_status: FileStatus): + """R-5: ack ставит `PROCESSING` только из `UPLOADED` — запоздавший ack не воскрешает `ERROR` и не трогает + `PROCESSED`/`SUPERSEDED`; задание при этом подтверждается как обычно.""" + file_id, job_id = await jobs.make_file_job(file_status=file_status, job_status=RecognitionStatus.published) + before = await _file_row(session_factory, file_id) - msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)) - await _handle(session_factory, msg, mapper=RecordingMapper(fail=True)) + await _handle(session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1))) - assert await _row(session_factory, job_id) == before - assert await _s3_total(session_factory) == total_before - (record,) = _logged(logs, "Ошибка маппинга", "ERROR") + assert await _file_row(session_factory, file_id) == before + assert (await _row(session_factory, job_id))["status"] is RecognitionStatus.processing + + +@pytest.mark.integration +@pytest.mark.parametrize("file_status", [FileStatus.PROCESSING, FileStatus.UPLOADED]) +async def test_error_result_moves_file_to_error(session_factory, jobs: CommittedJobs, file_status: FileStatus): + """Сценарий «Парсер вернул ошибку»: файл — `ERROR`, причина — в задании (`error_code`).""" + file_id, job_id = await jobs.make_file_job(file_status=file_status, job_status=RecognitionStatus.processing) + moment = datetime(2026, 10, 5, 9, 30, tzinfo=UTC) + + await _handle( + session_factory, + fake_msg(envelope(job_id, file_id, "error", error=_ERROR), result_msg_id(job_id, 1)), + now=lambda: moment, + ) + + file = await _file_row(session_factory, file_id) + assert (file["status"], file["updated_at"]) == (FileStatus.ERROR, moment) + assert (await _row(session_factory, job_id))["error_code"] == "UNSUPPORTED_FORMAT" + + +@pytest.mark.integration +@pytest.mark.parametrize("status", ["done", "needs_check"]) +async def test_success_result_leaves_processing_file_for_mapping(session_factory, jobs: CommittedJobs, status: str): + """Сценарий «Успешный результат»: файл остаётся `PROCESSING` до маппинга; строка файла не меняется вовсе.""" + file_id, job_id = await jobs.make_file_job( + file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing + ) + before = await _file_row(session_factory, file_id) + + await _handle(session_factory, fake_msg(envelope(job_id, file_id, status), result_msg_id(job_id, 1))) + + assert await _file_row(session_factory, file_id) == before + assert (await _row(session_factory, job_id))["status"] is RecognitionStatus(status) + + +@pytest.mark.integration +@pytest.mark.parametrize("status", ["done", "needs_check"]) +@pytest.mark.parametrize("job_status", [RecognitionStatus.published, RecognitionStatus.queued]) +async def test_success_result_without_ack_moves_uploaded_file_to_processing( + session_factory, jobs: CommittedJobs, status: str, job_status: RecognitionStatus +): + """R-4: ack потерялся (core NATS без JetStream) — результат `done`/`needs_check` переводит файл `UPLOADED → + PROCESSING`, иначе файл с готовым результатом не попал бы в цикл маппинга.""" + file_id, job_id = await jobs.make_file_job(job_status=job_status) + moment = datetime(2026, 10, 5, 10, 0, tzinfo=UTC) + + await _handle( + session_factory, fake_msg(envelope(job_id, file_id, status), result_msg_id(job_id, 1)), now=lambda: moment + ) + + file = await _file_row(session_factory, file_id) + assert (file["status"], file["updated_at"]) == (FileStatus.PROCESSING, moment) + assert (await _row(session_factory, job_id))["status"] is RecognitionStatus(status) + + +@pytest.mark.integration +@pytest.mark.parametrize("file_status", [FileStatus.PROCESSED, FileStatus.SUPERSEDED]) +@pytest.mark.parametrize("status", ["done", "error"]) +async def test_result_does_not_touch_processed_or_superseded_file( + session_factory, jobs: CommittedJobs, file_status: FileStatus, status: str +): + """`PROCESSED` и `SUPERSEDED` — итог маппинга / автозамены: результат обмена их не перезаписывает.""" + file_id, job_id = await jobs.make_file_job(file_status=file_status, job_status=RecognitionStatus.processing) + before = await _file_row(session_factory, file_id) + extra = {"error": _ERROR} if status == "error" else {} + + await _handle(session_factory, fake_msg(envelope(job_id, file_id, status, **extra), result_msg_id(job_id, 1))) + + assert await _file_row(session_factory, file_id) == before + assert (await _row(session_factory, job_id))["status"] is RecognitionStatus(status) + + +@pytest.mark.integration +@pytest.mark.parametrize( + ("status", "msg_id_fn", "expected_job"), + [ + ("processing", ack_msg_id, RecognitionStatus.processing), + ("done", result_msg_id, RecognitionStatus.done), + ("needs_check", result_msg_id, RecognitionStatus.needs_check), + ("error", result_msg_id, RecognitionStatus.error), + ], +) +async def test_message_of_non_current_job_does_not_touch_file( # noqa: PLR0913 - фикстуры и параметры + session_factory, jobs: CommittedJobs, status: str, msg_id_fn: Any, expected_job: RecognitionStatus +): + """Review Focus 1: у файла два задания, текущее — новое; ack/результат старого задания применяется к старому + заданию, но файл (статус, `updated_at`, `current_recognition_job_id`) не трогает.""" + file_id, old_job = await jobs.make_file_job(job_status=RecognitionStatus.published) + new_job = await jobs.add_job(file_id, status=RecognitionStatus.queued, current=True) + before = await _file_row(session_factory, file_id) + assert before["current_recognition_job_id"] == new_job + extra = {"error": _ERROR} if status == "error" else {} + + await _handle(session_factory, fake_msg(envelope(old_job, file_id, status, **extra), msg_id_fn(old_job, 1))) + + assert (await _row(session_factory, old_job))["status"] is expected_job + assert await _file_row(session_factory, file_id) == before + assert (await _row(session_factory, new_job))["status"] is RecognitionStatus.queued + + +@pytest.mark.integration +async def test_file_status_rolls_back_with_message_transaction( + session_factory, jobs: CommittedJobs, logs: list[Any], monkeypatch: pytest.MonkeyPatch +): + """Статус файла меняется в транзакции сообщения: ошибка после обновления задания откатывает и задание, и файл.""" + file_id, job_id = await jobs.make_file_job( + file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing + ) + job_before = await _row(session_factory, job_id) + file_before = await _file_row(session_factory, file_id) + original = rx.set_file_status_for_job + + async def failing(session: AsyncSession, job: RecognitionJob, status: FileStatus, **kwargs: Any) -> bool: + await original(session, job, status, **kwargs) + raise RuntimeError("подставная ошибка после обновления файла") + + monkeypatch.setattr(rx, "set_file_status_for_job", failing) + + await _handle(session_factory, fake_msg(envelope(job_id, file_id, "error", error=_ERROR), result_msg_id(job_id, 1))) + + assert await _row(session_factory, job_id) == job_before + assert await _file_row(session_factory, file_id) == file_before + (record,) = _logged(logs, "транзакция откачена", "ERROR") assert record["exception"] is not None @@ -627,17 +762,19 @@ async def test_mapper_error_rolls_back_whole_transaction(session_factory, jobs: @pytest.mark.integration -async def test_concurrent_same_message_applied_once(session_factory, jobs: CommittedJobs, logs: list[Any]): - job_id = await jobs.make_queued_job(status=RecognitionStatus.processing) - file_id = await _file_id(jobs, job_id) +async def test_concurrent_same_message_applied_once( + session_factory, jobs: CommittedJobs, logs: list[Any], monkeypatch: pytest.MonkeyPatch +): + file_id, job_id = await jobs.make_file_job(job_status=RecognitionStatus.processing) msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)) - mapper = RecordingMapper(delay=LOCK_HOLD_SECONDS) # первый держит строку, второй ждёт его commit + slow = SlowFileStatus(monkeypatch, delay=LOCK_HOLD_SECONDS) # первый держит строку, второй ждёт его commit - await asyncio.gather(*[_handle(session_factory, msg, mapper=mapper) for _ in range(2)]) + await asyncio.gather(*[_handle(session_factory, msg) for _ in range(2)]) assert len(await _s3_rows(session_factory, _result_key(file_id, job_id))) == 1 assert (await _row(session_factory, job_id))["result_msg_id"] == f"result:{job_id}:1" - assert len(mapper.calls) == 1 + assert slow.calls == [(job_id, FileStatus.PROCESSING)] # применён один раз + assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING assert len(_logged(logs, "дубликат", "DEBUG")) == 1 assert not _logged(logs, "неизвестного задания") @@ -684,26 +821,32 @@ async def test_ack_waits_for_publisher_lock(session_factory, jobs: CommittedJobs @pytest.mark.integration async def test_result_vs_watchdog_is_consistent(session_factory, jobs: CommittedJobs): - """Результат и вахтёр на заданиях с прошедшим сроком одновременно: каждое — либо done (с s3_object), либо - timeout (без результата), но не смесь. Порядок решает блокировка строки: кто первый, тот и финальный.""" + """Результат и вахтёр на заданиях с прошедшим сроком одновременно: каждое — либо done (с s3_object, файл + `PROCESSING`), либо timeout (без результата, файл `ERROR`), но не смесь. Порядок решает блокировка строки задания: + кто первый, тот и финальный (и задание, и файл меняются в порядке «задание → файл», без взаимоблокировок).""" past = datetime.now(UTC) - timedelta(minutes=1) - job_ids = [ - await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past) for _ in range(RACE_JOBS) - ] - messages = [ - fake_msg(envelope(job_id, await _file_id(jobs, job_id), "done"), result_msg_id(job_id, 1)) for job_id in job_ids + pairs = [ + await jobs.make_file_job( + file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing, reply_deadline=past + ) + for _ in range(RACE_JOBS) ] + job_ids = [job_id for _, job_id in pairs] + messages = [fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)) for file_id, job_id in pairs] await asyncio.gather(expire_deadlines(session_factory), *[_handle(session_factory, m) for m in messages]) for job_id in job_ids: job = await _row(session_factory, job_id) s3 = await _s3_rows(session_factory, _result_key(job["file_id"], job_id)) + file = await _file_row(session_factory, job["file_id"]) if job["status"] is RecognitionStatus.done: assert job["error_code"] is None assert job["result_msg_id"] == f"result:{job_id}:1" assert [r["id"] for r in s3] == [job["result_s3_object_id"]] + assert file["status"] is FileStatus.PROCESSING else: + assert file["status"] is FileStatus.ERROR assert job["status"] is RecognitionStatus.timeout assert job["error_code"] == "TIMEOUT" assert job["result_msg_id"] is None @@ -712,17 +855,21 @@ async def test_result_vs_watchdog_is_consistent(session_factory, jobs: Committed @pytest.mark.integration -async def test_watchdog_waits_for_result_transaction(session_factory, jobs: CommittedJobs): +async def test_watchdog_waits_for_result_transaction( + session_factory, jobs: CommittedJobs, monkeypatch: pytest.MonkeyPatch +): """Вахтёр запущен, пока обработчик результата держит строку: его UPDATE ждёт commit, перепроверяет статус на новой - версии строки (`done`) и задание не трогает.""" + версии строки (`done`) и задание не трогает; файл остаётся `PROCESSING` (не `ERROR`).""" past = datetime.now(UTC) - timedelta(minutes=1) - job_id = await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past) - mapper = RecordingMapper(delay=LOCK_HOLD_SECONDS) - msg = fake_msg(envelope(job_id, await _file_id(jobs, job_id), "done"), result_msg_id(job_id, 1)) - handler = asyncio.create_task(_handle(session_factory, msg, mapper=mapper)) + file_id, job_id = await jobs.make_file_job( + file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing, reply_deadline=past + ) + slow = SlowFileStatus(monkeypatch, delay=LOCK_HOLD_SECONDS) + msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)) + handler = asyncio.create_task(_handle(session_factory, msg)) async def locked() -> None: # обработчик внутри транзакции, строка заблокирована - while not mapper.calls: + while not slow.calls: await asyncio.sleep(0.01) await asyncio.wait_for(locked(), 5) @@ -734,6 +881,7 @@ async def test_watchdog_waits_for_result_transaction(session_factory, jobs: Comm job = await _row(session_factory, job_id) assert job["status"] is RecognitionStatus.done assert job["error_code"] is None + assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING @pytest.mark.integration @@ -805,17 +953,17 @@ class _FakeService: @pytest.mark.integration async def test_full_cycle_over_real_nats(session_factory, jobs: CommittedJobs, logs: list[Any]): - """queued → published (outbox) → processing (ack) → done + s3_object (результат) → повтор без изменений.""" + """queued → published (outbox) → processing (ack) → done + s3_object (результат) → повтор без изменений; + файл: UPLOADED → PROCESSING (ack) → PROCESSING (результат, ждёт цикл маппинга).""" settings = get_settings().OPENBAO.NATS result_subject = f"test.result.{uuid.uuid4().hex}" - handler = partial(handle_message, session_factory=session_factory, mapper=NoopMapper()) + handler = partial(handle_message, session_factory=session_factory) async def subscribe(conn: NatsConnection) -> None: await conn.subscribe_results(handler) core = NatsConnection(settings, on_connected=subscribe, result_subject=result_subject) - job_id = await jobs.make_queued_job() - file_id = await _file_id(jobs, job_id) + file_id, job_id = await jobs.make_file_job() service = _FakeService(job_id, result_subject) await core.start() try: @@ -834,6 +982,7 @@ async def test_full_cycle_over_real_nats(session_factory, jobs: CommittedJobs, l await service.send(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)) job = await _wait_status(session_factory, job_id, RecognitionStatus.processing) assert job["started_at"] is not None + assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING assert (job["service_name"], job["service_version"]) == ("tabular-parser", "0.1.0") result = envelope(job_id, file_id, "done", metrics={"duration_ms": 10}) @@ -843,6 +992,7 @@ async def test_full_cycle_over_real_nats(session_factory, jobs: CommittedJobs, l assert (job["result_s3_object_id"], job["result_msg_id"]) == (s3["id"], f"result:{job_id}:1") assert job["finished_at"] is not None assert s3["bucket_role"] is S3BucketRole.results + assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING await service.send(result, result_msg_id(job_id, 1)) diff --git a/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md b/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md index bb082c8..25d3639 100644 --- a/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md +++ b/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md @@ -137,9 +137,9 @@ async def set_file_status_for_job(session, job: RecognitionJob, status: FileStat ``` Ошибка задания (`error`/`timeout`/`dlq`) ставит файлу `ERROR`; ack — `PROCESSING` только из `UPLOADED`; `done`/`needs_check` файл не трогают. -- [ ] **Step 1: Тесты** (integration, `CommittedJobs` + `make_stored_file`): ack → `PROCESSING`; `error` → `ERROR`; `done` → файл `PROCESSING`; `timeout` вахтёром → `ERROR`; `dlq` публикатором → `ERROR`; результат задания, которое уже не `current_recognition_job_id` файла → файл не меняется (Review Focus 1); существующие тесты inbox без `mapper` зелёные. -- [ ] **Step 2–4:** падает → реализация → `pytest -m integration tests/test_recognition_inbox.py tests/test_recognition_exchange.py -q`, `make check`. -- [ ] **Step 5: Commit** `feat(exchange): статусы stored_file по ack/ошибке/таймауту/dlq; inbox без ResultMapper (OQ-67)`. +- [x] **Step 1: Тесты** (integration, `CommittedJobs` + `make_stored_file`): ack → `PROCESSING`; `error` → `ERROR`; `done` → файл `PROCESSING`; `timeout` вахтёром → `ERROR`; `dlq` публикатором → `ERROR`; результат задания, которое уже не `current_recognition_job_id` файла → файл не меняется (Review Focus 1); существующие тесты inbox без `mapper` зелёные. +- [x] **Step 2–4:** падает → реализация → `pytest -m integration tests/test_recognition_inbox.py tests/test_recognition_exchange.py -q`, `make check`. +- [x] **Step 5: Commit** `feat(exchange): статусы stored_file по ack/ошибке/таймауту/dlq; inbox без ResultMapper (OQ-67)`. --- diff --git a/openspec/changes/slice-2c-osv-ingest/tasks.md b/openspec/changes/slice-2c-osv-ingest/tasks.md index a153537..0b96d8c 100644 --- a/openspec/changes/slice-2c-osv-ingest/tasks.md +++ b/openspec/changes/slice-2c-osv-ingest/tasks.md @@ -23,7 +23,7 @@ `schemas/files.py` (`UploadResult{accepted,rejected}`, `FileOut`, `JobBrief`, `DocumentBrief`), эндпоинт `POST /legal-entities/{id}/files`, роутер. Тесты: сценарии спеки `primary-files` (принята, смесь, сигнатура, чужое ЮЛ, дубль); `fixtures`-тест: реальный xls принят и `publish_pending` на `FakeNats` публикует запрос с ключом `uploads/{file_id}/{job_id}/…`. -- [ ] 2.2 **Статусы файла в обмене.** `recognition_exchange.py`: ack → `PROCESSING`; `error` → `ERROR`; вахтёр `timeout` → `ERROR`; +- [x] 2.2 **Статусы файла в обмене.** `recognition_exchange.py`: ack → `PROCESSING`; `error` → `ERROR`; вахтёр `timeout` → `ERROR`; публикатор `dlq` → `ERROR` (все — только при `current_recognition_job_id = job`); `ResultMapper`/`NoopMapper` удалены, `_map_result` убран, тесты inbox обновлены (`RecordingMapper` → проверка статуса файла). Тесты: сценарии спеки `recognition-job-exchange` (ack → PROCESSING, error → ERROR, timeout → ERROR, done → файл остаётся PROCESSING).