From c9d60907d3383e300aa89fea81897ccf1efda207 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Tue, 13 Oct 2026 13:28:07 +0400 Subject: [PATCH] =?UTF-8?q?=D0=9F=D0=BE=D0=B2=D1=82=D0=BE=D1=80=D0=BD?= =?UTF-8?q?=D0=B0=D1=8F=20=D0=BF=D1=83=D0=B1=D0=BB=D0=B8=D0=BA=D0=B0=D1=86?= =?UTF-8?q?=D0=B8=D1=8F=20=D0=BE=D0=B1=D0=BD=D0=BE=D0=B2=D0=BB=D1=8F=D0=B5?= =?UTF-8?q?=D1=82=20request=5Fsubject;=20=D1=82=D0=B5=D0=BA=D1=81=D1=82?= =?UTF-8?q?=D1=8B=20=D0=B1=D0=B5=D0=B7=20=D1=83=D0=BC=D0=BE=D0=BB=D1=87?= =?UTF-8?q?=D0=B0=D0=BD=D0=B8=D0=B9=20=D0=BA=D0=B0=D0=BA=20=D1=84=D0=B0?= =?UTF-8?q?=D0=BA=D1=82=D0=B8=D1=87=D0=B5=D1=81=D0=BA=D0=B8=D1=85=20subjec?= =?UTF-8?q?t'=D0=BE=D0=B2=20(I-1,=20M-1,=20M-3,=20M-5,=20#106)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit I-1: ветка повторной публикации в _process_one записывает request_subject (как первая публикация); тест test_republish_updates_request_subject. M-1: примеры /healthcheck в README (ядро: три subject'а; tabular-parser: два). M-3: docstring'и и логи не выдают умолчание за фактический subject. M-5: CLAUDE.md — subject'ы задаются OPENBAO__NATS__NATS_*_SUBJECT. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 2 +- README.md | 4 +-- api/src/app.py | 17 ++++++----- api/src/core/nats.py | 2 +- api/src/services/recognition_exchange.py | 16 +++++----- api/tests/test_recognition_exchange.py | 38 ++++++++++++++++++++++++ 6 files changed, 61 insertions(+), 18 deletions(-) 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],