fix: воркер — задание не теряется молча при остановке во время ack, невалидный дубль job_id не публикует SCHEMA_MISMATCH
I9.1: после ack повторная проверка приёма — warning error_category=shutdown, job_id снимается из «в работе». T9.7: невалидное тело с job_id в работе/обработанным — warning и сброс (msg-id настоящего результата не занимается). T9.5: error.message исхода парсера — одной строкой, ≤ 1000 символов. T9.6: docstring stop() про to_thread. T9.1: тест конкуренции без опроса со sleep. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
82756b8c97
commit
18276e607e
@ -5,7 +5,9 @@ Design D-2, D-4, D-5, D-7; spec tabular-recognition-service «Подтвержд
|
||||
|
||||
- `on_request(msg)` — обработчик подписки NATS (queue group `tabular-parser`): разбор тела и валидация по контракту
|
||||
(pydantic `RecognitionRequest`, MAJOR `schema_version` = 1). Невалидный запрос — конверт `error SCHEMA_MISMATCH`
|
||||
без ack (если `job_id`, `file_id` и `attempt` читаемы; иначе собрать валидный конверт нельзя — error-лог и сброс).
|
||||
без ack (если `job_id`, `file_id` и `attempt` читаемы; иначе собрать валидный конверт нельзя — error-лог и сброс;
|
||||
если читаемый `job_id` уже в работе или обработан — warning и сброс: иначе `SCHEMA_MISMATCH` ушёл бы под тем же
|
||||
`result:<job_id>:<attempt>`, что и настоящий результат).
|
||||
Валидный: `job_id` уже в работе или обработан этим процессом — только повторный ack (`ack:<job_id>:<attempt>` этого
|
||||
запроса), второй обработки нет; иначе — `job_id` в множество «в работе», ack `processing` **до** чтения S3, задание
|
||||
во внутреннюю `asyncio.Queue`.
|
||||
@ -19,7 +21,8 @@ Design D-2, D-4, D-5, D-7; spec tabular-recognition-service «Подтвержд
|
||||
- Дедупликация: `job_id` в работе (очередь + обрабатываемые) — множество; завершённые (любой итог, в том числе ошибка и
|
||||
неудачная публикация результата) — LRU размера `done_cache_size`.
|
||||
- Остановка: новые запросы не принимаются (без ack — ядро переотправит другой инстанции), очередь и текущие задания
|
||||
дорабатываются не дольше `timeout`, остальное отменяется с warning-логом.
|
||||
дорабатываются не дольше `timeout`, остальное отменяется с warning-логом. Запрос, чей ack публиковался в момент
|
||||
остановки, в очередь не ставится (warning, `job_id` снимается из «в работе»).
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
@ -153,7 +156,11 @@ class RecognitionWorker:
|
||||
await self._queue.join()
|
||||
|
||||
async def stop(self, timeout: float = STOP_TIMEOUT) -> None:
|
||||
"""Перестаёт принимать запросы, дорабатывает очередь не дольше `timeout` секунд, затем отменяет остальное."""
|
||||
"""Перестаёт принимать запросы, дорабатывает очередь не дольше `timeout` секунд, затем отменяет остальное.
|
||||
|
||||
Отмена задания, которое сейчас внутри `asyncio.to_thread` (разбор), не останавливает сам поток: он доработает
|
||||
в фоне, а результат будет отброшен (для заглушки безвредно; для долгого разбора — учесть в Срезе 2).
|
||||
"""
|
||||
self._accepting = False
|
||||
if not self._tasks:
|
||||
return
|
||||
@ -198,8 +205,18 @@ class RecognitionWorker:
|
||||
else:
|
||||
self._in_progress.add(job_id)
|
||||
await self._publish_ack(request)
|
||||
if not duplicate:
|
||||
self._queue.put_nowait(_Job(request))
|
||||
if duplicate:
|
||||
return
|
||||
if not self._accepting:
|
||||
# stop() начался, пока публиковался ack: очередь уже никто не читает
|
||||
self._in_progress.discard(job_id)
|
||||
logger.warning(
|
||||
"Воркер остановлен во время публикации ack: ack опубликован, задание не будет обработано (ядро закроет "
|
||||
"его по reply_deadline). error_category=shutdown, job_id={job_id}",
|
||||
job_id=job_id,
|
||||
)
|
||||
return
|
||||
self._queue.put_nowait(_Job(request))
|
||||
|
||||
def _parse_request(self, data: bytes) -> RecognitionRequest:
|
||||
try:
|
||||
@ -227,6 +244,15 @@ class RecognitionWorker:
|
||||
job_id = self._readable_uuid(raw, "job_id")
|
||||
file_id = self._readable_uuid(raw, "file_id")
|
||||
attempt = self._readable_attempt(raw, msg.headers, job_id) if job_id is not None else None
|
||||
if job_id is not None and (job_id.lower() in self._in_progress or job_id.lower() in self._done):
|
||||
# Конверт SCHEMA_MISMATCH ушёл бы под тем же result:<job_id>:<attempt>, что и настоящий результат
|
||||
logger.warning(
|
||||
"Невалидный запрос с job_id, который уже в работе или обработан: отброшен без ответа. "
|
||||
"error_category=schema_mismatch, job_id={job_id}, reason={reason}",
|
||||
job_id=job_id,
|
||||
reason=reason,
|
||||
)
|
||||
return
|
||||
if job_id is None or file_id is None or attempt is None:
|
||||
logger.error(
|
||||
"Запрос отброшен: не проходит контракт, а job_id/file_id/attempt не читаемы — результат не "
|
||||
@ -336,8 +362,10 @@ class RecognitionWorker:
|
||||
return self._error(request, ErrorCode.UNKNOWN_ERROR, f"ошибка разбора: {_describe(exc)}", None, started)
|
||||
|
||||
if outcome.status is ResultStatus.ERROR:
|
||||
error = outcome.error or MessageError(
|
||||
code=ErrorCode.UNKNOWN_ERROR, message="парсер вернул ошибку без описания"
|
||||
error = (
|
||||
MessageError(code=outcome.error.code, message=_one_line(outcome.error.message))
|
||||
if outcome.error is not None
|
||||
else MessageError(code=ErrorCode.UNKNOWN_ERROR, message="парсер вернул ошибку без описания")
|
||||
)
|
||||
return self._envelope(
|
||||
job_id=request.job_id,
|
||||
|
||||
@ -98,14 +98,21 @@ class FakeStorage:
|
||||
|
||||
|
||||
class FakePublisher:
|
||||
"""`Publisher`: копит опубликованное; `fail_subjects_msg_ids` — msg_id, публикация которых падает."""
|
||||
"""`Publisher`: копит опубликованное; `fail_msg_ids` — msg_id, публикация которых падает; `gates` — msg_id,
|
||||
публикация которых ждёт события (`waiting` выставляется, когда публикация встала на ожидание)."""
|
||||
|
||||
def __init__(self, events: list[tuple[str, str]]) -> None:
|
||||
self.events = events
|
||||
self.published: list[tuple[str, dict[str, Any], str]] = []
|
||||
self.fail_msg_ids: set[str] = set()
|
||||
self.gates: dict[str, asyncio.Event] = {}
|
||||
self.waiting = asyncio.Event()
|
||||
|
||||
async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
|
||||
gate = self.gates.get(msg_id)
|
||||
if gate is not None:
|
||||
self.waiting.set()
|
||||
await gate.wait()
|
||||
if msg_id in self.fail_msg_ids:
|
||||
self.events.append(("publish_failed", msg_id))
|
||||
raise NatsPublishError(f"публикация {msg_id} не удалась")
|
||||
@ -679,10 +686,13 @@ async def test_concurrency_processes_jobs_in_parallel() -> None:
|
||||
gate = asyncio.Event()
|
||||
environment.storage.gate = gate
|
||||
started: list[str] = []
|
||||
all_started = asyncio.Event()
|
||||
original_get = environment.storage.get_bytes
|
||||
|
||||
async def tracking_get(bucket: str, key: str) -> bytes:
|
||||
started.append(key)
|
||||
if len(started) == JOBS:
|
||||
all_started.set()
|
||||
return await original_get(bucket, key)
|
||||
|
||||
environment.storage.get_bytes = tracking_get # type: ignore[method-assign]
|
||||
@ -693,11 +703,9 @@ async def test_concurrency_processes_jobs_in_parallel() -> None:
|
||||
await environment.worker.on_request(
|
||||
_msg(_request(job_id=OTHER_JOB_ID, s3={"bucket": BUCKET, "key": other_key}))
|
||||
)
|
||||
for _ in range(50):
|
||||
if len(started) == JOBS:
|
||||
break
|
||||
await asyncio.sleep(0.01)
|
||||
assert len(started) == JOBS # оба задания одновременно ждут S3
|
||||
# оба задания одновременно ждут S3 (первое не отпущено): при concurrency=1 второе не стартовало бы
|
||||
await asyncio.wait_for(all_started.wait(), 1)
|
||||
assert len(started) == JOBS
|
||||
gate.set()
|
||||
await environment.worker.join()
|
||||
finally:
|
||||
@ -739,3 +747,81 @@ async def test_stop_is_idempotent_and_safe_without_start() -> None:
|
||||
environment = Env()
|
||||
await environment.worker.stop(timeout=0.1)
|
||||
await environment.worker.stop(timeout=0.1)
|
||||
|
||||
|
||||
async def test_stop_during_ack_publish_drops_job_with_warning(log_records: list[Any]) -> None:
|
||||
environment = await _make_env()
|
||||
gate = asyncio.Event()
|
||||
environment.publisher.gates[f"ack:{JOB_ID}:1"] = gate
|
||||
request_task = asyncio.create_task(environment.worker.on_request(_msg(_request())))
|
||||
await asyncio.wait_for(environment.publisher.waiting.wait(), 1)
|
||||
|
||||
# stop() начался, пока on_request публикует ack: очередь пуста, корутины отменены
|
||||
await asyncio.wait_for(environment.worker.stop(timeout=1), 2)
|
||||
gate.set()
|
||||
await asyncio.wait_for(request_task, 1)
|
||||
|
||||
assert environment.msg_ids() == [f"ack:{JOB_ID}:1"]
|
||||
assert environment.storage.get_calls == []
|
||||
assert environment.worker._in_progress == set()
|
||||
assert environment.worker._queue.qsize() == 0
|
||||
warnings = [record["message"] for record in log_records if record["level"].name == "WARNING"]
|
||||
assert any("error_category=shutdown" in message and JOB_ID in message for message in warnings), warnings
|
||||
|
||||
|
||||
# --- нормализация сообщения ошибки парсера ------------------------------------------------------------------------
|
||||
async def test_parser_error_message_is_normalised_to_one_bounded_line() -> None:
|
||||
long_message = "строка 1\nстрока 2\t" + "x" * 3000
|
||||
|
||||
class VerboseErrorParser:
|
||||
def parse(self, data: bytes, file_type: FileType) -> ParseOutcome:
|
||||
return ParseOutcome(
|
||||
status=ResultStatus.ERROR,
|
||||
error=MessageError(code=ErrorCode.INVALID_FORMAT, message=long_message),
|
||||
)
|
||||
|
||||
environment = await _make_env(VerboseErrorParser())
|
||||
try:
|
||||
await environment.worker.on_request(_msg(_request()))
|
||||
await environment.worker.join()
|
||||
finally:
|
||||
await environment.worker.stop(timeout=1)
|
||||
|
||||
envelope = environment.bodies()[1]
|
||||
_assert_message_valid(envelope)
|
||||
message = envelope["error"]["message"]
|
||||
assert envelope["error"]["code"] == "INVALID_FORMAT"
|
||||
assert message.startswith("строка 1 строка 2 xxx")
|
||||
assert "\n" not in message
|
||||
assert "\t" not in message
|
||||
assert len(message) <= 1000 # noqa: PLR2004 - ERROR_MESSAGE_LIMIT
|
||||
|
||||
|
||||
# --- невалидное тело с job_id, который уже в работе или обработан ---------------------------------------------------
|
||||
async def test_malformed_request_for_job_in_progress_is_dropped(env: Env, log_records: list[Any]) -> None:
|
||||
gate = asyncio.Event()
|
||||
env.storage.gate = gate
|
||||
await env.worker.on_request(_msg(_request()))
|
||||
await asyncio.wait_for(env.storage.get_started.wait(), 1)
|
||||
|
||||
malformed = _request()
|
||||
malformed["s3"].pop("key")
|
||||
await env.worker.on_request(_msg(malformed))
|
||||
gate.set()
|
||||
await env.worker.join()
|
||||
|
||||
# SCHEMA_MISMATCH под тем же result:<job_id>:<attempt> не публикуется — только настоящий результат
|
||||
assert env.msg_ids() == [f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1"]
|
||||
assert env.bodies()[1]["error"]["code"] == "NOT_IMPLEMENTED"
|
||||
warnings = [record["message"] for record in log_records if record["level"].name == "WARNING"]
|
||||
assert any("error_category=schema_mismatch" in message and JOB_ID in message for message in warnings)
|
||||
|
||||
|
||||
async def test_malformed_request_for_done_job_is_dropped(env: Env) -> None:
|
||||
await env.worker.on_request(_msg(_request()))
|
||||
await env.worker.join()
|
||||
|
||||
await env.worker.on_request(_msg(_request(job_id=JOB_ID.upper(), schema_version="2.0")))
|
||||
await env.worker.join()
|
||||
|
||||
assert env.msg_ids() == [f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1"]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user