fix(mapping): пауза между повторами маппинга после сбоя

R-15 (design №28, находка T7.1): после повторяемой неудачи process_once возвращает False — run_mapping_loop
спит MAPPING_POLL_SECONDS перед следующей попыткой; кратковременный сбой S3/БД больше не сжигает все
MAPPING_MAX_ATTEMPTS за миллисекунды. Неповторяемая ошибка, исчерпание попыток (файл ERROR) и успешный
маппинг — по-прежнему True.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev 2026-10-05 22:54:11 +04:00
parent ecdf803bd4
commit 67e255d6c2
2 changed files with 53 additions and 13 deletions

View File

@ -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:

View File

@ -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]