Повторная публикация обновляет request_subject; тексты без умолчаний как фактических subject'ов (I-1, M-1, M-3, M-5, #106)
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 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
c14d5e3ffe
commit
c9d60907d3
@ -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` /
|
||||
|
||||
@ -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
|
||||
```
|
||||
|
||||
@ -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())
|
||||
|
||||
|
||||
|
||||
@ -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, именованные: подмены для тестов и тайминги
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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],
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user