From 18276e607e012f722204f4350575254c312ef541 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 4 Oct 2026 03:17:11 +0400 Subject: [PATCH] =?UTF-8?q?fix:=20=D0=B2=D0=BE=D1=80=D0=BA=D0=B5=D1=80=20?= =?UTF-8?q?=E2=80=94=20=D0=B7=D0=B0=D0=B4=D0=B0=D0=BD=D0=B8=D0=B5=20=D0=BD?= =?UTF-8?q?=D0=B5=20=D1=82=D0=B5=D1=80=D1=8F=D0=B5=D1=82=D1=81=D1=8F=20?= =?UTF-8?q?=D0=BC=D0=BE=D0=BB=D1=87=D0=B0=20=D0=BF=D1=80=D0=B8=20=D0=BE?= =?UTF-8?q?=D1=81=D1=82=D0=B0=D0=BD=D0=BE=D0=B2=D0=BA=D0=B5=20=D0=B2=D0=BE?= =?UTF-8?q?=20=D0=B2=D1=80=D0=B5=D0=BC=D1=8F=20ack,=20=D0=BD=D0=B5=D0=B2?= =?UTF-8?q?=D0=B0=D0=BB=D0=B8=D0=B4=D0=BD=D1=8B=D0=B9=20=D0=B4=D1=83=D0=B1?= =?UTF-8?q?=D0=BB=D1=8C=20job=5Fid=20=D0=BD=D0=B5=20=D0=BF=D1=83=D0=B1?= =?UTF-8?q?=D0=BB=D0=B8=D0=BA=D1=83=D0=B5=D1=82=20SCHEMA=5FMISMATCH?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- api/src/core/worker.py | 42 ++++++++++++--- api/tests/unit/test_worker.py | 98 ++++++++++++++++++++++++++++++++--- 2 files changed, 127 insertions(+), 13 deletions(-) diff --git a/api/src/core/worker.py b/api/src/core/worker.py index 4eaed96..83bb91e 100644 --- a/api/src/core/worker.py +++ b/api/src/core/worker.py @@ -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` уже в работе или обработан этим процессом — только повторный ack (`ack::` этого запроса), второй обработки нет; иначе — `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::, что и настоящий результат + 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, diff --git a/api/tests/unit/test_worker.py b/api/tests/unit/test_worker.py index a85e371..8035b00 100644 --- a/api/tests/unit/test_worker.py +++ b/api/tests/unit/test_worker.py @@ -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:: не публикуется — только настоящий результат + 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"]