diff --git a/api/src/services/recognition_exchange.py b/api/src/services/recognition_exchange.py index fc69e77..dbf4a24 100644 --- a/api/src/services/recognition_exchange.py +++ b/api/src/services/recognition_exchange.py @@ -79,8 +79,10 @@ from src.contract.versions import parse_message from src.core.config import Settings from src.core.nats import NatsPublishError from src.core.storage import ObjectStorage -from src.models.enums import FileStatus, RecognitionPipeline, RecognitionStatus, S3BucketRole +from src.models.directories import CompanyGroup, LegalEntity +from src.models.enums import FileStatus, QualityFailureKind, RecognitionPipeline, RecognitionStatus, S3BucketRole from src.models.files import RecognitionJob, S3Object, StoredFile +from src.repositories.quality_facts import insert_failure_fact from src.services.file_status import set_file_status_for_job, set_file_status_for_jobs from src.services.result_mapping import run_mapping_loop @@ -208,6 +210,121 @@ def _short(text: str) -> str: return text if len(text) <= ERROR_MESSAGE_MAX_LENGTH else text[: ERROR_MESSAGE_MAX_LENGTH - 1] + "…" +# --- факт отказа (quality_failure_fact, §14.5, design quality-metrics §4.3, AURORA-1759) ----------------------------- + + +@dataclass(frozen=True) +class _FailureDimensions: + """ЮЛ/ГК и признак скана файла на момент отказа — измерения `quality_failure_fact`, не хранимые в `recognition_job` + самом (design §4: факт без FK, имена и id ЮЛ/ГК — копия на момент отказа).""" + + legal_entity_id: uuid.UUID + legal_entity_name: str + group_id: uuid.UUID | None + group_name: str | None + is_scan: bool | None + + +async def _failure_dimensions_batch( + session: AsyncSession, file_ids: Sequence[uuid.UUID] +) -> dict[uuid.UUID, _FailureDimensions]: + """Измерения ЮЛ/ГК для набора `file_id` одним запросом (вахтёр переводит в `timeout` пачку заданий за проход, + §14.5) — без N+1 на задание. Файл без строки ЮЛ (не должно быть при штатной работе — ЮЛ не удаляются) просто + не попадает в результат.""" + if not file_ids: + return {} + stmt = ( + select( + StoredFile.id, + StoredFile.is_scan, + LegalEntity.id, + LegalEntity.name, + LegalEntity.company_group_id, + CompanyGroup.name, + ) + .join(LegalEntity, LegalEntity.id == StoredFile.legal_entity_id) + .outerjoin(CompanyGroup, CompanyGroup.id == LegalEntity.company_group_id) + .where(StoredFile.id.in_(list(file_ids))) + ) + rows = await session.execute(stmt) + return { + file_id: _FailureDimensions( + legal_entity_id=legal_entity_id, legal_entity_name=legal_entity_name, group_id=group_id, + group_name=group_name, is_scan=is_scan, + ) + for file_id, is_scan, legal_entity_id, legal_entity_name, group_id, group_name in rows + } + + +async def _failure_dimensions(session: AsyncSession, file_id: uuid.UUID) -> _FailureDimensions | None: + """Измерения ЮЛ/ГК одного файла (dlq / ошибка валидации запроса / ошибка сервиса — задание обрабатывается по + одному); тот же запрос, что у `_failure_dimensions_batch`, набор из одного элемента.""" + return (await _failure_dimensions_batch(session, [file_id])).get(file_id) + + +async def _record_failure_fact( # noqa: PLR0913 - все измерения факта отказа именованные + session: AsyncSession, + *, + job_id: uuid.UUID, + file_id: uuid.UUID, + attempt: int, + pipeline: RecognitionPipeline, + dims: _FailureDimensions, + failure_kind: QualityFailureKind, + error_code: str | None, + service_name: str | None, + failed_at: datetime, +) -> None: + """Тонкая обёртка `insert_failure_fact` (Task 4): измерения ЮЛ/ГК — из уже найденного `dims` (без повторного + запроса на каждый вызов у вызывающих с батчем).""" + await insert_failure_fact( + session, + job_id=job_id, + file_id=file_id, + attempt=attempt, + legal_entity_id=dims.legal_entity_id, + legal_entity_name=dims.legal_entity_name, + group_id=dims.group_id, + group_name=dims.group_name, + pipeline=pipeline, + is_scan=dims.is_scan, + service_name=service_name, + failure_kind=failure_kind, + error_code=error_code, + failed_at=failed_at, + ) + + +async def _write_timeout_failure_facts(session: AsyncSession, rows: Sequence[Any], *, failed_at: datetime) -> None: + """Факт отказа `TIMEOUT` на каждое задание, переведённое в `timeout` этим проходом вахтёра (§14.5): задание ещё + не получало ответа сервиса — `service_name = NULL`. Пишется для ВСЕХ строк `rows`, включая задания не текущие + для своего файла (у файла уже новое задание) — отказ конкретной попытки, а не статус файла.""" + if not rows: + return + dimensions = await _failure_dimensions_batch(session, sorted({row.file_id for row in rows})) + for row in rows: + dims = dimensions.get(row.file_id) + if dims is None: + logger.warning( + "Факт отказа TIMEOUT не записан: ЮЛ файла не найдено. job_id={job_id}, file_id={file_id}", + job_id=row.id, + file_id=row.file_id, + ) + continue + await _record_failure_fact( + session, + job_id=row.id, + file_id=row.file_id, + attempt=row.attempt, + pipeline=row.pipeline, + dims=dims, + failure_kind=QualityFailureKind.TIMEOUT, + error_code=ERROR_TIMEOUT, + service_name=None, + failed_at=failed_at, + ) + + async def _process_one( # noqa: PLR0913 - один шаг прохода: все параметры прохода session: AsyncSession, nats: RequestPublisher, @@ -239,6 +356,27 @@ async def _process_one( # noqa: PLR0913 - один шаг прохода: вс ) ) await set_file_status_for_job(session, row, FileStatus.ERROR, now=now) + # §14.5: факт отказа NO_ACK — задание так и не получило ответа сервиса, service_name неизвестен + dims = await _failure_dimensions(session, row.file_id) + if dims is None: + logger.warning( + "Факт отказа NO_ACK не записан: ЮЛ файла не найдено. job_id={job_id}, file_id={file_id}", + job_id=row.id, + file_id=row.file_id, + ) + else: + await _record_failure_fact( + session, + job_id=row.id, + file_id=row.file_id, + attempt=row.attempt, + pipeline=row.pipeline, + dims=dims, + failure_kind=QualityFailureKind.NO_ACK, + error_code=ERROR_NO_ACK, + service_name=None, + failed_at=now, + ) report.dlq += 1 logger.warning( "Запрос распознавания без ack: задание переведено в dlq. error_category=no_ack, job_id={job_id}, " @@ -263,6 +401,28 @@ async def _process_one( # noqa: PLR0913 - один шаг прохода: вс ) # финальная ошибка задания — файл текущего задания ERROR (spec primary-files «Жизненный цикл статуса файла») await set_file_status_for_job(session, row, FileStatus.ERROR, now=now) + # §14.5: любой переход recognition_job.status -> error — SERVICE_ERROR, независимо от причины (наша + # валидация запроса или ошибка, о которой сообщил сервис, design Review Focus основного плана) + dims = await _failure_dimensions(session, row.file_id) + if dims is None: + logger.warning( + "Факт отказа SERVICE_ERROR не записан: ЮЛ файла не найдено. job_id={job_id}, file_id={file_id}", + job_id=row.id, + file_id=row.file_id, + ) + else: + await _record_failure_fact( + session, + job_id=row.id, + file_id=row.file_id, + attempt=row.attempt, + pipeline=row.pipeline, + dims=dims, + failure_kind=QualityFailureKind.SERVICE_ERROR, + error_code=ERROR_INVALID_REQUEST, + service_name=None, + failed_at=now, + ) report.invalid += 1 logger.error( "Запрос распознавания не собран по контракту, задание переведено в error. " @@ -373,7 +533,8 @@ async def expire_deadlines(session_factory: async_sessionmaker[AsyncSession], *, """Вахтёр сроков (spec «Таймаут ожидания результата»): `processing`/`published` с `reply_deadline < now` → `timeout`, `error_code = TIMEOUT`, `finished_at = now`. Один атомарный `UPDATE … RETURNING id` с условием по статусу: результат, успевший перевести задание в финальный статус, не перетирается. В той же транзакции файлы этих - заданий (только где задание — текущее, `current_recognition_job_id`) → `ERROR`. Возвращает число заданий.""" + заданий (только где задание — текущее, `current_recognition_job_id`) → `ERROR`, и для КАЖДОГО (а не только + текущего) — факт отказа `TIMEOUT` (§14.5). Возвращает число заданий.""" moment = (now or _utcnow)() stmt = ( update(RecognitionJob) @@ -387,12 +548,14 @@ async def expire_deadlines(session_factory: async_sessionmaker[AsyncSession], *, error_message="Сервис распознавания не прислал результат до срока ответа (reply_deadline)", finished_at=moment, ) - .returning(RecognitionJob.id) + .returning(RecognitionJob.id, RecognitionJob.file_id, RecognitionJob.pipeline, RecognitionJob.attempt) .execution_options(synchronize_session=False) ) async with session_factory() as session, session.begin(): - job_ids = list((await session.execute(stmt)).scalars()) + rows = (await session.execute(stmt)).all() + job_ids = [row.id for row in rows] files = await set_file_status_for_jobs(session, job_ids, FileStatus.ERROR, now=moment) + await _write_timeout_failure_facts(session, rows, failed_at=moment) count = len(job_ids) if count: logger.warning( @@ -726,6 +889,29 @@ async def _apply_result( job.result_s3_object_id = result_s3_object_id job.result_msg_id = msg_id file_changed = await _apply_result_file_status(session, job, now) + if job.status is RecognitionStatus.error: + # §14.5: любой переход recognition_job.status -> error — SERVICE_ERROR; service_name/error_code уже + # заполнены выше из сообщения сервиса — здесь единственная точка отказа, где они известны + dims = await _failure_dimensions(session, job.file_id) + if dims is None: + logger.warning( + "Факт отказа SERVICE_ERROR не записан: ЮЛ файла не найдено. job_id={job_id}, file_id={file_id}", + job_id=job.id, + file_id=job.file_id, + ) + else: + await _record_failure_fact( + session, + job_id=job.id, + file_id=job.file_id, + attempt=job.attempt, + pipeline=job.pipeline, + dims=dims, + failure_kind=QualityFailureKind.SERVICE_ERROR, + error_code=job.error_code, + service_name=job.service_name, + failed_at=now, + ) logger.info( "Результат распознавания принят. job_id={job_id}, status={status}, service={service}, msg_id={msg_id}, " "error_code={error_code}, file_changed={file_changed}", diff --git a/api/src/services/result_mapping.py b/api/src/services/result_mapping.py index d882aeb..ca6cca9 100644 --- a/api/src/services/result_mapping.py +++ b/api/src/services/result_mapping.py @@ -73,6 +73,7 @@ from src.models.enums import ( OsvDataType, OsvGroupType, PeriodQuarter, + QualityFailureKind, QualityOutcome, RecognitionStatus, StatementLineStatus, @@ -82,7 +83,7 @@ from src.models.osv import OsvEntry from src.models.statements import DocumentIssue from src.repositories import statements as statements_repo from src.repositories.counterparties import CounterpartyIndex -from src.repositories.quality_facts import insert_intake_fact, insert_outcome_fact +from src.repositories.quality_facts import insert_failure_fact, insert_intake_fact, insert_outcome_fact from src.services import consolidation_documents from src.services.audit import AuditService from src.services.file_status import set_file_status_for_job @@ -129,6 +130,10 @@ INTAKE_FACT_DOCUMENT_TYPES = frozenset({DocumentType.BALANCE, DocumentType.INCOM STORED_FILE_OBJECT_TYPE = "stored_file" # `audit_event.object_type` события FILE_REPLACE SUPERSEDE_LOCK_PREFIX = "supersede" # пространство ключей advisory-блокировки автозамены в пределах ЮЛ _FORBIDDEN_KEY_SEGMENTS = frozenset({"", ".", ".."}) +# §14.5 (design quality-metrics §4.3): единый короткий error_code факта отказа MAPPING — что именно не так (схема, +# S3, объект не найден, БД) остаётся в `metrics.mapping_error`/логе (M-1), у `quality_failure_fact` нет текстовой +# колонки для детали; свой код, не пересекается с ERROR_TIMEOUT/ERROR_NO_ACK/ERROR_INVALID_REQUEST обмена +MAPPING_FAILURE_ERROR_CODE = "MAPPING_FAILED" Clock = Callable[[], datetime] SleepFn = Callable[[float], Awaitable[None]] @@ -1074,17 +1079,86 @@ async def _lock_file(session: AsyncSession, file_id: uuid.UUID) -> StoredFile | return (await session.execute(stmt)).scalar_one_or_none() +# --- факт отказа (quality_failure_fact, §14.5, design quality-metrics §4.3, AURORA-1759) ----------------------------- + + +@dataclass(frozen=True) +class _FailureDimensions: + """ЮЛ/ГК и признак скана файла на момент отказа маппинга — измерения `quality_failure_fact` (design §4: факт без + FK, имена и id ЮЛ/ГК — копия на момент отказа).""" + + legal_entity_id: uuid.UUID + legal_entity_name: str + group_id: uuid.UUID | None + group_name: str | None + is_scan: bool | None + + +async def _failure_dimensions(session: AsyncSession, file_id: uuid.UUID) -> _FailureDimensions | None: + """ЮЛ/ГК и признак скана файла по `file_id` (`session.get`, тот же приём, что у `_write_intake_facts`): файл или + ЮЛ не найдены (не должно быть при штатной работе) — `None`, факт отказа не пишется, остальная обработка не + прерывается.""" + file = await session.get(StoredFile, file_id) + if file is None: + return None + owner = await session.get(LegalEntity, file.legal_entity_id) + if owner is None: + return None + group_name: str | None = None + if owner.company_group_id is not None: + group = await session.get(CompanyGroup, owner.company_group_id) + group_name = group.name if group is not None else None + return _FailureDimensions( + legal_entity_id=owner.id, + legal_entity_name=owner.name, + group_id=owner.company_group_id, + group_name=group_name, + is_scan=file.is_scan, + ) + + +async def _write_mapping_failure_fact(session: AsyncSession, job: RecognitionJob, *, failed_at: datetime) -> None: + """Факт отказа `MAPPING` (§14.5) — `job` уже дошёл до `done`/`needs_check` (`service_name`/`attempt` сервиса + известны, в отличие от отказов обмена задачи 10 sites 1-4), а маппинг результата не удался.""" + dims = await _failure_dimensions(session, job.file_id) + if dims is None: + logger.warning( + "Факт отказа MAPPING не записан: ЮЛ файла не найдено. job_id={job_id}, file_id={file_id}", + job_id=job.id, + file_id=job.file_id, + ) + return + await insert_failure_fact( + session, + job_id=job.id, + file_id=job.file_id, + attempt=job.attempt, + legal_entity_id=dims.legal_entity_id, + legal_entity_name=dims.legal_entity_name, + group_id=dims.group_id, + group_name=dims.group_name, + pipeline=job.pipeline, + is_scan=dims.is_scan, + service_name=job.service_name, + failure_kind=QualityFailureKind.MAPPING, + error_code=MAPPING_FAILURE_ERROR_CODE, + failed_at=failed_at, + ) + + async def _fail( session: AsyncSession, candidate: Candidate, message: str, now: datetime, *, cause: BaseException | None = None ) -> None: """Неповторяемая ошибка: `metrics.mapping_error` и файл текущего задания → `ERROR` (транзакция вызывающего). - `cause` — исходное исключение (например, `ValidationError`): его полный текст пишется только в лог (M-1).""" + `cause` — исходное исключение (например, `ValidationError`): его полный текст пишется только в лог (M-1). + §14.5: факт отказа `MAPPING`.""" job = await _lock_job(session, candidate.job_id) if job is None: return job.metrics = {**_metrics(job), MAPPING_ERROR_KEY: _short(message)} await session.flush() changed = await set_file_status_for_job(session, job, FileStatus.ERROR, now=now) + await _write_mapping_failure_fact(session, job, failed_at=now) logger.opt(exception=cause).error( "Маппинг результата невозможен: файл переведён в ERROR. error_category=mapping_fatal, job_id={job_id}, " "file_id={file_id}, file_changed={changed}, error={error}", @@ -1103,8 +1177,9 @@ async def _register_failure( max_attempts: int, now: datetime, ) -> bool: - """D-2: после отката — `metrics.mapping_attempts + 1` отдельной транзакцией; на `max_attempts` — файл `ERROR`. - Возвращает `True`, если попытки исчерпаны (файл переведён в `ERROR` и покинул очередь).""" + """D-2: после отката — `metrics.mapping_attempts + 1` отдельной транзакцией; на `max_attempts` — файл `ERROR` и + факт отказа `MAPPING` (§14.5). Возвращает `True`, если попытки исчерпаны (файл переведён в `ERROR` и покинул + очередь).""" async with session_factory() as session, session.begin(): job = await _lock_job(session, candidate.job_id) if job is None: @@ -1122,6 +1197,7 @@ async def _register_failure( await session.flush() if exhausted: await set_file_status_for_job(session, job, FileStatus.ERROR, now=now) + await _write_mapping_failure_fact(session, job, failed_at=now) logger.opt(exception=exc).error( "Ошибка маппинга результата, транзакция откачена. error_category=mapping_error, job_id={job_id}, " "file_id={file_id}, attempts={attempts}, file_error={exhausted}", diff --git a/api/tests/test_recognition_exchange.py b/api/tests/test_recognition_exchange.py index 9892882..ca76048 100644 --- a/api/tests/test_recognition_exchange.py +++ b/api/tests/test_recognition_exchange.py @@ -23,17 +23,22 @@ 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 JSON, update +from sqlalchemy import JSON, select, 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 FileStatus, FileType, RecognitionPipeline, RecognitionStatus -from src.models.files import RecognitionJob +from src.models.directories import LegalEntity +from src.models.enums import FileStatus, FileType, QualityFailureKind, RecognitionPipeline, RecognitionStatus +from src.models.files import RecognitionJob, StoredFile +from src.models.quality_facts import QualityFailureFact from src.services import recognition_exchange as rx from src.services.recognition_exchange import ( + ERROR_INVALID_REQUEST, + ERROR_NO_ACK, + ERROR_TIMEOUT, WATCHDOG_PERIOD_SECONDS, ExchangeTasks, expire_deadlines, @@ -41,6 +46,7 @@ from src.services.recognition_exchange import ( run_publisher_loop, run_watchdog_loop, ) +from src.services.result_mapping import MAPPING_FAILURE_ERROR_CODE from tests.factories import CommittedJobs CONTRACTS_DIR = Path(__file__).resolve().parents[1] / "contracts" @@ -187,6 +193,25 @@ def _publish( return publish_pending(session_factory, nats, ack_timeout=ACK, max_republish=MAX_REPUBLISH, now=now) +async def _legal_entity(session_factory: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> LegalEntity: + """ЮЛ-владелец файла (для сравнения с измерениями `quality_failure_fact`, §14.5).""" + async with session_factory() as session: + file = await session.get(StoredFile, file_id) + assert file is not None + entity = await session.get(LegalEntity, file.legal_entity_id) + assert entity is not None + return entity + + +async def _failure_fact( + session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID +) -> QualityFailureFact | None: + """Факт отказа задания `job_id`, если записан (`job_id` — unique, не больше одного).""" + async with session_factory() as session: + stmt = select(QualityFailureFact).where(QualityFailureFact.job_id == job_id) + return (await session.execute(stmt)).scalar_one_or_none() + + # --- юнит: соответствие enum, константы ------------------------------------------------------------------------ @@ -200,6 +225,13 @@ def test_orm_pipeline_matches_request_subjects(): assert {m.value for m in RecognitionPipeline} == {"tabular", "image"} +def test_mapping_failure_error_code_is_distinct_from_exchange_codes(): + """§14.5: `error_code` отказа MAPPING (`result_mapping.py`, задача 10) — свой короткий код, не пересекается с + кодами обмена (ERROR_TIMEOUT/ERROR_NO_ACK/ERROR_INVALID_REQUEST) — иначе по `error_code` без `failure_kind` + нельзя было бы отличить отказ обмена от отказа маппинга. Чисто логическая проверка, без БД.""" + assert MAPPING_FAILURE_ERROR_CODE not in {ERROR_TIMEOUT, ERROR_NO_ACK, ERROR_INVALID_REQUEST} + + def test_watchdog_period_is_30_seconds(): assert WATCHDOG_PERIOD_SECONDS == 30 # noqa: PLR2004 - значение из spec @@ -395,6 +427,41 @@ async def test_invalid_request_marks_job_error_and_pass_continues( 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_invalid_request_writes_service_error_failure_fact( + session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, current: bool +): + """§14.5: запрос, не собранный по контракту, — тоже `SERVICE_ERROR` (любой переход `recognition_job.status -> + error` — SERVICE_ERROR, design Review Focus); факт пишется независимо от того, текущее это задание файла + (как и у TIMEOUT вахтёра) — отказ конкретной попытки. `nats.publish` до этой ветки не доходит — фейковый NATS + без реального соединения достаточно.""" + clock = Clock() + file_id, broken = await jobs.make_file_job(original_name=None) + if not current: + await jobs.add_job(file_id, status=RecognitionStatus.processing, current=True) + job = await jobs.get(broken) + entity = await _legal_entity(session_factory, file_id) + + report = await _publish(session_factory, FakeNats(), clock) + + assert report.invalid == 1 + fact = await _failure_fact(session_factory, broken) + assert fact is not None + assert fact.file_id == file_id + assert fact.attempt == job.attempt + assert fact.legal_entity_id == entity.id + assert fact.legal_entity_name == entity.name + assert fact.group_id is None + assert fact.group_name is None + assert fact.pipeline is RecognitionPipeline.tabular + assert fact.is_scan is None + assert fact.service_name is None + assert fact.failure_kind is QualityFailureKind.SERVICE_ERROR + assert fact.error_code == ERROR_INVALID_REQUEST + assert fact.failed_at == clock() + + # --- повторы без ack и dlq --------------------------------------------------------------------------------------- @@ -689,6 +756,62 @@ async def test_expire_deadlines_moves_current_file_to_error( assert (file.status, file.updated_at) == (FileStatus.ERROR, clock()) +@pytest.mark.integration +async def test_expire_deadlines_writes_timeout_failure_fact( + session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs +): + """§14.5: вахтёр переводит задание в `timeout` — факт отказа TIMEOUT; сервис так и не ответил — `service_name` + пуст. `.returning()` `expire_deadlines` отдаёт `file_id`/`pipeline`/`attempt` тем же `UPDATE`, без второго + запроса на задание.""" + clock = Clock() + file_id, job_id = await jobs.make_file_job( + job_status=RecognitionStatus.processing, reply_deadline=clock() - timedelta(minutes=1) + ) + job = await jobs.get(job_id) + entity = await _legal_entity(session_factory, file_id) + + assert await expire_deadlines(session_factory, now=clock) == 1 + + fact = await _failure_fact(session_factory, job_id) + assert fact is not None + assert fact.file_id == file_id + assert fact.attempt == job.attempt + assert fact.legal_entity_id == entity.id + assert fact.legal_entity_name == entity.name + assert fact.group_id is None + assert fact.group_name is None + assert fact.pipeline is RecognitionPipeline.tabular + assert fact.is_scan is None + assert fact.service_name is None + assert fact.failure_kind is QualityFailureKind.TIMEOUT + assert fact.error_code == ERROR_TIMEOUT + assert fact.failed_at == clock() + + +@pytest.mark.integration +async def test_expire_deadlines_writes_failure_fact_for_non_current_job( + session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs +): + """§14.5: факт отказа пишется на КАЖДОЕ просроченное задание, а не только на текущее задание файла (design + §14.5: «один файл может быть и в отказах, и в исходах»; 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), + ) + await jobs.add_job(file_id, status=RecognitionStatus.queued, current=True) + + assert await expire_deadlines(session_factory, now=clock) == 1 + + fact = await _failure_fact(session_factory, old_job) + assert fact is not None + assert fact.failure_kind is QualityFailureKind.TIMEOUT + file = await jobs.file(file_id) + assert file.status is FileStatus.UPLOADED # файл не трогается (текущее задание — другое) + + @pytest.mark.integration @pytest.mark.parametrize("file_status", [FileStatus.PROCESSED, FileStatus.SUPERSEDED]) async def test_expire_deadlines_keeps_processed_and_superseded_file( @@ -754,6 +877,40 @@ async def test_dlq_moves_current_file_to_error( assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None) +@pytest.mark.integration +async def test_dlq_writes_no_ack_failure_fact(session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs): + """§14.5: повторы без ack исчерпаны (`dlq`) — факт отказа NO_ACK; задание не получило ответа сервиса — + `service_name` пуст.""" + clock = Clock() + file_id, job_id = await jobs.make_file_job( + job_status=RecognitionStatus.published, + published_at=clock() - ACK * 2, + request_msg_id="dlq-fact-test:1", + reply_deadline=clock() + timedelta(hours=1), + metrics={"republish_count": MAX_REPUBLISH}, + ) + job = await jobs.get(job_id) + entity = await _legal_entity(session_factory, file_id) + + report = await _publish(session_factory, FakeNats(), clock) + + assert report.dlq == 1 + fact = await _failure_fact(session_factory, job_id) + assert fact is not None + assert fact.file_id == file_id + assert fact.attempt == job.attempt + assert fact.legal_entity_id == entity.id + assert fact.legal_entity_name == entity.name + assert fact.group_id is None + assert fact.group_name is None + assert fact.pipeline is RecognitionPipeline.tabular + assert fact.is_scan is None + assert fact.service_name is None + assert fact.failure_kind is QualityFailureKind.NO_ACK + assert fact.error_code == ERROR_NO_ACK + assert fact.failed_at == clock() + + # --- фоновые циклы ----------------------------------------------------------------------------------------------- diff --git a/api/tests/test_recognition_inbox.py b/api/tests/test_recognition_inbox.py index 289587a..65037bb 100644 --- a/api/tests/test_recognition_inbox.py +++ b/api/tests/test_recognition_inbox.py @@ -35,8 +35,10 @@ from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker 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 FileStatus, RecognitionStatus, S3BucketRole +from src.models.directories import LegalEntity +from src.models.enums import FileStatus, QualityFailureKind, RecognitionStatus, S3BucketRole from src.models.files import RecognitionJob, S3Object, StoredFile +from src.models.quality_facts import QualityFailureFact from src.services import recognition_exchange as rx from src.services.recognition_exchange import ( FINAL_STATUSES, @@ -176,6 +178,25 @@ async def _file_id(jobs: CommittedJobs, job_id: uuid.UUID) -> uuid.UUID: return (await jobs.get(job_id)).file_id +async def _legal_entity(session_factory: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> LegalEntity: + """ЮЛ-владелец файла (для сравнения с измерениями `quality_failure_fact`, §14.5).""" + async with session_factory() as session: + file = await session.get(StoredFile, file_id) + assert file is not None + entity = await session.get(LegalEntity, file.legal_entity_id) + assert entity is not None + return entity + + +async def _failure_fact( + session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID +) -> QualityFailureFact | None: + """Факт отказа задания `job_id`, если записан (`job_id` — unique, не больше одного).""" + async with session_factory() as session: + stmt = select(QualityFailureFact).where(QualityFailureFact.job_id == job_id) + return (await session.execute(stmt)).scalar_one_or_none() + + async def _handle(session_factory: async_sessionmaker[AsyncSession], msg: Any, **kwargs: Any) -> None: await handle_message(msg, session_factory=session_factory, **kwargs) @@ -535,6 +556,58 @@ async def test_service_error_without_result_s3(session_factory, jobs: CommittedJ assert await _s3_total(session_factory) == total_before +@pytest.mark.integration +async def test_service_error_result_writes_failure_fact(session_factory, jobs: CommittedJobs): + """§14.5 (задача 10, site 4): результат сервиса со `status=error` — факт отказа SERVICE_ERROR; в отличие от + отказов обмена без ответа (TIMEOUT/NO_ACK/INVALID_REQUEST — задачи 10 sites 1-3), здесь сервис уже ответил — + `service_name`/`error_code` берутся из уже заполненных полей задания, а не оставляются пустыми.""" + job_id = await jobs.make_queued_job(status=RecognitionStatus.processing) + file_id = await _file_id(jobs, job_id) + entity = await _legal_entity(session_factory, file_id) + body = envelope(job_id, file_id, "error", error={"code": "NOT_IMPLEMENTED", "message": "разбор ОСВ — Срез 1"}) + + await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1))) + + job = await _row(session_factory, job_id) + fact = await _failure_fact(session_factory, job_id) + assert fact is not None + assert fact.file_id == file_id + assert fact.attempt == job["attempt"] + assert fact.legal_entity_id == entity.id + assert fact.legal_entity_name == entity.name + assert fact.group_id is None + assert fact.group_name is None + assert fact.pipeline == job["pipeline"] + assert fact.is_scan is None + assert fact.service_name == "tabular-parser" + assert fact.failure_kind is QualityFailureKind.SERVICE_ERROR + assert fact.error_code == "NOT_IMPLEMENTED" + assert fact.failed_at == job["finished_at"] + + +@pytest.mark.integration +async def test_result_for_final_status_before_error_does_not_write_second_failure_fact( + session_factory, jobs: CommittedJobs +): + """Дубликат/запоздавший результат для задания, уже `error` (R-22 — без изменений): повторного факта нет — + `job_id` в `quality_failure_fact` unique, да и `_apply_result` для финального статуса возвращает `False` раньше + любых изменений (Review Focus).""" + job_id = await jobs.make_queued_job(status=RecognitionStatus.processing) + file_id = await _file_id(jobs, job_id) + body = envelope(job_id, file_id, "error", error={"code": "NOT_IMPLEMENTED", "message": "разбор ОСВ — Срез 1"}) + await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1))) + first = await _failure_fact(session_factory, job_id) + assert first is not None + + # повтор того же результата (дубликат по result_msg_id) — задание уже в финальном статусе, без изменений + await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1))) + + async with session_factory() as session: + stmt = select(func.count()).select_from(QualityFailureFact).where(QualityFailureFact.job_id == job_id) + count = (await session.execute(stmt)).scalar_one() + assert count == 1 + + @pytest.mark.integration @pytest.mark.parametrize( "status", diff --git a/api/tests/test_result_mapping.py b/api/tests/test_result_mapping.py index ea00a6b..e781248 100644 --- a/api/tests/test_result_mapping.py +++ b/api/tests/test_result_mapping.py @@ -50,12 +50,13 @@ from src.models.enums import ( OsvDataType, OsvGroupType, PeriodQuarter, + QualityFailureKind, QualityOutcome, RecognitionStatus, ) from src.models.files import Document, RecognitionJob, StoredFile from src.models.osv import OsvEntry -from src.models.quality_facts import QualityIntakeFact, QualityOutcomeFact +from src.models.quality_facts import QualityFailureFact, QualityIntakeFact, QualityOutcomeFact from src.models.statements import ( DocumentIssue, FinancialStatementLine, @@ -2317,4 +2318,141 @@ def test_intake_fact_document_types_excludes_disclosure_and_other() -> None: чистая проверка множества типов хука, без БД (в отличие от прочих тестов `_map_candidate` в этом файле).""" assert rm.INTAKE_FACT_DOCUMENT_TYPES == {DocumentType.BALANCE, DocumentType.INCOME, DocumentType.OSV} assert DocumentType.DISCLOSURE not in rm.INTAKE_FACT_DOCUMENT_TYPES + + +# --- факт отказа маппинга (quality_failure_fact, §14.5, design quality-metrics §4.3, задача 10 sites 5-6) ---------- + + +async def _failure_fact(maker: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> QualityFailureFact | None: + async with maker() as session: + return ( + await session.execute(select(QualityFailureFact).where(QualityFailureFact.job_id == job_id)) + ).scalar_one_or_none() + + +async def test_mapping_fatal_writes_mapping_failure_fact(committed_sessionmaker: async_sessionmaker[AsyncSession]): + """Задача 10, site 5 (§14.5): неповторяемая ошибка маппинга (`MappingFatal` — здесь ключ результата вне папки + задания, как в `test_result_key_outside_job_folder_is_error`) — факт отказа `MAPPING`. `job` в `_fail` уже + `done`/`needs_check` — `service_name` берётся из задания, а не оставляется пустым (в отличие от отказов обмена + задачи 10 sites 1-3, где сервис ещё не ответил).""" + maker, storage = committed_sessionmaker, FakeStorage() + file_id, job_id, entity_id = await _prepare( + maker, storage, _minimal(), key=f"results/{uuid.uuid4()}/{uuid.uuid4()}/result.json" + ) + job_before = await _job(maker, job_id) + + assert await process_once(maker, storage, _settings()) is True + + fact = await _failure_fact(maker, job_id) + assert fact is not None + assert fact.file_id == file_id + assert fact.attempt == job_before.attempt + assert fact.legal_entity_id == entity_id + assert fact.legal_entity_name == OWNER_NAME + assert fact.group_id is None + assert fact.group_name is None + assert fact.pipeline == job_before.pipeline + assert fact.is_scan is None + assert fact.service_name == job_before.service_name # `_prepare` не выставляет service_name — оба None + assert fact.failure_kind is QualityFailureKind.MAPPING + assert fact.error_code == rm.MAPPING_FAILURE_ERROR_CODE + assert fact.failed_at is not None + + +async def test_register_failure_writes_mapping_failure_fact_only_when_exhausted( + committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch +): + """Задача 10, site 6 (§14.5): повторяемая ошибка маппинга (как в `test_mapping_failure_retries_then_error`) — + факт отказа `MAPPING` пишется только когда попытки исчерпаны (`exhausted`), не на каждом откате транзакции.""" + maker, storage = committed_sessionmaker, FakeStorage() + file_id, job_id, entity_id = await _prepare(maker, storage, _minimal()) + + async def failing_map_file(session: AsyncSession, **kwargs: Any) -> list[str]: + raise RuntimeError("подставная ошибка БД при вставке строк") + + monkeypatch.setattr(rm, "map_file", failing_map_file) + + for _attempt in (1, 2): + assert await process_once(maker, storage, _settings()) is False + assert await _failure_fact(maker, job_id) is None # попытки не исчерпаны — факта ещё нет + + job_before = await _job(maker, job_id) + assert await process_once(maker, storage, _settings()) is True # третья попытка исчерпана — файл ERROR + + fact = await _failure_fact(maker, job_id) + assert fact is not None + assert fact.file_id == file_id + assert fact.attempt == job_before.attempt + assert fact.legal_entity_id == entity_id + assert fact.legal_entity_name == OWNER_NAME + assert fact.group_id is None + assert fact.group_name is None + assert fact.pipeline == job_before.pipeline + assert fact.is_scan is None + assert fact.service_name is None + assert fact.failure_kind is QualityFailureKind.MAPPING + assert fact.error_code == rm.MAPPING_FAILURE_ERROR_CODE + assert fact.failed_at is not None + + +async def test_file_reprocess_keeps_failure_fact_of_first_attempt_and_adds_new_intake( + committed_sessionmaker: async_sessionmaker[AsyncSession], +): + """Шаг 7 плана (§14.5): `FILE_REPROCESS` (новое `recognition_job` с `attempt + 1` на тот же файл — + `src/services/file_retry.py` `FileRetryService._new`, здесь воспроизведено напрямую вставкой задания — тем же + приёмом, что `test_result_of_non_current_job_is_ignored`/`test_job_replaced_while_reading_result` выше) не + теряет факт отказа первой попытки: он остаётся на `job_id` первого задания (`attempt=1`), а успешная вторая + попытка пишет свой `quality_intake_fact` (`attempt=2`, задача 5) — оба факта сосуществуют на одном `file_id`, + это не двойной учёт одного события (design §14.5: «один файл может быть и в отказах, и в исходах»).""" + maker, storage = committed_sessionmaker, FakeStorage() + async with maker() as session, session.begin(): + entity = await make_legal_entity(session, name=OWNER_NAME) + file, job1 = await make_stored_file( + session, entity=entity, status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing + ) + job1.reply_deadline = _now() - timedelta(minutes=1) + file_id, job1_id = file.id, job1.id + + # первая попытка: вахтёр переводит задание в timeout -> файл ERROR -> факт отказа TIMEOUT (attempt=1, задача 10 + # sites 1-4, уже на этой ветке) + assert await expire_deadlines(maker) == 1 + assert (await _file(maker, file_id)).status is FileStatus.ERROR + + # FILE_REPROCESS: новое задание attempt=2, файл снова ждёт результата (как `FileRetryService._new`) + async with maker() as session, session.begin(): + new_file = await session.get(StoredFile, file_id) + job1_fresh = await session.get(RecognitionJob, job1_id) + assert new_file is not None and job1_fresh is not None + job2 = RecognitionJob( + file_id=file_id, + pipeline=job1_fresh.pipeline, + request_subject=job1_fresh.request_subject, + schema_version="1.0", + status=RecognitionStatus.done, + attempt=2, + finished_at=_now(), + ) + session.add(job2) + await session.flush() + new_file.current_recognition_job_id = job2.id + new_file.status = FileStatus.PROCESSING + job2_id = job2.id + payload = _minimal() | {"job_id": str(job2.id), "file_id": str(file_id)} + await make_result_object(session, job=job2, storage=storage, result=payload) + + # вторая попытка — успех: маппинг создаёт документ и пишет свой факт приёма (attempt=2) + assert await process_once(maker, storage, _settings()) is True + + async with maker() as session: + facts = (await session.execute(select(QualityFailureFact))).scalars().all() + assert len(facts) == 1 + assert facts[0].job_id == job1_id + assert facts[0].attempt == 1 + assert facts[0].file_id == file_id + + [document] = await _documents(maker, file_id) + intake = await _intake_fact(maker, document.id) + assert intake.job_id == job2_id + assert intake.attempt == 2 # noqa: PLR2004 - вторая попытка (job2) этого файла + assert intake.file_id == file_id assert DocumentType.OTHER not in rm.INTAKE_FACT_DOCUMENT_TYPES