diff --git a/CLAUDE.md b/CLAUDE.md index 53ee62e..7fc949f 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` на `files.tabular_parsed`. Обмен — `services/recognition_exchange.py`: `publish_pending` + подписка queue group `arm-core` на `files.tabular_parsed` (по умолчанию; задаётся OPENBAO__NATS__NATS_*_SUBJECT). Обмен — `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 27c6621..adfc40e 100644 --- a/README.md +++ b/README.md @@ -129,7 +129,7 @@ curl -s localhost:8000/healthcheck ``` ```json -{"uptime":"2026-10-03T18:37:15.121394+00:00","version":"1.0.0","nats":"ok"} +{"uptime":"2026-10-03T18:37:15.121394+00:00","version":"1.0.0","nats":"ok","subjects":{"tabular_request":"files.tabular_uploaded","image_request":"files.img_uploaded","result":"files.tabular_parsed"}} ``` `nats`: `ok` — подключён, `down` — включён (`NATS_ENABLED=true`), но связи нет, `off` — выключен. @@ -1004,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", "subjects": {"tabular_request": "...", "image_request": "...", "result": "..."}} +curl -s localhost:8001/healthcheck # {"uptime": "...", "version": "...", "nats": "ok", "subjects": {"tabular_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/src/app.py b/api/src/app.py index 4a53798..8cb2c96 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -1,8 +1,9 @@ """Точка входа FastAPI. Каркас из шаблона команды: инфраструктура без доменного кода. Что здесь есть: настройки, подключение S3 (MinIO; `MINIO_SECURE` — https или http), NATS (`src/core/nats.py`: core -NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка `files.tabular_parsed` -в queue group `arm-core`), SMB (горячие папки ИС Сканирование; необязателен — при пустом `SMB_HOST` не подключается), +NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка на subject ответов +(по умолчанию `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, subjects), раздача собранного SPA. Документация API (`/docs`, `/redoc`, `/openapi.json`) — только при DEBUG (OQ-66). Приложение собирает @@ -11,8 +12,9 @@ SPA. Документация API (`/docs`, `/redoc`, `/openapi.json`) — то состояние. При 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. Сообщения -`files.tabular_parsed` (ack и результаты сервисов) обрабатывает inbox `handle_message` того же модуля с фабрикой сессий -приложения (inbox меняет задание и статус файла; маппинг результата в доменные таблицы — отдельный цикл, Срез 2c D-1). +subject ответов (по умолчанию `files.tabular_parsed`; ack и результаты сервисов) обрабатывает inbox +`handle_message` того же модуля с фабрикой сессий приложения (inbox меняет задание и статус файла; маппинг +результата в доменные таблицы — отдельный цикл, Срез 2c D-1). Остановка: фоновые задачи обмена → NATS → S3 → SMB → пул соединений БД; каждый шаг независим — ошибка одного логируется и не отменяет остальные. Логгер настраивает `setup_logger` при импорте модуля: без значений локальных переменных в трейсбеках (#63). @@ -110,13 +112,14 @@ async def _connect_s3() -> S3Storage | None: def _result_handler() -> MessageHandler: - """Inbox `files.tabular_parsed`: `handle_message` с фабрикой сессий приложения (маппинг результата — отдельный цикл, - Срез 2c design D-1).""" + """Inbox subject ответов (по умолчанию `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: `files.tabular_parsed` в queue group `arm-core`.""" + """Подписки после первого подключения к NATS: subject ответов (по умолчанию `files.tabular_parsed`) в queue + group `arm-core`.""" await nats.subscribe_results(_result_handler()) diff --git a/api/src/core/nats.py b/api/src/core/nats.py index 969c3a1..71c8833 100644 --- a/api/src/core/nats.py +++ b/api/src/core/nats.py @@ -105,7 +105,7 @@ class NatsConnection: `on_connected(conn)` вызывается один раз после первого успешного подключения (подписки); при исключении повторяется с той же экспоненциальной паузой. `result_subject` и `queue` параметризуются для тестов — - по умолчанию `files.tabular_parsed` и `arm-core`. + по умолчанию `files.tabular_parsed` (фактический subject задаётся настройками окружения) и `arm-core`. """ def __init__( # noqa: PLR0913 - все параметры, кроме settings, именованные: подмены для тестов и тайминги diff --git a/api/src/services/recognition_exchange.py b/api/src/services/recognition_exchange.py index 466af43..4e8ab30 100644 --- a/api/src/services/recognition_exchange.py +++ b/api/src/services/recognition_exchange.py @@ -26,7 +26,8 @@ SKIP LOCKED` → публикация → `UPDATE` → `COMMIT`). Блокиро Обмен не пишет `audit_event` (рулинг R-7, design #20; ASSUMPTION(OQ-55)). -Inbox (задача 24, рулинги R-12, R-22): `handle_message()` — обработчик подписки `files.tabular_parsed` +Inbox (задача 24, рулинги R-12, R-22): `handle_message()` — обработчик подписки на subject ответов +(по умолчанию `files.tabular_parsed`) (`src/core/nats.py`, queue group `arm-core`). Одно сообщение — одна короткая транзакция без сетевого IO: `SELECT … FOR UPDATE` строки задания (ЖДЁТ блокировку: ack, пришедший, пока публикатор держит строку до своего commit, применяется после него, а не теряется) → проверки статуса → изменения → `COMMIT`. @@ -299,7 +300,7 @@ async def _process_one( # noqa: PLR0913 - один шаг прохода: вс metrics = base.op("||", return_type=JSONB)(literal({REPUBLISH_COUNT_KEY: count}, JSONB)) await session.execute( _job_update(row.id, RecognitionStatus.published).values( - published_at=now, reply_deadline=reply_deadline, metrics=metrics + published_at=now, reply_deadline=reply_deadline, metrics=metrics, request_subject=subject ) ) report.republished += 1 @@ -508,7 +509,7 @@ class ExchangeTasks: ) -# --- inbox: files.tabular_parsed ---------------------------------------------------------------------------------- +# --- inbox: subject ответов ---------------------------------------------------------------------------------- # Финальные статусы для inbox (рулинг R-22, design #34): результат для них не применяется FINAL_STATUSES: frozenset[RecognitionStatus] = frozenset( @@ -752,7 +753,7 @@ async def _apply_message( job = await _lock_job(session, message.job_id) if job is None: logger.warning( - "Сообщение files.tabular_parsed для неизвестного задания, не применено. error_category=unknown_job, " + "Сообщение на subject ответа для неизвестного задания, не применено. error_category=unknown_job, " "job_id={job_id}, status={status}", job_id=message.job_id, status=message.status.value, @@ -772,7 +773,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( - "Сообщение files.tabular_parsed без заголовка Nats-Msg-Id: использован производный id. " + "Сообщение на subject ответа без заголовка 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, @@ -790,8 +791,9 @@ async def handle_message( session_factory: async_sessionmaker[AsyncSession], now: Clock | None = None, ) -> None: - """Обработчик `files.tabular_parsed` (spec «Ack сервиса переводит задание в `processing`», «Inbox принимает - результат идемпотентно»; design D-6). Модель транзакции и правила — в docstring модуля. + """Обработчик subject ответов (по умолчанию + `files.tabular_parsed`; spec «Ack сервиса переводит задание в `processing`», «Inbox принимает результат + идемпотентно»; design D-6). Модель транзакции и правила — в docstring модуля. Исключения наружу не выходят: невалидное сообщение — warning; ошибка БД — откат транзакции (задание и файл), `logger.exception`, без повторов (задание дойдёт до `timeout`). `audit_event` не пишется (R-7, diff --git a/api/tests/test_recognition_exchange.py b/api/tests/test_recognition_exchange.py index 8a6e216..9892882 100644 --- a/api/tests/test_recognition_exchange.py +++ b/api/tests/test_recognition_exchange.py @@ -285,6 +285,44 @@ async def test_publish_uses_configured_request_subject( assert (await jobs.get(job_id)).request_subject == subjects["tabular"] +@pytest.mark.integration +async def test_republish_updates_request_subject( + session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection +) -> None: + """I-1: повторная публикация уходит на актуальный subject и обновляет `request_subject` задания.""" + run = uuid.uuid4().hex[:8] + subjects = {"tabular": f"test-{run}.files.tabular_uploaded", "image": f"test-{run}.files.img_uploaded"} + clock = Clock() + job_id = await jobs.make_queued_job( + pipeline=RecognitionPipeline.tabular, + file_type=FileType.XLSX, + original_name="ОСВ 61.xlsx", + reply_deadline=clock() + timedelta(hours=1), + ) + assert (await _publish(session_factory, nats, clock)).published == 1 + old_subject = (await jobs.get(job_id)).request_subject + assert old_subject != subjects["tabular"] + + custom = RequestSubscriber(subjects=tuple(subjects.values())) + await custom.start() + try: + clock.advance(ACK + timedelta(seconds=1)) + report = await publish_pending( + session_factory, + nats, + ack_timeout=ACK, + max_republish=MAX_REPUBLISH, + now=clock, + request_subjects=subjects, + ) + assert report.republished == 1 + (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],