From d74b56a0d1e8d06903f2bb3a8f53d40fee9f969c Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 4 Oct 2026 04:07:38 +0400 Subject: [PATCH] =?UTF-8?q?=D0=A1=D1=80=D0=B5=D0=B7=200=20[5.4]:=20=D1=82?= =?UTF-8?q?=D0=B5=D1=81=D1=82=20=D0=BF=D0=B0=D1=80=D0=B0=D0=BB=D0=BB=D0=B5?= =?UTF-8?q?=D0=BB=D0=B8=D0=B7=D0=BC=D0=B0=20=D0=B2=D0=BE=D1=80=D0=BA=D0=B5?= =?UTF-8?q?=D1=80=D0=B0=20(test=5Fworker=5Fconcurrency.py),=20=D0=BE=D1=82?= =?UTF-8?q?=D0=BC=D0=B5=D1=82=D0=BA=D0=B0=20=D0=B7=D0=B0=D0=B4=D0=B0=D1=87?= =?UTF-8?q?=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Сценарий spec «Пять заданий подряд» переписан детерминированно: паузы опроса горячей папки управляются тестом (SteppedSleep), ожидания — по событиям с тайм-аутом. Проверяется: пять ack'ов до освобождения слота, ровно четыре файла и четыре чтения S3 при занятых слотах, пятый файл — только после итога освобождённого задания, у всех пяти — итоговый конверт. Добавлены строгая очерёдность при SCAN_MAX_INFLIGHT = 1 и остановка воркера при очереди на слот (T12.9): задачи отменены, утечек задач и слотов нет, новых файлов нет. Прежний test_five_jobs_with_four_slots удалён из test_worker.py (без дублирования). Co-Authored-By: Claude Fable 5.1 --- api/tests/unit/test_worker.py | 17 +- api/tests/unit/test_worker_concurrency.py | 238 ++++++++++++++++++ .../plans/2026-10-03-slice-0-skeleton.md | 2 +- openspec/changes/slice-0-skeleton/tasks.md | 2 +- 4 files changed, 241 insertions(+), 18 deletions(-) create mode 100644 api/tests/unit/test_worker_concurrency.py diff --git a/api/tests/unit/test_worker.py b/api/tests/unit/test_worker.py index ba9044b..0d33303 100644 --- a/api/tests/unit/test_worker.py +++ b/api/tests/unit/test_worker.py @@ -550,22 +550,7 @@ async def test_lone_surrogate_in_original_name_does_not_break_publishing(make_en assert body["status"] == "error" -# --- параллелизм и остановка -------------------------------------------------------------------------------------- -async def test_five_jobs_with_four_slots(make_env: EnvFactory) -> None: - env = await make_env(SCAN_MAX_INFLIGHT=4) - requests = [env.request() for _ in range(5)] - for request in requests: - await env.deliver(request) - await eventually(lambda: len(env.hotfolder.puts) == 4) - await eventually(lambda: sum(len(env.acks(r["job_id"])) for r in requests) == 5) - await asyncio.sleep(0.05) - assert len(env.hotfolder.puts) == 4 # пятый ждёт слот - first_job = env.hotfolder.puts[0].split("_", 1)[0] - env.answer(first_job, json.dumps({**SUCCESS_ANSWER, "file_uid": first_job}).encode()) - await eventually(lambda: len(env.hotfolder.puts) == 5) - assert len(env.results(first_job)) == 1 - - +# --- остановка и прямой вызов handle (параллелизм — test_worker_concurrency.py) --------------------------------------- async def test_stop_cancels_jobs_and_rejects_new_requests(make_env: EnvFactory) -> None: env = await make_env() request = env.request() diff --git a/api/tests/unit/test_worker_concurrency.py b/api/tests/unit/test_worker_concurrency.py new file mode 100644 index 0000000..58fe82d --- /dev/null +++ b/api/tests/unit/test_worker_concurrency.py @@ -0,0 +1,238 @@ +"""Параллельная обработка ограничена (spec image-recognition-gateway, задача 5.4; D-3, `SCAN_MAX_INFLIGHT`). + +Сценарий spec «Пять заданий подряд»: пять запросов при `SCAN_MAX_INFLIGHT = 4` → пять ack'ов сразу, четыре файла во +входящем каталоге, пятый — только после завершения одного из четырёх. Плюс строгая очерёдность при одном слоте и +остановка воркера, когда задания ждут слот (T12.9). + +Детерминизм без «поспать и посмотреть»: пауза между опросами горячей папки подменена (`SteppedSleep`) — задание, +записавшее файл и опросившее исходящий каталог, «паркуется» и опрашивает снова только по шагу теста. Ожидания — +по событиям (`asyncio.Condition`) с явным тайм-аутом. Ответ ИС кладёт сам тест (`env.answer`) и ровно для того +задания, слот которого освобождается. Чтение S3 у фейка не уступает цикл событий, поэтому задание, прошедшее мимо +семафора, читает S3 в том же проходе цикла, что и его ack, — счётчик чтений S3 различает «ждёт слот» и «слот не нужен» +без гонок. +""" + +import asyncio +import json +from collections import Counter +from typing import Any + +import pytest + +from tests.worker_helpers import SUCCESS_ANSWER, Env, EnvFactory + +# Логи воркера перехватываются (`log_lines` из conftest): вывод pytest чистый +pytestmark = pytest.mark.usefixtures("log_lines") + +WAIT_TIMEOUT = 3.0 + + +class SteppedSleep: + """Подмена `sleep` воркера: пауза между опросами длится, пока тест не сделает шаг (`step`) или не откроет её. + + Поколение растёт с каждым шагом: `wait_parked(n)` ждёт n заданий, припаркованных уже после последнего шага, — + так ожидание не срабатывает на заданиях, которые только что отпущены, но ещё не успели проснуться. + """ + + def __init__(self) -> None: + self._generation = 0 + self._parked: Counter[int] = Counter() + self._gate = asyncio.Event() + self._opened = False + self._changed = asyncio.Condition() + + @property + def parked(self) -> int: + """Сколько заданий припарковано после последнего шага.""" + return self._parked[self._generation] + + async def __call__(self, seconds: float) -> None: + if self._opened: + await asyncio.sleep(0) + return + generation, gate = self._generation, self._gate + async with self._changed: + self._parked[generation] += 1 + self._changed.notify_all() + try: + await gate.wait() + finally: + self._parked[generation] -= 1 + + def step(self) -> None: + """Отпустить всех припаркованных: каждое задание опросит исходящий каталог ещё раз.""" + self._gate.set() + self._gate = asyncio.Event() + self._generation += 1 + + def open(self) -> None: + """Снять паузы совсем: дальше опросы идут без ожидания.""" + self._opened = True + self._gate.set() + + async def wait_parked(self, count: int) -> None: + async with self._changed: + await asyncio.wait_for(self._changed.wait_for(lambda: self.parked >= count), WAIT_TIMEOUT) + + +def job_of(name: str) -> str: + """`job_id` из имени файла входящего каталога `_`.""" + return name.split("_", 1)[0] + + +def answer(env: Env, job_id: str) -> None: + """Ответ ИС для задания (синтетический успех R-10 → `NOT_IMPLEMENTED`).""" + env.answer(job_id, json.dumps({**SUCCESS_ANSWER, "file_uid": job_id}).encode()) + + +def index_of(env: Env, event: tuple[str, str]) -> int: + assert env.events.count(event) == 1, (event, env.events) + return env.events.index(event) + + +def source_key(request: dict[str, Any]) -> str: + return str(request["s3"]["key"]) + + +async def finish_all(env: Env, poll_sleep: SteppedSleep, jobs: list[str]) -> None: + """Ответить на все задания, снять паузы, дождаться конца; у каждого задания — ровно один ack и один итог.""" + for job in jobs: + if not env.local.answer_path(job).exists(): + answer(env, job) + poll_sleep.open() + await asyncio.wait_for(env.worker.wait_idle(), WAIT_TIMEOUT) + for job in jobs: + assert env.acks(job) == [f"ack:{job}:1"] + [(msg_id, body)] = env.results(job) + assert msg_id == f"result:{job}:1" + assert body["status"] == "error" + assert body["error"]["code"] == "NOT_IMPLEMENTED" + assert env.worker.active_jobs == 0 + + +# --- spec «Пять заданий подряд» ----------------------------------------------------------------------------------- +async def test_five_jobs_with_four_slots(make_env: EnvFactory) -> None: + poll_sleep = SteppedSleep() + env = await make_env(SCAN_MAX_INFLIGHT=4, SCAN_TIMEOUT_SECONDS=60, sleep=poll_sleep) + requests = [env.request() for _ in range(5)] + jobs = [request["job_id"] for request in requests] + for request in requests: + await env.deliver(request) + + # Четыре слота заняты: каждое из четырёх заданий записало файл и ждёт ответа ИС + await poll_sleep.wait_parked(4) + # пять ack'ов уже опубликованы — ни один слот ещё не освободился, итогов нет + assert [env.acks(job) for job in jobs] == [[f"ack:{job}:1"] for job in jobs] + assert all(env.results(job) == [] for job in jobs) + # во входящем каталоге ровно четыре файла, пятое задание даже не читало S3 + assert len(env.storage.get_calls) == 4 + assert len(env.hotfolder.puts) == 4 + assert env.inbox() == sorted(env.hotfolder.puts) + busy = [job_of(name) for name in env.hotfolder.puts] + [waiting] = [job for job in jobs if job not in busy] + assert poll_sleep.parked == 4 + assert env.worker.active_jobs == 5 + + # Ещё один круг опроса занятых слотов: пятое задание по-прежнему ждёт + poll_sleep.step() + await poll_sleep.wait_parked(4) + assert len(env.hotfolder.puts) == 4 + assert len(env.storage.get_calls) == 4 + assert waiting not in {job_of(name) for name in env.inbox()} + + # Ответ ИС для одного задания: его слот освобождается, и только тогда появляется пятый файл + freed = busy[0] + answer(env, freed) + poll_sleep.step() + await poll_sleep.wait_parked(4) # три прежних задания + пятое, записавшее файл + assert [job_of(name) for name in env.hotfolder.puts] == [*busy, waiting] + assert len(env.inbox()) == 5 + assert all(env.results(job) == [] for job in jobs if job != freed) + freed_result = index_of(env, ("publish", f"result:{freed}:1")) + waiting_request = requests[jobs.index(waiting)] + assert max(index_of(env, ("publish", f"ack:{job}:1")) for job in jobs) < freed_result + assert freed_result < index_of(env, ("s3_get", source_key(waiting_request))) + assert freed_result < index_of(env, ("put", f"{waiting}_ok_doc.pdf")) + + # Все пять заданий заканчиваются конвертами + await finish_all(env, poll_sleep, jobs) + assert len(env.inbox()) == 5 + + +async def test_single_slot_serialises_jobs(make_env: EnvFactory) -> None: + poll_sleep = SteppedSleep() + env = await make_env(SCAN_MAX_INFLIGHT=1, SCAN_TIMEOUT_SECONDS=60, sleep=poll_sleep) + requests = [env.request() for _ in range(3)] + jobs = [request["job_id"] for request in requests] + for request in requests: + await env.deliver(request) + + for done in range(3): + await poll_sleep.wait_parked(1) + # в работе ровно одно задание; ack'и всех трёх — сразу + assert len(env.hotfolder.puts) == done + 1 + assert len(env.storage.get_calls) == done + 1 + assert len(env.inbox()) == done + 1 + assert all(env.acks(job) == [f"ack:{job}:1"] for job in jobs) + current = job_of(env.hotfolder.puts[-1]) + assert env.results(current) == [] + answer(env, current) + poll_sleep.step() + await asyncio.wait_for(env.worker.wait_idle(), WAIT_TIMEOUT) + + # Строгая очерёдность в журнале: следующий файл записан только после итога предыдущего задания + order = [job_of(name) for name in env.hotfolder.puts] + assert sorted(order) == sorted(jobs) + files_and_results = [ + (kind, ref) for kind, ref in env.events if kind == "put" or (kind == "publish" and ref.startswith("result:")) + ] + expected: list[tuple[str, str]] = [] + for job in order: + expected += [("put", f"{job}_ok_doc.pdf"), ("publish", f"result:{job}:1")] + assert files_and_results == expected + await finish_all(env, poll_sleep, jobs) + + +# --- остановка при очереди на слот (T12.9) ------------------------------------------------------------------------ +async def test_stop_while_jobs_wait_for_slot(make_env: EnvFactory, log_lines: list[str]) -> None: + poll_sleep = SteppedSleep() + env = await make_env(SCAN_MAX_INFLIGHT=2, SCAN_TIMEOUT_SECONDS=60, sleep=poll_sleep) + tasks_before = asyncio.all_tasks() + requests = [env.request() for _ in range(5)] + for request in requests: + await env.deliver(request) + await poll_sleep.wait_parked(2) + assert env.worker.active_jobs == 5 + assert len(env.hotfolder.puts) == 2 # три задания ждут слот + assert all(env.acks(request["job_id"]) == [f"ack:{request['job_id']}:1"] for request in requests) + job_tasks = asyncio.all_tasks() - tasks_before + assert len(job_tasks) == 5 + + # без исключения; `asyncio.timeout`, а не `wait_for`: отдельная задача-обёртка дала бы отменённым заданиям + # лишние проходы цикла, и проверка «stop() дождался заданий» стала бы слепой + async with asyncio.timeout(WAIT_TIMEOUT): + await env.worker.stop() + + # Все задачи заданий завершены отменой, новых задач не осталось, итоговых конвертов нет + assert all(task.cancelled() for task in job_tasks) + assert asyncio.all_tasks() - tasks_before == set() + assert env.worker.active_jobs == 0 + assert all(env.results(request["job_id"]) == [] for request in requests) + assert not [line for line in log_lines if line.startswith(("ERROR|", "CRITICAL|"))] + # После остановки файлы во входящий каталог больше не пишутся и S3 не читается, даже если цикл событий крутится + for _ in range(20): + await asyncio.sleep(0) + assert len(env.hotfolder.puts) == 2 + assert len(env.inbox()) == 2 + assert len(env.storage.get_calls) == 2 + assert poll_sleep.parked == 0 + + # Слоты не утекли: после повторного запуска оба слота свободны + env.worker.start() + again = [env.request() for _ in range(3)] + for request in again: + await env.deliver(request) + await poll_sleep.wait_parked(2) + assert len(env.hotfolder.puts) == 4 + assert len(env.storage.get_calls) == 4 + await finish_all(env, poll_sleep, [request["job_id"] for request in again]) diff --git a/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md b/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md index 7f66842..fa7fa90 100644 --- a/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md +++ b/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md @@ -68,7 +68,7 @@ ### Task 13: [5.4] Тест параллелизма -- [ ] Тест параллелизма: пять заданий при `SCAN_MAX_INFLIGHT = 4` — пять ack сразу, четыре файла во входящем каталоге, пятый после освобождения слота (фейковые S3/NATS, `LocalHotFolder`); проверка: зелёный +- [x] Тест параллелизма: пять заданий при `SCAN_MAX_INFLIGHT = 4` — пять ack сразу, четыре файла во входящем каталоге, пятый после освобождения слота (фейковые S3/NATS, `LocalHotFolder`); проверка: зелёный ### Task 14: [6.1] `tests/integration/test_roundtrip.py` (маркер `integration`) diff --git a/openspec/changes/slice-0-skeleton/tasks.md b/openspec/changes/slice-0-skeleton/tasks.md index f4848b1..dc359de 100644 --- a/openspec/changes/slice-0-skeleton/tasks.md +++ b/openspec/changes/slice-0-skeleton/tasks.md @@ -56,7 +56,7 @@ фейками и `LocalHotFolder` + `process_once` эмулятора — сценарии spec (ack раньше файла, повторный запрос → один файл, успех → NOT_IMPLEMENTED с `raw_s3_key`, INVALID_DOCUMENT, TIMEOUT с укороченными таймаутами, RECOGNITION_FAILED при отсутствии каталога, SCHEMA_MISMATCH без ack, FILE_READ_ERROR) -- [ ] 5.4 Тест параллелизма: пять заданий при `SCAN_MAX_INFLIGHT = 4` — пять ack сразу, четыре файла во входящем каталоге, +- [x] 5.4 Тест параллелизма: пять заданий при `SCAN_MAX_INFLIGHT = 4` — пять ack сразу, четыре файла во входящем каталоге, пятый после освобождения слота (фейковые S3/NATS, `LocalHotFolder`); проверка: зелёный ## 6. Интеграция, проверки, завершение