Real-time и WebSocket

Перед началом: Модули 1-6. Инструменты: WebSocket, Yandex API Gateway. Время: ~6 ч.

Из M6: мы настроили YDB — таблицы, индексы, транзакции. Теперь разберём, как доставлять сообщения клиентам в реальном времени через WebSocket.

В этом модуле — WebSocket-воркер чата на Go в Serverless Container. Фокус: thousands of соединений, предсказуемая доставка, корректная остановка и масштабирование.

WebSocket — stateful, а Cloud Functions — stateless. Поэтому WS-хаб живёт в Serverless Container (долгоживущий процесс), а не в функции. REST-операции остаются в функциях.

🔥 Зачем это вообще нужно

5000 пользователей онлайн. Каждый держит WebSocket. Serverless Function умирает через 60 секунд (таймаут). Connection рвётся. Чат «моргает». Нужен WebSocket Gateway с long-lived соединениями — но Serverless не для этого.

📖 Теория

Почему WebSocket в Serverless Container

API Gateway терминирует соединение и проксирует его в контейнер, который держит постоянные сокеты и подписан на YDS. Контейнер масштабируется от 0; при росте — несколько реплик.

Масштабирование и fan-out через YDS

Одна публикация message_created попадает в YDS, и каждая реплика WS-контейнера, подписанная на стрим, получает событие и рассылает его своим подключённым клиентам. Поэтому не важно, на какой реплике сидит адресат — стрим доносит. Шардирование стрима по room_id гарантирует порядок сообщений в комнате.

WS Hub: управление соединениями

Hub хранит активные соединения и рассылает события. Каждое соединение — своя горутина чтения и записи.

go
type Hub struct {
    mu    sync.RWMutex
    conns map[string]*Client // connID → клиент
    sub   *yds.Subscriber    // подписка на стрим
}

func (h *Hub) register(c *Client) {
    h.mu.Lock()
    h.conns[c.id] = c
    h.mu.Unlock()
    metrics.WSInc() // gauge активных соединений
}

• Read pump — читает входящие фреймы от клиента (сообщения, ping).
• Write pump — отправляет исходящие (новые сообщения из YDS, pong, close).
• Heartbeat — ping/pong по таймеру, чтобы ловить «зависших» клиентов и закрывать неактивные соединения.

Read pump

Горутина, которая читает из сокета и пересылает в hub. Устанавливает дедлайн чтения, обновляемый pong-хендлером (если клиент не ответил на ping — соединение закрывается):

go
const (
    pongWait       = 60 * time.Second  // макс. ожидание pong
    pingPeriod     = 50 * time.Second  // отправляем ping чуть раньше pongWait
    writeWait      = 10 * time.Second
    maxMessageSize = 4096
)

func (c *Client) readPump() {
    defer c.hub.unregister(c)
    c.conn.SetReadLimit(maxMessageSize)
    c.conn.SetReadDeadline(time.Now().Add(pongWait))
    c.conn.SetPongHandler(func(string) error {
        c.conn.SetReadDeadline(time.Now().Add(pongWait))
        return nil
    })
    for {
        _, message, err := c.conn.ReadMessage()
        if err != nil {
            break // клиент отключился или таймаут
        }
        c.hub.inbound <- inboundMessage{client: c, data: message}
    }
}

Write pump

Горутина, которая читает из канала send и пишет в сокет. Тикер отправляет ping:

go
func (c *Client) writePump() {
    ticker := time.NewTicker(pingPeriod)
    defer func() { ticker.Stop(); c.conn.Close() }()
    for {
        select {
        case message, ok := <-c.send:
            c.conn.SetWriteDeadline(time.Now().Add(writeWait))
            if !ok {
                c.conn.WriteMessage(websocket.CloseMessage, []byte{})
                return // канал закрыт — завершаем
            }
            if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil {
                return
            }
        case <-ticker.C:
            c.conn.SetWriteDeadline(time.Now().Add(writeWait))
            if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
                return
            }
        }
    }
}

Жизненный цикл соединения

go
func (h *Hub) handleConnection(conn *websocket.Conn, userID string) {
    client := &Client{
        conn:    conn,
        send:    make(chan []byte, 256),
        userID:  userID,
    }
    h.register(client)
    metrics.WSConnectionsInc()

    go client.writePump()   // отдельная горутина для записи
    client.readPump()       // блокирует до отключения

    h.unregister(client)
    metrics.WSConnectionsDec()
}

При отключении readPump завершается → unregister убирает клиента из hub → writePump видит закрытый канал send и тоже завершается. Обе горутины освобождаются без утечек.

Подписка на YDS (получение сообщений)

Hub подписан на YDS-стрим и рассылает входящие сообщения подключённым клиентам. reader.Commit фиксирует offset — обработанное сообщение больше не придёт:

go
func (h *Hub) SubscribeYDS(ctx context.Context, streamName string) error {
    reader := yds.NewReader(ctx, streamName, yds.WithConsumer("ws-gateway"))

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
        }

        batch, err := reader.ReadBatch(ctx, 10) // батч до 10 записей
        if err != nil {
            log.Printf("YDS read error: %v", err)
            time.Sleep(time.Second) // backoff перед повтором
            continue
        }

        for _, record := range batch {
            var event domain.StreamEvent
            if json.Unmarshal(record.Data, &event) != nil {
                continue // пропускаем мусор
            }

            h.mu.RLock()
            for _, c := range h.conns {
                if c.userID == event.UserID {
                    select {
                    case c.send <- record.Data: // отправляем в writePump
                    default:
                        log.Printf("client %s buffer full, closing", c.userID)
                        go h.unregister(c) // клиент не успевает — закрываем
                    }
                }
            }
            h.mu.RUnlock()
        }

        reader.Commit(ctx) // фиксируем offset
    }
}

Этот цикл запускается один раз на старте контейнера (в горутине). При graceful shutdown ctx.Done() прерывает цикл, а неотправленные сообщения останутся в YDS до следующего старта.

Гарантии доставки

Полного exactly-once в распределённой системе без больших затрат не бывает — поэтому опираемся на at-least-once + идемпотентность по message_id на клиенте и сервере.

Graceful shutdown

При остановке реплики (деплой, масштабирование) нельзя рвать соединения «молча»:

1. Перестать принимать новые соединения.
2. Дочитать из YDS текущие события (с уже зафиксированным offset).
3. Отправить клиентам close-фреймы с кодом «try again later».
4. Закрыть соединения с таймаутом.

Клиенты при этом переподключаются к другой реплике, а пропущенные за момент переключения сообщения догоняются из истории по last_message_id (см. модуль 5).

В Dockerfile обязателен ENTRYPOINT в exec-форме, чтобы SIGTERM доходил до Go-процесса — иначе graceful shutdown не сработает (платформа убьёт процесс по таймауту).

Мониторинг

Ключевые метрики stateful-сервея:

• ws.connections.active — активные соединения (глобально и по комнатам);
• ws.ping.latency — задержка ping/pong;
• ws.reconnects — частота переподключений (много = проблемы);
• ws.messages_per_second — входящий/исходящий поток.

Таймауты пинга и рост reconnects — первые сигналы утечки соединений или перегрузки.

Нагрузочное тестирование (k6)

Реалистичные сценарии: подключение, отправка сообщения, автокомплит. Замеряем p95/p99 latency, CPU, память, поведение GC при росте одновременных соединений (100 / 500 / 1000). По результатам выбираем размер контейнера и число реплик.

💱 Компромиссы

• Cloud Functions vs Serverless Container: функция stateless и умирает по таймауту — не держит WS; контейнер держит long-lived соединения, но платится оверхедом в простое.
• At-least-once vs exactly-once: полный exactly-once в распределённой системе дорог; берём at-least-once + идемпотентность по message_id на клиенте и сервере.
• Ping/pong таймауты: короткий pongWait → ложные disconnects на медленном канале; длинный → утечка «зависших» соединений и памяти.
• Буфер канала send: маленький → отбрасываем медленных клиентов; большой → растёт память под тысячами соединений.

Практика

• Реализовать hub с register/unregister и рассылкой из подписки YDS.
• Нагрузочный тест k6 на 100/500/1000 одновременных WebSocket-соединений; проанализировать графики.
• Реализовать переподключение клиента и обработку «пропущенных» сообщений через last_message_id.

🔗 К следующему модулю: Real-time работает. Добавим асинхронность через очереди.

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

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

Real-time работает. Добавим асинхронность через очереди (YMQ): Handler → YDB → YMQ → воркеры (push, search, analytics).