From 0420c852195a65b27779e4c44b1b5bf9db323a3d Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 4 Oct 2026 05:02:28 +0400 Subject: [PATCH] =?UTF-8?q?=D0=A1=D1=80=D0=B5=D0=B7=200=20[fix=20T12.1]:?= =?UTF-8?q?=20SCHEMA=5FMISMATCH=20=D0=B4=D0=BB=D1=8F=20=D0=B7=D0=B0=D0=B4?= =?UTF-8?q?=D0=B0=D0=BD=D0=B8=D1=8F=20=D0=B2=20=D1=80=D0=B0=D0=B1=D0=BE?= =?UTF-8?q?=D1=82=D0=B5/=D0=BE=D0=B1=D1=80=D0=B0=D0=B1=D0=BE=D1=82=D0=B0?= =?UTF-8?q?=D0=BD=D0=BD=D0=BE=D0=B3=D0=BE=20=D0=BD=D0=B5=20=D0=BF=D1=83?= =?UTF-8?q?=D0=B1=D0=BB=D0=B8=D0=BA=D1=83=D0=B5=D1=82=D1=81=D1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Конверт `result::` битого запроса совпадал по Nats-Msg-Id с настоящим результатом, и ядро (дедуп по result_msg_id) отбросило бы настоящий итог. `_reject` нормализует job_id (`uuid.UUID`, нижний регистр) и при `_seen` только пишет предупреждение. Тест: битый запрос с тем же job_id во время ожидания ИС и после итога, в том числе UUID в верхнем регистре — опубликованы только ack и настоящий итог. Co-Authored-By: Claude Fable 5.1 --- api/src/core/worker.py | 17 ++++++++++++++++- api/tests/unit/test_worker.py | 27 +++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/api/src/core/worker.py b/api/src/core/worker.py index 8472043..89b96d3 100644 --- a/api/src/core/worker.py +++ b/api/src/core/worker.py @@ -19,6 +19,7 @@ Spec image-recognition-gateway; design D-2, D-3, D-4, D-7; решения R-14.. import asyncio import json import re +import uuid from collections import OrderedDict from collections.abc import Awaitable, Callable, Coroutine from dataclasses import dataclass @@ -284,7 +285,11 @@ class Worker: return _Rejection(f"запрос не соответствует схеме recognition-request: {details}", raw) def _reject(self, rejection: _Rejection) -> None: - """`SCHEMA_MISMATCH` без ack (R-16): конверт — только если в теле есть корректные `job_id` и `file_id`.""" + """`SCHEMA_MISMATCH` без ack (R-16): конверт — только если в теле есть корректные `job_id` и `file_id`. + + Для `job_id` в работе или уже обработанного конверт не публикуется (T12.1): его `Nats-Msg-Id` + `result::` совпал бы с настоящим результатом, и ядро отбросило бы настоящий как дубль. + """ raw = rejection.raw or {} job_id, file_id = raw.get("job_id"), raw.get("file_id") if not (_is_uuid(job_id) and _is_uuid(file_id)): @@ -293,6 +298,16 @@ class Worker: reason=clean_text(rejection.reason, MAX_FRAGMENT_CHARS), ) return + # `_active`/`_done` хранят канонический вид (нижний регистр), а UUID в теле может прийти в верхнем + key = str(uuid.UUID(str(job_id))) + if self._seen(key): + logger.warning( + "SCHEMA_MISMATCH для задания в работе/обработанного: конверт не публикуется. job_id={job_id}, " + "reason={reason}", + job_id=key, + reason=clean_text(rejection.reason, MAX_FRAGMENT_CHARS), + ) + return attempt = raw.get("attempt") if not (type(attempt) is int and 1 <= attempt <= _MAX_ATTEMPT): attempt = 1 diff --git a/api/tests/unit/test_worker.py b/api/tests/unit/test_worker.py index 0d33303..c4e865e 100644 --- a/api/tests/unit/test_worker.py +++ b/api/tests/unit/test_worker.py @@ -494,6 +494,33 @@ async def test_schema_mismatch_does_not_register_job_id(make_env: EnvFactory) -> assert len(env.hotfolder.puts) == 1 +async def test_schema_mismatch_for_active_or_done_job_id_is_not_published( + make_env: EnvFactory, log_lines: list[str] +) -> None: + """T12.1: битый запрос с job_id задания в работе или уже обработанного не публикует `result::`. + + Иначе ядро (дедуп по `result_msg_id`) отбросило бы настоящий итог с тем же Nats-Msg-Id. UUID в теле может прийти + в верхнем регистре — ключ дедупликации нормализуется. + """ + env = await make_env() + request = env.request() + job_id = request["job_id"] + broken = {**request, "schema_version": "2.0"} + await env.deliver(request) + await eventually(lambda: len(env.hotfolder.puts) == 1) # задание ждёт ответа ИС + await env.deliver(broken) + await env.deliver({**broken, "job_id": job_id.upper()}) + async with emulator(env): + await asyncio.wait_for(env.worker.wait_idle(), 5) + await env.deliver(broken) # задание уже обработано + await env.deliver({**broken, "job_id": job_id.upper()}) + await asyncio.wait_for(env.worker.wait_idle(), 5) + published = env.published() + assert [mid for mid, _ in published] == [f"ack:{job_id}:1", f"result:{job_id}:1"] + assert_error(published[1][1], ErrorCode.NOT_IMPLEMENTED, raw=True) + assert sum("SCHEMA_MISMATCH для задания в работе/обработанного" in line for line in log_lines) == 4 + + # --- публикация, размер и кодировка ------------------------------------------------------------------------------- async def test_publish_failure_is_logged_and_worker_keeps_working(make_env: EnvFactory, log_lines: list[str]) -> None: env = await make_env()