NATS: подписка, публикация и /healthcheck.subjects по настройкам; roundtrip на своих subject'ах (#31)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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, воркер не стартует |

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

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)

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} невозможна")
 

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. "

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

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

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()

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()))