Как проектировать distributed dead letter queue для сообщений?

Краткий тезис

Dead Letter Queue (DLQ) — это механизм изоляции сообщений, которые не удалось обработать после заданного числа повторных попыток. В распределённых системах, особенно в архитектуре RAG|agentic RAG, DLQ предотвращает потерю данных, даёт время на диагностику сбоев и позволяет автоматически или вручную перезапускать обработку. Правильное проектирование DLQ включает выбор схемы повторных попыток, структуру метаданных, политику хранения и интеграцию с мониторингом.

1. Термин: Dead Letter Queue (DLQ)

DLQ — это очередь (или топик в message broker), куда перемещаются сообщения, которые по какой-либо причине не могут быть обработаны после исчерпания попыток (retry). В контексте RAG|agentic RAG агенты часто общаются через очереди сообщений (например, Kafka, RabbitMQ), и сбой обработки шага агента (например, неудачный call|вызов LLM, недоступность векторной базы) должен корректно обрабатываться.

Зачем нужна DLQ

• Гарантия at‑least‑once delivery без потери данных.
• Возможность исследовать причину сбоя без влияния на основной поток.
• Сохранение сообщений для последующего перезапуска (replay) после исправления ошибки.

2. Компоненты распределённой системы с DLQ

На высоком уровне архитектура включает:

• Input topic (или очередь) – исходные сообщения.
• Consumer – процесс, который читает и обрабатывает сообщения.
• Retry mechanism – встроенная логика повторных попыток с нарастающей задержкой.
• DLQ topic – финальная очередь для проблемных сообщений.
• DLQ consumer – отдельный процесс (или воркер), предназначенный для анализа и перезапуска сообщений из DLQ.

В agentic RAG эта схема применяется, например, для обработки запросов пользователя, распределения их между агентами (planner, executor, validator) и сбора результатов.

3. Проектирование схемы повторных попыток (Retry Strategy)

Exponential backoff – стандартный подход, когда задержка между повторными попытками растёт экспоненциально, часто с добавлением случайного джиттера для предотвращения «шквала» соединений.

Пример конфигурации

• Первая попытка: немедленно.
• Retry 1: задержка 1 секунда.
• Retry 2: задержка 2 секунды.
• Retry 3: задержка 4 секунды.
• Retry 4: задержка 8 секунд.
• После 5-й попытки (или по достижении max retries = 5) → DLQ.

Вместо фиксированного числа попыток можно использовать порог по времени (например, максимум 60 секунд на все попытки).

Важно в agentic RAG повторная попытка может быть не только на уровне сообщения, но и на уровне шага агента: если LLM вернул ошибку, агент может повторно вызвать модель с другой температурой или разбить задачу.

4. Структура метаданных в DLQ

Каждое сообщение в DLQ должно быть обогащено метаданными для диагностики и replay.

В Kafka эти метаданные можно хранить в заголовках (headers) или добавлять в value сообщения. Рекомендуется использовать оба подхода: заголовки для быстрой фильтрации, value для полного контекста.

5. Политика хранения (Retention)

DLQ не является бесконечным хранилищем. Нужно определить:

• Maximum retention time (например, 7 дней).
• Maximum size (например, 10 ГБ).
• Action при достижении лимита: либо автоматически удалять самые старые сообщения, либо останавливать производителя (backpressure).

Для compliance или отладки можно комбинировать DLQ параллельной записью в объектное хранилище (S3) с более долгим сроком хранения.

6. Обработка DLQ: автоматическая и ручная

6.1 Автоматическая обработка

• Periodic reprocess: специальный scheduler читает сообщения из DLQ по расписанию (например, каждые 10 минут) и переотправляет их в исходный топик.
• Conditional reprocess: перед отправкой проверяется, исправлена ли причина сбоя (например, доступна ли база данных).
• Throttling: при восстановлении работы не заваливать систему всеми сообщениями сразу – использовать rate limiter.

6.2 Ручная обработка

• Web dashboard – интерфейс для просмотра содержимого DLQ, фильтрации по ошибке, ручного клика «Replay» или «Delete».
• API – программный доступ (REST/gRPC) для интеграции с системами мониторинга (PagerDuty, Opsgenie) или CI/CD.

6.3 Алерты

• Алерт при размере DLQ > N (например, > 100 сообщений).
• Алерт при появлении нового типа ошибки.
• Алерт при долгом невосстановлении (например, возраст старейшего сообщения > 1 часа).

В agentic RAG алерт может триггерить откат к fallback-агенту или переключение на упрощённый flow.

7. Интеграция с Kafka (как пример)

Kafka – популярный message broker для высоконагруженных распределённых систем. В Kafka DLQ обычно реализуется как отдельный топик с настраиваемой конфигурацией.

Пример настройки producer

# Используем confluent_kafka
from confluent_kafka import Producer, KafkaError

def delivery_report(err, msg):
    if err is not None:
        # При критических ошибках отправляем в DLQ
        if err.code() == KafkaError.MSG_SIZE_TOO_LARGE:
            dlq_producer.produce('dlq_topic', value=msg.value(), headers={
                'original_topic': msg.topic(),
                'error': str(err)
            })
        else:
            # Retry logic
            pass

producer = Producer({'bootstrap.servers': 'localhost:9092'})
producer.produce('input_topic', value=b'message', callback=delivery_report)
producer.flush()

Конфигурация consumer используем enable.auto.commit=False, чтобы точно управлять оффсетами при ошибке.

Consumer код с retry и DLQ:

from confluent_kafka import Consumer, KafkaException

def process_message(msg):
    # Попытка обработки, может выбросить исключение
    raise RuntimeError("LLM timeout")

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'agent_group',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False
})
consumer.subscribe(['input_topic'])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        raise KafkaException(msg.error())
    
    payload = msg.value()
    retry_count = int(msg.headers().get('retry_count', b'0').decode())
    
    if retry_count >= 5:
        # Отправляем в DLQ
        dlq_producer.produce('dlq_topic', value=payload, headers={
            'original_topic': msg.topic(),
            'retry_count': str(retry_count),
            'error': 'Max retries exceeded'
        })
        consumer.commit(msg)  # коммитим, чтобы сдвинуть offset
        continue
    
    try:
        process_message(payload)
        consumer.commit(msg)
    except Exception as e:
        # Увеличиваем retry_count и отправляем обратно в input_topic с задержкой
        import time
        delay = 2 ** retry_count
        time.sleep(delay)
        dlq_producer.produce('input_topic', value=payload, headers={
            'retry_count': str(retry_count + 1),
            'error': str(e)
        })
        consumer.commit(msg)

Рекомендуемая Kafka-конфигурация при включённой функции dlq.enabled=true (в некоторых Kafka Streams API) можно автоматически направлять сообщения, вызвавшие ошибку десериализации, в DLQ.

8. Аспекты мониторинга

Ключевые метрики

• Lag DLQ – количество необработанных сообщений в DLQ.
• Age – максимальный возраст сообщения в DLQ.
• Dead‑letter rate – количество сообщений, попавших в DLQ, за единицу времени.
• Replay rate – количество сообщений, успешно перезапущенных из DLQ.
• Top errors – частота различных типов ошибок.

Эти метрики удобно собирать в Prometheus и визуализировать в Grafana.

9. Связь с agentic RAG

В agentic RAG агенты работают в цикле планирования, выполнения и верификации. Сообщения между ними могут проходить через очереди. Если, например, агент-планировщик не может разбить запрос (слишком сложный), он отправляет сообщение в DLQ. DLQ позволяет не потерять запрос, а дать ему возможность быть обработанным после улучшения модели или переключения на агента‑запасного.

Кроме того, DLQ может собирать «hard cases» для дообучения: сообщения, которые не удалось обработать, можно аннотировать и использовать для fine-tuning LLM.

10. Проектирование масштабируемой DLQ

Горизонтальное масштабирование

• Разделите DLQ на несколько партиций (в Kafka – несколько партиций).
• Потребители DLQ могут быть распределены по группам (consumer group) с разным приоритетом.
• При реплее используйте idempotent producer, чтобы избежать дублей.

Backpressure: если система не справляется с реплеем, DLQ растёт. Можно ввести тротлинг на реплей (максимум N сообщений в секунду) и приоритизацию (новые ошибки важнее старых).

Отказоустойчивость сам DLQ должен быть реплицирован (в Kafka – replication factor > 1). Если DLQ недоступна, система должна либо заблокировать producer (circuit breaker), либо временно буферизировать в локальном логе.

11. Проектирование для нескольких уровней (multi‑layer DLQ)

В сложных системах имеет смысл использовать несколько слоёв DLQ:

• Layer 1 – DLQ для временных ошибок (например, таймаут сети), ожидается автоматический реплей.
• Layer 2 – DLQ для бизнес-ошибок (например, невалидный формат), требуется ручной разбор.
• Layer 3 – долговременное хранилище (S3) для аудита и анализа.

Пример routing: сообщение → Layer 1 → после 3 неудач → Layer 2 → после 10 дней → архивация в S3.

12. Практические рекомендации

• Используйте уникальные ID сообщений (UUID) для идемпотентности.
• Записывайте полный контекст ошибки (traceback, окружение).
• Не кладите в DLQ сообщения, которые никогда не смогут быть обработаны (например, повреждённый JSON) – сразу отправляйте в Layer 2 и алерт.
• Предусмотрите удаление из DLQ после успешного реплея (можно использовать compacted topic в Kafka с ключом = UUID сообщения).
• Обязательно логируйте действия с DLQ: кто и когда изменил статус, кто реплей.
• При запуске agentic RAG system проведите chaos engineering: искусственно вызывайте ошибки и наблюдайте за поведением DLQ.

Пет-проект для закрепления

Задача Разработать минимальную систему «Agent with DLQ» на Kafka, которая обрабатывает запросы пользователей к LLM, и при сбое отправляет запрос в DLQ с возможностью автоматического реплея.

Инструменты

• Python (confluent_kafka, asyncio)
• Kafka (локально через Docker compose)
• FastAPI (веб-интерфейс для ручного реплея)
• SQLite (для хранения метаданных обработки)

Шаги:

1. Запустите Kafka и создайте топики: user_requests, dlq_requests.
2. Producer: каждые 2 секунды отправляет случайный запрос (например, «What is X?» с вероятностью 20% ошибка формата – вызовет исключение при обработке).
3. Consumer (агент): имитирует вызов LLM с возможным исключением (timeout, invalid response). При успехе – коммитит оффсет. При ошибке – отправляет в DLQ с заголовками (retry_count=0). После 3 неудач – в DLQ с пометкой max_retries.
4. Отдельный scheduler (или consumer на DLQ) читает сообщения из dlq_requests, для которых retry_count < 3, и через exponential backoff переотправляет обратно в user_requests с увеличенным retry_count. Сообщения с max_retries сохраняет в SQLite.
5. FastAPI endpoint GET /dlq – показывает список сообщений из SQLite. POST /dlq/replay/{id} – переотправляет выбранное сообщение в user_requests.
6. Добавьте Prometheus метрики (lag, error rate) и Grafana дашборд.

Ожидаемый результат Работающая система, где при ошибках сообщения не теряются, а проходят через DLQ, и администратор может влиять на процесс через API. Вы сможете продемонстрировать знание retry, DLQ, мониторинга и архитектуры обработки ошибок.

Связь с другими вопросами

Навигация

• Предыдущий: 834
• Следующий: 836
• Индекс: 00. Индекс разборов