Модуль 3. Прод-консьюмер на go-redis: claim, идемпотентность, конкурентность, shutdown

Цель модуля: собрать воркер, который не теряет и не дублирует работу при повторах и отказах: цикл XReadGroup без потери сообщений, идемпотентный обработчик поверх Redis, спасатель зависших через XAutoClaim, воркер-пул и graceful shutdown. И проверить всё это ножом: make failover под нагрузкой.

Перед началом: Модули 1–2 курса, «Фундамент» (идемпотентность, graceful shutdown), github.com/redis/go-redis/v9. Трудоёмкость: ~8–10 ч (теория 4 ч · практика 5 ч).

📖 3.1. Цикл чтения: где обычно теряют сообщения

Базовый цикл кажется тривиальным — и именно в нём живут классические ошибки:

go
for {
    streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
        Group:    "billing",
        Consumer: consumerName, // billing-1, billing-2, ... — уникальное имя инстанса
        Streams:  []string{"orders", ">"},
        Count:    16,
        Block:    2 * time.Second,
    }).Result()

    if err == redis.Nil {
        continue // таймаут Block — это норма, просто читаем дальше
    }
    if err != nil {
        // Сетевые ошибки/фейловер: пауза с бэкоффом, не «continue горячим циклом»
        time.Sleep(backoff)
        continue
    }

    for _, s := range streams {
        for _, msg := range s.Messages {
            process(ctx, msg) // внутри — ack только после успеха
        }
    }
}

Грабли, которые здесь прячутся:

• redis.Nil — не ошибка. По таймауту Block пустой ответ нормален; трактовать его как сбой — словить ложную тревогу и паузу;
• Consumer должен быть уникальным на инстанс. Имя, сгенерированное один раз и захардкоженное, после рестарта «склеит» pending двух процессов: старые записи в PEL мертвеца и новые — под одним именем. Берите hostname-pid или pod-name;
• Обработка внутри цикла блокирует чтение. Пока process думает, новые сообщения не читаются (воркер пустого простоя ждёт нагрузку). Конкурентность — в §3.4;
• Count — ваш рычаг prefetch. 16 при обработке в 50 мс и 100 при в 5 с — разные системы. Меньше count = меньше работы в руках при падении и меньше памяти у консьюмера.

📖 3.2. Идемпотентность: подключаем ядро трека

Redis Streams даёт at-least-once: всё, что не XACK-нуто, может прийти снова — после падения, после claim'а, после фейловера. Значит обработчик обязан переживать повтор. Паттерн из «Фундамента» — «сначала работа, потом фиксация» — но фиксацию в Redis-мире удобно держать в том же Redis:

go
type IdempotentHandler struct {
    rdb    *redis.Client
    ttl    time.Duration // >= retention стрима, иначе дедуп «протухнет» раньше повторов
    handle func(ctx context.Context, msg redis.XMessage) error
}

func (h *IdempotentHandler) Handle(ctx context.Context, stream string, msg redis.XMessage) error {
    key := "processed:" + stream + ":" + msg.ID

    // Быстрый дедуп: уже прошло — подтверждаем и выходим.
    // SETNX с TTL и в том же контексте, что и подтверждение.
    ok, err := h.rdb.SetNX(ctx, key, "1", h.ttl).Result()
    if err != nil {
        return err // осторожно: не подтверждаем — запишется повтор
    }
    if !ok {
        return nil // дубль — обработка уже была
    }

    if err := h.handle(ctx, msg); err != nil {
        // Работа упала — снимаем метку, чтобы повтор попробовал снова
        h.rdb.Del(ctx, key)
        return err
    }
    return nil
}

Тонкости:

• TTL метки — не мейнстрим-настройка, а контракт: если retention стрима 24 часа, метка с TTL 1 час «забудет» старый дубль из DLQ-повтора. Возьмите TTL ≥ retention;
• Метка в Redis — та же незащищённая запись, что и всё остальное: при фейловере метка может пропасть (дедуп ослабнет — придёт дубль) или остаться при падении до обработки (дедуп окрепнет — потеряется работа). Для денег фиксацию держат в БД рядом с бизнес-эффектом (Transactional Outbox/applied_events — «Фундамент», М2), Redis-метка — для «бюджетного» дедупа: дешево, быстро, «обычно работает»;
• Если работает идемпотентно по природе (UPDATE ... SET status=...), внешний дедуп может не понадобиться вовсе. Метка — там, где повторы дороги (письма, списания, отправки).

📖 3.3. Спасатель зависших: XAutoClaim в отдельной горутине

Записи, за которыми умерший консьюмер не успел признаться, лежат в PEL. Их надо вернуть в работу — но не воровать слишком рано (иначе живой медленный обработчик получит дубль) и не поздно (иначе очередь встала).

go
go func() { // один спасатель на группу — не на каждого воркера
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            claimed, _, err := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{
                Stream:   "orders",
                Group:    "billing",
                Consumer: consumerName,   // кто именно перезабирает
                MinIdle:  5 * time.Minute, // «молчит дольше 5 минут — наш»
                Start:    "0-0",
                Count:    32,
            }).Result()
            if err != nil {
                metrics.ReclaimFailed.Inc()
                continue
            }
            for _, msg := range claimed {
                metrics.Reclaimed.Inc() // метрика обязательна — см. §3.6
                _ = handler.Handle(ctx, "orders", msg)
                rdb.XAck(ctx, "orders", "billing", msg.ID)
            }
        }
    }
}()

Правила гигиены:

• MinIdle — двойной таймаут вашего p99 обработки. p99 воркера 40 с → MinIdle минимум 2 минуты. Слишком маленький MinIdle — вечный шторм двойной обработки («ребаланс-шторм» Redis-мира);
• Спасатель — один на группу. Если каждый воркер делает XAutoClaim в цикле, они начнут отбирать сообщения друг у друга;
• Метрика reclaimed — индикатор проблем. Устойчиво ненулевая — у вас что-то падает/подвисает: смотрите логи и XPENDING-разрезы.

Посмотреть, кто и сколько держит, поможет:

bash
redis-cli XPENDING orders billing            # сводка: сколько, min/max ID, по консьюмерам
redis-cli XPENDING orders billing - + 10     # первые 10 зависших с idle-временем

Показать кодКопировать код

📖 3.4. Конкурентность: воркер-пул без сюрпризов

Порядок сообщений в Streams между записями не гарантирован (только по возрастанию ID при чтении), поэтому параллелить обработку можно — если ключ шардирования внутри держит порядок для связанных событий:

go
func (w *Worker) Run(ctx context.Context) {
    sem := make(chan struct{}, w.parallelism) // параллелизм воркера
    var wg sync.WaitGroup

    for {
        // команды из предыдущего раздела...
        for _, s := range streams {
            for _, msg := range s.Messages {
                sem <- struct{}{}
                wg.Add(1)
                go func(msg redis.XMessage) {
                    defer wg.Done()
                    defer func() { <-sem }()

                    if err := w.handler.Handle(ctx, "orders", msg); err != nil {
                        metrics.ProcessErrors.Inc()
                        return // НЕ ack: пусть повторят (или DLQ — по политике)
                    }
                    w.rdb.XAck(ctx, "orders", "billing", msg.ID)
                }(msg)
            }
        }
    }
}

• Порядок по ключу (order_id): если ваш process пишет в БД и требует порядка в рамках заказа — шардируйте по order_id внутри воркера (map[hash(key)] → канал → горутина), а не полагайтесь на чтение;
• Backpressure: sem ограничивает в вылет в памяти; ещё раньше его ограничивает Count на чтении. Оба должны быть осознаны, а не «16 и 32, потому что в примере»;
• Ошибка обработки ≠ ack. Решение про retry/DLQ — политика: счётчик попыток в Redis (INCR attempt:stream:id с TTL) и на N-й попытке движение в orders:dlq явным XADD — минимальный честный DLQ.

📖 3.5. Graceful shutdown: не притворяться живым

Классика Go: signal.NotifyContext + остановка по цепочке:

go
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()

// 1) перестать читать новые: цикл чтения завершается по ctx.Done()
// 2) дождаться, пока обработчики в полёте завершат работу
select {
case <-ctx.Done():
    w.stopReading()          // флаг/отменённый контекст чтения
    wg.Wait()                // обработчики дописывают и делают XACK
    log.Println("drain complete")
}

Что важно для Streams:

• Незавершённые и не-XACK-нутые записи — не потеря. Они остались в PEL; после рестарта (или из рук спасателя) придут заново. Именно поэтому shutdown — «дропнуть чтение и договорить», а не «успеть обработать всё»;
• Не делайте XACK «на глаз» при остановке, чтобы «не висело». Ack — это обещание «работа сделана»; обманув систему один раз, вы получите потерянное сообщение навсегда;
• Таймаут на drain: context.WithTimeout на 30–60 с поверх ожидания wg; не успевшие — остаются в PEL, их подхватит §3.3. Это записывается в лог, а не в тишину.

📖 3.6. Минимальный набор наблюдаемости воркера

Без этих чисел отладка «очередь встала» превращается в гадание:

• processed_total / process_errors_total — базовый конвейер;
• reclaimed_total — сколько записано спасателем; растёт → кто-то падает;
• pending_size (по группе, из XINFO GROUPS orders) — «займ» в руках воркеров: резкий рост = деградация обработчиков;
• stream_length и stream_age (ID первой записи от «сейчас») — длина и «возраст» хвоста; возраст важнее длины: он показывает, насколько вы отстали по времени;
• лаги обработки: время между ID сообщения и его обработкой — прямое измерение задержки очереди (XADD кладёт timestamp в ID — используйте его).

bash
redis-cli XINFO GROUPS orders
# name: billing  consumers: 3  pending: 12  last-delivered-id: 1712...  lag: 4

💻 Практика

Лаборатория «Воркер, который переживает фейловер» — целиком по разделам:

1. Соберите billing-worker: цикл XReadGroup (уникальный consumer), обработчик задержкой 50 мс с «падением» на каждом 7-м сообщении, IdempotentHandler из §3.2 (TTL метки ≥ retention).
2. Спасатель: горутина с XAutoClaim (MinIdle 2 мин), метрики processed/errors/reclaimed в лог каждые 10 с.
3. Проверьте штатный контур: продюсер пишет 50 сообщений/с; после «падения» на 7-х — сообщение переобрабатывается (лог + метрика reclaimed растёт, потерь нет: сколько XADD — столько processed при дедупе).
4. Фейловер под нагрузкой: make up-sentinel, запустите три воркера, дайте нагрузку и выполните make failover. Ожидания:
   • воркеры ловят ошибки соединения и переподключаются через Sentinel (в go-redis — redis.NewFailoverClient);
   • после промоушена поток восстанавливается (сколько секунд — замерьте);
   • «потерянные хвосты» переобрабатываются, дедуп не даёт двойных списаний.
5. Graceful shutdown: SIGTERM под нагрузкой — в логах drain complete, приложение вышло без потери незакрытой работы (проверьте XPENDING).

Артефакт: репозиторий billing-worker + отчёт «простой, потери, дубли в трёх режимах: штатно / failover / SIGTERM».

🧩 Quiz "Проверь себя"

✅ Итоги модуля

🔗 Мостик к Модулю 4

Воркер собран и переживает отказы. Но «переживает» нужно видеть на графиках: Opus-момент этого курса — мониторинг: какие метрики Redis отдаёт из INFO, как поднять redis-exporter с Prometheus, какие алерты отличить ложные от настоящих и как поймать деградацию по памяти раньше, чем она станет OOM.