feat: outbox-публикатор и вахтёр сроков recognition_job, фоновые задачи обмена в lifespan при NATS_ENABLED
- services/recognition_exchange.py: publish_pending — одно задание на короткую транзакцию (SELECT … LIMIT 1 FOR UPDATE OF recognition_job SKIP LOCKED → публикация → UPDATE → COMMIT): queued → published (request_msg_id = <job_id>:<attempt>, request_subject, published_at, reply_deadline — 5 мин по умолчанию, requested_at); published без ack дольше ACK_TIMEOUT → повтор с тем же Nats-Msg-Id и attempt, metrics.republish_count (jsonb ||, прочие ключи сохраняются); после OUTBOX_MAX_REPUBLISH → dlq/NO_ACK; запрос не собирается по контракту → error/INVALID_REQUEST; первая NatsPublishError откатывает задание и прерывает проход; нет соединения — проход пропускается. published с прошедшим reply_deadline оставлен вахтёру. - expire_deadlines — один UPDATE processing/published с reply_deadline < now → timeout/TIMEOUT. - run_publisher_loop / run_watchdog_loop (30 с), ExchangeTasks: запуск в lifespan вместе с NatsConnection, остановка до nats.stop(). audit_event не пишется (R-7). - Тесты: интеграционные против PostgreSQL и NATS (схема запроса, повторы и dlq на инжектируемых часах, ack после первого повтора, вахтёр, ошибка публикации, два конкурентных прохода, циклы), lifespan; фабрика CommittedJobs. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
a0de203d88
commit
2e96971475
@ -211,8 +211,8 @@ docker compose -f ../docs/dev/docker-compose.yml down -v # полный сб
|
||||
| `OPENBAO__SMB__SMB_PATH` | — (обязательно) | заглушка SMB |
|
||||
| `OPENBAO__SMB__SMB_LOGIN` | — (обязательно) | заглушка SMB |
|
||||
| `OPENBAO__SMB__SMB_PASSWORD` | — (обязательно) | заглушка SMB |
|
||||
| `NATS_ENABLED` | `false` | подключение к NATS и обмен через `recognition_job`; без него `/healthcheck` показывает `nats=off` |
|
||||
| `OUTBOX_POLL_SECONDS` | `5` | период опроса outbox |
|
||||
| `NATS_ENABLED` | `false` | подключение к NATS и обмен через `recognition_job`: фоновые outbox-публикатор и вахтёр сроков ответа (период 30 с, `processing`/`published` с прошедшим `reply_deadline` → `timeout`); без него фоновых задач нет, `/healthcheck` показывает `nats=off` |
|
||||
| `OUTBOX_POLL_SECONDS` | `5` | период опроса outbox (`queued` → публикация → `published`) |
|
||||
| `ACK_TIMEOUT_SECONDS` | `30` | ожидание ack сервиса распознавания до повторной публикации |
|
||||
| `OUTBOX_MAX_REPUBLISH` | `3` | число повторных публикаций до перевода в `dlq` |
|
||||
| `AUTH_DEV_ISSUER` | `false` | dev-эмитент токенов: работает **только** при `AUTH_DEV_ISSUER=true` **и** `DEBUG=true` |
|
||||
|
||||
@ -5,7 +5,8 @@ NATS без JetStream, подключение в фоне при NATS_ENABLED=tr
|
||||
group `arm-core`), SMB (горячие папки ИС Сканирование), middleware (gzip, proxy headers, CORS, РАИСА при DEBUG=False),
|
||||
healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA.
|
||||
Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает
|
||||
состояние. Outbox-паблишер и вахтёр сроков — задача 23 (docs/context/03-architecture.md).
|
||||
состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор и
|
||||
вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) — и останавливает их до закрытия NATS.
|
||||
"""
|
||||
|
||||
import os
|
||||
@ -28,7 +29,8 @@ from src.core.clients import S3Storage
|
||||
from src.core.config import get_settings
|
||||
from src.core.exception_handlers import register_exception_handlers
|
||||
from src.core.nats import NatsConnection
|
||||
from src.services.recognition_exchange import handle_message
|
||||
from src.db.db import async_session_maker
|
||||
from src.services.recognition_exchange import ExchangeTasks, handle_message
|
||||
|
||||
settings = get_settings()
|
||||
|
||||
@ -102,10 +104,16 @@ def _create_nats() -> NatsConnection | None:
|
||||
return NatsConnection(settings.OPENBAO.NATS, on_connected=_subscribe_results)
|
||||
|
||||
|
||||
def _start_exchange(nats: NatsConnection) -> ExchangeTasks:
|
||||
"""Фоновые задачи обмена: публикатор outbox и вахтёр сроков (только вместе с NATS, т.е. при NATS_ENABLED=true)."""
|
||||
return ExchangeTasks.start(async_session_maker, nats, settings)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
app.state.storage = None
|
||||
app.state.nats = None
|
||||
app.state.exchange = None
|
||||
|
||||
try:
|
||||
smbclient.register_session(
|
||||
@ -119,12 +127,15 @@ async def lifespan(app: FastAPI):
|
||||
app.state.nats = _create_nats()
|
||||
if app.state.nats is not None:
|
||||
await app.state.nats.start() # подключение в фоне: старт приложения не ждёт NATS
|
||||
app.state.exchange = _start_exchange(app.state.nats) # публикатор сам ждёт соединения
|
||||
|
||||
logger.info("Приложение запущено")
|
||||
yield
|
||||
finally:
|
||||
logger.info("Приложение закрывается")
|
||||
smbclient.reset_connection_cache()
|
||||
if app.state.exchange is not None: # до закрытия NATS: публикатор не должен писать в закрытое соединение
|
||||
await app.state.exchange.stop()
|
||||
if app.state.nats is not None:
|
||||
await app.state.nats.stop()
|
||||
if app.state.storage:
|
||||
|
||||
@ -1,12 +1,436 @@
|
||||
"""Обмен с сервисами распознавания через `recognition_job` (design D-6; docs/context/03-architecture.md).
|
||||
"""Обмен с сервисами распознавания через `recognition_job` (design D-6; spec recognition-job-exchange;
|
||||
docs/context/03-architecture.md, «Контракт обмена через NATS»).
|
||||
|
||||
Пока здесь только заглушка обработчика `recognition.result`: подписка в `src/core/nats.py` уже вызывает её.
|
||||
Полноценная обработка (ack `processing`, результат, дедуп по `result_msg_id`) — задача 24 (рулинг R-12, design #23);
|
||||
`publish_pending()` и `expire_deadlines()` — задача 23.
|
||||
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`. `attempt` не растёт:
|
||||
обработка не начиналась. Replay (`attempt + 1`) — вне Среза 0.
|
||||
- `expire_deadlines()` — вахтёр: `processing`/`published` с `reply_deadline < now` → `timeout`, `error_code = TIMEOUT`.
|
||||
- `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).
|
||||
|
||||
Inbox (`handle_message`) — пока заглушка: подписка в `src/core/nats.py` уже вызывает её; полноценная обработка
|
||||
(ack `processing`, результат, дедуп по `result_msg_id`) — задача 24 (рулинг R-12, design #23).
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import uuid
|
||||
from collections.abc import Awaitable, Callable, Sequence
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import Any, Protocol
|
||||
|
||||
from loguru import logger
|
||||
from nats.aio.msg import Msg
|
||||
from pydantic import ValidationError
|
||||
from sqlalchemy import CursorResult, and_, case, func, literal, or_, select, update
|
||||
from sqlalchemy.dialects.postgresql import JSONB
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
|
||||
from src.contract import CONTRACT_VERSION, RecognitionRequest, S3Ref, request_msg_id, request_subject
|
||||
from src.core.config import Settings
|
||||
from src.core.nats import NatsPublishError
|
||||
from src.models.enums import RecognitionPipeline, RecognitionStatus
|
||||
from src.models.files import RecognitionJob, S3Object, StoredFile
|
||||
|
||||
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,
|
||||
)
|
||||
)
|
||||
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,
|
||||
)
|
||||
)
|
||||
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 или JSON null — так ORM пишет None) → {}
|
||||
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` с условием по статусу: результат,
|
||||
успевший перевести задание в финальный статус, не перетирается. Возвращает число заданий."""
|
||||
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,
|
||||
)
|
||||
.execution_options(synchronize_session=False)
|
||||
)
|
||||
async with session_factory() as session, session.begin():
|
||||
result = await session.execute(stmt)
|
||||
count = result.rowcount if isinstance(result, CursorResult) else 0
|
||||
if count:
|
||||
logger.warning("Задания распознавания переведены в timeout по сроку ответа. count={count}", count=count)
|
||||
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): публикатор и вахтёр. Запускаются только при `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
|
||||
) -> "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",
|
||||
)
|
||||
logger.info(
|
||||
"Фоновые задачи обмена запущены: публикатор (период {poll} с), вахтёр сроков (период {watchdog} с)",
|
||||
poll=settings.OUTBOX_POLL_SECONDS,
|
||||
watchdog=WATCHDOG_PERIOD_SECONDS,
|
||||
)
|
||||
return cls([publisher, watchdog])
|
||||
|
||||
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()
|
||||
)
|
||||
|
||||
|
||||
async def handle_message(msg: Msg) -> None:
|
||||
|
||||
@ -1,11 +1,18 @@
|
||||
"""Фабрики тестовых данных для БД-тестов: минимальные цепочки FK поверх сессии (`flush`, без `commit`)."""
|
||||
"""Фабрики тестовых данных для БД-тестов.
|
||||
|
||||
- `make_osv_entry` — минимальная цепочка FK поверх сессии (`flush`, без `commit`): откатывается фикстурой `db_session`.
|
||||
- `CommittedJobs` — задания распознавания, ЗАФИКСИРОВАННЫЕ в БД (`commit`): для фоновых процессов обмена, которые
|
||||
работают в собственных транзакциях и не видят чужую незафиксированную транзакцию. Всё созданное удаляется
|
||||
`cleanup()` в порядке FK.
|
||||
"""
|
||||
|
||||
import datetime
|
||||
import decimal
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import delete
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
|
||||
from src.models.directories import LegalEntity
|
||||
from src.models.enums import (
|
||||
@ -20,6 +27,8 @@ from src.models.enums import (
|
||||
from src.models.files import Document, RecognitionJob, S3Object, StoredFile
|
||||
from src.models.osv import OsvEntry
|
||||
|
||||
DEFAULT_FILE_TYPES = {RecognitionPipeline.tabular: FileType.XLSX, RecognitionPipeline.image: FileType.PDF}
|
||||
|
||||
|
||||
async def make_osv_entry(session: AsyncSession, **overrides: Any) -> OsvEntry:
|
||||
"""Строка ОСВ с минимальной цепочкой FK (ЮЛ, s3_object, stored_file, recognition_job, document); `flush`.
|
||||
@ -69,3 +78,106 @@ async def make_osv_entry(session: AsyncSession, **overrides: Any) -> OsvEntry:
|
||||
session.add(entry)
|
||||
await session.flush()
|
||||
return entry
|
||||
|
||||
|
||||
class CommittedJobs:
|
||||
"""Задания `recognition_job` с цепочкой `legal_entity` + `s3_object` + `stored_file`, зафиксированные в БД.
|
||||
|
||||
Каждый `make_queued_job` — отдельная транзакция с `commit`; созданные id запоминаются, `cleanup()` удаляет их
|
||||
в порядке FK (задание → файл → объект S3 и ЮЛ). Вызывать `cleanup()` в teardown фикстуры — даже при падении теста.
|
||||
"""
|
||||
|
||||
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
|
||||
self._session_factory = session_factory
|
||||
self.job_ids: list[uuid.UUID] = []
|
||||
self._file_ids: list[uuid.UUID] = []
|
||||
self._s3_ids: list[uuid.UUID] = []
|
||||
self._entity_ids: list[uuid.UUID] = []
|
||||
|
||||
async def make_queued_job( # noqa: PLR0913 - каждое поле задания задаётся тестом явно
|
||||
self,
|
||||
*,
|
||||
pipeline: RecognitionPipeline = RecognitionPipeline.tabular,
|
||||
file_type: FileType | None = None,
|
||||
original_name: str | None = "osv.xlsx",
|
||||
status: RecognitionStatus = RecognitionStatus.queued,
|
||||
reply_deadline: datetime.datetime | None = None,
|
||||
published_at: datetime.datetime | None = None,
|
||||
request_msg_id: str | None = None,
|
||||
metrics: Any = None,
|
||||
attempt: int = 1,
|
||||
) -> uuid.UUID:
|
||||
"""Задание (по умолчанию `queued`, tabular, XLSX) со своей цепочкой FK; возвращает `recognition_job.id`."""
|
||||
now = datetime.datetime.now(datetime.UTC)
|
||||
async with self._session_factory() as session, session.begin():
|
||||
entity = LegalEntity(inn=str(uuid.uuid4().int)[:12], name="ООО Тест обмена", valid_from=now)
|
||||
s3 = S3Object(
|
||||
bucket_role=S3BucketRole.uploads,
|
||||
bucket="arm-ka",
|
||||
key=f"uploads/{uuid.uuid4()}/{original_name or 'file'}",
|
||||
created_at=now,
|
||||
)
|
||||
session.add_all([entity, s3])
|
||||
await session.flush()
|
||||
self._entity_ids.append(entity.id)
|
||||
self._s3_ids.append(s3.id)
|
||||
file = StoredFile(
|
||||
legal_entity_id=entity.id,
|
||||
file_type=file_type or DEFAULT_FILE_TYPES[pipeline],
|
||||
original_name=original_name,
|
||||
s3_object_id=s3.id,
|
||||
status=FileStatus.UPLOADED,
|
||||
)
|
||||
session.add(file)
|
||||
await session.flush()
|
||||
self._file_ids.append(file.id)
|
||||
job = RecognitionJob(
|
||||
file_id=file.id,
|
||||
pipeline=pipeline,
|
||||
request_subject=f"recognition.{pipeline.value}.request",
|
||||
schema_version="1.0",
|
||||
status=status,
|
||||
attempt=attempt,
|
||||
reply_deadline=reply_deadline,
|
||||
published_at=published_at,
|
||||
request_msg_id=request_msg_id,
|
||||
metrics=metrics,
|
||||
)
|
||||
session.add(job)
|
||||
await session.flush()
|
||||
self.job_ids.append(job.id)
|
||||
return job.id
|
||||
|
||||
async def s3_key(self, job_id: uuid.UUID) -> str:
|
||||
"""Ключ S3 исходного файла задания."""
|
||||
async with self._session_factory() as session:
|
||||
job = await session.get(RecognitionJob, job_id)
|
||||
assert job is not None
|
||||
file = await session.get(StoredFile, job.file_id)
|
||||
assert file is not None
|
||||
s3 = await session.get(S3Object, file.s3_object_id)
|
||||
assert s3 is not None
|
||||
return s3.key
|
||||
|
||||
async def get(self, job_id: uuid.UUID) -> RecognitionJob:
|
||||
"""Свежее состояние задания из БД (новая сессия)."""
|
||||
async with self._session_factory() as session:
|
||||
job = await session.get(RecognitionJob, job_id)
|
||||
assert job is not None
|
||||
return job
|
||||
|
||||
async def cleanup(self) -> None:
|
||||
"""Удаляет всё созданное в порядке FK."""
|
||||
async with self._session_factory() as session, session.begin():
|
||||
if self.job_ids:
|
||||
await session.execute(delete(RecognitionJob).where(RecognitionJob.id.in_(self.job_ids)))
|
||||
if self._file_ids:
|
||||
await session.execute(delete(StoredFile).where(StoredFile.id.in_(self._file_ids)))
|
||||
if self._s3_ids:
|
||||
await session.execute(delete(S3Object).where(S3Object.id.in_(self._s3_ids)))
|
||||
if self._entity_ids:
|
||||
await session.execute(delete(LegalEntity).where(LegalEntity.id.in_(self._entity_ids)))
|
||||
self.job_ids.clear()
|
||||
self._file_ids.clear()
|
||||
self._s3_ids.clear()
|
||||
self._entity_ids.clear()
|
||||
|
||||
@ -14,6 +14,7 @@ import pytest
|
||||
|
||||
from src.api.router import router
|
||||
from src.core.nats import NatsConnection
|
||||
from src.services import recognition_exchange
|
||||
|
||||
# src/__init__.py реэкспортирует объект app под именем модуля, поэтому модуль берём через importlib
|
||||
app_module = importlib.import_module("src.app")
|
||||
@ -93,6 +94,24 @@ def _no_smb(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(app_module.smbclient, "reset_connection_cache", lambda: None)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def exchange_loops(monkeypatch: pytest.MonkeyPatch) -> list[str]:
|
||||
"""Циклы публикатора и вахтёра подменены ожиданием: быстрые тесты lifespan не ходят в БД."""
|
||||
started: list[str] = []
|
||||
|
||||
async def publisher(session_factory: Any, nats: Any, **kwargs: Any) -> None:
|
||||
started.append("publisher")
|
||||
await asyncio.Event().wait()
|
||||
|
||||
async def watchdog(session_factory: Any, **kwargs: Any) -> None:
|
||||
started.append("watchdog")
|
||||
await asyncio.Event().wait()
|
||||
|
||||
monkeypatch.setattr(recognition_exchange, "run_publisher_loop", publisher)
|
||||
monkeypatch.setattr(recognition_exchange, "run_watchdog_loop", watchdog)
|
||||
return started
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_no_smb")
|
||||
async def test_lifespan_starts_when_s3_and_nats_unavailable(monkeypatch: pytest.MonkeyPatch):
|
||||
async def unavailable() -> None:
|
||||
@ -198,3 +217,61 @@ async def test_connect_s3_passes_secure_flag(monkeypatch: pytest.MonkeyPatch, se
|
||||
|
||||
assert storage is not None
|
||||
assert created["secure"] is secure
|
||||
|
||||
|
||||
class _RecordingNats(_FakeNats):
|
||||
"""Запоминает, остановлены ли фоновые задачи обмена к моменту `stop()` соединения NATS."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.exchange_done_at_stop: bool | None = None
|
||||
|
||||
async def stop(self) -> None:
|
||||
exchange = app_module.app.state.exchange
|
||||
self.exchange_done_at_stop = all(task.done() for task in exchange.tasks)
|
||||
await super().stop()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_no_smb")
|
||||
async def test_lifespan_starts_exchange_tasks_and_stops_them_before_nats(
|
||||
monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str]
|
||||
):
|
||||
"""NATS_ENABLED=true: публикатор и вахтёр — две фоновые задачи; остановка — до `nats.stop()`."""
|
||||
|
||||
async def unavailable() -> None:
|
||||
return None
|
||||
|
||||
fake_nats = _RecordingNats()
|
||||
monkeypatch.setattr(app_module, "_connect_s3", unavailable)
|
||||
monkeypatch.setattr(app_module, "_create_nats", lambda: fake_nats)
|
||||
|
||||
async with app_module.lifespan(app_module.app):
|
||||
exchange = app_module.app.state.exchange
|
||||
assert exchange is not None
|
||||
assert len(exchange.tasks) == 2 # noqa: PLR2004 - публикатор и вахтёр
|
||||
await asyncio.sleep(0)
|
||||
assert sorted(exchange_loops) == ["publisher", "watchdog"]
|
||||
assert not any(task.done() for task in exchange.tasks)
|
||||
|
||||
assert fake_nats.exchange_done_at_stop is True
|
||||
assert fake_nats.stopped
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("_no_smb")
|
||||
async def test_lifespan_without_nats_starts_no_exchange_tasks(
|
||||
monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str]
|
||||
):
|
||||
"""NATS_ENABLED=false (`_create_nats` → None): фоновые задачи обмена не запускаются."""
|
||||
|
||||
async def unavailable() -> None:
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(app_module, "_connect_s3", unavailable)
|
||||
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", False)
|
||||
tasks_before = asyncio.all_tasks()
|
||||
|
||||
async with app_module.lifespan(app_module.app):
|
||||
assert app_module.app.state.exchange is None
|
||||
await asyncio.sleep(0)
|
||||
assert exchange_loops == []
|
||||
assert asyncio.all_tasks() - tasks_before == set()
|
||||
|
||||
698
api/tests/test_recognition_exchange.py
Normal file
698
api/tests/test_recognition_exchange.py
Normal file
@ -0,0 +1,698 @@
|
||||
"""Outbox-публикатор и вахтёр сроков `recognition_job` (spec recognition-job-exchange: «Outbox публикует задания из
|
||||
`queued`», «Запрос без ack публикуется повторно, затем — `dlq`», «Таймаут ожидания результата»; design D-6).
|
||||
|
||||
Интеграционные тесты работают против реального PostgreSQL (БД сессии pytest) и NATS из `.env.test`. Публикатор и
|
||||
вахтёр ведут собственные транзакции, поэтому данные фиксируются реально (`CommittedJobs`) и удаляются в teardown.
|
||||
Subject'ы запросов общие (`recognition.*.request`): тестовый подписчик фильтрует сообщения по `job_id` своих заданий.
|
||||
Время в тестах повторов — инжектируемые часы `now`, без реальных ожиданий таймаутов.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import time
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator, Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any, get_args
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from jsonschema import Draft202012Validator, FormatChecker
|
||||
from nats.aio.client import Client as NATS
|
||||
from nats.aio.msg import Msg
|
||||
from sqlalchemy import update
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
|
||||
|
||||
from src.contract import NATS_MSG_ID_HEADER, RecognitionRequest
|
||||
from src.contract.request import FileType as ContractFileType
|
||||
from src.core.config import get_settings
|
||||
from src.core.nats import NatsConnection, NatsPublishError
|
||||
from src.models.enums import FileType, RecognitionPipeline, RecognitionStatus
|
||||
from src.models.files import RecognitionJob
|
||||
from src.services import recognition_exchange as rx
|
||||
from src.services.recognition_exchange import (
|
||||
WATCHDOG_PERIOD_SECONDS,
|
||||
ExchangeTasks,
|
||||
expire_deadlines,
|
||||
publish_pending,
|
||||
run_publisher_loop,
|
||||
run_watchdog_loop,
|
||||
)
|
||||
from tests.factories import CommittedJobs
|
||||
|
||||
CONTRACTS_DIR = Path(__file__).resolve().parents[1] / "contracts"
|
||||
REQUEST_SCHEMA = json.loads((CONTRACTS_DIR / "recognition-request.schema.json").read_text(encoding="utf-8"))
|
||||
REQUEST_VALIDATOR = Draft202012Validator(REQUEST_SCHEMA, format_checker=FormatChecker())
|
||||
REQUEST_SUBJECTS = ("recognition.tabular.request", "recognition.image.request")
|
||||
|
||||
ACK = timedelta(seconds=30)
|
||||
MAX_REPUBLISH = 3
|
||||
READY_TIMEOUT = 5.0
|
||||
CONCURRENT_JOBS = 20
|
||||
MIN_LOOP_PASSES = 3 # цикл пережил несколько ошибок прохода подряд
|
||||
EXCHANGE_TASKS = 2 # публикатор и вахтёр
|
||||
|
||||
|
||||
# --- инфраструктура ---------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class Clock:
|
||||
"""Инжектируемые часы: `now` для публикатора и вахтёра, `advance` — сдвиг без реального ожидания."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.current = datetime.now(UTC)
|
||||
|
||||
def __call__(self) -> datetime:
|
||||
return self.current
|
||||
|
||||
def advance(self, delta: timedelta) -> None:
|
||||
self.current += delta
|
||||
|
||||
|
||||
class RequestSubscriber:
|
||||
"""Тестовый подписчик на subject'ы запросов: «голый» клиент nats-py, сообщения фильтруются по `job_id`."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.messages: list[Msg] = []
|
||||
self._nc = NATS()
|
||||
|
||||
async def start(self) -> None:
|
||||
url = get_settings().OPENBAO.NATS.NATS_URL
|
||||
try:
|
||||
await self._nc.connect(url, connect_timeout=2, allow_reconnect=False)
|
||||
except Exception as exc:
|
||||
pytest.fail(f"NATS недоступен ({url}, .env.test): {type(exc).__name__}", pytrace=False)
|
||||
|
||||
async def collect(msg: Msg) -> None:
|
||||
self.messages.append(msg)
|
||||
|
||||
for subject in REQUEST_SUBJECTS:
|
||||
await self._nc.subscribe(subject, cb=collect)
|
||||
await self._nc.flush()
|
||||
|
||||
async def stop(self) -> None:
|
||||
await self._nc.close()
|
||||
|
||||
def for_job(self, job_id: uuid.UUID) -> list[Msg]:
|
||||
return [m for m in self.messages if json.loads(m.data).get("job_id") == str(job_id)]
|
||||
|
||||
async def settle(self) -> None:
|
||||
"""Всё, что сервер принял до этого момента, доставлено подписчику (PING/PONG после сообщений) и обработано."""
|
||||
await self._nc.flush()
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
async def wait_for(self, job_id: uuid.UUID, count: int, timeout: float = 10.0) -> list[Msg]:
|
||||
deadline = time.monotonic() + timeout
|
||||
while len(self.for_job(job_id)) < count:
|
||||
if time.monotonic() > deadline:
|
||||
raise AssertionError(f"за {timeout} с получено {len(self.for_job(job_id))} из {count} сообщений")
|
||||
await asyncio.sleep(0.02)
|
||||
return self.for_job(job_id)
|
||||
|
||||
|
||||
class FakeNats:
|
||||
"""Подставной NATS: состояние соединения и счётчик публикаций; первые `fail_first` публикаций — ошибка."""
|
||||
|
||||
def __init__(self, *, connected: bool = True, fail_first: int = 0, delegate: Any = None, delay: float = 0) -> None:
|
||||
self.is_connected_value = connected
|
||||
self.fail_first = fail_first
|
||||
self.delegate = delegate
|
||||
self.delay = delay
|
||||
self.calls: list[tuple[str, str]] = []
|
||||
|
||||
@property
|
||||
def is_connected(self) -> bool:
|
||||
return self.is_connected_value
|
||||
|
||||
async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
|
||||
self.calls.append((subject, msg_id))
|
||||
if len(self.calls) <= self.fail_first:
|
||||
raise NatsPublishError(f"подставная ошибка публикации {msg_id}")
|
||||
if self.delay:
|
||||
await asyncio.sleep(self.delay)
|
||||
if self.delegate is not None:
|
||||
await self.delegate.publish(subject, payload, msg_id)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def session_factory(db_engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
|
||||
return async_sessionmaker(db_engine, expire_on_commit=False)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def jobs(session_factory: async_sessionmaker[AsyncSession]) -> AsyncIterator[CommittedJobs]:
|
||||
committed = CommittedJobs(session_factory)
|
||||
try:
|
||||
yield committed
|
||||
finally:
|
||||
await committed.cleanup()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def nats() -> AsyncIterator[NatsConnection]:
|
||||
conn = NatsConnection(get_settings().OPENBAO.NATS)
|
||||
await conn.start()
|
||||
try:
|
||||
if not await conn.wait_ready(timeout=READY_TIMEOUT):
|
||||
pytest.fail(f"NATS недоступен ({get_settings().OPENBAO.NATS.NATS_URL}, .env.test)", pytrace=False)
|
||||
yield conn
|
||||
finally:
|
||||
await conn.stop()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def subscriber() -> AsyncIterator[RequestSubscriber]:
|
||||
sub = RequestSubscriber()
|
||||
await sub.start()
|
||||
try:
|
||||
yield sub
|
||||
finally:
|
||||
await sub.stop()
|
||||
|
||||
|
||||
async def _set_status(
|
||||
session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID, status: RecognitionStatus
|
||||
) -> None:
|
||||
async with session_factory() as session, session.begin():
|
||||
await session.execute(update(RecognitionJob).where(RecognitionJob.id == job_id).values(status=status))
|
||||
|
||||
|
||||
def _publish(
|
||||
session_factory: async_sessionmaker[AsyncSession], nats: Any, now: Callable[[], datetime] | None = None
|
||||
) -> Any:
|
||||
return publish_pending(session_factory, nats, ack_timeout=ACK, max_republish=MAX_REPUBLISH, now=now)
|
||||
|
||||
|
||||
# --- юнит: соответствие enum, константы ------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_orm_file_type_matches_contract_enum():
|
||||
"""`stored_file.file_type` (DBML) и `file_type` схемы запроса — одно множество значений."""
|
||||
assert {m.value for m in FileType} == set(get_args(ContractFileType))
|
||||
assert set(REQUEST_SCHEMA["properties"]["file_type"]["enum"]) == {m.value for m in FileType}
|
||||
|
||||
|
||||
def test_orm_pipeline_matches_request_subjects():
|
||||
assert {m.value for m in RecognitionPipeline} == {"tabular", "image"}
|
||||
|
||||
|
||||
def test_watchdog_period_is_30_seconds():
|
||||
assert WATCHDOG_PERIOD_SECONDS == 30 # noqa: PLR2004 - значение из spec
|
||||
|
||||
|
||||
# --- публикация -------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.parametrize(
|
||||
("pipeline", "file_type", "subject"),
|
||||
[
|
||||
(RecognitionPipeline.tabular, FileType.XLSX, "recognition.tabular.request"),
|
||||
(RecognitionPipeline.image, FileType.PDF, "recognition.image.request"),
|
||||
],
|
||||
)
|
||||
async def test_publish_request_to_service( # noqa: PLR0913,PLR0917 - фикстуры и параметры
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
nats: NatsConnection,
|
||||
subscriber: RequestSubscriber,
|
||||
pipeline: RecognitionPipeline,
|
||||
file_type: FileType,
|
||||
subject: str,
|
||||
):
|
||||
"""Сценарий «Публикация запроса парсеру»: сообщение по схеме, `attempt = 1`, строка `published`."""
|
||||
job_id = await jobs.make_queued_job(pipeline=pipeline, file_type=file_type, original_name="ОСВ 60.xlsx")
|
||||
before = datetime.now(UTC)
|
||||
|
||||
report = await _publish(session_factory, nats)
|
||||
|
||||
assert report.published == 1
|
||||
(msg,) = await subscriber.wait_for(job_id, 1)
|
||||
body = json.loads(msg.data)
|
||||
errors = [e.message for e in REQUEST_VALIDATOR.iter_errors(body)]
|
||||
assert errors == []
|
||||
request = RecognitionRequest.model_validate_json(msg.data)
|
||||
job = await jobs.get(job_id)
|
||||
assert msg.subject == subject
|
||||
assert msg.headers == {NATS_MSG_ID_HEADER: f"{job_id}:1"}
|
||||
assert request.job_id == job_id
|
||||
assert request.file_id == job.file_id
|
||||
assert request.pipeline == pipeline.value
|
||||
assert request.file_type == file_type.value
|
||||
assert request.original_name == "ОСВ 60.xlsx"
|
||||
assert request.s3.bucket == "arm-ka"
|
||||
assert request.s3.key == await jobs.s3_key(job_id)
|
||||
assert request.attempt == 1
|
||||
assert request.schema_version == "1.0"
|
||||
# reply_deadline не был задан: вычислен при первой публикации (5 мин) и сохранён в задание
|
||||
assert job.reply_deadline is not None
|
||||
assert request.reply_deadline == job.reply_deadline
|
||||
assert timedelta(minutes=4) < job.reply_deadline - before < timedelta(minutes=6)
|
||||
|
||||
assert job.status is RecognitionStatus.published
|
||||
assert job.request_msg_id == f"{job_id}:1"
|
||||
assert job.request_subject == subject
|
||||
assert job.published_at is not None and job.published_at >= before
|
||||
assert job.requested_at is not None
|
||||
assert job.attempt == 1
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_publish_keeps_existing_reply_deadline(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
nats: NatsConnection,
|
||||
subscriber: RequestSubscriber,
|
||||
):
|
||||
deadline = datetime.now(UTC).replace(microsecond=0) + timedelta(hours=2)
|
||||
job_id = await jobs.make_queued_job(reply_deadline=deadline)
|
||||
|
||||
await _publish(session_factory, nats)
|
||||
|
||||
(msg,) = await subscriber.wait_for(job_id, 1)
|
||||
assert RecognitionRequest.model_validate_json(msg.data).reply_deadline == deadline
|
||||
assert (await jobs.get(job_id)).reply_deadline == deadline
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_first_publication_msg_id_follows_attempt(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
nats: NatsConnection,
|
||||
subscriber: RequestSubscriber,
|
||||
):
|
||||
"""`queued` с `attempt = 2` и устаревшим `request_msg_id` (будущий replay): id и тело — по текущему attempt."""
|
||||
job_id = await jobs.make_queued_job(attempt=2, request_msg_id=f"stale-{uuid.uuid4()}")
|
||||
|
||||
await _publish(session_factory, nats)
|
||||
|
||||
(msg,) = await subscriber.wait_for(job_id, 1)
|
||||
assert msg.headers == {NATS_MSG_ID_HEADER: f"{job_id}:2"}
|
||||
assert RecognitionRequest.model_validate_json(msg.data).attempt == 2 # noqa: PLR2004 - attempt задания
|
||||
job = await jobs.get(job_id)
|
||||
assert (job.request_msg_id, job.attempt) == (f"{job_id}:2", 2)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_invalid_request_marks_job_error_and_pass_continues(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
nats: NatsConnection,
|
||||
subscriber: RequestSubscriber,
|
||||
):
|
||||
"""Запрос не собирается по схеме (нет `original_name`) — задание `error/INVALID_REQUEST`, проход идёт дальше."""
|
||||
broken = await jobs.make_queued_job(original_name=None)
|
||||
good = await jobs.make_queued_job()
|
||||
|
||||
report = await _publish(session_factory, nats)
|
||||
|
||||
assert (report.invalid, report.published) == (1, 1)
|
||||
await subscriber.wait_for(good, 1)
|
||||
await subscriber.settle()
|
||||
assert subscriber.for_job(broken) == []
|
||||
job = await jobs.get(broken)
|
||||
assert job.status is RecognitionStatus.error
|
||||
assert job.error_code == "INVALID_REQUEST"
|
||||
assert job.error_message and "original_name" in job.error_message
|
||||
assert job.finished_at is not None
|
||||
assert job.request_msg_id is None
|
||||
|
||||
|
||||
# --- повторы без ack и dlq ---------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_republish_without_ack_then_dlq(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
nats: NatsConnection,
|
||||
subscriber: RequestSubscriber,
|
||||
):
|
||||
"""Сценарий «Сервис недоступен»: 1 + 3 публикации с одним `Nats-Msg-Id`, затем `dlq / NO_ACK`."""
|
||||
clock = Clock()
|
||||
job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1), metrics={"note": "сохранить"})
|
||||
|
||||
assert (await _publish(session_factory, nats, clock)).published == 1
|
||||
first = await jobs.get(job_id)
|
||||
# раньше ACK_TIMEOUT повтора нет
|
||||
clock.advance(ACK - timedelta(seconds=1))
|
||||
assert (await _publish(session_factory, nats, clock)).republished == 0
|
||||
|
||||
for n in range(1, MAX_REPUBLISH + 1):
|
||||
clock.advance(timedelta(seconds=2)) # с последней публикации прошло больше ACK_TIMEOUT
|
||||
report = await _publish(session_factory, nats, clock)
|
||||
assert report.republished == 1
|
||||
job = await jobs.get(job_id)
|
||||
assert job.metrics == {"note": "сохранить", "republish_count": n}
|
||||
assert job.published_at == clock()
|
||||
assert job.status is RecognitionStatus.published
|
||||
clock.advance(ACK - timedelta(seconds=1))
|
||||
|
||||
messages = await subscriber.wait_for(job_id, 1 + MAX_REPUBLISH)
|
||||
assert {m.headers[NATS_MSG_ID_HEADER] for m in messages} == {f"{job_id}:1"}
|
||||
assert len({m.data for m in messages}) == 1 # то же тело: тот же attempt и reply_deadline
|
||||
|
||||
clock.advance(timedelta(seconds=2))
|
||||
report = await _publish(session_factory, nats, clock)
|
||||
assert (report.dlq, report.republished, report.published) == (1, 0, 0)
|
||||
job = await jobs.get(job_id)
|
||||
assert job.status is RecognitionStatus.dlq
|
||||
assert job.error_code == "NO_ACK"
|
||||
assert job.error_message
|
||||
assert job.finished_at == clock()
|
||||
assert job.request_msg_id == first.request_msg_id == f"{job_id}:1"
|
||||
assert job.attempt == 1
|
||||
assert job.metrics == {"note": "сохранить", "republish_count": MAX_REPUBLISH}
|
||||
|
||||
clock.advance(ACK * 2)
|
||||
assert (await _publish(session_factory, nats, clock)).total == 0
|
||||
await subscriber.settle()
|
||||
assert len(subscriber.for_job(job_id)) == 1 + MAX_REPUBLISH
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_ack_after_first_republish_stops_republishing(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
nats: NatsConnection,
|
||||
subscriber: RequestSubscriber,
|
||||
):
|
||||
"""Сценарий «Сервис появился после первого повтора»: после ack (`processing`) повторов нет."""
|
||||
clock = Clock()
|
||||
job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1))
|
||||
await _publish(session_factory, nats, clock)
|
||||
clock.advance(ACK + timedelta(seconds=1))
|
||||
assert (await _publish(session_factory, nats, clock)).republished == 1
|
||||
await subscriber.wait_for(job_id, 2)
|
||||
|
||||
await _set_status(session_factory, job_id, RecognitionStatus.processing) # имитация ack
|
||||
for _ in range(MAX_REPUBLISH + 1):
|
||||
clock.advance(ACK + timedelta(seconds=1))
|
||||
assert (await _publish(session_factory, nats, clock)).total == 0
|
||||
|
||||
await subscriber.settle()
|
||||
assert len(subscriber.for_job(job_id)) == 2 # noqa: PLR2004 - первая публикация и один повтор
|
||||
job = await jobs.get(job_id)
|
||||
assert job.status is RecognitionStatus.processing
|
||||
assert job.metrics == {"republish_count": 1}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_publisher_skips_published_job_with_passed_deadline(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
nats: NatsConnection,
|
||||
subscriber: RequestSubscriber,
|
||||
):
|
||||
"""`published` с прошедшим `reply_deadline` не републикуется и не уходит в dlq — его забирает вахтёр."""
|
||||
now = datetime.now(UTC)
|
||||
job_id = await jobs.make_queued_job(
|
||||
status=RecognitionStatus.published,
|
||||
published_at=now - timedelta(hours=1),
|
||||
reply_deadline=now - timedelta(minutes=1),
|
||||
request_msg_id=f"{uuid.uuid4()}:1",
|
||||
metrics={"republish_count": MAX_REPUBLISH},
|
||||
)
|
||||
|
||||
assert (await _publish(session_factory, nats)).total == 0
|
||||
await subscriber.settle()
|
||||
assert subscriber.for_job(job_id) == []
|
||||
assert (await jobs.get(job_id)).status is RecognitionStatus.published
|
||||
|
||||
assert await expire_deadlines(session_factory) == 1
|
||||
job = await jobs.get(job_id)
|
||||
assert (job.status, job.error_code) == (RecognitionStatus.timeout, "TIMEOUT")
|
||||
|
||||
|
||||
# --- NATS недоступен ---------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_pass_skipped_when_nats_disconnected(
|
||||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs
|
||||
):
|
||||
job_id = await jobs.make_queued_job()
|
||||
fake = FakeNats(connected=False)
|
||||
|
||||
report = await _publish(session_factory, fake)
|
||||
|
||||
assert report.skipped
|
||||
assert fake.calls == []
|
||||
assert (await jobs.get(job_id)).status is RecognitionStatus.queued
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_publish_error_keeps_queued_and_stops_pass(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
nats: NatsConnection,
|
||||
subscriber: RequestSubscriber,
|
||||
):
|
||||
"""Сценарий «NATS недоступен»: задание остаётся `queued`, проход прерван; следующий проход публикует."""
|
||||
first = await jobs.make_queued_job()
|
||||
second = await jobs.make_queued_job()
|
||||
fake = FakeNats(fail_first=1)
|
||||
|
||||
report = await _publish(session_factory, fake)
|
||||
|
||||
assert (report.failed, report.published) == (1, 0)
|
||||
assert len(fake.calls) == 1 # второе задание в этом проходе не пытались публиковать
|
||||
for job_id in (first, second):
|
||||
job = await jobs.get(job_id)
|
||||
assert job.status is RecognitionStatus.queued
|
||||
assert job.request_msg_id is None
|
||||
assert job.published_at is None
|
||||
assert job.reply_deadline is None # откат: срок не сохранён
|
||||
|
||||
report = await _publish(session_factory, nats)
|
||||
|
||||
assert report.published == 2 # noqa: PLR2004 - оба задания
|
||||
for job_id in (first, second):
|
||||
await subscriber.wait_for(job_id, 1)
|
||||
assert (await jobs.get(job_id)).status is RecognitionStatus.published
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_republish_error_keeps_published_state(
|
||||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection
|
||||
):
|
||||
clock = Clock()
|
||||
job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1))
|
||||
await _publish(session_factory, nats, clock)
|
||||
published_at = (await jobs.get(job_id)).published_at
|
||||
clock.advance(ACK + timedelta(seconds=1))
|
||||
|
||||
report = await _publish(session_factory, FakeNats(fail_first=1), clock)
|
||||
|
||||
assert report.failed == 1
|
||||
job = await jobs.get(job_id)
|
||||
assert job.status is RecognitionStatus.published
|
||||
assert job.published_at == published_at
|
||||
assert job.metrics is None
|
||||
|
||||
|
||||
# --- конкурентность -----------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_concurrent_publishers_publish_each_job_once(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
nats: NatsConnection,
|
||||
subscriber: RequestSubscriber,
|
||||
):
|
||||
"""Два одновременных прохода (SKIP LOCKED): каждое задание опубликовано ровно один раз."""
|
||||
job_ids = [await jobs.make_queued_job() for _ in range(CONCURRENT_JOBS)]
|
||||
# задержка внутри публикации держит блокировку строки, пока второй проход выбирает задания
|
||||
first, second = FakeNats(delegate=nats, delay=0.02), FakeNats(delegate=nats, delay=0.02)
|
||||
|
||||
reports = await asyncio.gather(_publish(session_factory, first), _publish(session_factory, second))
|
||||
|
||||
assert sum(r.published for r in reports) == CONCURRENT_JOBS
|
||||
assert all(r.published > 0 for r in reports) # оба прохода реально работали параллельно
|
||||
published_ids = [msg_id for fake in (first, second) for _, msg_id in fake.calls]
|
||||
assert sorted(published_ids) == sorted(f"{job_id}:1" for job_id in job_ids)
|
||||
await subscriber.settle()
|
||||
for job_id in job_ids:
|
||||
assert len(await subscriber.wait_for(job_id, 1)) == 1
|
||||
assert (await jobs.get(job_id)).status is RecognitionStatus.published
|
||||
|
||||
|
||||
# --- вахтёр ----------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.parametrize(
|
||||
("status", "deadline_offset", "expected"),
|
||||
[
|
||||
(RecognitionStatus.processing, timedelta(minutes=-1), RecognitionStatus.timeout),
|
||||
(RecognitionStatus.published, timedelta(minutes=-1), RecognitionStatus.timeout),
|
||||
(RecognitionStatus.processing, timedelta(minutes=5), RecognitionStatus.processing),
|
||||
(RecognitionStatus.published, timedelta(minutes=5), RecognitionStatus.published),
|
||||
(RecognitionStatus.processing, None, RecognitionStatus.processing),
|
||||
(RecognitionStatus.queued, timedelta(minutes=-1), RecognitionStatus.queued),
|
||||
(RecognitionStatus.done, timedelta(minutes=-1), RecognitionStatus.done),
|
||||
(RecognitionStatus.dlq, timedelta(minutes=-1), RecognitionStatus.dlq),
|
||||
],
|
||||
)
|
||||
async def test_expire_deadlines(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
status: RecognitionStatus,
|
||||
deadline_offset: timedelta | None,
|
||||
expected: RecognitionStatus,
|
||||
):
|
||||
"""Сценарий «Сервис взял, но не ответил»: `processing`/`published` с прошедшим сроком → `timeout`."""
|
||||
clock = Clock()
|
||||
deadline = None if deadline_offset is None else clock() + deadline_offset
|
||||
job_id = await jobs.make_queued_job(status=status, reply_deadline=deadline)
|
||||
|
||||
expired = await expire_deadlines(session_factory, now=clock)
|
||||
|
||||
job = await jobs.get(job_id)
|
||||
assert job.status is expected
|
||||
if expected is RecognitionStatus.timeout:
|
||||
assert expired == 1
|
||||
assert job.error_code == "TIMEOUT"
|
||||
assert job.error_message
|
||||
assert job.finished_at == clock()
|
||||
assert job.attempt == 1
|
||||
else:
|
||||
assert expired == 0
|
||||
assert job.error_code is None
|
||||
assert job.finished_at is None
|
||||
|
||||
|
||||
# --- фоновые циклы -----------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _wait_until(predicate: Callable[[], bool], timeout: float = 2.0) -> None:
|
||||
deadline = time.monotonic() + timeout
|
||||
while not predicate():
|
||||
if time.monotonic() > deadline:
|
||||
raise AssertionError("условие не выполнено за отведённое время")
|
||||
await asyncio.sleep(0.01)
|
||||
|
||||
|
||||
async def _cancel(task: asyncio.Task[Any]) -> None:
|
||||
task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await task
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_publisher_loop_publishes_and_stops_on_cancel(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
nats: NatsConnection,
|
||||
subscriber: RequestSubscriber,
|
||||
):
|
||||
job_id = await jobs.make_queued_job()
|
||||
task = asyncio.create_task(
|
||||
run_publisher_loop(session_factory, nats, poll_seconds=0.1, ack_timeout=ACK, max_republish=MAX_REPUBLISH)
|
||||
)
|
||||
try:
|
||||
await subscriber.wait_for(job_id, 1, timeout=2.0)
|
||||
finally:
|
||||
await _cancel(task)
|
||||
assert (await jobs.get(job_id)).status is RecognitionStatus.published
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_watchdog_loop_expires_and_stops_on_cancel(
|
||||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs
|
||||
):
|
||||
job_id = await jobs.make_queued_job(
|
||||
status=RecognitionStatus.processing, reply_deadline=datetime.now(UTC) - timedelta(seconds=1)
|
||||
)
|
||||
task = asyncio.create_task(run_watchdog_loop(session_factory, period_seconds=0.1))
|
||||
try:
|
||||
deadline = time.monotonic() + 2.0
|
||||
while (await jobs.get(job_id)).status is not RecognitionStatus.timeout:
|
||||
assert time.monotonic() < deadline, "вахтёр не перевёл задание в timeout за 2 с"
|
||||
await asyncio.sleep(0.05)
|
||||
finally:
|
||||
await _cancel(task)
|
||||
|
||||
|
||||
class RecordingSleep:
|
||||
"""Подмена `sleep` циклов: запоминает паузы и только уступает управление."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.delays: list[float] = []
|
||||
|
||||
async def __call__(self, delay: float) -> None:
|
||||
self.delays.append(delay)
|
||||
await asyncio.sleep(0)
|
||||
|
||||
|
||||
async def test_publisher_loop_survives_pass_errors(monkeypatch: pytest.MonkeyPatch):
|
||||
calls: list[int] = []
|
||||
|
||||
async def failing_pass(*args: Any, **kwargs: Any) -> Any:
|
||||
calls.append(1)
|
||||
raise RuntimeError("БД недоступна")
|
||||
|
||||
monkeypatch.setattr(rx, "publish_pending", failing_pass)
|
||||
sleep = RecordingSleep()
|
||||
task = asyncio.create_task(
|
||||
run_publisher_loop(
|
||||
object(), # type: ignore[arg-type] # проход подменён: фабрика сессий не используется
|
||||
FakeNats(),
|
||||
poll_seconds=5,
|
||||
ack_timeout=ACK,
|
||||
max_republish=MAX_REPUBLISH,
|
||||
sleep=sleep,
|
||||
)
|
||||
)
|
||||
try:
|
||||
await _wait_until(lambda: len(calls) >= MIN_LOOP_PASSES)
|
||||
finally:
|
||||
await _cancel(task)
|
||||
assert set(sleep.delays) == {5}
|
||||
|
||||
|
||||
async def test_watchdog_loop_survives_pass_errors(monkeypatch: pytest.MonkeyPatch):
|
||||
calls: list[int] = []
|
||||
|
||||
async def failing_pass(*args: Any, **kwargs: Any) -> int:
|
||||
calls.append(1)
|
||||
raise RuntimeError("БД недоступна")
|
||||
|
||||
monkeypatch.setattr(rx, "expire_deadlines", failing_pass)
|
||||
sleep = RecordingSleep()
|
||||
task = asyncio.create_task(run_watchdog_loop(object(), sleep=sleep)) # type: ignore[arg-type]
|
||||
try:
|
||||
await _wait_until(lambda: len(calls) >= MIN_LOOP_PASSES)
|
||||
finally:
|
||||
await _cancel(task)
|
||||
assert set(sleep.delays) == {WATCHDOG_PERIOD_SECONDS}
|
||||
|
||||
|
||||
async def test_exchange_tasks_start_two_loops_and_stop(monkeypatch: pytest.MonkeyPatch):
|
||||
started: dict[str, dict[str, Any]] = {}
|
||||
|
||||
async def fake_publisher(session_factory: Any, nats: Any, **kwargs: Any) -> None:
|
||||
started["publisher"] = kwargs
|
||||
await asyncio.Event().wait()
|
||||
|
||||
async def fake_watchdog(session_factory: Any, **kwargs: Any) -> None:
|
||||
started["watchdog"] = kwargs
|
||||
await asyncio.Event().wait()
|
||||
|
||||
monkeypatch.setattr(rx, "run_publisher_loop", fake_publisher)
|
||||
monkeypatch.setattr(rx, "run_watchdog_loop", fake_watchdog)
|
||||
settings = get_settings().model_copy(
|
||||
update={"OUTBOX_POLL_SECONDS": 7, "ACK_TIMEOUT_SECONDS": 11, "OUTBOX_MAX_REPUBLISH": 2}
|
||||
)
|
||||
|
||||
exchange = ExchangeTasks.start(object(), FakeNats(), settings) # type: ignore[arg-type]
|
||||
await _wait_until(lambda: len(started) == EXCHANGE_TASKS)
|
||||
|
||||
assert len(exchange.tasks) == EXCHANGE_TASKS
|
||||
assert started["publisher"] == {"poll_seconds": 7, "ack_timeout": timedelta(seconds=11), "max_republish": 2}
|
||||
assert started["watchdog"] == {"period_seconds": WATCHDOG_PERIOD_SECONDS}
|
||||
await exchange.stop()
|
||||
assert all(task.done() for task in exchange.tasks)
|
||||
await exchange.stop() # повторная остановка безопасна
|
||||
Loading…
x
Reference in New Issue
Block a user