diff --git a/README.md b/README.md index c21a76d..27c6621 100644 --- a/README.md +++ b/README.md @@ -922,6 +922,9 @@ email без учёта регистра проверяет только сер | `OPENBAO__NATS__NATS_URL` | — (обязательно) | адрес NATS (`nats://localhost:4222`) | | `OPENBAO__NATS__NATS_SEED` | пусто | nkey seed; пусто — подключение без учётных данных (compose, тесты). Задаётся вместе с `NATS_JWT` или не задаётся вовсе | | `OPENBAO__NATS__NATS_JWT` | пусто | JWT NATS; см. `NATS_SEED` | +| `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT` | `files.tabular_uploaded` | subject запроса распознавания таблиц (публикация) | +| `OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT` | `files.img_uploaded` | subject запроса распознавания изображений (публикация) | +| `OPENBAO__NATS__NATS_RESULT_SUBJECT` | `files.tabular_parsed` | subject ответов (подписка, queue group `arm-core`). Три имени различны, без пробелов, `*`, `>` и пустых токенов; нарушение — отказ старта. Действующие значения — в логе старта и в `/healthcheck.subjects` | | `OPENBAO__SMB__SMB_HOST` | пусто | SMB из шаблона, необязателен (ядру в Срезе 0 не нужен). Пусто — сессия SMB не регистрируется; задан — регистрация при старте (в отдельном потоке) и сброс кэша соединений при остановке | | `OPENBAO__SMB__SMB_PATH` | пусто | путь SMB (в Срезе 0 не используется) | | `OPENBAO__SMB__SMB_LOGIN` | пусто | логин SMB | @@ -1001,7 +1004,7 @@ tabular-parser (MinIO) → inbox → маппинг → `GET /files/{id}` = `PRO ```bash docker compose -f ../docs/dev/docker-compose.yml -p arm-ka --profile services up -d --build --no-deps tabular-parser -curl -s localhost:8001/healthcheck # {"uptime": "...", "version": "...", "nats": "ok"} +curl -s localhost:8001/healthcheck # {"uptime": "...", "version": "...", "nats": "ok", "subjects": {"tabular_request": "...", "image_request": "...", "result": "..."}} make e2e # PASSED docker compose -f ../docs/dev/docker-compose.yml -p arm-ka --profile services stop tabular-parser ``` diff --git a/api/.env.example b/api/.env.example index 3eb0770..5a14e50 100644 --- a/api/.env.example +++ b/api/.env.example @@ -37,6 +37,10 @@ OPENBAO__MINIO__MINIO_SECURE=false OPENBAO__NATS__NATS_URL=nats://localhost:4222 OPENBAO__NATS__NATS_SEED= OPENBAO__NATS__NATS_JWT= +# subject'ы обмена — по окружению (прод/тест), у трёх сервисов одинаковые; умолчания — имена контракта +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 # SMB необязателен (ядру в Срезе 0 не нужен): пустой SMB_HOST — сессия SMB не регистрируется OPENBAO__SMB__SMB_HOST= diff --git a/api/.env.test b/api/.env.test index b807ba7..522fdf0 100644 --- a/api/.env.test +++ b/api/.env.test @@ -23,6 +23,9 @@ OPENBAO__MINIO__MINIO_SECURE=false OPENBAO__NATS__NATS_URL=nats://localhost:4222 OPENBAO__NATS__NATS_SEED= OPENBAO__NATS__NATS_JWT= +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 # SMB необязателен: пустой SMB_HOST — сессия SMB не регистрируется OPENBAO__SMB__SMB_HOST= diff --git a/api/src/app.py b/api/src/app.py index 2e5afac..4a53798 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -4,8 +4,9 @@ NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка `files.tabular_parsed` в queue group `arm-core`), SMB (горячие папки ИС Сканирование; необязателен — при пустом `SMB_HOST` не подключается), middleware (gzip, proxy headers, CORS по `cors_options`; при DEBUG=False пакет РАИСА лишь условно импортируется, -middleware не подключается), healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. Документация API -(`/docs`, `/redoc`, `/openapi.json`) — только при DEBUG (OQ-66). Приложение собирает `create_app(settings)`. +middleware не подключается), healthcheck (uptime, version, nats: ok|down|off, subjects), раздача собранного +SPA. Документация API (`/docs`, `/redoc`, `/openapi.json`) — только при DEBUG (OQ-66). Приложение собирает +`create_app(settings)`. Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор, вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) и, если S3 подключено, цикл маппинга результатов @@ -43,7 +44,7 @@ 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.logger import setup_logger -from src.core.nats import MessageHandler, NatsConnection +from src.core.nats import MessageHandler, NatsConnection, used_subjects from src.db.db import async_session_maker, engine, session_transaction from src.services.recognition_exchange import ExchangeTasks, handle_message from src.services.session_audit import SessionAudit @@ -183,6 +184,7 @@ async def lifespan(app: FastAPI): app.state.nats = None app.state.exchange = None + logger.info("Subject'ы NATS. {subjects}", subjects=used_subjects(settings.OPENBAO.NATS)) await _register_smb() try: @@ -241,7 +243,12 @@ def create_app(app_settings: Settings, *, web_dir: str = "web") -> FastAPI: if app_settings.NATS_ENABLED: nc = getattr(request.app.state, "nats", None) nats_status = "ok" if nc is not None and nc.is_connected else "down" - return {"uptime": UP_TIME, "version": VERSION, "nats": nats_status} + return { + "uptime": UP_TIME, + "version": VERSION, + "nats": nats_status, + "subjects": used_subjects(app_settings.OPENBAO.NATS), + } application.include_router(build_api_router(app_settings)) # аудит входа (OQ-64): LOGIN первого запроса и LOGIN_FAILED — в отдельной транзакции, множество входов — на процесс diff --git a/api/src/core/nats.py b/api/src/core/nats.py index 292bca6..969c3a1 100644 --- a/api/src/core/nats.py +++ b/api/src/core/nats.py @@ -4,8 +4,9 @@ - Аутентификация как в шаблоне: заданы `NATS_SEED` и `NATS_JWT` — nkeys/JWT; оба пусты — без учётных данных. - `publish(subject, payload, msg_id)` — публикация с заголовком `Nats-Msg-Id` и `flush`: ошибка соединения проявляется сразу исключением `NatsPublishError` (outbox оставляет задание в `queued`). -- `subscribe_results(handler)` — подписка на `files.tabular_parsed` в queue group `arm-core`: из нескольких инстансов - ядра сообщение получает ровно один. Исключения обработчика логируются, подписка не падает. +- `subscribe_results(handler)` — подписка на `NATS_RESULT_SUBJECT` настроек (умолчание `files.tabular_parsed`) в queue + group `arm-core`: из нескольких инстансов ядра сообщение получает ровно один. Исключения обработчика + логируются, подписка не падает. - Stream'ов и consumer'ов JetStream нет. Переподключение после разрыва — встроенное в nats-py (`max_reconnect_attempts=-1`, `reconnect_time_wait=2`), подписки nats-py восстанавливает сам. - Первое подключение выполняет фоновый supervisor (`start()`): старт приложения не ждёт NATS. nats-py 2.14 с @@ -27,7 +28,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, RESULT_SUBJECT +from src.contract.subjects import NATS_MSG_ID_HEADER from src.core.config import NATSSettings RESULT_QUEUE_GROUP = "arm-core" @@ -60,6 +61,15 @@ class NatsPublishError(NatsError): """Публикация не подтверждена: соединения нет или сервер не ответил на flush.""" +def used_subjects(settings: NATSSettings) -> dict[str, str]: + """Subject'ы ядра для лога старта и `/healthcheck.subjects`: публикует оба запроса, подписано на ответ.""" + return { + "tabular_request": settings.NATS_TABULAR_REQUEST_SUBJECT, + "image_request": settings.NATS_IMAGE_REQUEST_SUBJECT, + "result": settings.NATS_RESULT_SUBJECT, + } + + def build_nats_auth(settings: NATSSettings) -> dict[str, Any]: """Параметры аутентификации для `Client.connect`. @@ -103,7 +113,7 @@ class NatsConnection: settings: NATSSettings, *, on_connected: Callable[["NatsConnection"], Awaitable[None]] | None = None, - result_subject: str = RESULT_SUBJECT, + result_subject: str | None = None, queue: str = RESULT_QUEUE_GROUP, client_factory: Callable[[], Any] = NATS, sleep: SleepFn = asyncio.sleep, @@ -114,7 +124,7 @@ class NatsConnection: ) -> None: self._settings = settings self._on_connected = on_connected - self.result_subject = result_subject + self.result_subject = result_subject or settings.NATS_RESULT_SUBJECT self.queue = queue self._client_factory = client_factory self._sleep = sleep diff --git a/api/src/services/primary_files.py b/api/src/services/primary_files.py index 366830b..ddbe5f3 100644 --- a/api/src/services/primary_files.py +++ b/api/src/services/primary_files.py @@ -480,7 +480,7 @@ class PrimaryFilesService: id=job_id, file_id=file_id, pipeline=detected.pipeline, - request_subject=request_subject(detected.pipeline.value), + request_subject=request_subject(detected.pipeline.value, self._settings.OPENBAO.NATS.request_subjects()), schema_version=CONTRACT_VERSION, status=RecognitionStatus.queued, attempt=1, diff --git a/api/src/services/recognition_exchange.py b/api/src/services/recognition_exchange.py index 25e918f..466af43 100644 --- a/api/src/services/recognition_exchange.py +++ b/api/src/services/recognition_exchange.py @@ -64,6 +64,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.contract import ( CONTRACT_VERSION, NATS_MSG_ID_HEADER, + REQUEST_SUBJECTS, MessageStatus, Metrics, RecognitionRequest, @@ -211,6 +212,7 @@ async def _process_one( # noqa: PLR0913 - один шаг прохода: вс ack_cutoff: datetime, pass_started: datetime, max_republish: int, + request_subjects: Mapping[str, str], ) -> bool: """Обработать одно задание в транзакции `session`; `False` — подходящих заданий больше нет.""" row = (await session.execute(_candidate_stmt(pass_started, ack_cutoff))).one_or_none() @@ -244,7 +246,7 @@ async def _process_one( # noqa: PLR0913 - один шаг прохода: вс reply_deadline = row.reply_deadline or now + DEFAULT_REPLY_TIMEOUT[row.pipeline] try: request = _build_request(row, reply_deadline) - subject = request_subject(row.pipeline.value) + subject = request_subject(row.pipeline.value, request_subjects) except (ValidationError, ValueError) as exc: await session.execute( _job_update(row.id, row.status).values( @@ -320,6 +322,7 @@ async def publish_pending( # noqa: PLR0913 - параметры прохода max_republish: int, batch_size: int = PUBLISH_BATCH_SIZE, now: Clock | None = None, + request_subjects: Mapping[str, str] = REQUEST_SUBJECTS, ) -> PublishReport: """Один проход outbox-публикатора (spec «Outbox публикует задания из `queued`», «Запрос без ack публикуется повторно, затем — `dlq`»; design D-6). @@ -346,6 +349,7 @@ async def publish_pending( # noqa: PLR0913 - параметры прохода ack_cutoff=ack_cutoff, pass_started=pass_started, max_republish=max_republish, + request_subjects=request_subjects, ) except NatsPublishError as exc: report.failed += 1 @@ -402,11 +406,18 @@ async def run_publisher_loop( # noqa: PLR0913 - настройки цикла ack_timeout: timedelta, max_republish: int, sleep: SleepFn = asyncio.sleep, + request_subjects: Mapping[str, str] = REQUEST_SUBJECTS, ) -> None: """Бесконечный цикл публикатора с периодом `poll_seconds`; ошибка прохода логируется, цикл продолжается.""" while True: try: - await publish_pending(session_factory, nats, ack_timeout=ack_timeout, max_republish=max_republish) + await publish_pending( + session_factory, + nats, + ack_timeout=ack_timeout, + max_republish=max_republish, + request_subjects=request_subjects, + ) except Exception as exc: logger.exception( "Ошибка прохода outbox-публикатора. error_category=outbox_error, error_type={error_type}, " @@ -459,6 +470,7 @@ class ExchangeTasks: poll_seconds=settings.OUTBOX_POLL_SECONDS, ack_timeout=timedelta(seconds=settings.ACK_TIMEOUT_SECONDS), max_republish=settings.OUTBOX_MAX_REPUBLISH, + request_subjects=settings.OPENBAO.NATS.request_subjects(), ), name="recognition-outbox-publisher", ) diff --git a/api/tests/test_api_files_upload.py b/api/tests/test_api_files_upload.py index c62c246..3eff62f 100644 --- a/api/tests/test_api_files_upload.py +++ b/api/tests/test_api_files_upload.py @@ -22,7 +22,7 @@ from sqlalchemy import func, select, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.api.dependencies.directories import get_app_settings -from src.core.config import get_settings +from src.core.config import NATSSettings, get_settings from src.models.access import AppUser from src.models.audit_ex import AuditEvent from src.models.consolidation import Consolidation, ConsolidationStatusHistory, ConsolidationVersion @@ -210,6 +210,32 @@ async def test_real_osv_is_accepted_and_published_from_outbox( # --- поля записей, имя файла, аудит ------------------------------------------------------------------------------ +@pytest.mark.integration +async def test_job_request_subject_comes_from_environment_settings( + app: FastAPI, auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage +) -> None: + """Review Focus 4: `recognition_job.request_subject` берётся из настроек окружения, а не из констант контракта.""" + base = get_settings() + nats = NATSSettings( + **{ + **base.OPENBAO.NATS.model_dump(), + "NATS_TABULAR_REQUEST_SUBJECT": "test.files.tabular_uploaded", + } + ) + app.dependency_overrides[get_app_settings] = lambda: base.model_copy( + update={"OPENBAO": base.OPENBAO.model_copy(update={"NATS": nats})} + ) + entity = await _rostov_entity(seeded) + client = await auth_client(KA_RF) + + response = await _post(client, entity.id, ("osv.xls", _xls("subject"))) + + assert response.status_code == HTTP_CREATED, response.text + job = await seeded.get(RecognitionJob, uuid.UUID(response.json()["accepted"][0]["job_id"])) + assert job is not None + assert job.request_subject == "test.files.tabular_uploaded" + + @pytest.mark.integration async def test_hostile_name_gives_flat_key_and_full_records( auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage diff --git a/api/tests/test_app.py b/api/tests/test_app.py index 4120751..59f6401 100644 --- a/api/tests/test_app.py +++ b/api/tests/test_app.py @@ -47,6 +47,25 @@ async def test_healthcheck_nats_off_when_disabled(client: httpx.AsyncClient, mon assert {"uptime", "version"} <= set(body) +def _expected_subjects() -> dict[str, str]: + nats = app_module.settings.OPENBAO.NATS + return { + "tabular_request": nats.NATS_TABULAR_REQUEST_SUBJECT, + "image_request": nats.NATS_IMAGE_REQUEST_SUBJECT, + "result": nats.NATS_RESULT_SUBJECT, + } + + +@pytest.mark.parametrize("enabled", [False, True], ids=["when_nats_off", "when_nats_down"]) +async def test_healthcheck_subjects(client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch, enabled: bool): + """Review Focus 5: subjects отдаются при любом состоянии NATS, JWT в ответе нет.""" + monkeypatch.setattr(app_module.settings, "NATS_ENABLED", enabled) + body = (await client.get("/healthcheck")).json() + assert body["subjects"] == _expected_subjects() + assert "NATS_JWT" not in str(body) + assert "jwt" not in str(body).lower() + + async def test_healthcheck_nats_down_when_enabled_without_connection( client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch ): diff --git a/api/tests/test_nats_core.py b/api/tests/test_nats_core.py index 4df623a..60b8e44 100644 --- a/api/tests/test_nats_core.py +++ b/api/tests/test_nats_core.py @@ -25,6 +25,7 @@ from src.core.nats import ( NatsNotConnectedError, NatsPublishError, build_nats_auth, + used_subjects, ) SRC_DIR = Path(__file__).resolve().parent.parent / "src" @@ -377,3 +378,18 @@ def test_src_does_not_use_jetstream(): if alias.name.startswith("nats.js") ) assert offenders == [] + + +def test_used_subjects_core() -> None: + nats = NATSSettings(NATS_URL="nats://x:4222", NATS_RESULT_SUBJECT="test.files.tabular_parsed") + assert used_subjects(nats) == { + "tabular_request": "files.tabular_uploaded", + "image_request": "files.img_uploaded", + "result": "test.files.tabular_parsed", + } + + +def test_connection_subscribes_to_configured_result_subject() -> None: + nats = NATSSettings(NATS_URL="nats://x:4222", NATS_RESULT_SUBJECT="test.files.tabular_parsed") + assert NatsConnection(nats).result_subject == "test.files.tabular_parsed" + assert NatsConnection(nats, result_subject="override").result_subject == "override" diff --git a/api/tests/test_recognition_exchange.py b/api/tests/test_recognition_exchange.py index ac4b0d1..8a6e216 100644 --- a/api/tests/test_recognition_exchange.py +++ b/api/tests/test_recognition_exchange.py @@ -75,7 +75,8 @@ class Clock: class RequestSubscriber: """Тестовый подписчик на subject'ы запросов: «голый» клиент nats-py, сообщения фильтруются по `job_id`.""" - def __init__(self) -> None: + def __init__(self, subjects: tuple[str, ...] = REQUEST_SUBJECTS) -> None: + self.subjects = subjects self.messages: list[Msg] = [] self._nc = NATS() @@ -89,7 +90,7 @@ class RequestSubscriber: async def collect(msg: Msg) -> None: self.messages.append(msg) - for subject in REQUEST_SUBJECTS: + for subject in self.subjects: await self._nc.subscribe(subject, cb=collect) await self._nc.flush() @@ -261,6 +262,29 @@ async def test_publish_request_to_service( # noqa: PLR0913,PLR0917 - фикст assert job.attempt == 1 +@pytest.mark.integration +async def test_publish_uses_configured_request_subject( + session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection +) -> None: + """Review Focus 4: subject — из настроек окружения, `request_subject` задания — фактический subject публикации.""" + run = uuid.uuid4().hex[:8] + subjects = {"tabular": f"test-{run}.files.tabular_uploaded", "image": f"test-{run}.files.img_uploaded"} + custom = RequestSubscriber(subjects=tuple(subjects.values())) + await custom.start() + try: + job_id = await jobs.make_queued_job( + pipeline=RecognitionPipeline.tabular, file_type=FileType.XLSX, original_name="ОСВ 60.xlsx" + ) + await publish_pending( + session_factory, nats, ack_timeout=ACK, max_republish=MAX_REPUBLISH, request_subjects=subjects + ) + (msg,) = await custom.wait_for(job_id, 1) + finally: + await custom.stop() + assert msg.subject == subjects["tabular"] + assert (await jobs.get(job_id)).request_subject == subjects["tabular"] + + @pytest.mark.integration async def test_publish_keeps_existing_reply_deadline( session_factory: async_sessionmaker[AsyncSession], @@ -819,7 +843,12 @@ async def test_exchange_tasks_start_two_loops_and_stop(monkeypatch: pytest.Monke await _wait_until(lambda: len(started) == EXCHANGE_TASKS) assert len(exchange.tasks) == EXCHANGE_TASKS - assert started["publisher"] == {"poll_seconds": 7, "ack_timeout": timedelta(seconds=11), "max_republish": 2} + assert started["publisher"] == { + "poll_seconds": 7, + "ack_timeout": timedelta(seconds=11), + "max_republish": 2, + "request_subjects": {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"}, + } assert started["watchdog"] == {"period_seconds": WATCHDOG_PERIOD_SECONDS} await exchange.stop() assert all(task.done() for task in exchange.tasks)