Срез 0 [5.4]: тест параллелизма воркера (test_worker_concurrency.py), отметка задачи
Сценарий 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 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
d199414e9c
commit
d74b56a0d1
@ -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()
|
||||
|
||||
238
api/tests/unit/test_worker_concurrency.py
Normal file
238
api/tests/unit/test_worker_concurrency.py
Normal file
@ -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` из имени файла входящего каталога `<job_id>_<original_name>`."""
|
||||
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])
|
||||
@ -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`)
|
||||
|
||||
|
||||
@ -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. Интеграция, проверки, завершение
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user