From 5ae8ff8a81077627347d5bcf9dfadb34f2271bf8 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Thu, 8 Oct 2026 23:46:18 +0400 Subject: [PATCH] =?UTF-8?q?nats:=20subject'=D1=8B=20=D0=BA=D0=B0=D0=BA=20?= =?UTF-8?q?=D0=B2=20=D1=81=D1=82=D0=B0=D1=80=D0=BE=D0=BC=20=D0=90=D0=A0?= =?UTF-8?q?=D0=9C=20=E2=80=94=20files.*=5Fuploaded,=20files.tabular=5Fpars?= =?UTF-8?q?ed=20(#77)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Решение владельца 08.10.2026: recognition.tabular.request → files.tabular_uploaded, recognition.image.request → files.img_uploaded, recognition.result → files.tabular_parsed (общий subject ack и результатов; files.img_parsed не используется). Тела сообщений, Nats-Msg-Id, queue group'ы и schema_version не меняются. Константы api/src/contract/subjects.py, докстринги, тесты, README, CLAUDE.md; копии схем api/contracts/ из docs/contracts/ побайтно; change nats-files-subjects с дельтой спеки. Сквозной change — RSHB/arm-ka-docs#74. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2 --- CLAUDE.md | 2 +- README.md | 8 ++-- api/contracts/recognition-request.schema.json | 2 +- .../recognition-result-message.schema.json | 2 +- api/src/app.py | 10 ++--- api/src/contract/message.py | 2 +- api/src/contract/subjects.py | 15 ++++--- api/src/core/nats.py | 6 +-- api/src/services/recognition_exchange.py | 12 ++--- api/tests/e2e/test_osv_ingest_e2e.py | 2 +- api/tests/factories.py | 9 ++-- api/tests/test_api_files_upload.py | 4 +- api/tests/test_app.py | 4 +- api/tests/test_contract.py | 9 ++-- api/tests/test_exchange_lifespan_nats.py | 2 +- api/tests/test_migrations.py | 2 +- api/tests/test_models_part_04_06.py | 2 +- api/tests/test_nats_core.py | 8 ++-- api/tests/test_nats_integration.py | 2 +- api/tests/test_recognition_exchange.py | 8 ++-- api/tests/test_recognition_inbox.py | 8 ++-- api/tests/test_result_mapping.py | 2 +- .../nats-files-subjects/.openspec.yaml | 2 + .../changes/nats-files-subjects/design.md | 14 ++++++ .../changes/nats-files-subjects/proposal.md | 33 ++++++++++++++ .../specs/recognition-job-exchange/spec.md | 44 +++++++++++++++++++ openspec/changes/nats-files-subjects/tasks.md | 11 +++++ 27 files changed, 167 insertions(+), 58 deletions(-) create mode 100644 openspec/changes/nats-files-subjects/.openspec.yaml create mode 100644 openspec/changes/nats-files-subjects/design.md create mode 100644 openspec/changes/nats-files-subjects/proposal.md create mode 100644 openspec/changes/nats-files-subjects/specs/recognition-job-exchange/spec.md create mode 100644 openspec/changes/nats-files-subjects/tasks.md diff --git a/CLAUDE.md b/CLAUDE.md index 8882051..644c08e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -67,7 +67,7 @@ (`docker build`; убран из `.gitignore` шаблона). CI в репозитории нет: перед PR — `make check-full`. Без Nexus: `make venv` ставит зависимости без `raisa_*`, npm-пакеты `raisa-*` — `optionalDependencies`. - **NATS без JetStream** (`NATS_ENABLED=true`): `core/nats.py` — подключение в фоне (supervisor повторяет попытки), - подписка queue group `arm-core` на `recognition.result`. Обмен — `services/recognition_exchange.py`: `publish_pending` + подписка queue group `arm-core` на `files.tabular_parsed`. Обмен — `services/recognition_exchange.py`: `publish_pending` (outbox), `expire_deadlines` (вахтёр сроков), `handle_message` (inbox: ack — конверт `processing`, результат); публикатор и вахтёр — фоновые задачи lifespan. Запрос без ack повторно публикуется (`ACK_TIMEOUT_SECONDS`=30 × `OUTBOX_MAX_REPUBLISH`=3, счётчик — `recognition_job.metrics.republish_count`) → `dlq`; дедуп — `request_msg_id` / diff --git a/README.md b/README.md index 979aa57..d7cf77a 100644 --- a/README.md +++ b/README.md @@ -190,7 +190,7 @@ Dev-пользователи (`api/seed/dev/users.yaml`, создаются `mak блокируется; старт ждёт её завершения не дольше тайм-аута `smbclient`, 60 с); ошибка подключения логируется трейсбэком `Ошибка при регистрации сессии SMB`, приложение продолжает работу. -Строки `Бакет проверен`, `Подключение к NATS установлено` и `Подписка NATS активна. subject=recognition.result, +Строки `Бакет проверен`, `Подключение к NATS установлено` и `Подписка NATS активна. subject=files.tabular_parsed, queue=arm-core` означают, что S3 и NATS из compose подключены (MinIO из compose слушает `http`, поэтому в `.env` задано `OPENBAO__MINIO__MINIO_SECURE=false`). @@ -420,8 +420,8 @@ cd web && npm install && npm run start # фронтенд (rsbuild dev), о ``` POST /legal-entities/{id}/files → S3 uploads/… + stored_file(UPLOADED) + recognition_job(queued) + audit FILE_UPLOAD - → публикатор (outbox) → NATS recognition.tabular.request → tabular-parser (читает S3, пишет results/…) - → recognition.result → inbox (задание done, файл PROCESSING) → цикл маппинга (document + osv_entry, файл PROCESSED) + → публикатор (outbox) → NATS files.tabular_uploaded → tabular-parser (читает S3, пишет results/…) + → files.tabular_parsed → inbox (задание done, файл PROCESSING) → цикл маппинга (document + osv_entry, файл PROCESSED) → автозамена прежних версий (SUPERSEDED) → GET /files/{id}, GET /documents/{id}/entries → POST /documents/{id}/confirm (только для NEEDS_CHECK и форм после правки) ``` @@ -994,7 +994,7 @@ docker compose -f ../docs/dev/docker-compose.yml -p arm-ka --profile services st Образ собирается из `../tabular-parser` (main соседнего репозитория workspace). -**Перед e2e остановите другие экземпляры ядра, подписанные на `recognition.result`** (контейнер `arm-core` из профиля +**Перед e2e остановите другие экземпляры ядра, подписанные на `files.tabular_parsed`** (контейнер `arm-core` из профиля `services`, локальный `make run`/`run_server.py` с `NATS_ENABLED=true`): тест подписывается на реальный subject в боевой queue group `arm-core`, подписчики группы получают сообщения по очереди (round-robin), и ack/результат парсера уйдут чужому экземпляру (он отбросит их как `unknown_job`, у него другая БД) — тест упадёт по таймауту 60 с. Поднимайте только diff --git a/api/contracts/recognition-request.schema.json b/api/contracts/recognition-request.schema.json index 780573c..9d59f2c 100644 --- a/api/contracts/recognition-request.schema.json +++ b/api/contracts/recognition-request.schema.json @@ -2,7 +2,7 @@ "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "arm-ka/recognition-request/v1", "title": "Запрос распознавания файла (arm-core → tabular-parser / scan-gateway), контракт v1.0", - "description": "Тело сообщения на subject recognition.tabular.request или recognition.image.request; subject должен соответствовать полю pipeline (tabular → recognition.tabular.request, image → recognition.image.request). Заголовок Nats-Msg-Id = : (не поле тела). Один файл (stored_file) → один job.", + "description": "Тело сообщения на subject files.tabular_uploaded или files.img_uploaded; subject должен соответствовать полю pipeline (tabular → files.tabular_uploaded, image → files.img_uploaded). Заголовок Nats-Msg-Id = : (не поле тела). Один файл (stored_file) → один job.", "$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.", "type": "object", "required": ["schema_version", "job_id", "file_id", "pipeline", "file_type", "original_name", "s3", "reply_deadline", "attempt"], diff --git a/api/contracts/recognition-result-message.schema.json b/api/contracts/recognition-result-message.schema.json index 1f20727..7c1cfb5 100644 --- a/api/contracts/recognition-result-message.schema.json +++ b/api/contracts/recognition-result-message.schema.json @@ -2,7 +2,7 @@ "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "arm-ka/recognition-result-message/v1", "title": "Сообщение результата распознавания (tabular-parser / scan-gateway → arm-core), контракт v1.0", - "description": "Тело сообщения на subject recognition.result. Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:: — для status = processing; result:: — для done, needs_check и error. Форма запроса : описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).", + "description": "Тело сообщения на subject files.tabular_parsed. Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:: — для status = processing; result:: — для done, needs_check и error. Форма запроса : описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).", "$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.", "type": "object", "required": ["schema_version", "job_id", "file_id", "status", "service"], diff --git a/api/src/app.py b/api/src/app.py index 6b42fc4..2e5afac 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -1,8 +1,8 @@ """Точка входа FastAPI. Каркас из шаблона команды: инфраструктура без доменного кода. Что здесь есть: настройки, подключение S3 (MinIO; `MINIO_SECURE` — https или http), NATS (`src/core/nats.py`: core -NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка `recognition.result` в queue -group `arm-core`), SMB (горячие папки ИС Сканирование; необязателен — при пустом `SMB_HOST` не подключается), +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)`. @@ -10,7 +10,7 @@ middleware не подключается), healthcheck (uptime, version, nats: o состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор, вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) и, если S3 подключено, цикл маппинга результатов (`src/services/result_mapping.py`, Срез 2c D-1) — и останавливает их до закрытия NATS. Сообщения -`recognition.result` (ack и результаты сервисов) обрабатывает inbox `handle_message` того же модуля с фабрикой сессий +`files.tabular_parsed` (ack и результаты сервисов) обрабатывает inbox `handle_message` того же модуля с фабрикой сессий приложения (inbox меняет задание и статус файла; маппинг результата в доменные таблицы — отдельный цикл, Срез 2c D-1). Остановка: фоновые задачи обмена → NATS → S3 → SMB → пул соединений БД; каждый шаг независим — ошибка одного логируется и не отменяет остальные. Логгер настраивает `setup_logger` при импорте модуля: без значений локальных @@ -109,13 +109,13 @@ async def _connect_s3() -> S3Storage | None: def _result_handler() -> MessageHandler: - """Inbox `recognition.result`: `handle_message` с фабрикой сессий приложения (маппинг результата — отдельный цикл, + """Inbox `files.tabular_parsed`: `handle_message` с фабрикой сессий приложения (маппинг результата — отдельный цикл, Срез 2c design D-1).""" return partial(handle_message, session_factory=async_session_maker) async def _subscribe_results(nats: NatsConnection) -> None: - """Подписки после первого подключения к NATS: `recognition.result` в queue group `arm-core`.""" + """Подписки после первого подключения к NATS: `files.tabular_parsed` в queue group `arm-core`.""" await nats.subscribe_results(_result_handler()) diff --git a/api/src/contract/message.py b/api/src/contract/message.py index 828b109..6e07b12 100644 --- a/api/src/contract/message.py +++ b/api/src/contract/message.py @@ -1,4 +1,4 @@ -"""Конверт сообщения на `recognition.result`: `recognition-result-message.schema.json` v1.0. +"""Конверт сообщения на `files.tabular_parsed`: `recognition-result-message.schema.json` v1.0. `processing` — ack «взял в работу» (до чтения файла из S3); `done | needs_check | error` — итог. Условная обязательность — буквально из `allOf` схемы: для `done | needs_check` обязательны `result_s3` и diff --git a/api/src/contract/subjects.py b/api/src/contract/subjects.py index be58d01..243b11d 100644 --- a/api/src/contract/subjects.py +++ b/api/src/contract/subjects.py @@ -12,9 +12,9 @@ from typing import Literal, get_args Pipeline = Literal["tabular", "image"] NATS_MSG_ID_HEADER = "Nats-Msg-Id" -RESULT_SUBJECT = "recognition.result" -TABULAR_REQUEST_SUBJECT = "recognition.tabular.request" -IMAGE_REQUEST_SUBJECT = "recognition.image.request" +RESULT_SUBJECT = "files.tabular_parsed" +TABULAR_REQUEST_SUBJECT = "files.tabular_uploaded" +IMAGE_REQUEST_SUBJECT = "files.img_uploaded" REQUEST_SUBJECTS: Mapping[str, str] = { "tabular": TABULAR_REQUEST_SUBJECT, @@ -23,7 +23,7 @@ REQUEST_SUBJECTS: Mapping[str, str] = { def request_subject(pipeline: str) -> str: - """Subject запроса по pipeline: `tabular` → `recognition.tabular.request`, `image` → `recognition.image.request`.""" + """Subject запроса по pipeline: `tabular` → `files.tabular_uploaded`, `image` → `files.img_uploaded`.""" try: return REQUEST_SUBJECTS[pipeline] except KeyError: @@ -36,10 +36,13 @@ def request_msg_id(job_id: uuid.UUID | str, attempt: int) -> str: def ack_msg_id(job_id: uuid.UUID | str, attempt: int) -> str: - """`Nats-Msg-Id` ack (`status = processing`) на `recognition.result`: `ack::`.""" + """`Nats-Msg-Id` ack (`status = processing`) на `files.tabular_parsed`: `ack::`.""" return f"ack:{job_id}:{attempt}" def result_msg_id(job_id: uuid.UUID | str, attempt: int) -> str: - """`Nats-Msg-Id` результата (`done | needs_check | error`) на `recognition.result`: `result::`.""" + """`Nats-Msg-Id` результата (`done | needs_check | error`) на `files.tabular_parsed`. + + Форма: `result::`. + """ return f"result:{job_id}:{attempt}" diff --git a/api/src/core/nats.py b/api/src/core/nats.py index a798c0c..536f278 100644 --- a/api/src/core/nats.py +++ b/api/src/core/nats.py @@ -4,8 +4,8 @@ - Аутентификация как в шаблоне: заданы `NATS_SEED` и `NATS_JWT` — nkeys/JWT; оба пусты — без учётных данных. - `publish(subject, payload, msg_id)` — публикация с заголовком `Nats-Msg-Id` и `flush`: ошибка соединения проявляется сразу исключением `NatsPublishError` (outbox оставляет задание в `queued`). -- `subscribe_results(handler)` — подписка на `recognition.result` в queue group `arm-core`: из нескольких инстансов ядра - сообщение получает ровно один. Исключения обработчика логируются, подписка не падает. +- `subscribe_results(handler)` — подписка на `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 с @@ -95,7 +95,7 @@ class NatsConnection: `on_connected(conn)` вызывается один раз после первого успешного подключения (подписки); при исключении повторяется с той же экспоненциальной паузой. `result_subject` и `queue` параметризуются для тестов — - по умолчанию `recognition.result` и `arm-core`. + по умолчанию `files.tabular_parsed` и `arm-core`. """ def __init__( # noqa: PLR0913 - все параметры, кроме settings, именованные: подмены для тестов и тайминги diff --git a/api/src/services/recognition_exchange.py b/api/src/services/recognition_exchange.py index dec11f8..4754774 100644 --- a/api/src/services/recognition_exchange.py +++ b/api/src/services/recognition_exchange.py @@ -26,7 +26,7 @@ SKIP LOCKED` → публикация → `UPDATE` → `COMMIT`). Блокиро Обмен не пишет `audit_event` (рулинг R-7, design #20; ASSUMPTION(OQ-55)). -Inbox (задача 24, рулинги R-12, R-22): `handle_message()` — обработчик подписки `recognition.result` +Inbox (задача 24, рулинги R-12, R-22): `handle_message()` — обработчик подписки `files.tabular_parsed` (`src/core/nats.py`, queue group `arm-core`). Одно сообщение — одна короткая транзакция без сетевого IO: `SELECT … FOR UPDATE` строки задания (ЖДЁТ блокировку: ack, пришедший, пока публикатор держит строку до своего commit, применяется после него, а не теряется) → проверки статуса → изменения → `COMMIT`. @@ -495,7 +495,7 @@ class ExchangeTasks: ) -# --- inbox: recognition.result ---------------------------------------------------------------------------------- +# --- inbox: files.tabular_parsed ---------------------------------------------------------------------------------- # Финальные статусы для inbox (рулинг R-22, design #34): результат для них не применяется FINAL_STATUSES: frozenset[RecognitionStatus] = frozenset( @@ -739,7 +739,7 @@ async def _apply_message( job = await _lock_job(session, message.job_id) if job is None: logger.warning( - "Сообщение recognition.result для неизвестного задания, не применено. error_category=unknown_job, " + "Сообщение files.tabular_parsed для неизвестного задания, не применено. error_category=unknown_job, " "job_id={job_id}, status={status}", job_id=message.job_id, status=message.status.value, @@ -747,7 +747,7 @@ async def _apply_message( return if job.file_id != message.file_id: logger.warning( - "Сообщение recognition.result: file_id не совпадает с заданием, не применено. " + "Сообщение files.tabular_parsed: file_id не совпадает с заданием, не применено. " "error_category=file_mismatch, job_id={job_id}, file_id={file_id}, message_file_id={message_file_id}", job_id=job.id, file_id=job.file_id, @@ -759,7 +759,7 @@ async def _apply_message( if msg_id is None: msg_id = ack_msg_id(job.id, job.attempt) if is_ack else result_msg_id(job.id, job.attempt) logger.warning( - "Сообщение recognition.result без заголовка Nats-Msg-Id: использован производный id. " + "Сообщение files.tabular_parsed без заголовка Nats-Msg-Id: использован производный id. " "error_category=missing_msg_id, job_id={job_id}, msg_id={msg_id}", job_id=job.id, msg_id=msg_id, @@ -777,7 +777,7 @@ async def handle_message( session_factory: async_sessionmaker[AsyncSession], now: Clock | None = None, ) -> None: - """Обработчик `recognition.result` (spec «Ack сервиса переводит задание в `processing`», «Inbox принимает + """Обработчик `files.tabular_parsed` (spec «Ack сервиса переводит задание в `processing`», «Inbox принимает результат идемпотентно»; design D-6). Модель транзакции и правила — в docstring модуля. Исключения наружу не выходят: невалидное сообщение — warning; ошибка БД — откат транзакции (задание и файл), diff --git a/api/tests/e2e/test_osv_ingest_e2e.py b/api/tests/e2e/test_osv_ingest_e2e.py index e28a4ec..4e30262 100644 --- a/api/tests/e2e/test_osv_ingest_e2e.py +++ b/api/tests/e2e/test_osv_ingest_e2e.py @@ -6,7 +6,7 @@ tabular-parser (читает файл из MinIO, пишет результат `PROCESSED`, документ и строки ОСВ. Предусловие — профиль `services` (`tests/e2e/conftest.py`); иначе тест пропускается. Очистка: `DELETE /files/{id}` через API и удаление временной БД (фикстура `committed_sessionmaker`). -ВНИМАНИЕ: приложение подписывается на реальный `recognition.result` в боевой queue group `arm-core`; если в это время +ВНИМАНИЕ: приложение подписывается на реальный `files.tabular_parsed` в боевой queue group `arm-core`; если в это время работает другой экземпляр ядра (контейнер `arm-core` профиля `services`, локальный сервер с `NATS_ENABLED=true`), часть ack/результатов уйдёт ему и тест упадёт по таймауту. Перед прогоном остановите такие экземпляры (README). """ diff --git a/api/tests/factories.py b/api/tests/factories.py index 0fc5f18..7de3d6a 100644 --- a/api/tests/factories.py +++ b/api/tests/factories.py @@ -27,6 +27,7 @@ from typing import Any from sqlalchemy import delete, func, insert, select, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from src.contract import request_subject from src.contract.result import OsvDocument, RecognitionResult from src.domain.consolidation_periods import reporting_date from src.domain.osv_mapping import build_entries, document_fields @@ -235,7 +236,7 @@ async def make_stored_file( # noqa: PLR0913 - сигнатура фабрики job = RecognitionJob( file_id=file.id, pipeline=pipeline, - request_subject=f"recognition.{pipeline.value}.request", + request_subject=request_subject(pipeline.value), schema_version="1.0", status=job_status, ) @@ -535,7 +536,7 @@ async def make_osv_entry(session: AsyncSession, **overrides: Any) -> OsvEntry: job = RecognitionJob( file_id=file.id, pipeline=RecognitionPipeline.tabular, - request_subject="recognition.tabular.request", + request_subject="files.tabular_uploaded", schema_version="1", status=RecognitionStatus.queued, ) @@ -621,7 +622,7 @@ class CommittedJobs: job = RecognitionJob( file_id=file.id, pipeline=pipeline, - request_subject=f"recognition.{pipeline.value}.request", + request_subject=request_subject(pipeline.value), schema_version="1.0", status=status, attempt=attempt, @@ -692,7 +693,7 @@ class CommittedJobs: job = RecognitionJob( file_id=file_id, pipeline=pipeline, - request_subject=f"recognition.{pipeline.value}.request", + request_subject=request_subject(pipeline.value), schema_version="1.0", status=status, **job_fields, diff --git a/api/tests/test_api_files_upload.py b/api/tests/test_api_files_upload.py index 9d9d053..16534ef 100644 --- a/api/tests/test_api_files_upload.py +++ b/api/tests/test_api_files_upload.py @@ -248,7 +248,7 @@ async def test_hostile_name_gives_flat_key_and_full_records( assert file.uploaded_by == user_id assert file.uploaded_at is not None assert file.current_recognition_job_id == job_id - assert job.request_subject == "recognition.tabular.request" + assert job.request_subject == "files.tabular_uploaded" assert job.schema_version == "1.0" assert (job.status, job.attempt, job.max_attempts) == (RecognitionStatus.queued, 1, 3) assert job.requested_at is not None @@ -276,7 +276,7 @@ async def test_image_file_goes_to_image_pipeline_with_one_attempt( assert (job.pipeline, job.max_attempts, job.request_subject) == ( RecognitionPipeline.image, 1, - "recognition.image.request", + "files.img_uploaded", ) assert len(storage.objects) == 1 diff --git a/api/tests/test_app.py b/api/tests/test_app.py index 3436cc6..4120751 100644 --- a/api/tests/test_app.py +++ b/api/tests/test_app.py @@ -229,12 +229,12 @@ def test_create_nats_enabled_returns_connection_for_results(monkeypatch: pytest. monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True) nats = app_module._create_nats() assert isinstance(nats, NatsConnection) - assert (nats.result_subject, nats.queue) == ("recognition.result", "arm-core") + assert (nats.result_subject, nats.queue) == ("files.tabular_parsed", "arm-core") assert not nats.is_connected # создание не подключает: подключение — в start() async def test_result_subscription_uses_inbox_handler_with_dependencies(): - """Подписка `recognition.result` получает `handle_message` с фабрикой сессий приложения (без `mapper`: маппинг + """Подписка `files.tabular_parsed` получает `handle_message` с фабрикой сессий приложения (без `mapper`: маппинг результата — отдельный цикл, Срез 2c D-1).""" subscribed: list[Any] = [] diff --git a/api/tests/test_contract.py b/api/tests/test_contract.py index 2f408a1..cb732c0 100644 --- a/api/tests/test_contract.py +++ b/api/tests/test_contract.py @@ -620,9 +620,9 @@ def test_ack_and_result_msg_ids() -> None: def test_request_subject_by_pipeline() -> None: - assert request_subject("tabular") == "recognition.tabular.request" - assert request_subject("image") == "recognition.image.request" - assert RESULT_SUBJECT == "recognition.result" + assert request_subject("tabular") == "files.tabular_uploaded" + assert request_subject("image") == "files.img_uploaded" + assert RESULT_SUBJECT == "files.tabular_parsed" assert NATS_MSG_ID_HEADER == "Nats-Msg-Id" @@ -634,7 +634,8 @@ def test_request_subject_rejects_unknown_pipeline() -> None: def test_request_subject_matches_example_pipeline() -> None: for name in ("request-tabular.json", "request-image.json"): request = RecognitionRequest.model_validate(_example(name)) - assert request_subject(request.pipeline) == f"recognition.{request.pipeline}.request" + expected = {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"}[request.pipeline] + assert request_subject(request.pipeline) == expected def test_integer_valued_float_is_normalized_to_int() -> None: diff --git a/api/tests/test_exchange_lifespan_nats.py b/api/tests/test_exchange_lifespan_nats.py index eea74f7..ac97ea1 100644 --- a/api/tests/test_exchange_lifespan_nats.py +++ b/api/tests/test_exchange_lifespan_nats.py @@ -3,7 +3,7 @@ Интеграционные тесты против NATS и PostgreSQL из `.env.test` (БД сессии pytest; задания фиксируются `CommittedJobs` и удаляются в teardown). Subject результатов — уникальный на тест (`NatsConnection(result_subject=...)`): общий -`recognition.result` не трогаем, queue group — боевая `arm-core`. S3 не нужен (`_connect_s3` → None), SMB выключен +`files.tabular_parsed` не трогаем, queue group — боевая `arm-core`. S3 не нужен (`_connect_s3` → None), SMB выключен (`.env.test`), пул движка БД приложения подменён (lifespan закрывает его последним шагом). Ожидания — с условием и ограничением по времени, без «голых» пауз. """ diff --git a/api/tests/test_migrations.py b/api/tests/test_migrations.py index 5e26103..695a9eb 100644 --- a/api/tests/test_migrations.py +++ b/api/tests/test_migrations.py @@ -382,7 +382,7 @@ async def test_upgrade_0004_backfills_counterparty_match_kind(scratch_db_url: st await conn.execute( text( "INSERT INTO dkb.recognition_job (id, file_id, pipeline, request_subject, schema_version, status) " - "VALUES (:job, :file, 'tabular', 'recognition.tabular.request', '1', 'queued')" + "VALUES (:job, :file, 'tabular', 'files.tabular_uploaded', '1', 'queued')" ), ids, ) diff --git a/api/tests/test_models_part_04_06.py b/api/tests/test_models_part_04_06.py index e931f0b..99a029b 100644 --- a/api/tests/test_models_part_04_06.py +++ b/api/tests/test_models_part_04_06.py @@ -154,7 +154,7 @@ async def _make_file_with_job(session: AsyncSession) -> Chain: job = RecognitionJob( file_id=file.id, pipeline=RecognitionPipeline.tabular, - request_subject="recognition.tabular.request", + request_subject="files.tabular_uploaded", schema_version="1", status=RecognitionStatus.queued, ) diff --git a/api/tests/test_nats_core.py b/api/tests/test_nats_core.py index 176d242..4df623a 100644 --- a/api/tests/test_nats_core.py +++ b/api/tests/test_nats_core.py @@ -289,7 +289,7 @@ async def test_publish_without_connection_raises_publish_error(): conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld()) with pytest.raises(NatsPublishError, match="NATS"): - await conn.publish("recognition.tabular.request", b"{}", "job:1") + await conn.publish("files.tabular_uploaded", b"{}", "job:1") async def test_publish_sends_msg_id_header_and_flushes(): @@ -298,10 +298,10 @@ async def test_publish_sends_msg_id_header_and_flushes(): await conn.start() assert await conn.wait_ready(timeout=2) - await conn.publish("recognition.tabular.request", b'{"a": 1}', "job:1") + await conn.publish("files.tabular_uploaded", b'{"a": 1}', "job:1") assert world.clients[-1].published == [ - {"subject": "recognition.tabular.request", "payload": b'{"a": 1}', "headers": {NATS_MSG_ID_HEADER: "job:1"}} + {"subject": "files.tabular_uploaded", "payload": b'{"a": 1}', "headers": {NATS_MSG_ID_HEADER: "job:1"}} ] await conn.stop() @@ -314,7 +314,7 @@ async def test_publish_flush_failure_raises_publish_error(): world.clients[-1].flush_error = TimeoutError() with pytest.raises(NatsPublishError): - await conn.publish("recognition.tabular.request", b"{}", "job:1") + await conn.publish("files.tabular_uploaded", b"{}", "job:1") await conn.stop() diff --git a/api/tests/test_nats_integration.py b/api/tests/test_nats_integration.py index 0c224d3..e725d63 100644 --- a/api/tests/test_nats_integration.py +++ b/api/tests/test_nats_integration.py @@ -1,7 +1,7 @@ """Интеграционные тесты `src/core/nats.py` против NATS без JetStream из `.env.test` (spec recognition-job-exchange, «Соединение с NATS по шаблону, без JetStream»; design D-6). -Subject'ы — уникальные на тест (`test.`): общий `recognition.result` не трогаем, чтобы не пересекаться +Subject'ы — уникальные на тест (`test.`): общий `files.tabular_parsed` не трогаем, чтобы не пересекаться с другими запущенными инстансами ядра. """ diff --git a/api/tests/test_recognition_exchange.py b/api/tests/test_recognition_exchange.py index b61be01..f777ca5 100644 --- a/api/tests/test_recognition_exchange.py +++ b/api/tests/test_recognition_exchange.py @@ -5,7 +5,7 @@ Интеграционные тесты работают против реального PostgreSQL (БД сессии pytest) и NATS из `.env.test`. Публикатор и вахтёр ведут собственные транзакции, поэтому данные фиксируются реально (`CommittedJobs`) и удаляются в teardown. -Subject'ы запросов общие (`recognition.*.request`): тестовый подписчик фильтрует сообщения по `job_id` своих заданий. +Subject'ы запросов общие (`files.*_uploaded`): тестовый подписчик фильтрует сообщения по `job_id` своих заданий. Время в тестах повторов — инжектируемые часы `now`, без реальных ожиданий таймаутов. """ @@ -46,7 +46,7 @@ from tests.factories import CommittedJobs CONTRACTS_DIR = Path(__file__).resolve().parents[1] / "contracts" REQUEST_SCHEMA = json.loads((CONTRACTS_DIR / "recognition-request.schema.json").read_text(encoding="utf-8")) REQUEST_VALIDATOR = Draft202012Validator(REQUEST_SCHEMA, format_checker=FormatChecker()) -REQUEST_SUBJECTS = ("recognition.tabular.request", "recognition.image.request") +REQUEST_SUBJECTS = ("files.tabular_uploaded", "files.img_uploaded") ACK = timedelta(seconds=30) MAX_REPUBLISH = 3 @@ -210,8 +210,8 @@ def test_watchdog_period_is_30_seconds(): @pytest.mark.parametrize( ("pipeline", "file_type", "subject"), [ - (RecognitionPipeline.tabular, FileType.XLSX, "recognition.tabular.request"), - (RecognitionPipeline.image, FileType.PDF, "recognition.image.request"), + (RecognitionPipeline.tabular, FileType.XLSX, "files.tabular_uploaded"), + (RecognitionPipeline.image, FileType.PDF, "files.img_uploaded"), ], ) async def test_publish_request_to_service( # noqa: PLR0913,PLR0917 - фикстуры и параметры diff --git a/api/tests/test_recognition_inbox.py b/api/tests/test_recognition_inbox.py index 5a1bec3..c578302 100644 --- a/api/tests/test_recognition_inbox.py +++ b/api/tests/test_recognition_inbox.py @@ -7,7 +7,7 @@ обращения к БД, ошибка БД не выходит наружу, шаблонный `nats_callback` удалён. Интеграционные — `handle_message` с фейковым Msg (`SimpleNamespace(data, headers, subject)`) на реальном PostgreSQL (БД сессии pytest; данные фиксируются `CommittedJobs` и удаляются в teardown) и полный цикл через реальный NATS -(уникальный result-subject: общий `recognition.result` не трогаем). +(уникальный result-subject: общий `files.tabular_parsed` не трогаем). """ import asyncio @@ -104,7 +104,7 @@ def fake_msg(body: Any, msg_id: str | None, *, header: str = NATS_MSG_ID_HEADER, MESSAGE_VALIDATOR.validate(body) data = body if isinstance(body, bytes) else json.dumps(body, ensure_ascii=False).encode("utf-8") headers = None if msg_id is None else {header: msg_id} - return SimpleNamespace(data=data, headers=headers, subject="recognition.result") + return SimpleNamespace(data=data, headers=headers, subject="files.tabular_parsed") class SlowFileStatus: @@ -232,7 +232,7 @@ _FILE = uuid.UUID("22222222-2222-4222-8222-222222222222") ], ) async def test_invalid_message_is_logged_and_does_not_touch_db(data: bytes, logs: list[Any]): - msg = SimpleNamespace(data=data, headers={NATS_MSG_ID_HEADER: "result:x:1"}, subject="recognition.result") + msg = SimpleNamespace(data=data, headers={NATS_MSG_ID_HEADER: "result:x:1"}, subject="files.tabular_parsed") await handle_message(msg, session_factory=_UntouchableFactory()) # type: ignore[arg-type] rejected = _logged(logs, "отклонено", "WARNING") assert len(rejected) == 1 @@ -939,7 +939,7 @@ class _FakeService: if json.loads(msg.data).get("job_id") == str(self.job_id): self.requests.append(msg) - await self._nc.subscribe("recognition.tabular.request", cb=collect) + await self._nc.subscribe("files.tabular_uploaded", cb=collect) await self._nc.flush() async def send(self, body: dict[str, Any], msg_id: str) -> None: diff --git a/api/tests/test_result_mapping.py b/api/tests/test_result_mapping.py index 973f7ab..48ac9bb 100644 --- a/api/tests/test_result_mapping.py +++ b/api/tests/test_result_mapping.py @@ -687,7 +687,7 @@ async def test_result_of_non_current_job_is_ignored(committed_sessionmaker: asyn new_job = RecognitionJob( file_id=file_id, pipeline=(await session.get(RecognitionJob, old_job_id)).pipeline, # type: ignore[union-attr] - request_subject="recognition.tabular.request", + request_subject="files.tabular_uploaded", schema_version="1.0", status=RecognitionStatus.processing, attempt=2, diff --git a/openspec/changes/nats-files-subjects/.openspec.yaml b/openspec/changes/nats-files-subjects/.openspec.yaml new file mode 100644 index 0000000..01660ba --- /dev/null +++ b/openspec/changes/nats-files-subjects/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-10-08 diff --git a/openspec/changes/nats-files-subjects/design.md b/openspec/changes/nats-files-subjects/design.md new file mode 100644 index 0000000..63d5945 --- /dev/null +++ b/openspec/changes/nats-files-subjects/design.md @@ -0,0 +1,14 @@ +## Context + +Переименование subject'ов по решению владельца 08.10.2026; схема обмена (два subject'а запроса по `pipeline`, один +общий subject ответа) не меняется. Подробности и риски — design сквозного change'а `nats-files-subjects` в docs. + +## Decisions + +Решения R-1…R-5 приняты в сквозном change'е (docs, `openspec/changes/nats-files-subjects/design.md`) и здесь не +дублируются. + +## Decisions made without discussion + +Своих решений у этого change'а нет: маршрут (R-1), версия контракта (R-2), DBML (R-3), константы вместо настроек (R-4), +историю не переписываем (R-5) — см. design docs. diff --git a/openspec/changes/nats-files-subjects/proposal.md b/openspec/changes/nats-files-subjects/proposal.md new file mode 100644 index 0000000..0aae6e2 --- /dev/null +++ b/openspec/changes/nats-files-subjects/proposal.md @@ -0,0 +1,33 @@ +## Why + +Решение владельца 08.10.2026: subject'ы NATS — «правильно как в старом АРМ». На проде заведены `files.tabular_uploaded`, +`files.img_uploaded`, `files.tabular_parsed`, `files.img_parsed`; используем три, `files.img_parsed` не используем, +старый АРМ гасится. Сквозной change — `nats-files-subjects` в RSHB/arm-ka-docs (issue #74, design с решениями R-1…R-5). +Issue этого репозитория — RSHB/arm-core#77. + +## What Changes + +- `recognition.tabular.request` → `files.tabular_uploaded`, `recognition.image.request` → `files.img_uploaded`, + `recognition.result` → `files.tabular_parsed` (константы `api/src/contract/subjects.py`, докстринги, тесты, README, + `CLAUDE.md`). +- После изменения ядро публикует запросы в `files.tabular_uploaded` / `files.img_uploaded` (outbox) и подписано на `files.tabular_parsed` в queue group `arm-core` (inbox). +- Копии схем `api/contracts/recognition-request.schema.json` и `recognition-result-message.schema.json` — побайтно из + `docs/contracts/` (изменены только описания subject'ов). +- `recognition_job.request_subject` новых заданий получает новые имена; уже созданные задания сохраняют старое значение (поле справочное: при повторной публикации subject вычисляется по `pipeline`). +- Тестовая фабрика `tests/factories.py` берёт subject из `request_subject(pipeline)`, а не собирает строку сама. + +## Non-goals + +- Тела сообщений, `Nats-Msg-Id`, queue group'ы и `schema_version` не меняются. +- Subject'ы не выносятся в настройки (OQ-83). + +## Capabilities + +### New Capabilities + +### Modified Capabilities +- `recognition-job-exchange`: имена subject'ов запроса и ответа. + +## Impact + +Выкатывается одновременно с остальными сервисами: сервис со старыми именами не увидит сообщений с новыми. diff --git a/openspec/changes/nats-files-subjects/specs/recognition-job-exchange/spec.md b/openspec/changes/nats-files-subjects/specs/recognition-job-exchange/spec.md new file mode 100644 index 0000000..1b5a650 --- /dev/null +++ b/openspec/changes/nats-files-subjects/specs/recognition-job-exchange/spec.md @@ -0,0 +1,44 @@ +## MODIFIED Requirements + +### Requirement: Outbox публикует задания из `queued` +Фоновый публикатор SHALL с периодом `OUTBOX_POLL_SECONDS` (по умолчанию 5) выбирать задания `status = queued` +(с блокировкой `FOR UPDATE SKIP LOCKED`), формировать сообщение по `recognition-request.schema.json` v1.0 из +`recognition_job ⋈ stored_file ⋈ s3_object` и публиковать в subject по `pipeline` с заголовком +`Nats-Msg-Id = :`. После успешной публикации задание SHALL получить `status = published`, +`published_at`, `request_msg_id`, `request_subject`; при ошибке публикации задание SHALL остаться `queued`. + +#### Scenario: Публикация запроса парсеру +- **WHEN** в `recognition_job` вставлена строка `queued` с `pipeline = tabular`, связанная со `stored_file` (XLSX) и его `s3_object` +- **THEN** в течение 10 с в `files.tabular_uploaded` появляется сообщение, проходящее схему запроса, с `job_id`, + `file_id`, `s3.key` исходника, `reply_deadline`, `attempt = 1`, а строка переходит в `published` с заполненным `request_msg_id` + +#### Scenario: NATS недоступен +- **WHEN** публикация завершается ошибкой соединения +- **THEN** задание остаётся `queued`, ошибка залогирована, следующий цикл повторяет публикацию + +### Requirement: Ack сервиса переводит задание в `processing` +Получив в `files.tabular_parsed` конверт со `status = processing` для задания в `published`, ядро SHALL установить +`status = processing`, `started_at`, `service_name`, `service_version` и перевести файл задания в +`stored_file.status = PROCESSING`, если `current_recognition_job_id` файла равен этому заданию. Повторный ack для +задания в `processing` SHALL игнорироваться; ack для задания в другом статусе SHALL логироваться и не менять задание. + +#### Scenario: Сервис подтвердил +- **WHEN** сервис опубликовал конверт `processing` с `job_id` опубликованного задания +- **THEN** задание имеет `status = processing`, `started_at` заполнен, `service_name = tabular-parser`, файл — `PROCESSING` + +### Requirement: Соединение с NATS по шаблону, без JetStream +Ядро SHALL подключаться к NATS с nkeys-seed и JWT, если заданы `OPENBAO__NATS__NATS_SEED/NATS_JWT`, и без учётных данных, +если они пусты; подписываться на `files.tabular_parsed` в queue group `arm-core`; переподключаться автоматически. +Stream'ы и consumer'ы JetStream MUST NOT требоваться. Фоновые задачи SHALL запускаться при `NATS_ENABLED=true`. + +#### Scenario: Dev-окружение compose +- **WHEN** ядро стартует с `DEBUG=true`, `NATS_ENABLED=true` и пустыми учётными данными против NATS без JetStream +- **THEN** подписка активна, `/healthcheck` отдаёт `nats = ok`, в логах нет ошибок подключения + +#### Scenario: Два инстанса ядра +- **WHEN** запущены два инстанса с `NATS_ENABLED=true` и опубликован один результат +- **THEN** результат обработан ровно одним инстансом (queue group), строка задания изменена один раз + +#### Scenario: NATS выключен +- **WHEN** `NATS_ENABLED=false` +- **THEN** приложение стартует, фоновые задачи не запускаются, `/healthcheck` отвечает 200 с `nats = off` diff --git a/openspec/changes/nats-files-subjects/tasks.md b/openspec/changes/nats-files-subjects/tasks.md new file mode 100644 index 0000000..788fdc5 --- /dev/null +++ b/openspec/changes/nats-files-subjects/tasks.md @@ -0,0 +1,11 @@ +## 1. Переименование + +- [x] 1.1 `api/src/contract/subjects.py`: `files.tabular_uploaded`, `files.img_uploaded`, `files.tabular_parsed` +- [x] 1.2 Докстринги, логи, тесты, README, `CLAUDE.md` — новые имена +- [x] 1.3 Копии схем `api/contracts/` побайтно из `docs/contracts/` +- [x] 1.4 Дельта спеки `recognition-job-exchange` + +## 2. Проверка + +- [x] 2.1 `make check` +- [x] 2.2 Интеграционные тесты NATS