Consumer patterns: идемпотентность, retry, outbox, graceful shutdown

Это последний урок трека. В предыдущих изучили RabbitMQ и Kafka. Но просто знать API недостаточно для production - нужны patterns для reliability. В этом уроке - идемпотентность, retry с backoff, outbox pattern для consistency, graceful shutdown, plus обзор task queues (Celery, ARQ).

Идемпотентность - foundational principle

Поскольку message systems дают at-least-once гарантию, consumer может получить одно message несколько раз (retry после network blip, restart consumer, reprocessing). Идемпотентный consumer обрабатывает duplicate безопасно - second processing даёт same result что и первое.

Не идемпотентно:

async def process(message):
    user_id = message["user_id"]
    amount = message["amount"]

    # Каждый вызов добавляет к balance
    await db.execute("UPDATE users SET balance = balance + ? WHERE id = ?", amount, user_id)

При retry баланс прибавится дважды - bug.

Идемпотентно через message ID:

async def process(message):
    message_id = message["message_id"]   # unique per message

    async with db.transaction():
        # Check уже processed
        existing = await db.fetchone("SELECT 1 FROM processed_messages WHERE id = ?", message_id)
        if existing:
            logger.info(f"Skip duplicate: {message_id}")
            return

        # Process
        user_id = message["user_id"]
        amount = message["amount"]
        await db.execute("UPDATE users SET balance = balance + ? WHERE id = ?", amount, user_id)

        # Mark as processed
        await db.execute("INSERT INTO processed_messages (id, processed_at) VALUES (?, NOW())", message_id)

Transaction атомарно: process + mark, либо ничего. Repeat ничего не делает.

Idempotency через UPSERT

Для некоторых операций можно использовать idempotent SQL:

# Создание user (идемпотентно через unique email)
await db.execute("""
    INSERT INTO users (email, name) VALUES (?, ?)
    ON CONFLICT (email) DO NOTHING
""", email, name)

# Status update (set value, не increment)
await db.execute("UPDATE orders SET status = 'completed' WHERE id = ?", order_id)
# Repeat = same result

Inherent idempotency предпочтительнее explicit tracking - проще.

Retry с exponential backoff

При transient failures (network blip, downstream temporarily slow) - повторить:

import asyncio
import logging

logger = logging.getLogger(__name__)

async def process_with_retry(message, max_retries=3, base_delay=1.0):
    for attempt in range(max_retries):
        try:
            return await actual_process(message)
        except TransientError as e:
            if attempt == max_retries - 1:
                raise   # дальше пусть DLQ обрабатывает
            delay = base_delay * (2 ** attempt)   # 1s, 2s, 4s
            logger.warning(f"Attempt {attempt + 1} failed: {e}, retry in {delay}s")
            await asyncio.sleep(delay)
        except PermanentError:
            # бессмысленно retry - сразу dead letter
            logger.error(f"Permanent failure: {message}")
            raise

Различай transient (retry поможет) и permanent (retry бесполезен). Catch broadly типы исключений.

Готовые библиотеки:

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type(TransientError),
)
async def process(message):
    # ...

tenacity - mature library с many options. Cleaner чем self-rolled retry.

Outbox pattern - consistency

Проблема dual write: хочется сохранить в БД + publish в broker. Если делать раздельно, возможны inconsistency states:

# ПЛОХО - возможна inconsistency
async def create_order(data):
    order = await db.save(data)   # OK
    await broker.publish("order.created", order)   # network fail - event потерян
    return order

Outbox pattern:

# В одной transaction
async def create_order(data):
    async with db.transaction():
        order = await db.save_order(data)
        await db.insert("outbox", {
            "id": uuid4(),
            "topic": "order.created",
            "payload": order.json(),
            "created_at": utcnow(),
        })
    return order

# Background worker
async def outbox_publisher():
    while True:
        pending = await db.fetch("SELECT * FROM outbox WHERE published_at IS NULL LIMIT 100")
        for row in pending:
            try:
                await broker.publish(row.topic, row.payload)
                await db.update("UPDATE outbox SET published_at = NOW() WHERE id = ?", row.id)
            except Exception:
                logger.exception("Failed to publish from outbox")
        await asyncio.sleep(1)

Atomicity гарантируется БД. Worker periodically publishes. Гарантия: либо order saved и event eventually published, либо нет ни того ни другого.

Аналогичный pattern для consumer (inbox pattern) - сохранить incoming message в БД, потом обрабатывать.

Dead Letter Queue (DLQ) pattern

Для messages которые систематически fail - в DLQ:

async def process_with_dlq(message):
    try:
        await actual_process(message)
    except Exception as e:
        logger.exception(f"Failed to process {message}")
        await dlq.publish({
            "original_message": message,
            "error": str(e),
            "failed_at": utcnow().isoformat(),
            "attempt_count": message.get("attempt_count", 0) + 1,
        })
        # ACK original чтобы не retry endlessly

DLQ для inspection: что ломается? data corruption? bug в processing? Можно после фикса replay из DLQ.

Graceful shutdown

Container/process termination (Docker stop, k8s rolling update) шлёт SIGTERM. Consumer должен dorobотать текущее и завершиться:

import asyncio
import signal
import logging

logger = logging.getLogger(__name__)

class Consumer:
    def __init__(self):
        self.stop_event = asyncio.Event()

    async def start(self):
        loop = asyncio.get_running_loop()
        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(sig, self.stop_event.set)

        # init broker, queue
        consumer = await create_consumer()
        try:
            async for message in consumer:
                if self.stop_event.is_set():
                    logger.info("Stop requested, finishing current message")
                    break

                async with message.process():
                    await self.handle(message)
        finally:
            await consumer.stop()
            logger.info("Consumer stopped cleanly")

asyncio.run(Consumer().start())

В Docker shutdown timeout обычно 10s. В k8s terminationGracePeriodSeconds (default 30s). Если processing slow - больше timeout или break processing на checkpoints.

Concurrent processing с capped concurrency

Один consumer может обрабатывать messages параллельно через asyncio:

import asyncio

async def consume_concurrent():
    consumer = await create_consumer()
    semaphore = asyncio.Semaphore(10)   # max 10 concurrent

    async def process_one(message):
        async with semaphore:
            await actual_process(message.body)
            await message.ack()

    tasks = set()
    async for message in consumer:
        task = asyncio.create_task(process_one(message))
        tasks.add(task)
        task.add_done_callback(tasks.discard)

        # Prevent unbounded tasks
        if len(tasks) >= 100:
            await asyncio.sleep(0.1)

Идеально когда processing IO-bound (внешние API calls). CPU-bound - используй ProcessPoolExecutor.

Task queues - high-level abstractions

Для tasks (одна функция processed asynchronously) - frameworks упрощают:

Celery (классический)

from celery import Celery

app = Celery("tasks", broker="amqp://localhost", backend="redis://localhost")

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def send_email(self, user_id):
    try:
        user = User.get(user_id)
        send_mail(user.email, "Hello!")
    except SMTPException as e:
        raise self.retry(exc=e, countdown=2 ** self.request.retries)

# Usage
send_email.delay(user_id=42)

# Worker
# celery -A tasks worker --loglevel=info

Celery features:

• Retries с backoff
• Scheduled tasks (Celery Beat)
• Result backend (получить result async)
• Chains, groups, chords (workflows)
• Mature и stable

ARQ (async-first)

from arq import create_pool
from arq.connections import RedisSettings

async def send_email(ctx, user_id):
    user = await User.get(user_id)
    await async_send_mail(user.email, "Hello!")

class WorkerSettings:
    functions = [send_email]
    redis_settings = RedisSettings()

# Schedule
async def main():
    redis = await create_pool(RedisSettings())
    await redis.enqueue_job("send_email", user_id=42)

# Worker
# arq tasks.WorkerSettings

ARQ async-первое, использует Redis. Меньше features чем Celery, но проще для async stack. Что бы ты ни выбрал, консумер обязан писать структурированные логи и метрики - см. уроки про logging и метрики в Prometheus.

Dramatiq

import dramatiq
from dramatiq.brokers.rabbitmq import RabbitmqBroker

broker = RabbitmqBroker(host="localhost")
dramatiq.set_broker(broker)

@dramatiq.actor(max_retries=3)
def send_email(user_id):
    user = User.get(user_id)
    send_mail(user.email, "Hello!")

# Schedule
send_email.send(user_id=42)

Dramatiq - middle ground: проще Celery, более mature чем ARQ. Sync only основной, для async требует доработки.

Раздельная задача vs in-process

Когда использовать task queue framework vs raw broker?

Task queue (Celery, ARQ):

• Discrete tasks (send email, generate PDF, process upload)
• Retry logic нужна
• Scheduled tasks
• Result tracking
• Workflows (task A → task B → task C)

Raw broker (aio-pika, aiokafka):

• Event-driven architecture
• High throughput / streaming
• Complex routing
• Tight control
• Не нужны task framework features

Producer-consumer rate limiting

Если producer медленный, consumer wait. Если producer fast - может overwhelm consumer. Backpressure:

# RabbitMQ - publisher flow control автоматически (broker сообщает)

# Kafka producer - settings
producer = AIOKafkaProducer(
    bootstrap_servers="...",
    max_in_flight_requests_per_connection=5,   # ограничение
    buffer_memory=32 * 1024 * 1024,              # 32MB buffer
)

Consumer prefetch ограничивает (RabbitMQ) или Kafka rebalance distributes work. Это automatic для большинства случаев.

Monitoring и observability

Critical metrics:

Tools:

• Prometheus + RabbitMQ exporter / Kafka exporter
• Grafana dashboards
• Application metrics через prometheus_client
• Alerts на anomalies

Структурированное логирование

import structlog

logger = structlog.get_logger()

async def process(message):
    log = logger.bind(message_id=message["id"], topic="orders")
    log.info("processing")
    try:
        await actual_process(message)
        log.info("processed")
    except Exception as e:
        log.error("failed", error=str(e), exc_info=True)
        raise

Structured logs (JSON) с context (message_id, topic, user_id) - easy фильтровать в Loki/ELK при debugging.

Schema evolution

Если payload format меняется (new fields, deprecated fields):

# Producer
event = {
    "version": 2,
    "user_id": user.id,
    "email": user.email,
    "new_field": "value",   # added in v2
}

# Consumer должен handle обе версии
def process(event):
    if event.get("version", 1) == 1:
        return process_v1(event)
    else:
        return process_v2(event)

Стратегии:

• Avro + Schema Registry (Kafka standard) - automatic schema evolution
• Protobuf - backward-compatible если не change required fields
• Manual versioning - в payload и code paths

Schema evolution критична для long-lived event streams. План эволюции заранее.

Testing consumers

import pytest
import aio_pika

@pytest.mark.asyncio
async def test_consumer_processes_message():
    # Mock broker через test connection
    connection = await aio_pika.connect_robust("amqp://localhost/test")
    async with connection:
        channel = await connection.channel()
        queue = await channel.declare_queue("test_queue")

        # Publish test message
        await channel.default_exchange.publish(
            aio_pika.Message(b'{"user_id": 1}'),
            routing_key="test_queue",
        )

        # Consumer reads и processes
        async with queue.iterator() as it:
            async for message in it:
                async with message.process():
                    data = json.loads(message.body)
                    assert data["user_id"] == 1
                    break

Для unit tests: mock broker (testcontainers), test processing logic отдельно от broker integration.

Production checklist

Перед deploy consumer в production:

• Идемпотентность через message_id или upsert
• Retry logic для transient failures
• DLQ для permanent failures
• Graceful shutdown с SIGTERM handler
• Resource limits (prefetch, concurrency)
• Monitoring и alerts
• Structured logs с context
• Tests для critical paths
• Schema management plan
• Outbox pattern если consistency критична
• Documented operational runbook

Полный пример - production-ready consumer

import asyncio
import json
import logging
import signal
from uuid import uuid4

import aio_pika
import structlog
from tenacity import retry, stop_after_attempt, wait_exponential

from todo_api.db import async_session
from todo_api.models import ProcessedMessage

logger = structlog.get_logger()

class OrderConsumer:
    def __init__(self, amqp_url: str):
        self.amqp_url = amqp_url
        self.stop_event = asyncio.Event()

    async def start(self):
        loop = asyncio.get_running_loop()
        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(sig, self.stop_event.set)

        connection = await aio_pika.connect_robust(self.amqp_url)
        async with connection:
            channel = await connection.channel()
            await channel.set_qos(prefetch_count=20)

            queue = await channel.declare_queue(
                "orders.created",
                durable=True,
                arguments={
                    "x-dead-letter-exchange": "orders.dlx",
                },
            )

            async with queue.iterator() as it:
                async for message in it:
                    if self.stop_event.is_set():
                        logger.info("Shutdown requested")
                        break

                    log = logger.bind(message_id=message.message_id or "unknown")
                    async with message.process(requeue=True):
                        try:
                            await self.handle_message(message, log)
                        except Exception:
                            log.exception("Processing failed - requeue or DLQ")
                            raise

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
    )
    async def handle_message(self, message, log):
        data = json.loads(message.body)
        message_id = data.get("message_id", str(uuid4()))

        async with async_session() as session:
            # Idempotency check
            existing = await session.get(ProcessedMessage, message_id)
            if existing:
                log.info("Skipping duplicate")
                return

            # Process
            log.info("Processing", order_id=data["order_id"])
            await self.do_business_logic(data, session)

            # Mark processed (in same transaction)
            session.add(ProcessedMessage(id=message_id))
            await session.commit()

    async def do_business_logic(self, data, session):
        # Real work
        await send_confirmation_email(data["user_id"], data["order_id"])
        await update_inventory(data["items"], session)

async def main():
    consumer = OrderConsumer("amqp://localhost/")
    await consumer.start()

if __name__ == "__main__":
    asyncio.run(main())

Это template для production consumer. Все patterns: idempotency, retry, DLQ, graceful shutdown, logging.

Заключение по Модулю 11 и треку

Modуль 11 завершён - освоили message brokers (RabbitMQ, Kafka), производственные паттерны для consumers.

Весь трек 11 модулей (59 уроков):

1. Старт и инструменты (5)
2. Типы и значения (5)
3. Поток выполнения (6)
4. Функции (5)
5. Коллекции (5)
6. ООП (6)
7. Итерация и асинхронность (6)
8. Модули, packaging, stdlib (5)
9. Тестирование и качество (4)
10. Веб-бэкенд и продакшн (8)
11. Async messaging и consumers (4)

Что ты теперь умеешь:

• Идиоматичный Python код (modules 1-8)
• Тестируешь и поддерживаешь качество (module 9)
• Создаёшь REST API на FastAPI с auth, БД, миграциями, Docker (module 10)
• Используешь message brokers для async architectures (module 11)

Куда расти дальше:

• Реальный production проект (best teacher)
• Distributed systems (микросервисы, distributed tracing, observability)
• Performance optimization, profiling
• DevOps: Kubernetes, monitoring, GitOps
• Specific domains: ML serving, streaming analytics, fintech

Поздравляю с прохождением трека. У тебя сейчас базовый стек modern Python backend разработчика. Дальше - применять на практике в real проектах.

Мини-задание (финальное)

Возьми финальный проект из урока 55 (TODO API) и:

1. Добавь outbox таблицу:

# models.py
class OutboxEvent(Base):
    __tablename__ = "outbox"
    id: Mapped[UUID] = mapped_column(primary_key=True)
    topic: Mapped[str]
    payload: Mapped[str]   # JSON
    created_at: Mapped[datetime]
    published_at: Mapped[datetime | None] = mapped_column(default=None)

2. При создании todo - сохраняй event в outbox в той же transaction:

async def create_todo(user_id, data, session):
    todo = Todo(user_id=user_id, **data.model_dump())
    session.add(todo)
    await session.flush()   # получить todo.id

    session.add(OutboxEvent(
        id=uuid4(),
        topic="todo.created",
        payload=json.dumps({"todo_id": todo.id, "user_id": user_id}),
        created_at=datetime.utcnow(),
    ))
    await session.commit()
    return todo

3. Напиши background worker (отдельный process) - читает outbox и publishes в RabbitMQ, помечает published.

4. Напиши consumer - читает события из RabbitMQ, делает что-то (например, отправляет email уведомление).

Это полный production messaging stack: outbox → broker → consumer с идемпотентностью.

Заключение

Поздравляю! Ты прошёл финальный урок Python-трека от начала до конца. 59 уроков, от установки до production-ready backend с messaging.

Если выполнишь финальное задание - у тебя на руках будет complete backend система с auth, БД, async messaging поверх финального проекта - хорошее портфолио. Удачи в дальнейшем!