Обмен NATS: лог file_mismatch без литерала subject'а, переносы docstring'ов (RM-1, RM-2)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
c9d60907d3
commit
eaa563c10f
@ -1,13 +1,12 @@
|
||||
"""Точка входа FastAPI. Каркас из шаблона команды: инфраструктура без доменного кода.
|
||||
|
||||
Что здесь есть: настройки, подключение S3 (MinIO; `MINIO_SECURE` — https или http), NATS (`src/core/nats.py`: core
|
||||
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). Приложение собирает
|
||||
`create_app(settings)`.
|
||||
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).
|
||||
Приложение собирает `create_app(settings)`.
|
||||
Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает
|
||||
состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор,
|
||||
вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) и, если S3 подключено, цикл маппинга результатов
|
||||
|
||||
@ -26,11 +26,10 @@ SKIP LOCKED` → публикация → `UPDATE` → `COMMIT`). Блокиро
|
||||
|
||||
Обмен не пишет `audit_event` (рулинг R-7, design #20; ASSUMPTION(OQ-55)).
|
||||
|
||||
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`.
|
||||
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`.
|
||||
- ack (`status = processing`): `published` → `processing`, `started_at`, `service_*`, файл `UPLOADED → PROCESSING`
|
||||
(R-5: только из `UPLOADED`); повторный ack для `processing` — тихо игнорируется; ack для другого статуса — warning,
|
||||
задание и файл не меняются.
|
||||
@ -761,7 +760,7 @@ async def _apply_message(
|
||||
return
|
||||
if job.file_id != message.file_id:
|
||||
logger.warning(
|
||||
"Сообщение files.tabular_parsed: file_id не совпадает с заданием, не применено. "
|
||||
"Сообщение на subject ответа: 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,
|
||||
@ -791,9 +790,9 @@ async def handle_message(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
now: Clock | None = None,
|
||||
) -> None:
|
||||
"""Обработчик subject ответов (по умолчанию
|
||||
`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,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user