5ae8ff8a81
Решение владельца 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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
802 lines
40 KiB
Python
802 lines
40 KiB
Python
"""Обмен с сервисами распознавания через `recognition_job` (design D-6; spec recognition-job-exchange;
|
||
docs/context/03-architecture.md, «Контракт обмена через NATS»).
|
||
|
||
Outbox (задача 23):
|
||
- `publish_pending()` — один проход публикатора: `queued` → первая публикация → `published`; `published` без ack дольше
|
||
`ACK_TIMEOUT_SECONDS` → повторная публикация с тем же `Nats-Msg-Id` и тем же `attempt` (счётчик —
|
||
`metrics.republish_count`); после `OUTBOX_MAX_REPUBLISH` повторов → `dlq`, `error_code = NO_ACK`, файл текущего
|
||
задания → `ERROR`. `attempt` не растёт: обработка не начиналась. Replay (`attempt + 1`) — вне Среза 2c.
|
||
|
||
Статусы `stored_file` (Срез 2c; spec primary-files «Жизненный цикл статуса файла», design D-5, рулинги R-4, R-5) —
|
||
только через `services/file_status.py` и только для текущего задания файла (`current_recognition_job_id = job.id`).
|
||
- `expire_deadlines()` — вахтёр: `processing`/`published` с `reply_deadline < now` → `timeout`, `error_code = TIMEOUT`;
|
||
файлы этих заданий (по `current_recognition_job_id`) → `ERROR` в той же транзакции.
|
||
- `run_publisher_loop()` / `run_watchdog_loop()` — фоновые циклы; `ExchangeTasks` запускает их в lifespan при
|
||
`NATS_ENABLED=true` и останавливает до закрытия NATS.
|
||
|
||
Гарантия доставки — at-least-once за счёт повторов без ack, а не за счёт `flush`: успешный `NatsConnection.publish`
|
||
означает лишь, что сервер ответил на PING после отправки, но не что принял именно это сообщение (ошибка прав, например,
|
||
приходит асинхронно в `error_cb`). Сбой commit после успешной публикации тоже даёт повтор того же сообщения. Поэтому
|
||
сервисы распознавания идемпотентны по `job_id` + `attempt` (`Nats-Msg-Id`).
|
||
|
||
Модель транзакций публикатора: одно задание — одна короткая транзакция (`SELECT … LIMIT 1 FOR UPDATE OF recognition_job
|
||
SKIP LOCKED` → публикация → `UPDATE` → `COMMIT`). Блокировка строки держится на время одной публикации, а не пачки;
|
||
параллельный проход (второй инстанс ядра) пропускает заблокированную строку и не публикует её повторно, а после commit
|
||
строка уже не подходит под условие выборки.
|
||
|
||
Обмен не пишет `audit_event` (рулинг R-7, design #20; ASSUMPTION(OQ-55)).
|
||
|
||
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`.
|
||
- ack (`status = processing`): `published` → `processing`, `started_at`, `service_*`, файл `UPLOADED → PROCESSING`
|
||
(R-5: только из `UPLOADED`); повторный ack для `processing` — тихо игнорируется; ack для другого статуса — warning,
|
||
задание и файл не меняются.
|
||
- результат (`done | needs_check | error`): дубликат определяется ТОЛЬКО по `recognition_job.result_msg_id ==
|
||
Nats-Msg-Id` (at-least-once) — без изменений; задание в финальном статусе (`FINAL_STATUSES`) — warning, без изменений;
|
||
иначе (`queued | published | processing`: потерянный ack не блокирует результат) — регистрация файла результата в
|
||
`s3_object` (`bucket_role = results`), обновление задания, статус файла и commit. `error` → файл `ERROR`;
|
||
`done`/`needs_check` → файл `UPLOADED → PROCESSING` (R-4: ack мог потеряться — core NATS без JetStream), из
|
||
`PROCESSING` файл не меняется и ждёт цикла маппинга (design D-1). Inbox НЕ читает файл результата и не создаёт
|
||
документы: `ResultMapper` удалён (OQ-67), маппинг — отдельный цикл `services/result_mapping.py`.
|
||
- Гонка с вахтёром (`expire_deadlines` — один `UPDATE` по статусу) решается блокировкой строки: кто первым взял
|
||
строку, тот и задаёт финальный статус; второй видит финальный статус и ничего не меняет.
|
||
- Невалидное сообщение, неизвестный `job_id`, ошибка БД — лог без повторов; исключения наружу не выходят.
|
||
Задание без применённого результата дойдёт до `timeout` (осознанный компромисс D-6).
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import uuid
|
||
from collections.abc import Awaitable, Callable, Mapping, Sequence
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any, Protocol
|
||
|
||
from loguru import logger
|
||
from pydantic import ValidationError
|
||
from sqlalchemy import and_, case, func, literal, or_, select, update
|
||
from sqlalchemy.dialects.postgresql import JSONB
|
||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||
|
||
from src.contract import (
|
||
CONTRACT_VERSION,
|
||
NATS_MSG_ID_HEADER,
|
||
MessageStatus,
|
||
Metrics,
|
||
RecognitionRequest,
|
||
RecognitionResultMessage,
|
||
S3Ref,
|
||
ack_msg_id,
|
||
request_msg_id,
|
||
request_subject,
|
||
result_msg_id,
|
||
)
|
||
from src.core.config import Settings
|
||
from src.core.nats import NatsPublishError
|
||
from src.core.storage import ObjectStorage
|
||
from src.models.enums import FileStatus, RecognitionPipeline, RecognitionStatus, S3BucketRole
|
||
from src.models.files import RecognitionJob, S3Object, StoredFile
|
||
from src.services.file_status import set_file_status_for_job, set_file_status_for_jobs
|
||
from src.services.result_mapping import run_mapping_loop
|
||
|
||
WATCHDOG_PERIOD_SECONDS = 30 # период вахтёра сроков (spec «Таймаут ожидания результата»)
|
||
PUBLISH_BATCH_SIZE = 50 # заданий за один проход публикатора (design D-6, LIMIT 50)
|
||
# Срок ответа сервиса, если задание создано без `reply_deadline`: OCR — 5 мин (DBML note), парсер — так же
|
||
DEFAULT_REPLY_TIMEOUT: dict[RecognitionPipeline, timedelta] = {
|
||
RecognitionPipeline.tabular: timedelta(minutes=5),
|
||
RecognitionPipeline.image: timedelta(minutes=5),
|
||
}
|
||
REPUBLISH_COUNT_KEY = "republish_count" # счётчик повторов в `recognition_job.metrics` (design #6)
|
||
|
||
ERROR_NO_ACK = "NO_ACK"
|
||
ERROR_TIMEOUT = "TIMEOUT"
|
||
ERROR_INVALID_REQUEST = "INVALID_REQUEST"
|
||
ERROR_MESSAGE_MAX_LENGTH = 2000
|
||
|
||
SleepFn = Callable[[float], Awaitable[None]]
|
||
Clock = Callable[[], datetime]
|
||
|
||
|
||
class RequestPublisher(Protocol):
|
||
"""То, что публикатору нужно от соединения NATS (`src.core.nats.NatsConnection`)."""
|
||
|
||
@property
|
||
def is_connected(self) -> bool: ...
|
||
|
||
async def publish(self, subject: str, payload: bytes, msg_id: str) -> None: ...
|
||
|
||
|
||
@dataclass
|
||
class PublishReport:
|
||
"""Итог одного прохода публикатора."""
|
||
|
||
published: int = 0 # первая публикация: queued → published
|
||
republished: int = 0 # повторная публикация без ack
|
||
dlq: int = 0 # повторы исчерпаны: published → dlq
|
||
invalid: int = 0 # запрос не собирается по контракту: → error / INVALID_REQUEST
|
||
failed: int = 0 # ошибка публикации (проход прерван)
|
||
skipped: bool = False # нет соединения с NATS: проход не выполнялся
|
||
|
||
@property
|
||
def total(self) -> int:
|
||
"""Сколько заданий изменено за проход."""
|
||
return self.published + self.republished + self.dlq + self.invalid
|
||
|
||
|
||
def _utcnow() -> datetime:
|
||
return datetime.now(UTC)
|
||
|
||
|
||
def _republish_count(metrics: Any) -> int:
|
||
"""`metrics.republish_count` (нет поля / `metrics` NULL / не число — 0)."""
|
||
if isinstance(metrics, dict):
|
||
value = metrics.get(REPUBLISH_COUNT_KEY)
|
||
if isinstance(value, int) and not isinstance(value, bool):
|
||
return value
|
||
return 0
|
||
|
||
|
||
def _candidate_stmt(now: datetime, ack_cutoff: datetime) -> Any:
|
||
"""Следующее задание прохода: `queued` или `published` без ack дольше таймаута (срок ответа ещё не прошёл —
|
||
иначе задание за вахтёром). Строка задания блокируется, занятые другим проходом пропускаются."""
|
||
job = RecognitionJob
|
||
awaiting_ack = and_(
|
||
job.status == RecognitionStatus.published,
|
||
or_(job.published_at.is_(None), job.published_at < ack_cutoff),
|
||
or_(job.reply_deadline.is_(None), job.reply_deadline >= now),
|
||
)
|
||
return (
|
||
select(
|
||
job.id,
|
||
job.status,
|
||
job.pipeline,
|
||
job.attempt,
|
||
job.reply_deadline,
|
||
job.requested_at,
|
||
job.metrics,
|
||
job.request_msg_id,
|
||
StoredFile.id.label("file_id"),
|
||
StoredFile.file_type,
|
||
StoredFile.original_name,
|
||
S3Object.bucket,
|
||
S3Object.key,
|
||
)
|
||
.join(StoredFile, StoredFile.id == job.file_id)
|
||
.join(S3Object, S3Object.id == StoredFile.s3_object_id)
|
||
.where(or_(job.status == RecognitionStatus.queued, awaiting_ack))
|
||
.order_by(job.requested_at.asc().nulls_first(), job.id)
|
||
.limit(1)
|
||
.with_for_update(of=job, skip_locked=True)
|
||
)
|
||
|
||
|
||
def _job_update(job_id: uuid.UUID, expected: RecognitionStatus) -> Any:
|
||
"""`UPDATE recognition_job` только из ожидаемого статуса (страховка от гонки с обработчиком результата)."""
|
||
return (
|
||
update(RecognitionJob)
|
||
.where(RecognitionJob.id == job_id, RecognitionJob.status == expected)
|
||
.execution_options(synchronize_session=False)
|
||
)
|
||
|
||
|
||
def _build_request(row: Any, reply_deadline: datetime) -> RecognitionRequest:
|
||
"""Запрос `recognition-request.schema.json` из `recognition_job ⋈ stored_file ⋈ s3_object`."""
|
||
return RecognitionRequest(
|
||
schema_version=CONTRACT_VERSION,
|
||
job_id=row.id,
|
||
file_id=row.file_id,
|
||
pipeline=row.pipeline.value,
|
||
file_type=row.file_type.value,
|
||
original_name=row.original_name,
|
||
s3=S3Ref(bucket=row.bucket, key=row.key),
|
||
reply_deadline=reply_deadline,
|
||
attempt=row.attempt,
|
||
)
|
||
|
||
|
||
def _short(text: str) -> str:
|
||
return text if len(text) <= ERROR_MESSAGE_MAX_LENGTH else text[: ERROR_MESSAGE_MAX_LENGTH - 1] + "…"
|
||
|
||
|
||
async def _process_one( # noqa: PLR0913 - один шаг прохода: все параметры прохода
|
||
session: AsyncSession,
|
||
nats: RequestPublisher,
|
||
report: PublishReport,
|
||
*,
|
||
clock: Clock,
|
||
ack_cutoff: datetime,
|
||
pass_started: datetime,
|
||
max_republish: int,
|
||
) -> bool:
|
||
"""Обработать одно задание в транзакции `session`; `False` — подходящих заданий больше нет."""
|
||
row = (await session.execute(_candidate_stmt(pass_started, ack_cutoff))).one_or_none()
|
||
if row is None:
|
||
return False
|
||
now = clock()
|
||
first = row.status is RecognitionStatus.queued
|
||
|
||
if not first and _republish_count(row.metrics) >= max_republish:
|
||
await session.execute(
|
||
_job_update(row.id, RecognitionStatus.published).values(
|
||
status=RecognitionStatus.dlq,
|
||
error_code=ERROR_NO_ACK,
|
||
error_message=(
|
||
f"Сервис распознавания не подтвердил получение запроса (ack) после {max_republish} "
|
||
"повторных публикаций"
|
||
),
|
||
finished_at=now,
|
||
)
|
||
)
|
||
await set_file_status_for_job(session, row, FileStatus.ERROR, now=now)
|
||
report.dlq += 1
|
||
logger.warning(
|
||
"Запрос распознавания без ack: задание переведено в dlq. error_category=no_ack, job_id={job_id}, "
|
||
"republish_count={count}",
|
||
job_id=row.id,
|
||
count=_republish_count(row.metrics),
|
||
)
|
||
return True
|
||
|
||
reply_deadline = row.reply_deadline or now + DEFAULT_REPLY_TIMEOUT[row.pipeline]
|
||
try:
|
||
request = _build_request(row, reply_deadline)
|
||
subject = request_subject(row.pipeline.value)
|
||
except (ValidationError, ValueError) as exc:
|
||
await session.execute(
|
||
_job_update(row.id, row.status).values(
|
||
status=RecognitionStatus.error,
|
||
error_code=ERROR_INVALID_REQUEST,
|
||
error_message=_short(f"Запрос распознавания не соответствует контракту: {exc}"),
|
||
finished_at=now,
|
||
)
|
||
)
|
||
# финальная ошибка задания — файл текущего задания ERROR (spec primary-files «Жизненный цикл статуса файла»)
|
||
await set_file_status_for_job(session, row, FileStatus.ERROR, now=now)
|
||
report.invalid += 1
|
||
logger.error(
|
||
"Запрос распознавания не собран по контракту, задание переведено в error. "
|
||
"error_category=invalid_request, job_id={job_id}, error={error}",
|
||
job_id=row.id,
|
||
error=str(exc),
|
||
)
|
||
return True
|
||
|
||
# первая публикация — id из текущего attempt; повтор — сохранённый id (тот же Nats-Msg-Id)
|
||
msg_id = request_msg_id(row.id, row.attempt) if first else row.request_msg_id or request_msg_id(row.id, row.attempt)
|
||
payload = json.dumps(request.to_wire(), ensure_ascii=False).encode("utf-8")
|
||
await nats.publish(subject, payload, msg_id) # NatsPublishError → откат транзакции у вызывающего
|
||
|
||
if first:
|
||
await session.execute(
|
||
_job_update(row.id, RecognitionStatus.queued).values(
|
||
status=RecognitionStatus.published,
|
||
published_at=now,
|
||
request_msg_id=msg_id,
|
||
request_subject=subject,
|
||
reply_deadline=reply_deadline,
|
||
requested_at=row.requested_at or now,
|
||
)
|
||
)
|
||
report.published += 1
|
||
logger.info(
|
||
"Запрос распознавания опубликован. job_id={job_id}, subject={subject}, msg_id={msg_id}",
|
||
job_id=row.id,
|
||
subject=subject,
|
||
msg_id=msg_id,
|
||
)
|
||
else:
|
||
count = _republish_count(row.metrics) + 1
|
||
# jsonb `||`: прочие ключи metrics сохраняются; не объект (SQL NULL или jsonb null — строки, записанные до
|
||
# JSONB(none_as_null=True) или сырым SQL) → {}
|
||
is_object = func.jsonb_typeof(RecognitionJob.metrics) == "object"
|
||
base = case((is_object, RecognitionJob.metrics), else_=literal({}, JSONB))
|
||
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
|
||
)
|
||
)
|
||
report.republished += 1
|
||
logger.warning(
|
||
"Нет ack от сервиса распознавания, запрос опубликован повторно. job_id={job_id}, subject={subject}, "
|
||
"msg_id={msg_id}, republish_count={count}",
|
||
job_id=row.id,
|
||
subject=subject,
|
||
msg_id=msg_id,
|
||
count=count,
|
||
)
|
||
return True
|
||
|
||
|
||
async def publish_pending( # noqa: PLR0913 - параметры прохода именованные: настройки и подмены для тестов
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
nats: RequestPublisher,
|
||
*,
|
||
ack_timeout: timedelta,
|
||
max_republish: int,
|
||
batch_size: int = PUBLISH_BATCH_SIZE,
|
||
now: Clock | None = None,
|
||
) -> PublishReport:
|
||
"""Один проход outbox-публикатора (spec «Outbox публикует задания из `queued`», «Запрос без ack публикуется
|
||
повторно, затем — `dlq`»; design D-6).
|
||
|
||
До `batch_size` заданий, каждое — в своей транзакции (см. docstring модуля). Нет соединения с NATS — проход
|
||
пропускается (задания остаются как есть). Первая `NatsPublishError` откатывает транзакцию задания (оно остаётся
|
||
`queued` / `published` без изменений) и прерывает проход: следующий проход повторит публикацию.
|
||
"""
|
||
clock = now or _utcnow
|
||
report = PublishReport()
|
||
if not nats.is_connected:
|
||
report.skipped = True
|
||
return report
|
||
pass_started = clock()
|
||
ack_cutoff = pass_started - ack_timeout
|
||
for _ in range(batch_size):
|
||
try:
|
||
async with session_factory() as session, session.begin():
|
||
found = await _process_one(
|
||
session,
|
||
nats,
|
||
report,
|
||
clock=clock,
|
||
ack_cutoff=ack_cutoff,
|
||
pass_started=pass_started,
|
||
max_republish=max_republish,
|
||
)
|
||
except NatsPublishError as exc:
|
||
report.failed += 1
|
||
logger.warning(
|
||
"Публикация запроса распознавания не удалась, повтор в следующем проходе. error_category=nats_error, "
|
||
"error={error}",
|
||
error=str(exc),
|
||
)
|
||
break
|
||
if not found:
|
||
break
|
||
return report
|
||
|
||
|
||
async def expire_deadlines(session_factory: async_sessionmaker[AsyncSession], *, now: Clock | None = None) -> int:
|
||
"""Вахтёр сроков (spec «Таймаут ожидания результата»): `processing`/`published` с `reply_deadline < now` →
|
||
`timeout`, `error_code = TIMEOUT`, `finished_at = now`. Один атомарный `UPDATE … RETURNING id` с условием по
|
||
статусу: результат, успевший перевести задание в финальный статус, не перетирается. В той же транзакции файлы этих
|
||
заданий (только где задание — текущее, `current_recognition_job_id`) → `ERROR`. Возвращает число заданий."""
|
||
moment = (now or _utcnow)()
|
||
stmt = (
|
||
update(RecognitionJob)
|
||
.where(
|
||
RecognitionJob.status.in_([RecognitionStatus.processing, RecognitionStatus.published]),
|
||
RecognitionJob.reply_deadline < moment,
|
||
)
|
||
.values(
|
||
status=RecognitionStatus.timeout,
|
||
error_code=ERROR_TIMEOUT,
|
||
error_message="Сервис распознавания не прислал результат до срока ответа (reply_deadline)",
|
||
finished_at=moment,
|
||
)
|
||
.returning(RecognitionJob.id)
|
||
.execution_options(synchronize_session=False)
|
||
)
|
||
async with session_factory() as session, session.begin():
|
||
job_ids = list((await session.execute(stmt)).scalars())
|
||
files = await set_file_status_for_jobs(session, job_ids, FileStatus.ERROR, now=moment)
|
||
count = len(job_ids)
|
||
if count:
|
||
logger.warning(
|
||
"Задания распознавания переведены в timeout по сроку ответа. count={count}, files_error={files}",
|
||
count=count,
|
||
files=files,
|
||
)
|
||
return count
|
||
|
||
|
||
async def run_publisher_loop( # noqa: PLR0913 - настройки цикла именованные
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
nats: RequestPublisher,
|
||
*,
|
||
poll_seconds: float,
|
||
ack_timeout: timedelta,
|
||
max_republish: int,
|
||
sleep: SleepFn = asyncio.sleep,
|
||
) -> None:
|
||
"""Бесконечный цикл публикатора с периодом `poll_seconds`; ошибка прохода логируется, цикл продолжается."""
|
||
while True:
|
||
try:
|
||
await publish_pending(session_factory, nats, ack_timeout=ack_timeout, max_republish=max_republish)
|
||
except Exception as exc:
|
||
logger.exception(
|
||
"Ошибка прохода outbox-публикатора. error_category=outbox_error, error_type={error_type}, "
|
||
"error={error}",
|
||
error_type=type(exc).__name__,
|
||
error=repr(exc),
|
||
)
|
||
await sleep(poll_seconds)
|
||
|
||
|
||
async def run_watchdog_loop(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
*,
|
||
period_seconds: float = WATCHDOG_PERIOD_SECONDS,
|
||
sleep: SleepFn = asyncio.sleep,
|
||
) -> None:
|
||
"""Бесконечный цикл вахтёра сроков с периодом `period_seconds`; ошибка прохода логируется, цикл продолжается."""
|
||
while True:
|
||
try:
|
||
await expire_deadlines(session_factory)
|
||
except Exception as exc:
|
||
logger.exception(
|
||
"Ошибка прохода вахтёра сроков. error_category=watchdog_error, error_type={error_type}, error={error}",
|
||
error_type=type(exc).__name__,
|
||
error=repr(exc),
|
||
)
|
||
await sleep(period_seconds)
|
||
|
||
|
||
class ExchangeTasks:
|
||
"""Фоновые задачи обмена в lifespan (design D-6; Срез 2c D-1): публикатор, вахтёр и — при подключённом S3 — цикл
|
||
маппинга результатов (`services/result_mapping.py`). Запускаются только при `NATS_ENABLED=true` (вызывающий код
|
||
создаёт их вместе с `NatsConnection`) и останавливаются до закрытия NATS."""
|
||
|
||
def __init__(self, tasks: Sequence[asyncio.Task[None]]) -> None:
|
||
self.tasks = tuple(tasks)
|
||
|
||
@classmethod
|
||
def start(
|
||
cls,
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
nats: RequestPublisher,
|
||
settings: Settings,
|
||
storage: ObjectStorage | None = None,
|
||
) -> "ExchangeTasks":
|
||
publisher = asyncio.create_task(
|
||
run_publisher_loop(
|
||
session_factory,
|
||
nats,
|
||
poll_seconds=settings.OUTBOX_POLL_SECONDS,
|
||
ack_timeout=timedelta(seconds=settings.ACK_TIMEOUT_SECONDS),
|
||
max_republish=settings.OUTBOX_MAX_REPUBLISH,
|
||
),
|
||
name="recognition-outbox-publisher",
|
||
)
|
||
watchdog = asyncio.create_task(
|
||
run_watchdog_loop(session_factory, period_seconds=WATCHDOG_PERIOD_SECONDS),
|
||
name="recognition-deadline-watchdog",
|
||
)
|
||
tasks = [publisher, watchdog]
|
||
if storage is not None:
|
||
tasks.append(
|
||
asyncio.create_task(
|
||
run_mapping_loop(session_factory, storage, settings), name="recognition-result-mapper"
|
||
)
|
||
)
|
||
else:
|
||
logger.warning("S3 не подключено: маппинг результатов не запущен")
|
||
logger.info(
|
||
"Фоновые задачи обмена запущены: публикатор (период {poll} с), вахтёр сроков (период {watchdog} с), "
|
||
"маппинг результатов: {mapping}",
|
||
poll=settings.OUTBOX_POLL_SECONDS,
|
||
watchdog=WATCHDOG_PERIOD_SECONDS,
|
||
mapping=f"период {settings.MAPPING_POLL_SECONDS} с" if storage is not None else "нет",
|
||
)
|
||
return cls(tasks)
|
||
|
||
async def stop(self) -> None:
|
||
"""Отменяет задачи и дожидается их завершения; исключения наружу не выходят."""
|
||
for task in self.tasks:
|
||
task.cancel()
|
||
results = await asyncio.gather(*self.tasks, return_exceptions=True)
|
||
for task, result in zip(self.tasks, results, strict=True):
|
||
if isinstance(result, Exception): # CancelledError — штатная остановка, не Exception
|
||
logger.opt(exception=result).error(
|
||
"Фоновая задача обмена завершилась с ошибкой. task={name}", name=task.get_name()
|
||
)
|
||
|
||
|
||
# --- inbox: files.tabular_parsed ----------------------------------------------------------------------------------
|
||
|
||
# Финальные статусы для inbox (рулинг R-22, design #34): результат для них не применяется
|
||
FINAL_STATUSES: frozenset[RecognitionStatus] = frozenset(
|
||
{
|
||
RecognitionStatus.done,
|
||
RecognitionStatus.needs_check,
|
||
RecognitionStatus.error,
|
||
RecognitionStatus.timeout,
|
||
RecognitionStatus.dlq,
|
||
}
|
||
)
|
||
# Итог сервиса → статус задания
|
||
RESULT_STATUSES: Mapping[MessageStatus, RecognitionStatus] = {
|
||
MessageStatus.DONE: RecognitionStatus.done,
|
||
MessageStatus.NEEDS_CHECK: RecognitionStatus.needs_check,
|
||
MessageStatus.ERROR: RecognitionStatus.error,
|
||
}
|
||
# Счётчики значений из метрик сервиса → одноимённые колонки задания (DBML v5.7)
|
||
FIELD_COUNTERS = ("fields_total", "fields_low_confidence", "fields_unrecognized")
|
||
LOG_ID_MAX_LENGTH = 64 # job_id невалидного сообщения в логе — не длиннее (тело в лог не пишется)
|
||
|
||
|
||
class InboundMessage(Protocol):
|
||
"""То, что inbox читает из сообщения nats-py (`nats.aio.msg.Msg`)."""
|
||
|
||
@property
|
||
def data(self) -> bytes: ...
|
||
|
||
@property
|
||
def headers(self) -> Mapping[str, str] | None: ...
|
||
|
||
@property
|
||
def subject(self) -> str: ...
|
||
|
||
|
||
def header_msg_id(headers: Mapping[str, str] | None) -> str | None:
|
||
"""Значение заголовка `Nats-Msg-Id` без учёта регистра имени; нет заголовка или пустое значение — `None`."""
|
||
if not headers:
|
||
return None
|
||
target = NATS_MSG_ID_HEADER.lower()
|
||
for name, value in headers.items():
|
||
if name.lower() == target and value:
|
||
return value
|
||
return None
|
||
|
||
|
||
def merge_metrics(existing: Any, incoming: Mapping[str, Any] | None) -> Any:
|
||
"""Метрики сервиса поверх метрик задания: ключи ядра (`republish_count`) сохраняются, совпадающие — от сервиса.
|
||
|
||
`incoming is None` — метрики задания как есть. Не объект (SQL NULL; jsonb `null` в строках, записанных до
|
||
`JSONB(none_as_null=True)` или сырым SQL) — как `{}`.
|
||
"""
|
||
if incoming is None:
|
||
return existing
|
||
base = dict(existing) if isinstance(existing, dict) else {}
|
||
return {**base, **incoming}
|
||
|
||
|
||
def _parse_message(msg: InboundMessage) -> RecognitionResultMessage | None:
|
||
"""Конверт `recognition-result-message` v1.0 или `None` (warning в лог без тела сообщения)."""
|
||
try:
|
||
data = json.loads(msg.data)
|
||
except ValueError: # JSONDecodeError и UnicodeDecodeError
|
||
data = None
|
||
if not isinstance(data, dict):
|
||
logger.warning(
|
||
"Сообщение {subject} отклонено: тело не JSON-объект. error_category=invalid_message, size={size}",
|
||
subject=msg.subject,
|
||
size=len(msg.data),
|
||
)
|
||
return None
|
||
try:
|
||
return RecognitionResultMessage.model_validate(data)
|
||
except ValidationError as exc:
|
||
raw_job_id = data.get("job_id")
|
||
# job_id невалидного сообщения — данные отправителя: repr экранирует переводы строк и управляющие символы
|
||
job_id = repr(raw_job_id[:LOG_ID_MAX_LENGTH]) if isinstance(raw_job_id, str) else None
|
||
errors = "; ".join(
|
||
f"{'.'.join(str(part) for part in error['loc']) or '<root>'}: {error['msg']}"
|
||
for error in exc.errors(include_url=False, include_input=False, include_context=False)
|
||
)
|
||
logger.warning(
|
||
"Сообщение {subject} отклонено: не соответствует контракту recognition-result-message. "
|
||
"error_category=invalid_message, job_id={job_id}, errors={errors}",
|
||
subject=msg.subject,
|
||
job_id=job_id,
|
||
errors=errors,
|
||
)
|
||
return None
|
||
|
||
|
||
async def _lock_job(session: AsyncSession, job_id: uuid.UUID) -> RecognitionJob | None:
|
||
"""Строка задания под `FOR UPDATE` — ЖДЁТ блокировку (не SKIP LOCKED / NOWAIT): ack или результат, пришедший, пока
|
||
публикатор или вахтёр держат строку, применяется после их commit и видит их итог."""
|
||
stmt = (
|
||
select(RecognitionJob)
|
||
.where(RecognitionJob.id == job_id)
|
||
.with_for_update()
|
||
.execution_options(populate_existing=True)
|
||
)
|
||
return (await session.execute(stmt)).scalar_one_or_none()
|
||
|
||
|
||
async def _apply_ack(
|
||
session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage, msg_id: str, now: datetime
|
||
) -> None:
|
||
"""Ack (`status = processing`): `published` → `processing` и файл `UPLOADED → PROCESSING` (R-5); повтор для
|
||
`processing` — тихо; иначе — warning (запоздавший ack после `timeout`/`dlq` файл не трогает)."""
|
||
if job.status is RecognitionStatus.published:
|
||
job.status = RecognitionStatus.processing
|
||
job.started_at = now
|
||
job.service_name = message.service.name.value
|
||
job.service_version = message.service.version
|
||
file_changed = await set_file_status_for_job(
|
||
session, job, FileStatus.PROCESSING, now=now, only_from={FileStatus.UPLOADED}
|
||
)
|
||
logger.info(
|
||
"Сервис подтвердил получение запроса (ack): задание в processing. job_id={job_id}, service={service}, "
|
||
"msg_id={msg_id}, file_processing={file_changed}",
|
||
job_id=job.id,
|
||
service=message.service.name.value,
|
||
msg_id=msg_id,
|
||
file_changed=file_changed,
|
||
)
|
||
elif job.status is RecognitionStatus.processing:
|
||
logger.debug("Повторный ack для задания в processing, игнорируется. job_id={job_id}", job_id=job.id)
|
||
else:
|
||
logger.warning(
|
||
"Ack для задания не в статусе published: задание не изменено. error_category=unexpected_ack, "
|
||
"job_id={job_id}, status={status}, msg_id={msg_id}",
|
||
job_id=job.id,
|
||
status=job.status.value,
|
||
msg_id=msg_id,
|
||
)
|
||
|
||
|
||
async def _register_result_object(session: AsyncSession, message: RecognitionResultMessage, now: datetime) -> uuid.UUID:
|
||
"""Файл результата в реестре `s3_object` (`bucket_role = results`, `created_by = service.name`).
|
||
|
||
Уникальность `(bucket, key)`: строка с этим ключом уже есть — переиспользуется (дубль не вставляется, не падаем).
|
||
"""
|
||
assert message.result_s3 is not None
|
||
bucket, key = message.result_s3.bucket, message.result_s3.key
|
||
stmt = (
|
||
pg_insert(S3Object)
|
||
.values(
|
||
bucket_role=S3BucketRole.results,
|
||
bucket=bucket,
|
||
key=key,
|
||
created_by=message.service.name.value,
|
||
created_at=now,
|
||
)
|
||
.on_conflict_do_nothing(index_elements=[S3Object.bucket, S3Object.key])
|
||
.returning(S3Object.id)
|
||
)
|
||
inserted = (await session.execute(stmt)).scalar_one_or_none()
|
||
if inserted is not None:
|
||
return inserted
|
||
existing = (
|
||
await session.execute(select(S3Object.id).where(S3Object.bucket == bucket, S3Object.key == key))
|
||
).scalar_one()
|
||
logger.warning(
|
||
"Файл результата уже зарегистрирован в s3_object, строка переиспользована. job_id={job_id}, bucket={bucket}, "
|
||
"key={key}",
|
||
job_id=message.job_id,
|
||
bucket=bucket,
|
||
key=key,
|
||
)
|
||
return existing
|
||
|
||
|
||
async def _apply_result(
|
||
session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage, msg_id: str, now: datetime
|
||
) -> bool:
|
||
"""Результат (`done | needs_check | error`): дубликат и финальный статус — без изменений (`False`); иначе —
|
||
регистрация файла результата, обновление задания и статуса файла (`True`, дальше — commit)."""
|
||
if job.result_msg_id == msg_id:
|
||
logger.debug(
|
||
"Повторная доставка результата (дубликат по result_msg_id), изменений нет. job_id={job_id}, "
|
||
"msg_id={msg_id}",
|
||
job_id=job.id,
|
||
msg_id=msg_id,
|
||
)
|
||
return False
|
||
if job.status in FINAL_STATUSES:
|
||
logger.warning(
|
||
"Результат для задания в финальном статусе не применён. error_category=late_result, job_id={job_id}, "
|
||
"status={status}, result_status={result_status}, msg_id={msg_id}",
|
||
job_id=job.id,
|
||
status=job.status.value,
|
||
result_status=message.status.value,
|
||
msg_id=msg_id,
|
||
)
|
||
return False
|
||
|
||
result_s3_object_id = await _register_result_object(session, message, now) if message.result_s3 else None
|
||
job.status = RESULT_STATUSES[message.status]
|
||
job.service_name = message.service.name.value
|
||
job.service_version = message.service.version
|
||
job.finished_at = now
|
||
if message.metrics is not None:
|
||
job.metrics = merge_metrics(job.metrics, _metrics_json(message.metrics))
|
||
for name in FIELD_COUNTERS:
|
||
value = getattr(message.metrics, name)
|
||
if value is not None:
|
||
setattr(job, name, value)
|
||
job.error_code = message.error.code.value if message.error is not None else None
|
||
job.error_message = message.error.message if message.error is not None else None
|
||
job.result_schema_version = message.result_schema_version
|
||
job.result_s3_object_id = result_s3_object_id
|
||
job.result_msg_id = msg_id
|
||
file_changed = await _apply_result_file_status(session, job, now)
|
||
logger.info(
|
||
"Результат распознавания принят. job_id={job_id}, status={status}, service={service}, msg_id={msg_id}, "
|
||
"error_code={error_code}, file_changed={file_changed}",
|
||
job_id=job.id,
|
||
status=job.status.value,
|
||
service=message.service.name.value,
|
||
msg_id=msg_id,
|
||
error_code=job.error_code,
|
||
file_changed=file_changed,
|
||
)
|
||
return True
|
||
|
||
|
||
async def _apply_result_file_status(session: AsyncSession, job: RecognitionJob, now: datetime) -> bool:
|
||
"""Статус файла по принятому результату (design D-5): `error` → `ERROR`; `done`/`needs_check` → `PROCESSING` только
|
||
из `UPLOADED` (R-4: ack мог потеряться), иначе файл ждёт цикла маппинга как есть. Только текущее задание файла."""
|
||
if job.status is RecognitionStatus.error:
|
||
return await set_file_status_for_job(session, job, FileStatus.ERROR, now=now)
|
||
return await set_file_status_for_job(session, job, FileStatus.PROCESSING, now=now, only_from={FileStatus.UPLOADED})
|
||
|
||
|
||
def _metrics_json(metrics: Metrics) -> dict[str, Any]:
|
||
return metrics.model_dump(mode="json", exclude_none=True)
|
||
|
||
|
||
async def _apply_message(
|
||
session: AsyncSession, message: RecognitionResultMessage, header_id: str | None, clock: Clock
|
||
) -> None:
|
||
job = await _lock_job(session, message.job_id)
|
||
if job is None:
|
||
logger.warning(
|
||
"Сообщение files.tabular_parsed для неизвестного задания, не применено. error_category=unknown_job, "
|
||
"job_id={job_id}, status={status}",
|
||
job_id=message.job_id,
|
||
status=message.status.value,
|
||
)
|
||
return
|
||
if job.file_id != message.file_id:
|
||
logger.warning(
|
||
"Сообщение 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,
|
||
message_file_id=message.file_id,
|
||
)
|
||
return
|
||
is_ack = message.status is MessageStatus.PROCESSING
|
||
msg_id = header_id
|
||
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. "
|
||
"error_category=missing_msg_id, job_id={job_id}, msg_id={msg_id}",
|
||
job_id=job.id,
|
||
msg_id=msg_id,
|
||
)
|
||
now = clock()
|
||
if is_ack:
|
||
await _apply_ack(session, job, message, msg_id, now)
|
||
else:
|
||
await _apply_result(session, job, message, msg_id, now)
|
||
|
||
|
||
async def handle_message(
|
||
msg: InboundMessage,
|
||
*,
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
now: Clock | None = None,
|
||
) -> None:
|
||
"""Обработчик `files.tabular_parsed` (spec «Ack сервиса переводит задание в `processing`», «Inbox принимает
|
||
результат идемпотентно»; design D-6). Модель транзакции и правила — в docstring модуля.
|
||
|
||
Исключения наружу не выходят: невалидное сообщение — warning; ошибка БД — откат транзакции (задание и файл),
|
||
`logger.exception`, без повторов (задание дойдёт до `timeout`). `audit_event` не пишется (R-7,
|
||
ASSUMPTION(OQ-55)). Файл результата не читается: маппинг — отдельный цикл (design D-1).
|
||
"""
|
||
message = _parse_message(msg)
|
||
if message is None:
|
||
return
|
||
try:
|
||
async with session_factory() as session, session.begin():
|
||
await _apply_message(session, message, header_msg_id(msg.headers), now or _utcnow)
|
||
except Exception as exc:
|
||
logger.exception(
|
||
"Ошибка обработки сообщения {subject}, транзакция откачена (без повторов). error_category=inbox_error, "
|
||
"job_id={job_id}, error_type={error_type}, error={error}",
|
||
subject=msg.subject,
|
||
job_id=message.job_id,
|
||
error_type=type(exc).__name__,
|
||
error=repr(exc),
|
||
)
|