diff --git a/README.md b/README.md index 793221d..c070aeb 100644 --- a/README.md +++ b/README.md @@ -48,8 +48,9 @@ Stateless-сервис распознавания табличных файло ``` api/ бэкенд (все команды разработки — отсюда или через make из корня) run_server.py точка входа платформы (uvicorn; SERVER__HOST, SERVER__PORT) - src/app.py FastAPI: lifespan (подключение к MinIO и NATS), GET /healthcheck, middleware, обработчики ошибок - src/core/ config.py (настройки), clients.py (S3), logger.py, exception_handlers.py + src/app.py FastAPI: lifespan (MinIO, NATS, воркер), GET /healthcheck, middleware, обработчики ошибок + src/core/ config.py (настройки), clients.py (S3), nats.py (соединение NATS), worker.py (воркер + распознавания), logger.py, exception_handlers.py src/parser/ интерфейс Parser и реализации (добавляется задачей 4.1) src/contract/ pydantic-модели контракта v1.0 (добавляется задачей 3.1) contracts/ копии JSON-схем контракта (добавляется задачей 3.1) @@ -108,8 +109,8 @@ cd api && .venv/bin/python run_server.py curl -s localhost:8000/healthcheck ``` -В ответе — `uptime`, `version` и (после добавления воркера, задача 4.3) `nats`: `ok` — подключён, `down` — включён -(`NATS_ENABLED=true`), но связи нет, `off` — выключен. +В ответе — `uptime`, `version` и `nats`: `ok` — подключён, `down` — включён (`NATS_ENABLED=true`), но связи нет, +`off` — выключен (`NATS_ENABLED=false`, воркер не запущен). Если NATS или MinIO недоступны при старте, приложение всё равно поднимается: ошибка логируется, соединение с NATS восстанавливается автоматически (бесконечное автопереподключение). diff --git a/api/.env.example b/api/.env.example index 17adc3a..fa55ae2 100644 --- a/api/.env.example +++ b/api/.env.example @@ -27,7 +27,7 @@ ROOT_PATH=/ VERSION=1.0.0 # `service.name` конвертов контракта и имя queue group NATS; менять не нужно, пока контракт v1.x SERVICE_NAME=tabular-parser -# false — не подключаться к NATS (воркер не стартует; /healthcheck отдаёт nats = down) +# false — не подключаться к NATS (воркер не стартует; /healthcheck отдаёт nats = off) NATS_ENABLED=true # Число заданий, обрабатываемых одновременно WORKER_CONCURRENCY=1 diff --git a/api/src/app.py b/api/src/app.py index b3e2592..1b4ca99 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -1,15 +1,18 @@ """Точка входа FastAPI сервиса tabular-parser. Что здесь есть: настройки, подключение S3 (MinIO), NATS (`src/core/nats.py`: nkeys JWT, пустые seed/jwt — без -аутентификации, фоновое подключение — старт не ждёт NATS), middleware (gzip, proxy headers), healthcheck, обработчики -исключений. Сервис без БД и фронта (DECISION-4). Воркер обработки `recognition.tabular.request` и поле `nats` -в healthcheck добавляются в задаче 4.3. +аутентификации, фоновое подключение — старт не ждёт NATS), воркер `src/core/worker.py` (подписка на +`recognition.tabular.request` после подключения к NATS), middleware (gzip, proxy headers), `/healthcheck` +(`nats = ok | down | off`, R-17), обработчики исключений. Сервис без БД и фронта (DECISION-4). + +При `NATS_ENABLED=false` ни NATS, ни воркер не запускаются. Остановка: воркер перестаёт принимать запросы и +дорабатывает очередь (NATS ещё нужен для публикации результатов), затем закрывается NATS (drain), затем сессия S3. """ from contextlib import asynccontextmanager from datetime import UTC, datetime -from fastapi import FastAPI +from fastapi import FastAPI, Request from loguru import logger from miniopy_async import Minio from starlette.middleware.gzip import GZipMiddleware @@ -21,6 +24,8 @@ from src.core.config import get_settings from src.core.exception_handlers import register_exception_handlers from src.core.logger import setup_logger from src.core.nats import NatsConnection +from src.core.worker import STOP_TIMEOUT, RecognitionWorker +from src.parser import get_parser settings = get_settings() @@ -30,8 +35,12 @@ UP_TIME = datetime.now(UTC) setup_logger(level="DEBUG" if settings.DEBUG else "INFO") -async def _connect_s3() -> S3Storage | None: - """Клиент S3 с проверкой бакета; при недоступности — ``None`` (приложение всё равно стартует).""" +async def _connect_s3() -> S3Storage: + """Клиент S3 с проверкой бакета. + + Проверка — только диагностика: при недоступном MinIO или отсутствующем бакете ошибка логируется, а клиент + остаётся (соединения открываются на каждый запрос) — когда MinIO поднимется, воркер прочитает файл без рестарта. + """ minio = settings.OPENBAO.MINIO s3_client = Minio( minio.MINIO_URL, @@ -48,40 +57,79 @@ async def _connect_s3() -> S3Storage | None: error_type=type(exc).__name__, error=repr(exc), ) - await s3_client.close_session() - return None - logger.info("Бакет проверен. bucket_exists={bucket_exists}", bucket_exists=bucket_exists) + else: + if bucket_exists: + logger.info("Бакет проверен. bucket={bucket}", bucket=minio.MINIO_BUCKET) + else: + logger.warning( + "Бакет не найден: чтение исходников и запись результатов будут падать. error_category=s3_error, " + "bucket={bucket}", + bucket=minio.MINIO_BUCKET, + ) return S3Storage(client=s3_client, bucket_name=minio.MINIO_BUCKET) +def _build_worker(storage: S3Storage) -> tuple[NatsConnection, RecognitionWorker]: + """Соединение NATS и воркер; подписка на запросы — в `on_connected` (после первого подключения).""" + + async def subscribe(conn: NatsConnection) -> None: + await conn.subscribe_requests(worker.on_request) + + nats = NatsConnection(settings.OPENBAO.NATS, on_connected=subscribe, publish_retries=settings.PUBLISH_RETRIES) + worker = RecognitionWorker( + storage=storage, + publisher=nats, + parser=get_parser(), + service_name=settings.SERVICE_NAME, + service_version=VERSION, + concurrency=settings.WORKER_CONCURRENCY, + done_cache_size=settings.DONE_CACHE_SIZE, + ) + return nats, worker + + @asynccontextmanager async def lifespan(app: FastAPI): - app.state.storage = await _connect_s3() - app.state.nats = None + storage = await _connect_s3() + nats: NatsConnection | None = None + worker: RecognitionWorker | None = None if settings.NATS_ENABLED: + nats, worker = _build_worker(storage) + worker.start() # start() не ждёт NATS: при недоступном сервере подключение продолжается в фоне (design D-6) - app.state.nats = NatsConnection(settings.OPENBAO.NATS, publish_retries=settings.PUBLISH_RETRIES) - await app.state.nats.start() + await nats.start() else: - logger.info("NATS выключен (NATS_ENABLED=false)") + logger.info("NATS выключен (NATS_ENABLED=false): воркер не запускается") + app.state.storage = storage + app.state.nats = nats + app.state.worker = worker try: logger.info("Приложение запущено") yield finally: logger.info("Приложение закрывается") - if app.state.nats: - await app.state.nats.stop() - if app.state.storage: - await app.state.storage.client.close_session() + if worker is not None: + await worker.stop(STOP_TIMEOUT) # результаты ещё публикуются: NATS закрывается после воркера + if nats is not None: + await nats.stop() + await storage.client.close_session() app = FastAPI(title="tabular-parser", version=VERSION, root_path=settings.ROOT_PATH, lifespan=lifespan) +def _nats_state(application: FastAPI) -> str: + """`off` — NATS выключен флагом (R-17); `ok` — соединение есть; `down` — включён, но связи нет.""" + if not settings.NATS_ENABLED: + return "off" + nats = getattr(application.state, "nats", None) + return "ok" if nats is not None and nats.is_connected else "down" + + @app.get("/healthcheck", status_code=200) -async def healthcheck(): - return {"uptime": UP_TIME, "version": VERSION} +async def healthcheck(request: Request): + return {"uptime": UP_TIME, "version": VERSION, "nats": _nats_state(request.app)} app.include_router(router) diff --git a/api/src/core/nats_callback.py b/api/src/core/nats_callback.py deleted file mode 100644 index e3420b5..0000000 --- a/api/src/core/nats_callback.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Обработчики сообщений NATS. - -Каркас: доменный обработчик удалён. Consumer `recognition.result` (дедуп по recognition_job.result_msg_id, -маппинг JSON из S3 в document/osv_entry/financial_statement_*) реализуется в Срезе 0–2 по -docs/context/03-architecture.md. Для референса — reference-old-arm/api/src/core/nats_callback.py. -""" - -from loguru import logger -from nats.aio.client import Client as NATS - -from src.core.clients import S3Storage - - -def build_result_handler(nc: NATS, storage: S3Storage): - async def handler(msg): - logger.info("NATS message received. subject={subject}, size={size}", subject=msg.subject, size=len(msg.data)) - await msg.ack() - - return handler diff --git a/api/tests/unit/test_app.py b/api/tests/unit/test_app.py new file mode 100644 index 0000000..f89ff75 --- /dev/null +++ b/api/tests/unit/test_app.py @@ -0,0 +1,179 @@ +"""Unit-тесты lifespan и `/healthcheck` приложения без NATS и MinIO (design D-6, R-17; carry-over T8.7). + +`NatsConnection`, `_connect_s3` и `RecognitionWorker` модуля `src.app` подменяются фейками: проверяется, что при +`NATS_ENABLED=true` воркер запускается и подписывается через `on_connected`, порядок остановки +(воркер → NATS → S3) и значения `nats = ok | down | off`. +""" + +import importlib +import json +from collections.abc import AsyncIterator, Awaitable, Callable, Iterator +from contextlib import asynccontextmanager +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import httpx +import pytest + +from src.contract import NATS_MSG_ID_HEADER, RESULT_SUBJECT +from src.core.worker import RecognitionWorker + +EXAMPLES_DIR = Path(__file__).resolve().parents[2] / "contracts" / "examples" +REQUEST = json.loads((EXAMPLES_DIR / "request-tabular.json").read_text(encoding="utf-8")) +JOB_ID = REQUEST["job_id"] +app_module = importlib.import_module("src.app") # модуль, а не объект `app`, который реэкспортирует пакет `src` + + +@asynccontextmanager +async def running_app(module: Any) -> AsyncIterator[httpx.AsyncClient]: + """Lifespan приложения + HTTP-клиент поверх ASGI (без сети).""" + application = module.app + async with application.router.lifespan_context(application): + transport = httpx.ASGITransport(app=application) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + yield client + + +class FakeStorage: + def __init__(self, events: list[str]) -> None: + self.events = events + self.client = SimpleNamespace(close_session=self._close) + self.objects = {(REQUEST["s3"]["bucket"], REQUEST["s3"]["key"]): b"xlsx"} + + async def _close(self) -> None: + self.events.append("s3.close") + + async def get_bytes(self, bucket: str, key: str) -> bytes: + return self.objects[(bucket, key)] + + async def put_json(self, bucket: str, key: str, obj: Any) -> None: + raise AssertionError("NullParser не пишет файл результата") + + +class FakeNatsConnection: + """Повторяет интерфейс `NatsConnection`, нужный lifespan и воркеру.""" + + instances: list["FakeNatsConnection"] = [] + connect_on_start = True + events: list[str] = [] # общий журнал порядка запуска и остановки (задаётся фикстурой) + + def __init__( + self, + settings: Any, + *, + on_connected: Callable[["FakeNatsConnection"], Awaitable[None]] | None = None, + publish_retries: int = 3, + ) -> None: + self.settings = settings + self.on_connected = on_connected + self.publish_retries = publish_retries + self.is_connected = False + self.handlers: list[tuple[str, Callable[[Any], Awaitable[None]]]] = [] + self.published: list[tuple[str, dict[str, Any], str]] = [] + FakeNatsConnection.instances.append(self) + + async def start(self) -> None: + self.events.append("nats.start") + if self.connect_on_start: + self.is_connected = True + if self.on_connected is not None: + await self.on_connected(self) + + async def stop(self) -> None: + self.events.append("nats.stop") + self.is_connected = False + + async def subscribe_requests(self, cb: Callable[[Any], Awaitable[None]]) -> None: + self.handlers.append(("recognition.tabular.request", cb)) + + async def publish(self, subject: str, payload: bytes, msg_id: str) -> None: + self.published.append((subject, json.loads(payload), msg_id)) + + +@pytest.fixture +def events() -> list[str]: + return [] + + +@pytest.fixture +def patched_app(monkeypatch: pytest.MonkeyPatch, events: list[str]) -> Iterator[Any]: + """`src.app` с фейковыми S3, NATS и записывающим порядок остановки воркером.""" + FakeNatsConnection.instances = [] + FakeNatsConnection.connect_on_start = True + FakeNatsConnection.events = events + storage = FakeStorage(events) + + async def fake_connect_s3() -> FakeStorage: + return storage + + class RecordingWorker(RecognitionWorker): + async def stop(self, timeout: float = 10.0) -> None: + events.append("worker.stop") + await super().stop(timeout) + + monkeypatch.setattr(app_module, "_connect_s3", fake_connect_s3) + monkeypatch.setattr(app_module, "NatsConnection", FakeNatsConnection) + monkeypatch.setattr(app_module, "RecognitionWorker", RecordingWorker) + yield app_module + + +async def test_nats_disabled_starts_nothing_and_healthcheck_reports_off( + patched_app: Any, monkeypatch: pytest.MonkeyPatch, events: list[str] +) -> None: + monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", False) + + async with running_app(patched_app) as client: + response = await client.get("/healthcheck") + assert patched_app.app.state.worker is None + assert patched_app.app.state.nats is None + + assert response.status_code == 200 # noqa: PLR2004 - HTTP 200 + body = response.json() + assert set(body) == {"uptime", "version", "nats"} + assert body["nats"] == "off" + assert body["version"] == patched_app.settings.VERSION + assert FakeNatsConnection.instances == [] + assert events == ["s3.close"] + + +async def test_nats_enabled_subscribes_worker_and_reports_ok( + patched_app: Any, monkeypatch: pytest.MonkeyPatch, events: list[str] +) -> None: + monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", True) + + async with running_app(patched_app) as client: + assert (await client.get("/healthcheck")).json()["nats"] == "ok" + (connection,) = FakeNatsConnection.instances + assert connection.publish_retries == patched_app.settings.PUBLISH_RETRIES + assert connection.handlers, "воркер должен подписаться через on_connected" + worker = patched_app.app.state.worker + assert isinstance(worker, RecognitionWorker) + + # Запрос, доставленный подписке, проходит через воркер: ack и результат NOT_IMPLEMENTED в recognition.result + _subject, handler = connection.handlers[0] + message = SimpleNamespace( + data=json.dumps(REQUEST).encode(), headers={NATS_MSG_ID_HEADER: f"{JOB_ID}:1"}, subject=_subject + ) + await handler(message) + await worker.join() + + assert [(subject, msg_id) for subject, _body, msg_id in connection.published] == [ + (RESULT_SUBJECT, f"ack:{JOB_ID}:1"), + (RESULT_SUBJECT, f"result:{JOB_ID}:1"), + ] + assert connection.published[1][1]["error"]["code"] == "NOT_IMPLEMENTED" + assert connection.published[1][1]["service"]["name"] == "tabular-parser" + # Остановка: воркер дорабатывает очередь, пока NATS ещё жив, затем NATS, затем S3 + assert events == ["nats.start", "worker.stop", "nats.stop", "s3.close"] + + +async def test_nats_enabled_but_not_connected_reports_down(patched_app: Any, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", True) + FakeNatsConnection.connect_on_start = False + + async with running_app(patched_app) as client: + assert (await client.get("/healthcheck")).json()["nats"] == "down" + (connection,) = FakeNatsConnection.instances + assert connection.handlers == [] # подписка — только после подключения + assert patched_app.app.state.worker is not None diff --git a/openspec/changes/slice-0-skeleton/specs/tabular-recognition-service/spec.md b/openspec/changes/slice-0-skeleton/specs/tabular-recognition-service/spec.md index 51282d1..f3b8f8f 100644 --- a/openspec/changes/slice-0-skeleton/specs/tabular-recognition-service/spec.md +++ b/openspec/changes/slice-0-skeleton/specs/tabular-recognition-service/spec.md @@ -80,7 +80,8 @@ NATS и S3; воркер MUST NOT зависеть от конкретной р ### Requirement: Соединение с NATS по правилам шаблона, без JetStream При пустых `NATS_SEED`/`NATS_JWT` сервис SHALL подключаться без учётных данных, иначе — с nkeys/JWT. Сервис MUST NOT требовать stream'ов или consumer'ов JetStream; при разрыве соединения SHALL переподключаться автоматически и восстанавливать -подписку; `/healthcheck` SHALL отражать состояние (`nats = ok | down`). +подписку; `/healthcheck` SHALL отражать состояние (`nats = ok | down`; `off` — NATS выключен флагом +`NATS_ENABLED=false`, R-17). #### Scenario: NATS перезапущен - **WHEN** контейнер NATS перезапущен во время работы сервиса