notification-worker: своя группа, правила пуша, идемпотентность

Перед началом: event-service и fan-out (Модуль 5); правила уведомлений курса 1 (Модуль 10) — они переезжают как есть. Инструменты: репозиторий трека (boilerplate), Go, Redis Streams, PostgreSQL, Web Push (VAPID). Трудоёмкость: ~5 ч.

🔥 Инцидент: «провайдер пушей лёг — и подтормозил чат»

Пятница, push-провайдер начал отвечать таймаутами. В монолите пуши отправлял тот же процесс, что публиковал события в шину: пачка ретраев на мёртвый эндпоинт забила общий пул воркеров, публикация message.new встала в очередь — и WS-кадры стали задерживаться на минуты, хотя БД и сокеты были здоровы. Хуже: после рестарта воркера шина переотдала необработанные события — и часть пользователей получила пуши-дубли к сообщениям, которые уже прочитали. Один сбой вскрыл две проблемы: судьбы WS-доставки и пушей не должны быть связаны, а at-least-once шины нельзя пускать в телефон. В этом модуле уведомления переезжают в собственный процесс со своей группой на шине — и дублям объявляется война по event_id.

📖 Теория

6.1 Зачем пуши — отдельный процесс

Из четырёх сервисов notification-worker вынести проще всех: это уже воркер. В курсе 1 он жил внутри монолита; вынос покупает две вещи:

• изоляция сбоев. Push-провайдер — внешняя зависимость с чужим SLA; его таймауты и 5xx — рутина, а не инцидент. Отдельный процесс держит ретраи и очереди при себе. Связь с чатом односторонняя: воркер читает шину, но на пути messenger → шина → event-service → клиент его нет — отказ провайдера не трогает WS-доставку;
• независимый деплой. Правила пуша — самое часто меняемое: какие события будят, тексты, окна агрегации. Это катится без релиза ядра.

Своё состояние — ровно то, что и положено воркеру: таблица push_subscriptions (endpoint + ключи, курс 1, Модуль 10) и состояние агрегационных окон; своя база notifications_db — как и у каждого сервиса топологии.

6.2 Своя группа: work-queue против fan-out

Топик один — messenger.events, но группы выражают разные роли потребителей:

• gateway-<replica-id> (Модуль 5): каждая реплика читает всё — broadcast живой доставки;
• notifications: work-queue — событие читает ровно один воркер. Масштабирование воркера добавляет рук одной очереди, а не копий пуша.

ACK — после успешной обработки; воркер умер посреди — событие остаётся в pending и доотправляется при возврате (механика Streams, Модуль 5). Отсюда честное следствие: на входе воркера — at-least-once, и §6.5 расскажет, почему это не приводит к двойным пушам.

go
const stream = "messenger.events"
const group  = "notifications" // work-queue: событие читает ОДИН воркер

for {
    res, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
        Group: group, Consumer: name,
        Streams: []string{stream, ">"}, Count: 16, Block: 5 * time.Second,
    }).Result()
    if err != nil {
        continue
    }
    for _, m := range res[0].Messages {
        if err := handle(ctx, m); err != nil {
            continue // без ACK: вернётся из pending и будет обработан заново
        }
        rdb.XAck(ctx, stream, group, m.ID)
    }
}

6.3 Правила «кого будить»

Правила зафиксированы в курсе 1 (Модуль 10) и переезжают без изменений — меняется только источник ответа на вопрос «онлайн?»:

«Офлайн» воркер проверяет не догадкой, а у владельца соединений: GET /internal/presence?users=… у event-service (матрица вызовов Модуля 2). Это единственный синхронный вызов воркера, и он не на пути клиента — WS-доставке его не нужно. Правила живут на сервере: клиент к моменту пуша может быть закрыт — в этом весь смысл push.

6.4 Web Push: подписки, агрегация, чистка

• Web Push (VAPID) — как в курсе 1: подписка (endpoint, keys.p256dh, keys.auth) — одна запись на устройство; механику шифрования не повторяем.
• Агрегация окна ~30 с. Активная группа рожает десятки сообщений в минуту; пуш на каждое — сирена и отписки (инцидент курса 1). Воркер копит события по пользователю и через ~30 секунд шлёт один пуш вида «5 новых сообщений». Состояние окон — данные воркера: процесс умер до отправки — события вернутся из pending, окно пересоберётся, дубль отсечёт журнал (§6.5).
• Чистка протухших подписок. Ответ провайдера 404/410 означает, что подписка мертва (приложение снесли, браузер сменил endpoint). Запись удаляется — иначе воркер будет вечно ретраить мёртвый endpoint.

6.5 Идемпотентность: at-least-once шины ≠ двойные пуши

Шина даёт at-least-once (Модуль 4): ретрай outbox-воркера, рестарт консьюмера, pending-redelivery — событие может прийти дважды. Для WS-кадра дубль почти безвреден: клиент дедуплицирует по message.id (курс 1, Модуль 6). Пуш — не клиент: телефон зазвонит дважды, отменить уведомление нельзя. Поэтому:

• каждый конверт несёт event_id — он и есть ключ идемпотентности;
• воркер ведёт собственный журнал отправок с уникальным индексом: перед отправкой — вставка, конфликт = уже отправляли, молча пропускаем.

sql
CREATE TABLE notification_log (
    event_id UUID        NOT NULL,
    user_id  UUID        NOT NULL,
    sent_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (event_id, user_id)  -- уникальный индекс = барьер дублей
);

go
tag, err := pg.Exec(ctx, `
    INSERT INTO notification_log (event_id, user_id)
    VALUES ($1, $2) ON CONFLICT DO NOTHING`, ev.EventID, userID)
if err != nil {
    return err
}
if tag.RowsAffected() == 0 {
    return nil // дубль из шины: пуш уже уходил
}
sendPush(ctx, sub, ev) // вставили — значит, наш

Между вставкой и отправкой есть щель: падение в ней потеряет пуш, но не удвоит его. Для уведомлений это правильная сторона ошибки — тишина лучше сирены; недостающее клиент доберёт sync-ом по seq при входе.

💻 Практика

1. Воркер со своей группой (45 мин). Поднимите notification-worker, создайте группу notifications на messenger.events. Проверьте: событие message.new обработал ровно один воркер (вторая копия — в простое), а реплики event-service получили своё — группы не мешают друг другу.
2. Сценарий правил (45 мин). Три состояния получателя: офлайн → пуш; онлайн (открыт WS) → тишина; диалог в mute → тишина, а mention будит даже mute. Проверку офлайна делайте через GET /internal/presence.
3. Агрегация (30 мин). Отправьте 5 сообщений за 30 секунд в один диалог: получателю уходит ровно один пуш («5 новых сообщений»), а не пять.
4. Сломать провайдера (45 мин). Фейковый эндпоинт (локальный стаб) отвечает 500: события копятся в pending, воркер ретраит — и всё это время WS-доставка идёт как ни в чём не бывало. Верните стабу 200: pending доотправляется, дублей нет — журнал отсёк повторные вставки.
5. Чистка (15 мин). Заставьте стаб ответить 410: подписка удаляется, следующие события уходят по оставшимся подпискам, ретраев мёртвого endpoint нет.

🧩 Quiz

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

• Объяснить изоляцию: сбой push-провайдера — личная проблема воркера, WS-доставка от него не зависит.
• Отличать work-queue (одна группа notifications) от fan-out групп event-реплик на том же топике.
• Применять правила «кого будить» из курса 1 с проверкой presence через internal-API event-service.
• Агрегировать пуши в окно ~30 с и чистить протухшие подписки по 404/410.
• Гасить at-least-once шины: event_id + журнал отправок с уникальным индексом.

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

Четыре сервиса стоят, но впереди эксплуатация: рестарты под нагрузкой, долгая деградация Redis (обещали в Модуле 5), SLO на лаги групп и 10 тысяч соединений на стенде. Научимся не только переживать отказы, но и измерять их до того, как их заметят пользователи.