diff --git a/api/src/services/result_mapping.py b/api/src/services/result_mapping.py index 63cb0a2..4ec14fa 100644 --- a/api/src/services/result_mapping.py +++ b/api/src/services/result_mapping.py @@ -15,9 +15,9 @@ PROCESSING` у файла, чьё ТЕКУЩЕЕ задание (`current_recogn Ошибки (D-2): неповторяемые (`MappingFatal`: ключ вне папки, объекта нет, JSON не проходит схему, нет документов ОСВ) — сразу файл `ERROR` и `metrics.mapping_error`; прочие — откат транзакции маппинга, в отдельной транзакции -`metrics.mapping_attempts + 1`, на `MAPPING_MAX_ATTEMPTS` — `ERROR`. Статус файла — только через -`services/file_status.py` (только текущее задание). `audit_event` не пишется: обмен и маппинг не аудируются -(ASSUMPTION(OQ-55)); `FILE_REPLACE` автозамены — отдельно. +`metrics.mapping_attempts + 1`, на `MAPPING_MAX_ATTEMPTS` — `ERROR`; следующая попытка — после паузы цикла (R-15). +Статус файла — через `services/file_status.py` (только текущее задание). `audit_event` не пишется: обмен и маппинг +не аудируются (ASSUMPTION(OQ-55)); `FILE_REPLACE` автозамены — отдельно. """ import asyncio @@ -456,12 +456,13 @@ async def _register_failure( *, max_attempts: int, now: datetime, -) -> None: - """D-2: после отката — `metrics.mapping_attempts + 1` отдельной транзакцией; на `max_attempts` — файл `ERROR`.""" +) -> bool: + """D-2: после отката — `metrics.mapping_attempts + 1` отдельной транзакцией; на `max_attempts` — файл `ERROR`. + Возвращает `True`, если попытки исчерпаны (файл переведён в `ERROR` и покинул очередь).""" async with session_factory() as session, session.begin(): job = await _lock_job(session, candidate.job_id) if job is None: - return + return False metrics = _metrics(job) attempts = _attempts(metrics) + 1 metrics[MAPPING_ATTEMPTS_KEY] = attempts @@ -482,6 +483,7 @@ async def _register_failure( attempts=attempts, exhausted=exhausted, ) + return exhausted async def _map_candidate( @@ -524,8 +526,11 @@ async def process_once( *, now: Clock | None = None, ) -> bool: - """Один проход цикла маппинга (модель шагов — в docstring модуля). `False` — кандидатов нет; `True` — кандидат - обработан (смапплен, отклонён, отложен до следующей попытки или пропущен, потому что его опередили).""" + """Один проход цикла маппинга (модель шагов — в docstring модуля). Результат — выполнена ли полезная работа, после + которой цикл может не спать: `True` — кандидат смапплен, отклонён неповторяемой ошибкой или исчерпал попытки + (файл `ERROR`), либо пропущен, потому что его опередили; `False` — кандидатов нет или повторяемая неудача + (R-15: пауза `MAPPING_POLL_SECONDS` перед следующей попыткой, иначе кратковременный сбой S3/БД сжёг бы все попытки + за миллисекунды).""" clock = now or _utcnow async with session_factory() as session, session.begin(): candidate = await pick_candidate(session, max_attempts=settings.MAPPING_MAX_ATTEMPTS) @@ -543,8 +548,8 @@ async def process_once( except MappingFatal as exc: async with session_factory() as session, session.begin(): await _fail(session, candidate, str(exc), clock()) - except Exception as exc: # сбой S3 или БД: транзакция маппинга откачена, повтор в следующем проходе - await _register_failure( + except Exception as exc: # сбой S3 или БД: транзакция маппинга откачена, повтор после паузы цикла (R-15) + return await _register_failure( session_factory, candidate, exc, max_attempts=settings.MAPPING_MAX_ATTEMPTS, now=clock() ) return True @@ -557,8 +562,9 @@ async def run_mapping_loop( *, sleep: SleepFn = asyncio.sleep, ) -> None: - """Бесконечный цикл маппинга: после обработанного кандидата — следующий сразу, без кандидатов — пауза - `MAPPING_POLL_SECONDS`; ошибка прохода логируется, цикл продолжается.""" + """Бесконечный цикл маппинга: после полезной работы (`process_once` → `True`) — следующий кандидат сразу; без + кандидатов или после повторяемой неудачи — пауза `MAPPING_POLL_SECONDS` (R-15); ошибка прохода логируется, цикл + продолжается.""" while True: processed = False try: diff --git a/api/tests/test_result_mapping.py b/api/tests/test_result_mapping.py index 14e1854..2fa6397 100644 --- a/api/tests/test_result_mapping.py +++ b/api/tests/test_result_mapping.py @@ -355,13 +355,15 @@ async def test_mapping_failure_retries_then_error( monkeypatch.setattr(rm, "map_file", failing_map_file) for attempt in (1, 2): - assert await process_once(maker, storage, _settings()) is True + # R-15: повторяемая неудача — `False`, цикл выдерживает паузу MAPPING_POLL_SECONDS до следующей попытки + assert await process_once(maker, storage, _settings()) is False job = await _job(maker, job_id) assert job.metrics[MAPPING_ATTEMPTS_KEY] == attempt assert MAPPING_ERROR_KEY not in job.metrics assert (await _file(maker, file_id)).status is FileStatus.PROCESSING assert await _documents(maker, file_id) == [] + # третья неудача переводит файл в ERROR: кандидат покинул очередь, следующий можно брать без паузы — `True` assert await process_once(maker, storage, _settings()) is True job = await _job(maker, job_id) assert job.metrics[MAPPING_ATTEMPTS_KEY] == 3 # noqa: PLR2004 - MAPPING_MAX_ATTEMPTS по умолчанию @@ -663,6 +665,38 @@ async def test_run_mapping_loop_processes_two_candidates(committed_sessionmaker: assert (await _file(maker, second)).status is FileStatus.PROCESSED +async def test_run_mapping_loop_pauses_after_retryable_failure( + committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch +): + """R-15: после повторяемой неудачи цикл спит `MAPPING_POLL_SECONDS` (попытки не сгорают за миллисекунды), после + успешного маппинга — `sleep(0)`.""" + maker, storage = committed_sessionmaker, FakeStorage() + file_id, job_id, _ = await _prepare(maker, storage, _minimal()) + real_map_file = rm.map_file + failures = [RuntimeError("кратковременный сбой БД")] + + async def flaky_map_file(session: AsyncSession, **kwargs: Any) -> list[str]: + if failures: + raise failures.pop() + return await real_map_file(session, **kwargs) + + monkeypatch.setattr(rm, "map_file", flaky_map_file) + settings = _settings() + sleeps: list[float] = [] + + async def fake_sleep(seconds: float) -> None: + sleeps.append(seconds) + if len(sleeps) == 3: # noqa: PLR2004 - неудача, успех, пустой проход + raise asyncio.CancelledError + + with pytest.raises(asyncio.CancelledError): + await run_mapping_loop(maker, storage, settings, sleep=fake_sleep) + + assert sleeps == [settings.MAPPING_POLL_SECONDS, 0, settings.MAPPING_POLL_SECONDS] + assert (await _file(maker, file_id)).status is FileStatus.PROCESSED + assert (await _job(maker, job_id)).metrics[MAPPING_ATTEMPTS_KEY] == 1 + + async def test_run_mapping_loop_survives_errors(monkeypatch: pytest.MonkeyPatch): """Ошибка прохода логируется, цикл не умирает и ждёт `MAPPING_POLL_SECONDS`.""" outcomes: list[Callable[[], bool]] = [lambda: (_ for _ in ()).throw(RuntimeError("сбой")), lambda: True]