event-service ×N: fan-out на репликах и насыщение кадров

Перед началом: целевая топология и матрица вызовов (Модуль 2), конверт события и гарантии доставки (Модуль 4). Sync-протокол и presence курса 1 (Модули 6–7) — механика не повторяется, она переезжает как есть. Инструменты: репозиторий трека (boilerplate), Go, Redis (Streams + pub/sub), docker. Трудоёмкость: ~6 ч.

🔥 Инцидент: «половина пользователей не видит сообщения live»

Первую реплику event-service выкатили — и стали масштабировать: подняли вторую за тем же nginx-upstream. Через час — тикет «сообщения доходят только после перезахода»: клиенты первой реплики видят ленту живой, клиенты второй — лишь после переподключения. Причина в конфиге: консьюмеров завели в одну общую consumer group на обе реплики. Redis Streams внутри группы — work-queue: каждое событие достаётся ровно одному консьюмеру. Событие читала одна реплика, а соединения получателя висели на другой — кадр уходил в пустоту. Спасал только sync по seq после перезахода: механика курса 1 отработала честно, но «живой» чат она не заменяет. В этом модуле — fan-out, при котором событие попадает на каждую реплику с получателем, и всё, что event-service делает с событием до отправки в сокет.

📖 Теория

5.1 Технический шов: состояние, обязанности, запреты

event-service — единственный stateful компонент топологии: десятки тысяч long-lived соединений, которые нельзя «перезалить рестартом». Из этого следуют три решения Модуля 2:

  • свой профиль нагрузки — память под буферы соединений и файловые дескрипторы, а не CPU запросов; реплики масштабируются отдельно от API;
  • деплой независимо от бизнес-логики: релиз messenger-service не рвёт соединения (инцидент Модуля 1 — «релиз стикеров уронил чат всем»);
  • ноль доменных данных: никаких запросов к БД messenger. Адресацию решил отправитель — recipients уже в конверте; event-service тупо матчит recipient → мои соединения → send.

Его собственное состояние перечислимо: реестр соединений user → conns с лимитами, подписки conversation → соединения (ими управляют сами конверты), presence-ключи и каталог-кэши (§5.4). Обязанностей ровно четыре:

  1. WS-терминация: GET /api/v1/ws?token= — JWT валидируется публичным ключом из JWKS до апгрейда (RS256, Модуль 3), лимиты соединений, sync-протокол курса 1 (догон по seq, маркер sync.complete).
  2. Fan-out: события шины → свои соединения (§5.2).
  3. Насыщение: конверт → клиентский кадр (§5.4) — главная ценность.
  4. Presence/typing relay: владелец присутствия; typing — ретрансляция с rate-limit мимо БД (курс 1, Модуль 7).

Чего event-service не делает: проверок прав (адресацию решил messenger), запросов к БД домена, бизнес-логики.

5.2 Fan-out на репликах: три механизма по нарастающей

Требование: событие должно попасть на каждую реплику, где висит хотя бы один получатель, — именно это нарушил инцидент. Три механизма:

  1. Redis Pub/Sub — этап переезда, простейший: все реплики подписаны на канал, каждая получает всё; гарантия at-most-once — разрыв с Redis = потеря, клиент догонит по seq. Рекомендация для старта.
  2. Streams + consumer group на реплику — штатный режим. Топик messenger.events, у каждой реплики своя группа gateway-<replica-id>: каждая читает все события, у каждой свой pending и свой ACK (после доставки в соединения — at-least-once, Модуль 4). Группа — единица доставки, реплика — единица группы: инцидент был нарушением именно этого правила. Бонусы: краткое падение Redis переживается (pending дочитывается), лаг группы — наблюдаемая задержка доставки.
  3. Routing registry — гигантский масштаб, ВНЕ плана: реестр user → replica, точечная доставка. Понадобится, когда «каждая реплика читает всё» перестанет успевать разбирать поток по CPU. Не делаем.
go
group := "gateway-" + f.replicaID // своя группа НА РЕПЛИКУ, не общая
f.rdb.XGroupCreateMkStream(ctx, "messenger.events", group, "0")
for {
    res, _ := f.rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
        Group: group, Consumer: "worker",
        Streams: []string{"messenger.events", ">"}, Count: 64, Block: time.Second,
    }).Result()
    for _, m := range res[0].Messages {
        f.route(m) // матч recipients → conns
        f.rdb.XAck(ctx, "messenger.events", group, m.ID)
    }
}
Fan-out между репликами

На том же топике живёт и notification-worker — но всегда со своей группой notifications, обычный work-queue: событие читает один воркер. Это не ошибка конфигурации, как в инциденте, а другой контракт потребления — предмет Модуля 6.

Про деградацию: pub/sub молча теряет при разрыве, Streams держит pending; сценарий долгой недоступности Redis — отдельный разбор, ему посвящён Модуль 7.

5.3 Presence, когда реплик несколько

Механика курса 1 не меняется — меняется раскладка по процессам:

  • владелец соединения обновляет TTL-ключ presence:{user} своим heartbeat'ом: пользователь «живёт» на той реплике, где его соединения;
  • presence-событие публикуется в отдельный pub/sub-канал — не в Streams: событие эфемерное, хранить и переигрывать нечего;
  • все реплики читают канал и доставляют событие своим соединениям — presence кросс-реплический (event ⇄ event, матрица вызовов Модуля 2);
  • offline — только при закрытии последнего соединения пользователя на реплике-владельце: счётчик соединений тот же, что в курсе 1 (DEL presence:{user} + публикация offline — на самой реплике).

Наружу у event-service один internal-эндпоинт по этой теме: GET /internal/presence?users=…. Его клиент — notification-worker Модуля 6: правило «кого будить» начинается с вопроса «онлайн?».

5.4 Насыщение: из конверта — клиентский кадр

Доменный конверт не пригоден для показа: в нём sender_id, но не sender_name. Насыщение (enrichment) — сборка клиентского кадра:

jsonc
// конверт (шина)                →  WS-кадр (клиенту)
{ "type": "message.new",           { "type": "message.new",
  "conversation_id": "c1",           "conversation_id": "c1",
  "recipients": ["u1", "u2"],        "message": {
  "payload": {                         "id": "…", "seq": 42,
    "id": "…", "seq": 42,              "sender_id": "u1",
    "sender_id": "u1",                 "sender_name": "Алиса",  // ← каталог
    "text": "…"                        "avatar_url": "…",       // ← каталог
  }                                    "text": "…"
}                                    },
                                     "conversation": { "title": "Команда" } // ← карточка
                                   }

Кадр — не конверт: recipients наружу не уходят. Данные для показа — из каталога, кэшей event-service:

КэшКлючЗначениеИнвалидация
Карточка пользователяuser_iddisplay_name, avatar, statususer.updated (pub/sub) + TTL-подстраховка
Карточка диалогаconversation_idtype, title/peer, участникиconversation.updated, member.changed

Правила насыщения:

  • кэш — in-process LRU + pub/sub-инвалидация; холодный промах — один запрос карточки у владельца (пользователь — user-service, диалог — GET /internal/conversations/{id}/card у messenger), дальше из памяти;
  • насыщение не имеет права падать: карточки нет — кадр уходит с fallback-значением и флагом stale:true, догрузка на клиенте. Доставка важнее полноты;
  • насыщение идёт после матчитга recipients → conns — сначала дешёвая адресация, потом сборка кадра; имена берутся из каталога, не из шины.
go
name, stale := "…", true // насыщение не имеет права падать
if card, ok := c.lru.Get(senderID); ok {
    name, stale = card.DisplayName, false
} else if card, err := c.fetchCard(ctx, senderID); err == nil { // GET владельцу
    c.lru.Add(senderID, card)
    name, stale = card.DisplayName, false
}

Typing-кадры насыщаются так же, но живут только в памяти реплики-владельца: ретрансляция идёт мимо БД и outbox (курс 1, Модуль 7).

💻 Практика

  1. Две реплики за одним upstream (45 мин). Поднимите два инстанса event-service (REPLICA_ID=gateway-1 и gateway-2) за nginx-upstream /api/v1/ws (Модуль 2). Откройте несколько соединений тестовым клиентом из курса 1 — они должны распределиться по обеим репликам.
  2. Одна публикация — обе группы (30 мин). Опубликуйте message.new в messenger.events ровно один раз: в логах обеих реплик событие появилось, кадры ушли каждой. Затем воспроизведите инцидент: общая группа на обе реплики — событие достанется одной, соединения второй останутся без кадра. Разведите группы обратно.
  3. Сломать реплику (30 мин). Убейте одну реплику при живых клиентах: соединения рвутся, клиенты переподключаются и догоняют по seq (sync-протокол курса 1) — без потерь. Поднимите реплику: её группа дочитает pending за время простоя; возможные дубли кадров клиенты гасят дедупликацией по message.id (at-least-once, Модуль 4).
  4. Замерить лаг группы (45 мин). Синтетика: генератор держит 1000 соединений, фоновый скрипт публикует события. Снимайте лаг группы (XINFO GROUPS, экспорт в /metrics) — это и есть задержка доставки. Запишите число: в Модуле 7 оно станет SLO на лаги.

🧩 Quiz

1. Почему у каждой реплики event-service своя consumer group gateway-<replica-id>, а не одна общая?
2. Какая гарантия у fan-out через Redis Pub/Sub и кто спасает при потере?
3. Пользователь открыл три вкладки и закрыл две. Когда другие увидят «offline»?
4. Карточки отправителя нет в кэше, владелец карточки не отвечает. Что уйдёт клиенту?
5. Почему событие presence публикуется в pub/sub, а не в Streams?

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

  • Объяснить, почему event-service — технический шов: stateful-соединения, профиль памяти/дескрипторов, независимый деплой — и почему он не владеет доменными данными.
  • Выбирать механизм fan-out: pub/sub на старте, Streams с группой gateway-<replica-id> в штате, routing registry — вне плана.
  • Держать presence на N репликах: TTL-ключи и счётчик соединений, события presence — через pub/sub, offline — на последнем соединении.
  • Собирать клиентский кадр из каталога-кэша: LRU, pub/sub-инвалидация, fallback stale:true.
  • Знать, где смотреть задержку доставки: лаг consumer-группы.

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

Шина раздала событие всем репликам — но спящему телефону нужен не кадр, а пуш. Дальше — notification-worker со своей группой notifications: work-queue вместо broadcast, правила «кого будить», агрегация и идемпотентность — чтобы падение провайдера осталось его личной проблемой.