diff --git a/api/src/app.py b/api/src/app.py index cd9ab60..f5537bc 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -1,12 +1,18 @@ """Точка входа FastAPI сервиса scan-gateway (адаптер ИС Сканирование). Без БД и без фронта. -Что здесь есть: настройки, регистрация SMB-сессии (только при `SCAN_HOTFOLDER_MODE=smb`), middleware (gzip, proxy -headers), обработчики ошибок, `GET /healthcheck` (`uptime`, `version`). -Чего здесь пока нет: NATS-воркера и S3-клиента (подключаются в lifespan задачами 5.2/5.3 change slice-0-skeleton), -полей `nats`/`hotfolder` в healthcheck. +- `create_app(settings)` строит горячую папку (`create_hotfolder`, без обращения к ФС/шаре) и `HealthProvider`; + `GET /healthcheck` отдаёт `uptime`, `version`, `nats` (`ok`/`down`/`disabled`), `hotfolder` (`ok`/`down`). +- lifespan: регистрация SMB-сессии (только `SCAN_HOTFOLDER_MODE=smb`), S3 (`S3Storage`), NATS (`NatsGateway`), воркер + (`Worker` с `NullNormalizer`). Подключение к NATS блокирует до доступности сервера, поэтому идёт фоновой задачей: + HTTP отвечает сразу (`nats = down`), подписка `recognition.image.request` появляется после подключения. Сбой этой + задачи (например, `NatsConfigError`) — лог error с трассировкой и `nats = down`, HTTP-сервер жив. + Остановка: отмена подключения, остановка воркера (задания в работе отменяются), закрытие NATS и S3. +- Объекты lifespan доступны в `app.state`: `worker`, `nats`, `storage`, `health` (тесты, задачи 13–14). +- `NATS_ENABLED=false` — подключения нет (`nats = disabled`), воркер запросов не получает (dev без NATS). """ -from contextlib import asynccontextmanager +import asyncio +from contextlib import asynccontextmanager, suppress from datetime import UTC, datetime import smbclient @@ -16,8 +22,14 @@ from starlette.middleware.gzip import GZipMiddleware from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware from src.api.router import router +from src.core.clients import S3Storage from src.core.config import Settings, get_settings from src.core.exception_handlers import register_exception_handlers +from src.core.health import HealthProvider +from src.core.nats import NatsGateway +from src.core.worker import Worker +from src.gateway import create_hotfolder +from src.normalizer import NullNormalizer def _normalize_root_path(value: str) -> str: @@ -29,10 +41,35 @@ def _normalize_root_path(value: str) -> str: return "/" + value.strip("/") +async def run_nats(gateway: NatsGateway, worker: Worker) -> None: + """Фоновая задача: подключение к NATS (ждёт сервер сколь угодно долго) и подписка воркера на запросы. + + Ошибка не роняет приложение: лог error с трассировкой; если соединение установлено, а подписка не удалась, + соединение закрывается, чтобы `/healthcheck.nats = down` честно показывал, что запросы не принимаются. + """ + try: + if not await gateway.connect(): + return + except asyncio.CancelledError: + raise + except Exception: + logger.exception("NATS: подключение не выполнено, запросы не принимаются (nats=down) до перезапуска") + return + try: + await gateway.subscribe_requests(worker.on_request) + except asyncio.CancelledError: + raise + except Exception: + logger.exception("NATS: подписка на запросы не оформлена, соединение закрывается (nats=down)") + await gateway.close() + + def create_app(settings: Settings | None = None) -> FastAPI: settings = settings or get_settings() version = settings.VERSION up_time = datetime.now(UTC) + hotfolder = create_hotfolder(settings) + health = HealthProvider(nats_enabled=settings.NATS_ENABLED, hotfolder=hotfolder) @asynccontextmanager async def lifespan(app: FastAPI): @@ -43,20 +80,38 @@ def create_app(settings: Settings | None = None) -> FastAPI: smbclient.register_session(smb.SMB_HOST, smb.SMB_LOGIN, smb.SMB_PASSWORD) smb_session = True except Exception: + # учётные данные передаются в каждый вызов smbclient (SmbHotFolder), сессия восстановится при + # доступной шаре; пока шара недоступна, /healthcheck.hotfolder = down logger.exception("Ошибка при регистрации сессии SMB") + storage = S3Storage.from_settings(settings.OPENBAO.MINIO) + gateway = NatsGateway(settings) + worker = Worker(settings, storage=storage, publisher=gateway, hotfolder=hotfolder, normalizer=NullNormalizer()) + health.nats = gateway + app.state.storage = storage + app.state.nats = gateway + app.state.worker = worker + worker.start() + nats_task = asyncio.create_task(run_nats(gateway, worker), name="nats-connect") try: logger.info("Приложение запущено") yield finally: logger.info("Приложение закрывается") + nats_task.cancel() + with suppress(asyncio.CancelledError): + await nats_task + await worker.stop() + await gateway.close() + await storage.close() if smb_session: smbclient.reset_connection_cache() app = FastAPI(title="scan-gateway", version=version, root_path=settings.ROOT_PATH, lifespan=lifespan) + app.state.health = health @app.get("/healthcheck", status_code=200) async def healthcheck(): - return {"uptime": up_time, "version": version} + return {"uptime": up_time, "version": version, **await health.snapshot()} app.include_router(router) register_exception_handlers(app) diff --git a/api/src/core/health.py b/api/src/core/health.py new file mode 100644 index 0000000..20f4c06 --- /dev/null +++ b/api/src/core/health.py @@ -0,0 +1,96 @@ +"""Состояние сервиса для `GET /healthcheck`: поля `nats` и `hotfolder` (D-8; spec image-recognition-gateway). + +- `nats`: `disabled` при `NATS_ENABLED=false`; `ok`, если соединение установлено (`is_connected`), иначе `down` — в том + числе пока идёт первое подключение, во время переподключения и после сбоя фоновой задачи подключения. +- `hotfolder`: `HotFolder.healthy()` в потоке (`asyncio.to_thread`) под `wait_for` с коротким таймаутом и с кэшем на + короткое время (T8.4: подключение к недоступной SMB-шаре ждёт до 60 с, проба не должна висеть и плодить потоки). + Пока прежняя проверка не вернулась, новая не запускается — ответ `down`. + В режиме smb сбой регистрации сессии при старте виден естественно: `SmbHotFolder.healthy()` обращается к шаре с теми + же учётными данными и при недоступной шаре или неверном логине возвращает `False` (T2.3). +""" + +import asyncio +import time +from collections.abc import Callable +from typing import Literal, Protocol + +from loguru import logger + +from src.gateway.hotfolder import HotFolder + +__all__ = ["HOTFOLDER_CACHE_TTL", "HOTFOLDER_CHECK_TIMEOUT", "ConnectionState", "HealthProvider"] + +HOTFOLDER_CHECK_TIMEOUT = 2.0 # секунд на `healthy()`; дольше — `down` +HOTFOLDER_CACHE_TTL = 5.0 # секунд, в течение которых повторная проба отдаёт прежний результат + +Status = Literal["ok", "down"] +NatsStatus = Literal["ok", "down", "disabled"] + + +class ConnectionState(Protocol): + """Источник состояния NATS (`NatsGateway`).""" + + @property + def is_connected(self) -> bool: ... + + +class HealthProvider: + """Поля `nats`/`hotfolder` для `/healthcheck`. `nats` присваивается, когда соединение создано (lifespan).""" + + def __init__( + self, + *, + nats_enabled: bool, + hotfolder: HotFolder | None, + check_timeout: float = HOTFOLDER_CHECK_TIMEOUT, + cache_ttl: float = HOTFOLDER_CACHE_TTL, + clock: Callable[[], float] = time.monotonic, + ) -> None: + self.nats: ConnectionState | None = None + self._nats_enabled = nats_enabled + self._hotfolder = hotfolder + self._check_timeout = check_timeout + self._cache_ttl = cache_ttl + self._clock = clock + self._cached: Status | None = None + self._cached_at = 0.0 + self._inflight: asyncio.Task[bool] | None = None + + def nats_status(self) -> NatsStatus: + if not self._nats_enabled: + return "disabled" + if self.nats is not None and self.nats.is_connected: + return "ok" + return "down" + + async def hotfolder_status(self) -> Status: + if self._hotfolder is None: + return "down" + if self._cached is not None and self._clock() - self._cached_at < self._cache_ttl: + return self._cached + if self._inflight is None or self._inflight.done(): + self._inflight = asyncio.create_task(asyncio.to_thread(self._hotfolder.healthy)) + self._inflight.add_done_callback(_consume_exception) + status: Status + try: + healthy = await asyncio.wait_for(asyncio.shield(self._inflight), self._check_timeout) + status = "ok" if healthy is True else "down" + except TimeoutError: + logger.warning( + "Проверка горячей папки не уложилась в {timeout} с: hotfolder=down", timeout=self._check_timeout + ) + status = "down" + except Exception as exc: + logger.warning("Проверка горячей папки упала: hotfolder=down. error_type={type}", type=type(exc).__name__) + status = "down" + self._cached, self._cached_at = status, self._clock() + return status + + async def snapshot(self) -> dict[str, str]: + return {"nats": self.nats_status(), "hotfolder": await self.hotfolder_status()} + + +def _consume_exception(task: "asyncio.Task[bool]") -> None: + """Исключение пробы, которую никто не дождался (таймаут), не должно попадать в лог asyncio как «не извлечённое».""" + if not task.cancelled(): + task.exception() diff --git a/api/tests/unit/test_app.py b/api/tests/unit/test_app.py index b3ffae6..41587c9 100644 --- a/api/tests/unit/test_app.py +++ b/api/tests/unit/test_app.py @@ -1,18 +1,46 @@ +"""Приложение: `/healthcheck` (`uptime`, `version`, `nats`, `hotfolder`), lifespan (SMB-сессия, S3, NATS в фоне, +воркер). NATS и S3 не нужны: в тестах lifespan `NATS_ENABLED=false` либо методы `NatsGateway` подменены.""" + +import asyncio from datetime import datetime +from pathlib import Path +from typing import Any from unittest.mock import MagicMock import httpx +import pytest import smbclient from src.app import create_app from src.core.config import get_settings +from src.core.nats import NatsGateway +from src.core.worker import Worker + +pytestmark = pytest.mark.usefixtures("log_lines") def _client(app) -> httpx.AsyncClient: return httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") -async def test_healthcheck_returns_uptime_and_version_without_infrastructure(monkeypatch): +@pytest.fixture +def local_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Режим local с существующим корнем и без NATS: lifespan не ходит в сеть.""" + monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "local") + monkeypatch.setenv("SCAN_HOTFOLDER_PATH", str(tmp_path)) + monkeypatch.setenv("NATS_ENABLED", "false") + return tmp_path + + +async def eventually(predicate, timeout: float = 2.0) -> None: + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while not predicate(): + assert loop.time() < deadline, "условие не выполнилось вовремя" + await asyncio.sleep(0.005) + + +async def test_healthcheck_without_lifespan_has_all_fields(local_env: Path) -> None: app = create_app(get_settings()) # lifespan не запускается: подключений к NATS/S3 нет @@ -21,13 +49,42 @@ async def test_healthcheck_returns_uptime_and_version_without_infrastructure(mon assert response.status_code == 200 body = response.json() - assert set(body) == {"uptime", "version"} + assert set(body) == {"uptime", "version", "nats", "hotfolder"} assert body["version"] == get_settings().VERSION datetime.fromisoformat(body["uptime"]) -async def test_lifespan_does_not_register_smb_session_in_local_mode(monkeypatch): - monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "local") +async def test_start_in_local_mode(local_env: Path) -> None: + """Spec «Старт в режиме локальных папок»: 200 с uptime, version, nats, hotfolder = ok; OPENBAO__SMB__* не нужны.""" + app = create_app(get_settings()) + async with app.router.lifespan_context(app), _client(app) as client: + response = await client.get("/healthcheck") + assert response.status_code == 200 + body = response.json() + assert body["hotfolder"] == "ok" + assert body["nats"] == "disabled" + assert body["version"] == get_settings().VERSION + + +async def test_missing_hotfolder_root_is_down(local_env: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("SCAN_HOTFOLDER_PATH", str(local_env / "missing")) + app = create_app(get_settings()) + async with app.router.lifespan_context(app), _client(app) as client: + assert (await client.get("/healthcheck")).json()["hotfolder"] == "down" + + +async def test_lifespan_starts_and_stops_worker(local_env: Path) -> None: + app = create_app(get_settings()) + async with app.router.lifespan_context(app): + worker = app.state.worker + assert isinstance(worker, Worker) + assert isinstance(app.state.nats, NatsGateway) + # после остановки воркер запросов не принимает + await worker.on_request(type("Msg", (), {"data": b"{}"})()) + assert worker.active_jobs == 0 + + +async def test_lifespan_does_not_register_smb_session_in_local_mode(local_env: Path, monkeypatch) -> None: register = MagicMock() monkeypatch.setattr(smbclient, "register_session", register) app = create_app(get_settings()) @@ -38,26 +95,114 @@ async def test_lifespan_does_not_register_smb_session_in_local_mode(monkeypatch) register.assert_not_called() -async def test_lifespan_registers_smb_session_in_smb_mode(monkeypatch): +def _smb_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "smb") monkeypatch.setenv("OPENBAO__SMB__SMB_HOST", "smb.example") monkeypatch.setenv("OPENBAO__SMB__SMB_PATH", "share") monkeypatch.setenv("OPENBAO__SMB__SMB_LOGIN", "user") monkeypatch.setenv("OPENBAO__SMB__SMB_PASSWORD", "secret") + monkeypatch.setenv("NATS_ENABLED", "false") + + +async def test_lifespan_registers_smb_session_in_smb_mode(monkeypatch) -> None: + _smb_env(monkeypatch) register = MagicMock() reset = MagicMock() monkeypatch.setattr(smbclient, "register_session", register) monkeypatch.setattr(smbclient, "reset_connection_cache", reset) + monkeypatch.setattr(smbclient, "listdir", MagicMock(return_value=["Input", "Output"])) app = create_app(get_settings()) async with app.router.lifespan_context(app), _client(app) as client: - assert (await client.get("/healthcheck")).status_code == 200 + response = await client.get("/healthcheck") + assert response.status_code == 200 + assert response.json()["hotfolder"] == "ok" register.assert_called_once_with("smb.example", "user", "secret") reset.assert_called_once() -async def test_unhandled_error_has_json_500(monkeypatch): +async def test_smb_session_failure_shows_hotfolder_down(monkeypatch, log_lines: list[str]) -> None: + """T2.3: регистрация SMB-сессии при старте не удалась (шара недоступна) — `hotfolder = down`, сервис жив.""" + _smb_env(monkeypatch) + monkeypatch.setattr(smbclient, "register_session", MagicMock(side_effect=OSError("сервер недоступен"))) + monkeypatch.setattr(smbclient, "listdir", MagicMock(side_effect=OSError("сервер недоступен"))) + app = create_app(get_settings()) + + async with app.router.lifespan_context(app), _client(app) as client: + response = await client.get("/healthcheck") + + assert response.status_code == 200 + assert response.json()["hotfolder"] == "down" + assert any("SMB" in line for line in log_lines if line.startswith("ERROR|")) + + +async def test_nats_connects_in_background_and_subscribes_worker(local_env: Path, monkeypatch) -> None: + monkeypatch.setenv("NATS_ENABLED", "true") + subscribed: list[Any] = [] + connected = asyncio.Event() + + async def fake_connect(self: NatsGateway) -> bool: + await connected.wait() # подключение «висит», а HTTP уже отвечает + return True + + async def fake_subscribe(self: NatsGateway, cb: Any, queue: str = "scan-gateway") -> None: + subscribed.append(cb) + + async def fake_close(self: NatsGateway) -> None: + return None + + monkeypatch.setattr(NatsGateway, "connect", fake_connect) + monkeypatch.setattr(NatsGateway, "subscribe_requests", fake_subscribe) + monkeypatch.setattr(NatsGateway, "close", fake_close) + app = create_app(get_settings()) + + async with app.router.lifespan_context(app), _client(app) as client: + assert (await client.get("/healthcheck")).json()["nats"] == "down" + connected.set() + await eventually(lambda: bool(subscribed)) + assert subscribed == [app.state.worker.on_request] + + +async def test_nats_connect_hanging_is_cancelled_on_shutdown(local_env: Path, monkeypatch) -> None: + monkeypatch.setenv("NATS_ENABLED", "true") + cancelled = asyncio.Event() + + async def fake_connect(self: NatsGateway) -> bool: + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelled.set() + raise + return True + + async def fake_close(self: NatsGateway) -> None: + return None + + monkeypatch.setattr(NatsGateway, "connect", fake_connect) + monkeypatch.setattr(NatsGateway, "close", fake_close) + app = create_app(get_settings()) + async with app.router.lifespan_context(app): + await asyncio.sleep(0.01) + assert cancelled.is_set() + + +async def test_nats_config_error_is_loud_and_shows_down(local_env: Path, monkeypatch, log_lines: list[str]) -> None: + """Неполные учётные данные NATS (`NatsConfigError`): лог error с трассировкой, `nats = down`, HTTP жив.""" + monkeypatch.setenv("NATS_ENABLED", "true") + monkeypatch.setenv("OPENBAO__NATS__NATS_SEED", "SUAINCOMPLETESEED") + monkeypatch.setenv("OPENBAO__NATS__NATS_JWT", "") + app = create_app(get_settings()) + + async with app.router.lifespan_context(app), _client(app) as client: + await eventually(lambda: any(line.startswith("ERROR|") and "NATS" in line for line in log_lines)) + response = await client.get("/healthcheck") + + assert response.status_code == 200 + assert response.json()["nats"] == "down" + + +async def test_unhandled_error_has_json_500(local_env: Path) -> None: app = create_app(get_settings()) @app.get("/boom") diff --git a/api/tests/unit/test_health.py b/api/tests/unit/test_health.py new file mode 100644 index 0000000..9c95782 --- /dev/null +++ b/api/tests/unit/test_health.py @@ -0,0 +1,120 @@ +"""Состояние для `/healthcheck` (`src.core.health`): поля `nats` и `hotfolder` (spec «Старт в режиме локальных папок», +«Горячая папка недоступна», «Соединение с NATS без JetStream»; D-8; T8.4 — проверка не зависает, T2.3 — сбой SMB).""" + +import threading +from pathlib import Path + +import pytest + +from src.core.health import HealthProvider +from src.gateway.hotfolder import LocalHotFolder + +pytestmark = pytest.mark.usefixtures("log_lines") + + +class Connection: + def __init__(self, connected: bool) -> None: + self.is_connected = connected + + +class CountingHotFolder: + """`healthy()` считает вызовы; может зависнуть до события или бросить исключение.""" + + def __init__(self, result: bool = True) -> None: + self.result = result + self.calls = 0 + self.release: threading.Event | None = None + self.error: BaseException | None = None + + def put(self, name: str, data: bytes) -> None: + raise NotImplementedError + + def poll(self, job_id: str) -> bytes | None: + raise NotImplementedError + + def healthy(self) -> bool: + self.calls += 1 + if self.release is not None: + self.release.wait(5) + if self.error is not None: + raise self.error + return self.result + + +class Clock: + def __init__(self) -> None: + self.now = 1000.0 + + def __call__(self) -> float: + return self.now + + +async def test_local_root_exists_is_ok(tmp_path: Path) -> None: + health = HealthProvider(nats_enabled=False, hotfolder=LocalHotFolder(tmp_path)) + assert await health.snapshot() == {"nats": "disabled", "hotfolder": "ok"} + + +async def test_missing_root_is_down(tmp_path: Path) -> None: + health = HealthProvider(nats_enabled=False, hotfolder=LocalHotFolder(tmp_path / "missing")) + assert await health.hotfolder_status() == "down" + + +async def test_no_hotfolder_is_down() -> None: + assert await HealthProvider(nats_enabled=False, hotfolder=None).hotfolder_status() == "down" + + +@pytest.mark.parametrize( + ("enabled", "connection", "expected"), + [ + (False, None, "disabled"), + (False, Connection(True), "disabled"), + (True, None, "down"), + (True, Connection(False), "down"), + (True, Connection(True), "ok"), + ], +) +def test_nats_status(enabled: bool, connection: Connection | None, expected: str) -> None: + health = HealthProvider(nats_enabled=enabled, hotfolder=None) + health.nats = connection + assert health.nats_status() == expected + + +async def test_hotfolder_result_is_cached_for_ttl() -> None: + hotfolder = CountingHotFolder(result=True) + clock = Clock() + health = HealthProvider(nats_enabled=False, hotfolder=hotfolder, cache_ttl=5, clock=clock) + assert await health.hotfolder_status() == "ok" + hotfolder.result = False + clock.now += 4.9 + assert await health.hotfolder_status() == "ok" + assert hotfolder.calls == 1 + clock.now += 0.2 + assert await health.hotfolder_status() == "down" + assert hotfolder.calls == 2 + + +async def test_hanging_check_is_down_within_timeout_and_not_repeated() -> None: + hotfolder = CountingHotFolder() + hotfolder.release = threading.Event() + clock = Clock() + health = HealthProvider(nats_enabled=False, hotfolder=hotfolder, check_timeout=0.05, cache_ttl=5, clock=clock) + try: + assert await health.hotfolder_status() == "down" + clock.now += 10 # кэш истёк, но прежняя проверка ещё висит — новый поток не запускается + assert await health.hotfolder_status() == "down" + assert hotfolder.calls == 1 + finally: + hotfolder.release.set() + + +async def test_check_raising_is_down() -> None: + hotfolder = CountingHotFolder() + hotfolder.error = OSError("smb") + health = HealthProvider(nats_enabled=False, hotfolder=hotfolder) + assert await health.hotfolder_status() == "down" + + +async def test_snapshot_with_connected_nats(tmp_path: Path) -> None: + health = HealthProvider(nats_enabled=True, hotfolder=LocalHotFolder(tmp_path)) + health.nats = Connection(True) + assert await health.snapshot() == {"nats": "ok", "hotfolder": "ok"} 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 2e9f4cc..7f66842 100644 --- a/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md +++ b/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md @@ -64,7 +64,7 @@ ### Task 12: [5.3] `src/core/worker.py` -- [ ] `src/core/worker.py`: `on_request(msg)`: валидация → LRU `job_id` → ack `processing` → очередь; семафор `SCAN_MAX_INFLIGHT`; `handle(job)`: S3 get → `hotfolder.put` → опрос до дедлайна → `raw/{job_id}.json` → `Normalizer` → результат/конверт → publish; таблица ошибок D-7; `/healthcheck.nats`, `.hotfolder`; проверка: unit-тесты `handle` с фейками и `LocalHotFolder` + `process_once` эмулятора — сценарии spec (ack раньше файла, повторный запрос → один файл, успех → NOT_IMPLEMENTED с `raw_s3_key`, INVALID_DOCUMENT, TIMEOUT с укороченными таймаутами, RECOGNITION_FAILED при отсутствии каталога, SCHEMA_MISMATCH без ack, FILE_READ_ERROR) +- [x] `src/core/worker.py`: `on_request(msg)`: валидация → LRU `job_id` → ack `processing` → очередь; семафор `SCAN_MAX_INFLIGHT`; `handle(job)`: S3 get → `hotfolder.put` → опрос до дедлайна → `raw/{job_id}.json` → `Normalizer` → результат/конверт → publish; таблица ошибок D-7; `/healthcheck.nats`, `.hotfolder`; проверка: unit-тесты `handle` с фейками и `LocalHotFolder` + `process_once` эмулятора — сценарии spec (ack раньше файла, повторный запрос → один файл, успех → NOT_IMPLEMENTED с `raw_s3_key`, INVALID_DOCUMENT, TIMEOUT с укороченными таймаутами, RECOGNITION_FAILED при отсутствии каталога, SCHEMA_MISMATCH без ack, FILE_READ_ERROR) ### Task 13: [5.4] Тест параллелизма diff --git a/openspec/changes/slice-0-skeleton/tasks.md b/openspec/changes/slice-0-skeleton/tasks.md index 1500bfa..f4848b1 100644 --- a/openspec/changes/slice-0-skeleton/tasks.md +++ b/openspec/changes/slice-0-skeleton/tasks.md @@ -50,7 +50,7 @@ - [x] 5.2 `src/core/clients.py` (`get_bytes`, `put_json`, `put_bytes`), `src/core/nats.py` (подключение с/без nkeys, автопереподключение, `subscribe_requests(queue="scan-gateway", cb)`, `publish` с `Nats-Msg-Id` и повторами); проверка: unit с фейками, интеграционный тест подключения к compose-NATS без JetStream -- [ ] 5.3 `src/core/worker.py`: `on_request(msg)`: валидация → LRU `job_id` → ack `processing` → очередь; семафор +- [x] 5.3 `src/core/worker.py`: `on_request(msg)`: валидация → LRU `job_id` → ack `processing` → очередь; семафор `SCAN_MAX_INFLIGHT`; `handle(job)`: S3 get → `hotfolder.put` → опрос до дедлайна → `raw/{job_id}.json` → `Normalizer` → результат/конверт → publish; таблица ошибок D-7; `/healthcheck.nats`, `.hotfolder`; проверка: unit-тесты `handle` с фейками и `LocalHotFolder` + `process_once` эмулятора — сценарии spec (ack раньше файла, повторный запрос → один файл,