Асинхронность и очереди (YMQ)

Перед началом: Модули 1-7. Инструменты: Yandex Message Queue (YMQ), SQS. Время: ~5 ч.

Из M7: WebSocket-воркер получает сообщения из YDS и рассылает клиентам. Теперь разберём, как сообщения попадают в YMQ (откуда триггерится Broadcast Worker) и как обеспечить надёжную доставку с retry и DLQ.

Фокус: развязка — это устойчивость. Ingestion быстро отвечает клиенту 202 Accepted, а тяжёлая работа (найти участников, разослать, push) идёт в фоне через очередь.

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

Пользователь отправляет сообщение. Handler пишет в YDB → отправляет push-уведомление → индексирует для поиска. Три операции синхронно: если push падает — сообщение «потерялось» (хотя в YDB оно есть). Нужна очередь: Handler → YDB → YMQ → воркеры (push, search, analytics).

📖 Теория

Жизненный цикл сообщения в YMQ

1. Ingestion кладёт событие в очередь (SendMessage) и сразу отвечает клиенту 202 Accepted.
2. YMQ-триггер запускает консьюмер (Broadcast Function) — сообщение на это время невидимо для других (visibility timeout).
3. При успехе — сообщение удаляется. При ошибке — возвращается в очередь и повторяется.
4. После N неудач (maxReceiveCount) — уходит в DLQ → алерт → разбор вручную.

Стандартная очередь vs FIFO

Для нашего чата отправка/рассылка идёт через стандартную очередь (порядок нам даёт YDS по room_id, а не YMQ). FIFO пригодился бы для упорядоченных команд «присоединиться/покинуть».

Семантика и идемпотентность

YMQ — at-least-once: сообщение может быть доставлено больше одного раза. Поэтому консьюмер обязан быть идемпотентным (проверка по request_id — см. модуль 6).

Visibility timeout — время, пока сообщение невидимо другим консьюмерам после выдачи. Если консьюмер не успел — сообщение снова видимо и его заберёт другой.

Конкретный пример: Broadcast Worker обрабатывает сообщение за ~2 с. Ставим visibility_timeout = 30s (запас 15×). Если за 30 с функция не ответила (зависла, OOM) — сообщение вернётся и переобработается. Если поставить 5s — сообщение вернётся ДО завершения нормальной обработки → ложные повторы.

Dead Letter Queue (DLQ)

DLQ — отдельная очередь для «битых» сообщений. YMQ сама переносит туда сообщение после maxReceiveCount неудач. Рост глубины DLQ = баг в консьюмере или «отравленное» сообщение. Алерт на глубину DLQ — обязателен.

Publisher: отправка событий в YMQ

YMQ совместима с AWS SQS, поэтому используем стандартный AWS SDK для Go:

go
// Инициализация SQS-клиента для Yandex Message Queue
func NewYMQPublisher(awsCfg aws.Config, queueURL string) *Publisher {
    client := sqs.NewFromConfig(awsCfg, func(o *sqs.Options) {
        o.BaseEndpoint = aws.String("https://message-queue.api.cloud.yandex.net")
        o.Region = "ru-central1"
    })
    return &Publisher{client: client, queueURL: queueURL}
}

func (p *Publisher) Publish(ctx context.Context, event domain.MessageEvent) error {
    payload, err := json.Marshal(event)
    if err != nil {
        return fmt.Errorf("marshal event: %w", err)
    }

    _, err = p.client.SendMessage(ctx, &sqs.SendMessageInput{
        QueueUrl:    &p.queueURL,
        MessageBody: aws.String(string(payload)),
    })
    if err != nil {
        return fmt.Errorf("send to YMQ: %w", err)
    }
    return nil
}

В Ingestion Function после сохранения в YDB: publisher.Publish(ctx, event) → ответ клиенту 202. Если паблиш упал — возвращаем ошибку клиенту (сообщение не отправилось). Но можно и «принять и забыть» — если YDB-запись прошла, сообщение не потеряется (inbox-паттерн: отдельный воркер дочитает неопубликованные).

Consumer: обработка в Broadcast Function

YMQ-триггер передаёт сообщения в Cloud Function как HTTP-запрос. Важно: вернули ошибку → сообщение вернётся в очередь (retry). Вернули успех → сообщение удалится.

go
// HandleYMQTrigger — точка входа Broadcast Function (триггер YMQ).
func HandleYMQTrigger(ctx context.Context, body []byte) error {
    var batch []ymqMessage
    if err := json.Unmarshal(body, &batch); err != nil {
        log.Printf("skip malformed batch: %v", err)
        return nil // не возвращаем ошибку → битый батч не будет вечно ретраиться
    }

    for _, m := range batch {
        var event domain.MessageEvent
        if err := json.Unmarshal([]byte(m.Body), &event); err != nil {
            continue // пропускаем мусор
        }

        if err := processMessage(ctx, event); err != nil {
            // Возвращаем ошибку → ВСЁ сообщение вернётся в очередь (retry)
            return fmt.Errorf("process %s: %w", m.MessageID, err)
        }
    }
    return nil // успех → все сообщения батча удаляются
}

func processMessage(ctx context.Context, event domain.MessageEvent) error {
    // 1. Идемпотентность: уже обработали?
    done, err := repo.ExistsByRequestID(ctx, event.RoomID, event.RequestID)
    if err != nil {
        return err
    }
    if done {
        return nil // уже обработано — удаляем (не ошибка)
    }

    // 2. Найти активных участников комнаты
    participants, err := repo.GetActiveParticipants(ctx, event.RoomID)
    if err != nil {
        return err
    }

    // 3. Online → YDS (real-time), offline → очередь пушей
    for _, p := range participants {
        if p.Online {
            ydsPub.Publish(ctx, event.ToStream(p.UserID))
        } else {
            pushQueue.Publish(ctx, event.ToPush(p.UserID))
        }
    }
    return nil
}

Паттерн «всё или ничего» в батче: если хотя бы одно сообщение из батча не обработалось — возвращаем ошибку. YMQ вернёт ВЕСЬ батч. Если сообщения независимы — можно обрабатывать каждое отдельно и возвращать ошибку только для непрошедших (но YMQ работает с батчами как с единицей).

Terraform: очередь + DLQ + триггер

hcl
# Основная очередь
resource "yandex_message_queue" "chat_messages" {
  name               = "chat-messages"
  visibility_timeout = 30          # секунд (консьюмер должен успеть)
  receive_wait_time  = 10          # long polling (снижает пустые запросы)
  message_retention  = 86400       # 24 часа
}

# Dead Letter Queue
resource "yandex_message_queue" "chat_dlq" {
  name = "chat-dlq"
}

# Redrive: после max_receive_count → DLQ
resource "yandex_message_queue_redrive_policy" "redrive" {
  queue_url         = yandex_message_queue.chat_messages.id
  source_queue_urls = [yandex_message_queue.chat_messages.id]
  dead_letter_queue = yandex_message_queue.chat_dlq.id
  max_receive_count = 5  # после 5 неудач → DLQ
}

# Триггер: запускать Broadcast Function при появлении сообщения
resource "yandex_function_trigger" "mq_to_broadcast" {
  name = "mq-to-broadcast"
  message_queue {
    queue_id        = yandex_message_queue.chat_messages.id
    service_account = yandex_iam_service_account.worker.id
    batch_size      = 10  # обрабатывать по 10 за раз
    cutoff          = 2   # секунд между батчами
  }
  function {
    function_id     = yandex_function.broadcast.id
    service_account = yandex_iam_service_account.worker.id
  }
}

YMQ vs Kafka vs Redis: когда что

Стоимость

YMQ тарифицирует количество запросов и объём данных. Для MVP (1000 сообщений/день × 30 дней ≈ 30 000 запросов) — укладывается в free-tier (100 000/мес). Следите за метриками ymq_depth (глубина) и NumberOfMessagesDeleted (успешная обработка).

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

• At-least-once vs exactly-once: дубли возможны → идемпотентность по request_id обязательна; exactly-once дорог.
• Visibility timeout: короткий → ложные повторы (сообщение возвращается до завершения обработки); длинный → задержка переобработки при реальном сбое консьюмера.
• Стандартная vs FIFO: FIFO даёт порядок и дедупликацию, но режет пропускную до ~30/сек; стандартная — быстрая, без порядка.
• Синхронный publish vs fire-and-forget: синхронный — клиент ждёт и знает об ошибке; fire-and-forget — быстрее, но требует inbox-паттерна для надёжности.

Практика

• Сравнить стоимость отправки 1 млн уведомлений: YMQ vs Cloud Functions vs Kafka.
• Реализовать консьюмер с идемпотентной обработкой; сэмулировать «отравленное» сообщение → проверить попадание в DLQ.
• Подобрать visibility_timeout под длительность обработки; проверить поведение при таймауте (сообщение вернулось).

🔗 К следующему модулю: Очереди работают. Развернём всё в облаке через Terraform.

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

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

Очереди работают. Развернём всё в облаке через Terraform — от ClickOps к IaC.