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:
parent
ecdf803bd4
commit
67e255d6c2
@ -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:
|
||||
|
||||
@ -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]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user