> Как проверять корректность работы сервиса и обработку сообщений по разным приоритетам вовремя (Python)

Короткий ответ

Проверка корректности работы сервиса с приоритетами сводится к трём уровням: unit-тесты на логику приоритизации, интеграционные тесты на обработку сообщений в правильном порядке, и observability (метрики, логи, трейсы) для контроля в проде. Для проверки "вовремя" нужно определить SLO на latency по каждому приоритету и настроить алерты на их превышение. Ключевой инструмент - тесты с контролем времени (fake time, freeze_time) и нагрузочные тесты с реальными задержками.

Подробное объяснение

Корректность работы сервиса с приоритетами включает два аспекта: правильность (сообщение обработано ровно один раз, не потеряно, не продублировано) и своевременность (сообщение с высоким приоритетом обработано раньше и быстрее, чем с низким).

Для проверки правильности:

• юнит-тесты на компонент приоритизации (очередь, планировщик, выбор следующего сообщения);
• тесты на идемпотентность обработчиков - повторная доставка не приводит к побочным эффектам;
• тесты на обработку ошибок: сообщение не теряется при сбое, уходит в DLQ (dead letter queue) после N ретраев.

Для проверки своевременности:

• тесты с контролем времени: замораживаем часы, отправляем сообщения разных приоритетов, проверяем порядок обработки;
• тесты на starvation: низкоприоритетные сообщения не блокируются бесконечно высокоприоритетными (нужен механизм aging или весовой планировщик);
• нагрузочные тесты: под нагрузкой замеряем p99 latency по каждому приоритету и сравниваем с SLO.

Важно разделять порядок обработки (ordering) и latency. Порядок гарантируется очередью, latency - планировщиком и ресурсами. Для проверки "вовремя" нужно:

• определить SLO: например, p99 для high - 100ms, для low - 10s;
• настроить метрики: latency по приоритету, количество сообщений в очереди по приоритету, время ожидания в очереди;
• алерты на превышение SLO и на рост очереди.

На практике

Начинаем с определения контракта: какие приоритеты есть, какие гарантии по latency и ordering. Затем пишем тесты в порядке: юнит → интеграционные → нагрузочные.

Для интеграционных тестов используем реальную очередь (RabbitMQ, Redis, Kafka) в Docker, но с контролем времени через freezegun или time_machine. Это позволяет проверить, что сообщение с high приоритетом, отправленное позже, обработается раньше low.

Для проверки "вовремя" в проде - метрики в Prometheus + Grafana, алерты в Alertmanager. Ключевые метрики: message_processing_duration_seconds{priority="high"} (histogram), queue_depth{priority="high"}, message_wait_time_seconds{priority="low"}.

Также полезно добавить сквозной тест (end-to-end): отправляем сообщения всех приоритетов, ждём, проверяем, что все обработаны, и что порядок и время соответствуют ожиданиям. Такой тест можно запускать по расписанию в CI или отдельным прогоном перед релизом.

Пример кода

PYTHON
import asyncio
from time_machine import travel
from unittest.mock import AsyncMock, patch

class PriorityQueue:
    def __init__(self):
        self.queues = {1: [], 2: [], 3: []}  # 1 - high, 3 - low

    def push(self, priority, message):
        self.queues[priority].append(message)

    async def pop(self):
        for priority in sorted(self.queues):
            if self.queues[priority]:
                return priority, self.queues[priority].pop(0)
        return None

async def test_high_priority_processed_first():
    queue = PriorityQueue()
    queue.push(3, "low")
    queue.push(1, "high")

    with travel("2024-01-01 00:00:00", tick=False):
        priority, msg = await queue.pop()
        assert priority == 1
        assert msg == "high"

async def test_sla_for_high_priority():
    queue = PriorityQueue()
    queue.push(1, "urgent")
    start = asyncio.get_event_loop().time()

    with patch("asyncio.sleep", new=AsyncMock()) as mock_sleep:
        priority, msg = await queue.pop()
        # имитация обработки
        await asyncio.sleep(0.05)
        elapsed = asyncio.get_event_loop().time() - start
        assert elapsed < 0.1  # SLO для high

Как отвечать на собеседовании

Начни с трёхуровневой схемы: юнит-тесты, интеграционные, observability. Подчеркни, что "вовремя" - это не только порядок, но и метрики latency. Упомяни trade-off: строгий порядок по приоритетам может привести к starvation, поэтому нужен aging или весовой планировщик. Приведи пример метрик и алертов. Если спросят про конкретную очередь - расскажи про RabbitMQ (priority queues) или Kafka (партиции по приоритету). Заверши мыслью, что без SLO и алертов проверка "вовремя" невозможна.

Что проверяет интервьюер

• понимание разницы между порядком обработки и latency;
• умение проектировать тесты с контролем времени;
• знание паттернов observability для очередей;
• осознание проблемы starvation и способов её решения;
• практический опыт с очередями и приоритизацией.

Типичные ошибки

• путать порядок обработки с гарантией "вовремя" - порядок не гарантирует latency;
• не учитывать starvation: тест только на то, что high обрабатывается первым, но не проверяет, что low вообще обработается;
• использовать time.sleep в тестах вместо контроля времени - тесты становятся медленными и флаки;
• забывать про метрики в проде: тесты проверяют логику, но не реальную нагрузку;
• не определять SLO заранее - без них невозможно сказать, "вовремя" или нет.