diff --git a/api/src/core/worker.py b/api/src/core/worker.py index cab3f91..aaad8f4 100644 --- a/api/src/core/worker.py +++ b/api/src/core/worker.py @@ -542,10 +542,13 @@ class Worker: job_id = str(request.job_id) # 4. Склейка всех файлов набора (не JSON — parse_error файла, R-24) и сырой ответ в S3 до нормализации (D-4) - merged = merge_answers(job_id, answer) + # (склейка и сериализация — CPU-работа над произвольным JSON: в потоке, как нормализатор; сбой сериализации, + # например RecursionError на очень глубокой вложенности, — такой же сбой записи сырого ответа) + merged = await asyncio.to_thread(merge_answers, job_id, answer) key = raw_key(job_id) try: - await self._storage.put_bytes(key, dump_merged(merged), JSON_CONTENT_TYPE) + payload = await asyncio.to_thread(dump_merged, merged) + await self._storage.put_bytes(key, payload, JSON_CONTENT_TYPE) except Exception as exc: logger.error( "Сырой ответ ИС не сохранён. job_id={job_id}, error={error}", job_id=job_id, error=_describe(exc) diff --git a/api/tests/unit/test_worker.py b/api/tests/unit/test_worker.py index 5209f5f..69be5e8 100644 --- a/api/tests/unit/test_worker.py +++ b/api/tests/unit/test_worker.py @@ -26,6 +26,7 @@ from src.contract import ( from src.core.clients import JSON_CONTENT_TYPE from src.core.worker import MAX_ERROR_MESSAGE_CHARS, MAX_WARNING_CHARS, MAX_WARNINGS from src.gateway.hotfolder import AnswerFile +from src.gateway.merge import dump_merged, merge_answers from src.normalizer import NOT_IMPLEMENTED_MESSAGE, NOT_OBJECT_MESSAGE, NormalizeOutcome from tests.worker_helpers import ( API_DIR, @@ -418,6 +419,39 @@ async def test_unknown_code_is_unknown_error_with_original_in_warnings(make_env: assert body["metrics"] == {"warnings": [f"{job_id}.json: SOME_NEW_CODE: новое"]} +async def _unserialisable_depth() -> bytes: + """JSON вложенности, который разбирается, но после обёртки склейки не сериализуется (`RecursionError`). + + Глубина подбирается в потоке тем же способом, каким воркер вызывает склейку и сериализацию (`to_thread`): окно + ~2 уровня зависит от глубины стека потока и версии интерпретатора. + """ + for depth in range(900, 1000): + data = ("[" * depth + "]" * depth).encode() + merged = await asyncio.to_thread(merge_answers, "probe", [AnswerFile("p.json", data)]) + if merged["files"][0]["parse_error"] is not None: + continue + try: + await asyncio.to_thread(dump_merged, merged) + except RecursionError: + return data + raise AssertionError("не найдена глубина: разбирается, но не сериализуется") + + +async def test_unserialisable_answer_is_unknown_error_without_raw_and_worker_survives(make_env: EnvFactory) -> None: + """M-2 (T3.1): `RecursionError` при сериализации склейки — как сбой записи сырого ответа (`UNKNOWN_ERROR`, без + `raw_s3_key`), а не непредвиденная ошибка; воркер продолжает работать.""" + env = await make_env() + request = env.request() + job_id = request["job_id"] + env.hotfolder.poll_script = [[AnswerFile(f"{job_id}.json", await _unserialisable_depth())]] + body = await env.run(request) + assert_error(body, ErrorCode.UNKNOWN_ERROR, raw=False) + assert "сырой ответ ИС не сохранён" in body["error"]["message"] + assert (BUCKET, f"raw/{job_id}.json") not in env.storage.objects + async with emulator(env): # воркер жив: следующее задание обрабатывается штатно + assert_error(await env.run(env.request()), ErrorCode.NOT_IMPLEMENTED, raw=True) + + @pytest.mark.fixtures async def test_real_aep_6_success_is_not_implemented_without_warnings(make_env: EnvFactory) -> None: """Регрессия: реальный успешный ответ `aep_6.json` (`error = {"code": "", "message": ""}`) — не ошибка ИС."""