diff --git a/README.md b/README.md index cb98e34..7de7c2a 100644 --- a/README.md +++ b/README.md @@ -382,7 +382,7 @@ cd api && .venv/bin/python run_server.py curl -s localhost:8000/healthcheck ``` -В ответе — `uptime`, `version` и `nats`: `ok` — подключён, `down` — включён (`NATS_ENABLED=true`), но связи нет, +В ответе — `uptime`, `version`, `subjects` (используемые subject'ы: `tabular_request`, `result`; JWT там нет) и `nats`: `ok` — подключён, `down` — включён (`NATS_ENABLED=true`), но связи нет, `off` — выключен (`NATS_ENABLED=false`, воркер не запущен). Если NATS или MinIO недоступны при старте, приложение всё равно поднимается: ошибка логируется, соединение с NATS @@ -468,6 +468,9 @@ ENV_FILE=.env.test .venv/bin/python -m src.parser.cli <файл> [--file-type XL | `OPENBAO__NATS__NATS_URL` | — (обязательна) | адрес NATS, `nats://host:4222` | | `OPENBAO__NATS__NATS_SEED` | пусто | nkeys seed; пусто вместе с JWT — подключение без аутентификации (dev) | | `OPENBAO__NATS__NATS_JWT` | пусто | JWT пользователя NATS | +| `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT` | `files.tabular_uploaded` | subject запроса `tabular`, на который подписан сервис | +| `OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT` | `files.img_uploaded` | subject запроса `image` (слушает scan-gateway; здесь — единый набор ключей) | +| `OPENBAO__NATS__NATS_RESULT_SUBJECT` | `files.tabular_parsed` | subject, в который сервис публикует ack и результат | | `DEBUG` | `false` | уровень логов DEBUG | | `ENV_FILE` | `.env` | путь к файлу окружения (от `api/`) | | `NATS_ENABLED` | `true` | `false` — не подключаться к NATS, воркер не стартует | diff --git a/api/.env.example b/api/.env.example index 106f24d..db873e1 100644 --- a/api/.env.example +++ b/api/.env.example @@ -27,6 +27,11 @@ OPENBAO__MINIO__MINIO_SECURE=false OPENBAO__NATS__NATS_URL=nats://localhost:4222 OPENBAO__NATS__NATS_SEED= OPENBAO__NATS__NATS_JWT= +# Subject'ы (необязательно; значения ниже — по умолчанию). Запрос tabular — подписка, результат — публикация. +# Image-запрос слушает scan-gateway; здесь ключ нужен только для единого набора OPENBAO__NATS__* +# OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT=files.tabular_uploaded +# OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT=files.img_uploaded +# OPENBAO__NATS__NATS_RESULT_SUBJECT=files.tabular_parsed # Необязательные настройки (значения ниже — значения по умолчанию из src/core/config.py) # Префикс путей приложения за прокси и версия в OpenAPI diff --git a/api/src/app.py b/api/src/app.py index aaf88db..609c904 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -2,7 +2,7 @@ Что здесь есть: настройки, подключение S3 (MinIO), NATS (`src/core/nats.py`: nkeys JWT, пустые seed/jwt — без аутентификации, фоновое подключение — старт не ждёт NATS), воркер `src/core/worker.py` (подписка на -`files.tabular_uploaded` после подключения к NATS), middleware (gzip, proxy headers), `/healthcheck` +subject запроса `tabular` из настроек после подключения к NATS), middleware (gzip, proxy headers), `/healthcheck` (`nats = ok | down | off`, R-17), обработчики исключений. Сервис без БД и фронта (DECISION-4). При `NATS_ENABLED=false` ни NATS, ни воркер не запускаются. Остановка: воркер перестаёт принимать запросы и @@ -24,7 +24,7 @@ from src.core.clients import S3Storage 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.nats import NatsConnection, used_subjects from src.core.worker import STOP_TIMEOUT, RecognitionWorker from src.parser import get_parser @@ -85,12 +85,14 @@ def _build_worker(storage: S3Storage) -> tuple[NatsConnection, RecognitionWorker service_version=VERSION, concurrency=settings.WORKER_CONCURRENCY, done_cache_size=settings.DONE_CACHE_SIZE, + result_subject=settings.OPENBAO.NATS.NATS_RESULT_SUBJECT, ) return nats, worker @asynccontextmanager async def lifespan(app: FastAPI): + logger.info("Subject'ы NATS. {subjects}", subjects=used_subjects(settings.OPENBAO.NATS)) storage = await _connect_s3() nats: NatsConnection | None = None worker: RecognitionWorker | None = None @@ -130,7 +132,12 @@ def _nats_state(application: FastAPI) -> str: @app.get("/healthcheck", status_code=200) async def healthcheck(request: Request): - return {"uptime": UP_TIME, "version": VERSION, "nats": _nats_state(request.app)} + return { + "uptime": UP_TIME, + "version": VERSION, + "nats": _nats_state(request.app), + "subjects": used_subjects(settings.OPENBAO.NATS), + } app.include_router(router) diff --git a/api/src/core/nats.py b/api/src/core/nats.py index cd7c4c7..d18530d 100644 --- a/api/src/core/nats.py +++ b/api/src/core/nats.py @@ -10,7 +10,8 @@ завершился исключением, supervisor повторяет попытку с паузой 1 → 2 → 4 → … ≤ 30 с. После подключения один раз вызывается `on_connected(conn)` (подписки воркера); если он упал, созданные в этой попытке подписки снимаются и вызов повторяется с той же паузой. Подписки после разрыва nats-py восстанавливает сам. -- `subscribe_requests(cb)` — подписка на `files.tabular_uploaded` в queue group `tabular-parser`: из нескольких +- `subscribe_requests(cb)` — подписка на subject запроса `tabular` из настроек (по умолчанию + `files.tabular_uploaded`) в queue group `tabular-parser`: из нескольких инстанций сообщение получает одна. После `subscribe` — `flush`: сервер обработал SUB, ошибки соединения проявляются здесь. Отказ в правах (`Permissions Violation`) nats-py 2.14 так не проявляет: соединение остаётся открытым, `subscribe`/`flush`/`publish` проходят, сервер лишь присылает `-ERR` в `error_cb` — он логируется отдельно уровнем @@ -37,7 +38,7 @@ from loguru import logger from nats.aio.client import Client as NATS from nats.aio.msg import Msg -from src.contract.subjects import NATS_MSG_ID_HEADER, TABULAR_REQUEST_SUBJECT +from src.contract.subjects import NATS_MSG_ID_HEADER from src.core.config import NATSSettings REQUEST_QUEUE_GROUP = "tabular-parser" @@ -112,6 +113,11 @@ def build_nats_auth(settings: NATSSettings) -> dict[str, Any]: return {"user_jwt_cb": user_jwt_cb, "signature_cb": signature_cb} +def used_subjects(settings: NATSSettings) -> dict[str, str]: + """Subject'ы сервиса для лога старта и `/healthcheck.subjects`: подписан на запрос `tabular`, публикует ответ.""" + return {"tabular_request": settings.NATS_TABULAR_REQUEST_SUBJECT, "result": settings.NATS_RESULT_SUBJECT} + + class NatsConnection: """Соединение сервиса с NATS: фоновое подключение, подписка на запросы, публикация с повторами. @@ -276,9 +282,13 @@ class NatsConnection: # --- подписка и публикация ------------------------------------------------------------------ async def subscribe_requests( - self, cb: MessageHandler, *, queue: str = REQUEST_QUEUE_GROUP, subject: str = TABULAR_REQUEST_SUBJECT + self, cb: MessageHandler, *, queue: str = REQUEST_QUEUE_GROUP, subject: str | None = None ) -> Any: - """Подписка на запросы (`files.tabular_uploaded`) в queue group; исключения `cb` логируются.""" + """Подписка на запросы в queue group; исключения `cb` логируются. + + Без `subject` — `NATS_TABULAR_REQUEST_SUBJECT` настроек (по умолчанию `files.tabular_uploaded`). + """ + subject = subject or self._settings.NATS_TABULAR_REQUEST_SUBJECT if not self.is_connected: raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {subject} невозможна") diff --git a/api/src/core/worker.py b/api/src/core/worker.py index 6ce6372..81652a2 100644 --- a/api/src/core/worker.py +++ b/api/src/core/worker.py @@ -141,11 +141,13 @@ class RecognitionWorker: concurrency: int = 1, done_cache_size: int = 1000, clock: Clock = time.monotonic, + result_subject: str = RESULT_SUBJECT, ) -> None: if concurrency < 1 or done_cache_size < 1: raise ValueError("concurrency и done_cache_size должны быть ≥ 1") self._storage = storage self._publisher = publisher + self._result_subject = result_subject self._parser = parser self._service = ServiceInfo.model_validate({"name": service_name, "version": service_version}) self._concurrency = concurrency @@ -464,7 +466,7 @@ class RecognitionWorker: """Публикация конверта; исчерпанные повторы и прочие сбои — error-лог, не исключение (D-5).""" payload = json.dumps(envelope.to_wire(), ensure_ascii=False).encode("utf-8") try: - await self._publisher.publish(RESULT_SUBJECT, payload, msg_id) + await self._publisher.publish(self._result_subject, payload, msg_id) except NatsPublishError as exc: logger.error( "Конверт не опубликован после всех повторов; ядро закроет задание по reply_deadline. " diff --git a/api/tests/integration/test_roundtrip.py b/api/tests/integration/test_roundtrip.py index 6c817b5..ae47aaa 100644 --- a/api/tests/integration/test_roundtrip.py +++ b/api/tests/integration/test_roundtrip.py @@ -12,7 +12,8 @@ Spec tabular-recognition-service: «Ack раньше результата», « «Реальная ОСВ через воркер» (маркер `fixtures`: реальный XLS из `../fixtures` → `done` и файл результата в MinIO), «Реальные формы через воркер» (книга НПЗ → `done`, документы BALANCE и INCOME, `result_schema_version 1.2`). -Изоляция: `job_id`/`file_id`/ключ S3 уникальны на прогон (uuid4), исходник удаляется после теста; на шине может +Изоляция: subject'ы прогона уникальны — контейнер стенда сообщения теста не получает (OQ-83); +`job_id`/`file_id`/ключ S3 уникальны на прогон (uuid4), исходник удаляется после теста; на шине может работать другой сервис — сообщения `files.tabular_parsed` фильтруются по своему `job_id`. Ожидания — по событиям с таймаутом; единственная фиксированная пауза — явное окно тишины `QUIET_WINDOW` для отрицательных проверок. """ @@ -37,8 +38,6 @@ from result_contract import schema_errors from src.contract import ( NATS_MSG_ID_HEADER, - RESULT_SUBJECT, - TABULAR_REQUEST_SUBJECT, RecognitionResult, request_msg_id, ) @@ -71,6 +70,14 @@ SOURCE_BYTES = b"PK\x03\x04 round-trip: broken zip, the default parser answers U XLSX_CONTENT_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" XLS_CONTENT_TYPE = "application/vnd.ms-excel" +# Subject'ы прогона уникальны (OQ-83): контейнер tabular-parser стенда слушает умолчания и сообщения теста не получает +RUN = uuid.uuid4().hex[:8] +TEST_SUBJECTS = { + "NATS_TABULAR_REQUEST_SUBJECT": f"test-{RUN}.files.tabular_uploaded", + "NATS_IMAGE_REQUEST_SUBJECT": f"test-{RUN}.files.img_uploaded", + "NATS_RESULT_SUBJECT": f"test-{RUN}.files.tabular_parsed", +} + app_module = importlib.import_module("src.app") # модуль, а не объект `app`, который реэкспортирует пакет `src` @@ -136,7 +143,7 @@ class Core: self._inboxes.setdefault(request["job_id"], asyncio.Queue()) payload = json.dumps(request, ensure_ascii=False).encode("utf-8") headers = {NATS_MSG_ID_HEADER: request_msg_id(request["job_id"], request["attempt"])} - await self.client.publish(TABULAR_REQUEST_SUBJECT, payload, headers=headers) + await self.client.publish(TEST_SUBJECTS["NATS_TABULAR_REQUEST_SUBJECT"], payload, headers=headers) await self.client.flush() async def next_message(self, request: dict[str, Any]) -> Msg: @@ -216,6 +223,8 @@ async def uploads() -> AsyncIterator[Uploads]: async def service(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[None]: """Приложение в процессе теста: lifespan с `NATS_ENABLED=True` (в `.env.test` — false); выход — остановка.""" monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True) + nats_settings = app_module.settings.OPENBAO.NATS + monkeypatch.setattr(app_module.settings.OPENBAO, "NATS", nats_settings.model_copy(update=TEST_SUBJECTS)) application = app_module.app tasks: list[asyncio.Task[None]] = [] async with application.router.lifespan_context(application): @@ -236,7 +245,7 @@ async def core(service: None) -> AsyncIterator[Core]: client = NATS() await client.connect(get_settings().OPENBAO.NATS.NATS_URL, connect_timeout=2, max_reconnect_attempts=1) core = Core(client) - await client.subscribe(RESULT_SUBJECT, cb=core.on_result) + await client.subscribe(TEST_SUBJECTS["NATS_RESULT_SUBJECT"], cb=core.on_result) await client.flush() try: yield core diff --git a/api/tests/unit/test_app.py b/api/tests/unit/test_app.py index c4a79f9..78d7e20 100644 --- a/api/tests/unit/test_app.py +++ b/api/tests/unit/test_app.py @@ -124,6 +124,11 @@ def patched_app(monkeypatch: pytest.MonkeyPatch, events: list[str]) -> Iterator[ yield app_module +def _expected_subjects(module: Any) -> dict[str, str]: + nats = module.settings.OPENBAO.NATS + return {"tabular_request": nats.NATS_TABULAR_REQUEST_SUBJECT, "result": nats.NATS_RESULT_SUBJECT} + + async def test_nats_disabled_starts_nothing_and_healthcheck_reports_off( patched_app: Any, monkeypatch: pytest.MonkeyPatch, events: list[str] ) -> None: @@ -136,7 +141,8 @@ async def test_nats_disabled_starts_nothing_and_healthcheck_reports_off( assert response.status_code == 200 # noqa: PLR2004 - HTTP 200 body = response.json() - assert set(body) == {"uptime", "version", "nats"} + assert set(body) == {"uptime", "version", "nats", "subjects"} + assert body["subjects"] == _expected_subjects(patched_app) assert body["nats"] == "off" assert body["version"] == patched_app.settings.VERSION assert FakeNatsConnection.instances == [] @@ -154,7 +160,7 @@ async def test_minimal_environment_starts_and_healthcheck_answers( response = await client.get("/healthcheck") assert response.status_code == 200 # noqa: PLR2004 - HTTP 200 - assert set(response.json()) == {"uptime", "version", "nats"} + assert set(response.json()) == {"uptime", "version", "nats", "subjects"} async def test_nats_enabled_subscribes_worker_and_reports_ok( @@ -195,7 +201,10 @@ async def test_nats_enabled_but_not_connected_reports_down(patched_app: Any, mon FakeNatsConnection.connect_on_start = False async with running_app(patched_app) as client: - assert (await client.get("/healthcheck")).json()["nats"] == "down" + response = await client.get("/healthcheck") + assert response.json()["nats"] == "down" + assert response.json()["subjects"] == _expected_subjects(patched_app) + assert "jwt" not in response.text.lower() (connection,) = FakeNatsConnection.instances assert connection.handlers == [] # подписка — только после подключения assert patched_app.app.state.worker is not None diff --git a/api/tests/unit/test_nats.py b/api/tests/unit/test_nats.py index bd4f4b4..8349a9e 100644 --- a/api/tests/unit/test_nats.py +++ b/api/tests/unit/test_nats.py @@ -26,6 +26,7 @@ from src.core.nats import ( NatsPublishError, Publisher, build_nats_auth, + used_subjects, ) from src.core.worker import STOP_TIMEOUT @@ -347,6 +348,26 @@ async def test_subscribe_requests_uses_tabular_subject_and_queue_group(): await conn.stop() +async def test_subscribe_requests_default_subject_comes_from_settings(): + world = _FakeWorld() + settings = NATSSettings( + NATS_URL="nats://localhost:4222", NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded" + ) + conn = NatsConnection(settings, client_factory=world, sleep=_RecordingSleep()) + await conn.start() + assert await conn.wait_ready(1) + + await conn.subscribe_requests(_noop_handler) + + assert world.clients[-1].subscriptions[0]["subject"] == "test.files.tabular_uploaded" + await conn.stop() + + +def test_used_subjects_tabular_parser() -> None: + nats = NATSSettings(NATS_URL="nats://x:4222", NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded") + assert used_subjects(nats) == {"tabular_request": "test.files.tabular_uploaded", "result": "files.tabular_parsed"} + + async def test_subscribe_requests_flushes_after_subscribe(): """T8.3: подписка считается активной, когда сервер обработал SUB (flush после subscribe).""" world = _FakeWorld() diff --git a/api/tests/unit/test_worker.py b/api/tests/unit/test_worker.py index 5a0fbb4..b75ed8a 100644 --- a/api/tests/unit/test_worker.py +++ b/api/tests/unit/test_worker.py @@ -174,7 +174,14 @@ def log_records() -> Iterator[list[Any]]: class Env: - def __init__(self, parser: Any = None, *, concurrency: int = 1, done_cache_size: int = 1000) -> None: + def __init__( + self, + parser: Any = None, + *, + concurrency: int = 1, + done_cache_size: int = 1000, + result_subject: str | None = None, + ) -> None: self.events: list[tuple[str, str]] = [] self.storage = FakeStorage(self.events) self.storage.objects[(BUCKET, SOURCE_KEY)] = b"PK\x03\x04 xlsx bytes" @@ -188,6 +195,7 @@ class Env: concurrency=concurrency, done_cache_size=done_cache_size, clock=FakeClock(), + **({} if result_subject is None else {"result_subject": result_subject}), ) def bodies(self) -> list[dict[str, Any]]: @@ -217,6 +225,20 @@ def _assert_message_valid(body: dict[str, Any]) -> None: RecognitionResultMessage.model_validate(body) +async def test_ack_and_result_go_to_configured_result_subject() -> None: + env = await _make_env(result_subject="test.files.tabular_parsed") + try: + await env.worker.on_request(_msg(_request())) + await env.worker.join() + finally: + await env.worker.stop(timeout=1) + + assert [(subject, msg_id) for subject, _body, msg_id in env.publisher.published] == [ + ("test.files.tabular_parsed", f"ack:{JOB_ID}:1"), + ("test.files.tabular_parsed", f"result:{JOB_ID}:1"), + ] + + # --- ack раньше результата ------------------------------------------------------------------------------------------ async def test_ack_is_published_before_s3_read_and_result(env: Env) -> None: await env.worker.on_request(_msg(_request()))