Модуль 4: Transactional Outbox — YDB + Message Queue (доставка событий без FOR UPDATE) — Git Hosting на Go: Serverless MVP на Yandex Cloud

Модуль 4. Transactional Outbox: YDB + Message Queue (доставка событий без FOR UPDATE)

Цель модуля: Гарантировать, что событие RepositoryPushed будет доставлено в CI/CD, даже если Message Queue недоступна или консьюмер упал. Реализовать Transactional Outbox на YDB (без FOR UPDATE) + Message Queue с DLQ. Outbox Relay — Cloud Function с Timer Trigger каждые 5 секунд.

Стадия роста проекта: Интеграции. Git Smart HTTP работает, метаданные в YDB. Теперь система должна уведомлять внешние сервисы: CI/CD, аналитику, уведомления. И первое, что ломается — события теряются.

Перед началом: Модули 1–3 (Setup, Smart HTTP, YDB)

Инструменты: YDB Serverless, Message Queue (SQS), Cloud Functions (Timer Trigger), Terraform, Go 1.22+

Трудоёмкость: ~7 ч (теория 1.5 ч · практика 4.5 ч · Terraform DLQ 1 ч)

🔥 Инцидент: push прошёл, а CI/CD не знает

Вася сделал git push origin main. Packfile на NFS, метаданные в YDB, refs обновлены. Всё хорошо. Но CI/CD не запустился. Вебхук не ушёл. Вася ждёт 10 минут, перезапускает pipeline вручную. «Почему не сработало автоматически?»

В коде привычный Dual-Write:

go
// ❌ Антипаттерн: событие публикуется после коммита YDB
func (h *Handler) Push(w http.ResponseWriter, r *http.Request) {
    // (1) Обновляем refs в YDB
    h.ydb.UpdateRef(ctx, repoID, "refs/heads/main", oldSHA, newSHA)

    // (2) Публикуем событие в Message Queue — ПОСЛЕ коммита
    h.mq.SendMessage(ctx, "outbox-events", RepositoryPushed{...})
    // Если MQ недоступна 2 секунды — событие потеряно навсегда
}

Message Queue была недоступна ровно 2 секунды (кратковременный сбой Yandex Cloud). Публикация упала после коммита YDB. Откатывать уже принятый push бессмысленно. Событие потеряно, CI/CD не знает о push.

Знакомая история из Модуля 3: две независимые операции, одна прошла — вторая нет. Решение то же — атомарность, но на уровне «YDB + Message Queue».

Событие сохраняется в той же транзакции YDB, что и обновление refs. Доставкой занимается отдельная Cloud Function (Outbox Relay) с Timer Trigger каждые 5 секунд.

📖 Теория: Transactional Outbox без FOR UPDATE

Глоссарий модуля:

Transactional Outbox — событие сохраняется в той же транзакции БД, что и бизнес-данные; отдельный воркер гарантирует доставку.

Dual-Write — запись в два разных хранилища (YDB + Message Queue) без общей транзакции: одно может зафиксироваться, а второе — нет.

Outbox Relay — Cloud Function с Timer Trigger: каждые 5 секунд поллит outbox_events и публикует в Message Queue.

Timer Trigger — триггер Yandex Cloud: запускает Cloud Function по cron-расписанию. Замена goroutine-воркера в Serverless.

Visibility Timeout — время, в течение которого сообщение в Message Queue «невидимо» для других консьюмеров после первого чтения. Если консьюмер не удалил сообщение за это время — оно возвращается в очередь.

DLQ (Dead-Letter Queue) — очередь для сообщений, не доставленных после всех попыток. Разбираются вручную.

Redrive Policy — правило: после N неудачных попыток доставки сообщение перемещается в DLQ.

At-least-once Delivery — Outbox гарантирует, что событие не потеряется, но НЕ гарантирует однократную отправку. Консьюмер обязан быть идемпотентным (Модуль 5).

Ментальная модель: «Представьте, что вы пишете письмо и кладёте его в исходящий лоток на столе. Вы не идёте на почту сами — секретарь (Outbox Relay) каждые 5 минут забирает письма из лотка и несёт на почту. Если почта закрыта — секретарь вернёт письмо обратно в лоток и попробует в следующий раз. Если письмо не удаётся отправить 5 раз подряд — оно попадает в "мёртвый ящик" (DLQ), и начальник разбирается вручную.»

4.1. Проблема Dual-Write в Serverless

В классической архитектуре (Архитектура v2, Модуль 3) Outbox-воркер — это goroutine с бесконечным циклом:

go
// Классический Outbox-воркер (K8s / VPS)
func (w *Worker) Run(ctx context.Context) {
    ticker := time.NewTicker(1 * time.Second)
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            w.poll(ctx)
        }
    }
}

В Serverless нет долгоживущих процессов. Cloud Function живёт максимум 10 минут и умирает. Замена — Timer Trigger: Yandex Cloud запускает функцию по cron каждые 5 секунд. Функция поллит один раз и умирает. Следующий запуск — новый инстанс.

100%

🖱 Перетащите • Колесо для зума

4.2. Outbox в YDB: атомарная запись

Критично: событие outbox_events создаётся в той же транзакции YDB, что и обновление refs. YDB поддерживает интерактивные транзакции через SDK:

go
// internal/git/push_handler.go
func (h *PushHandler) HandlePush(ctx context.Context, req PushRequest) error {
    // Одна транзакция YDB: UPDATE refs + UPSERT outbox_events
    err := h.ydb.DoTx(ctx, func(ctx context.Context, tx ydb.TableTransaction) error {
        // (1) Обновляем refs (атомарный UPDATE с условием из Модуля 3)
        res, err := tx.Execute(ctx, `
            UPDATE refs
            SET sha = $new_sha,
                updated_at = CurrentUtcTimestamp(),
                version = version + 1
            WHERE repo_id = $repo_id
              AND ref_name = $ref_name
              AND sha = $expected_sha
        `, map[string]any{
            "$repo_id":      req.RepoID,
            "$ref_name":     req.RefName,
            "$expected_sha": req.ExpectedSHA,
            "$new_sha":      req.NewSHA,
        })
        if err != nil {
            return err
        }
        if res.RowsAffected() == 0 {
            return ErrConflict // Конкурентный push
        }

        // (2) Создаём outbox-событие В ТОЙ ЖЕ ТРАНЗАКЦИИ
        eventID := uuid.NewString()
        payload, _ := json.Marshal(RepositoryPushed{
            RepoID: req.RepoID,
            Ref:    req.RefName,
            SHA:    req.NewSHA,
            User:   req.UserID,
        })
        _, err = tx.Execute(ctx, `
            UPSERT INTO outbox_events (event_id, repo_id, event_type, payload, status, created_at)
            VALUES ($event_id, $repo_id, 'RepositoryPushed', $payload, 'pending', CurrentUtcTimestamp())
        `, map[string]any{
            "$event_id": eventID,
            "$repo_id":  req.RepoID,
            "$payload":  string(payload),
        })
        return err
    })
    return err
}

Если функция упадёт между UPDATE refs и UPSERT outbox_events — транзакция откатится целиком. Атомарность гарантирована.

4.3. Outbox Relay: Cloud Function с Timer Trigger

go
// cmd/outbox-relay/main.go
package main

import (
    "context"
    "encoding/json"
    "log"

    "github.com/yandex-cloud/go-genproto/yandex/cloud/serverless/functions/v1"
)

func main() {
    // Cloud Function入口: Yandex Cloud вызывает эту функцию по Timer Trigger
    functions.Start(func(ctx context.Context, req *functions.Request) (*functions.Response, error) {
        return handleRelay(ctx, req)
    })
}

func handleRelay(ctx context.Context, req *functions.Request) (*functions.Response, error) {
    // (1) Поллим pending-события через Index Scan
    events, err := pollPendingEvents(ctx, 10)
    if err != nil {
        return nil, err
    }

    for _, ev := range events {
        // (2) Публикуем в Message Queue
        if err := publishToMQ(ctx, ev); err != nil {
            // (3) Ошибка: инкрементируем attempts
            if ev.Attempts >= 5 {
                markFailed(ctx, ev.EventID)
                // Событие останется в таблице со статусом 'failed'
                // DLQ на стороне MQ — для событий, которые MQ не смогла доставить консьюмеру
            } else {
                incrementAttempts(ctx, ev.EventID)
            }
            continue
        }

        // (4) Успех: помечаем published
        markPublished(ctx, ev.EventID)
    }

    return &functions.Response{StatusCode: 200}, nil
}

func pollPendingEvents(ctx context.Context, limit int) ([]OutboxEvent, error) {
    // Index Scan через idx_outbox_status — НЕ Full Scan!
    res, err := ydbClient.Execute(ctx, `
        SELECT event_id, repo_id, event_type, payload, attempts
        FROM outbox_events
        VIEW idx_outbox_status
        WHERE status = 'pending'
        ORDER BY created_at
        LIMIT $limit
    `, map[string]any{"$limit": limit})
    // ...
}

Ключевое отличие от Архитектуры v2: нет FOR UPDATE SKIP LOCKED. В Serverless Timer Trigger запускает одну функцию за раз (concurrency = 1 для outbox-relay), поэтому конкурентных воркеров нет. Если нужна параллельная обработка — разбиваем по event_id % N и запускаем N функций.

4.4. DLQ и Redrive Policy в Terraform

Студенты не любят настраивать очереди вручную. Показываем в Terraform:

hcl
# terraform/message_queue.tf

# Основная очередь
resource "yandex_message_queue" "outbox-events" {
  name                       = "outbox-events"
  visibility_timeout_seconds = 30
  message_retention_seconds  = 86400  # 24 часа
}

# DLQ: мёртвая очередь
resource "yandex_message_queue" "outbox-events-dlq" {
  name                       = "outbox-events-dlq"
  visibility_timeout_seconds = 60
  message_retention_seconds  = 1209600  # 14 дней
}

# Redrive Policy: после 5 неудач → DLQ
# Это настройка на стороне Message Queue (не outbox_events в YDB)
# Если консьюмер CI/CD не смог обработать сообщение 5 раз — оно уходит в DLQ
resource "yandex_message_queue" "outbox-events-redrive" {
  name                     = "outbox-events-redrive"
  redrive_policy {
    max_receive_count     = 5
    dead_letter_queue_arn = yandex_message_queue.outbox-events-dlq.arn
  }
}

Два уровня DLQ:

YDB outbox_events.attempts >= 5 — Outbox Relay не смог опубликовать в MQ (MQ недоступна). Событие остаётся в YDB со статусом failed.
MQ Redrive Policy — MQ не смогла доставить консьюмеру (CI/CD упал). Сообщение уходит в DLQ-очередь.

4.5. Идемпотентность консьюмера (связь с Модулем 5)

Outbox гарантирует At-least-once. Если Relay упадёт между publishToMQ и markPublished — при следующем запуске событие опубликуется повторно. Консьюмер (CI/CD) обязан быть идемпотентным:

go
// Консьюмер CI/CD: идемпотентность через event_id
func (c *CICDConsumer) handle(ctx context.Context, msg SQSMessage) error {
    var ev RepositoryPushed
    json.Unmarshal(msg.Body, &ev)

    // Идемпотентность: проверяем, не обрабатывали ли уже
    // В YDB: таблица processed_events с UNIQUE(event_id)
    _, err := c.ydb.Execute(ctx, `
        UPSERT INTO processed_events (event_id, processed_at)
        VALUES ($event_id, CurrentUtcTimestamp())
    `, map[string]any{"$event_id": ev.EventID})

    if isDuplicate(err) {
        return nil // Уже обработано — no-op
    }

    // Запускаем CI/CD pipeline
    return c.triggerPipeline(ctx, ev)
}

4.6. Этот паттерн в архетипах

Outbox проявляется по-разному в четырёх архетипах проектов:

Архетип	Событие	Консьюмер	Воркер	Почему критично
Финтех	`PaymentCreated`	Уведомления, антифрод	goroutine (K8s)	Деньги списаны, но система не знает
Git-хостинг	`RepositoryPushed`	CI/CD вебхуки	Timer Trigger (Serverless)	Push прошёл, но сборка не запустилась
Мессенджер	`MessageSent`	Push-уведомления	goroutine (K8s)	Сообщение отправлено, но получатель не уведомлён
Лента	`PostCreated`	Поисковый индекс	Timer Trigger (Serverless)	Пост опубликован, но не виден в поиске

Общий принцип: бизнес-операция и событие должны быть атомарны. Различается только воркер: goroutine для K8s, Timer Trigger для Serverless.

4.7. Компромиссы (Trade-offs)

Решение	Плюс	Минус	Когда выбирать
Outbox + Timer Trigger (Serverless)	Нет долгоживущих процессов; 0₽ при малой нагрузке	Задержка до 5 сек (интервал Timer); нет FOR UPDATE	Serverless MVP
Outbox + goroutine (K8s)	Минимальная задержка (~1 сек); FOR UPDATE SKIP LOCKED	Нужен долгоживущий процесс	K8s / VPS
Прямая публикация в MQ после коммита	Просто	Dual-Write: событие теряется при сбое MQ	Антипаттерн (никогда)
CDC (Debezium) вместо поллинга	Real-time; нет поллинга	Сложная инфраструктура; не работает с YDB	High-load, PostgreSQL
DLQ на стороне MQ	Автоматическое перемещение «мёртвых» сообщений	Доп. очередь; нужно мониторить DLQ	Всегда для production

Вывод: для Serverless MVP — Outbox + Timer Trigger + DLQ в Terraform. При миграции на K8s — замена Timer Trigger на goroutine, паттерн тот же. См. ADR-SV4.

🛠 Практические задания

Сценарий: Вася сделал git push origin main. Нужно гарантировать, что CI/CD получит событие RepositoryPushed даже при сбое Message Queue. Цель: 0 потерянных событий при 50% недоступности MQ.

Задача 1: Атомарная запись Outbox в YDB

Реализовать HandlePush: UPDATE refs + UPSERT outbox_events в одной транзакции YDB.

✅ Готово, когда: [ ] UPDATE refs + UPSERT outbox_events в одной DoTx [ ] Конфликт SHA → ErrConflict → 409 (событие не создаётся) [ ] Тест: транзакция откатывается целиком при ошибке на любом шаге

Задача 2: Outbox Relay (Cloud Function + Timer Trigger)

Реализовать outbox-relay: поллинг pending → publish в MQ → mark published/failed.

✅ Готово, когда: [ ] Timer Trigger: cron */5 * * * * ? (каждые 5 секунд) [ ] Поллинг через Index Scan (VIEW idx_outbox_status), НЕ Full Scan [ ] Успех → status='published'; ошибка → attempts+1 [ ] attempts >= 5 → status='failed' [ ] EXPLAIN подтверждает Index Scan

Задача 3: DLQ + Redrive Policy в Terraform

Настроить в Terraform: основная очередь, DLQ, Redrive Policy (max_receive_count = 5).

✅ Готово, когда: [ ] Terraform: yandex_message_queue для основной очереди и DLQ [ ] Redrive Policy: max_receive_count = 5 → DLQ [ ] terraform plan показывает создание обеих очередей [ ] DLQ не пустая после теста (сообщения реально перемещаются)

Задача 4: Тест с эмуляцией сбоя MQ

Поднять elasticmq локально. Отключить elasticmq на 10 секунд. Проверить: события не теряются, attempts инкрементируются, после восстановления MQ — события доставляются.

✅ Готово, когда: [ ] При недоступности MQ: события остаются pending, attempts растёт [ ] После восстановления MQ: все pending-события доставляются [ ] Потерянных событий = 0 [ ] При attempts >= 5: событие failed, не публикуется повторно

⚠️ Антипаттерн для разбора

go
// Go Cloud Function (Антипаттерн)
func (h *Handler) Push(w http.ResponseWriter, r *http.Request) {
    // ❌ 1. Обновляем refs
    h.ydb.UpdateRef(ctx, repoID, refName, oldSHA, newSHA)

    // ❌ 2. Публикуем ПОСЛЕ коммита — Dual-Write
    h.mq.SendMessage(ctx, "outbox-events", event)

    // ❌ 3. Нет DLQ: если консьюмер не обработал — сообщение теряется
    // ❌ 4. Нет идемпотентности: повторная доставка → повторный запуск CI/CD
}

Вопросы студенту:

Что произойдёт, если MQ недоступна 2 секунды после коммита YDB? (Событие потеряно навсегда)
Что произойдёт, если Relay упадёт между publish и markPublished? (Событие опубликуется повторно при следующем запуске)
Почему консьюмер обязан быть идемпотентным? (At-least-once: дубли неизбежны)

📐 Задание по визуализации (Mermaid)

Студенту необходимо нарисовать Sequence Diagram работы Outbox Relay с обработкой ошибок.

Обязательные элементы:

Timer Trigger запускает outbox-relay каждые 5 секунд.
Relay поллит YDB через Index Scan.
Для каждого события: publish в MQ → mark published.
При ошибке MQ: attempts+1, при attempts >= 5: mark failed.
Redrive Policy: 5 неудач на стороне консьюмера → DLQ.

Шаблон для ориентира: /serverless-mvp/diagrams/04-outbox-relay.svg

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

Q: Почему событие outbox_events создаётся в той же транзакции YDB, что и UPDATE refs?

Для скорости
Атомарность: если функция упадёт между двумя операциями — транзакция откатится целиком*
Это не обязательно

Q: Чем Timer Trigger отличается от goroutine-воркера?

Ничем
Timer Trigger запускает функцию по cron; goroutine живёт в долгоживущем процессе*
Timer Trigger быстрее

Q: Зачем нужен FOR UPDATE SKIP LOCKED в классическом Outbox и почему его нет в Serverless?

В Serverless нет конкурентных воркеров (Timer Trigger запускает одну функцию за раз)*
FOR UPDATE не нужен вообще
YDB поддерживает FOR UPDATE

Q: Два уровня DLQ — зачем?

Один уровень достаточен
YDB attempts >= 5 (Relay не смог опубликовать) + MQ Redrive Policy (консьюмер не смог обработать)*
Это дублирование

Q: Почему консьюмер обязан быть идемпотентным?

Outbox гарантирует Exactly-once
Outbox гарантирует At-least-once: дубли неизбежны при повторной доставке*
Это не обязательно

Q: Как настроить DLQ в Terraform?

Вручную через консоль Yandex Cloud
yandex_message_queue + redrive_policy с max_receive_count и dead_letter_queue_arn*
DLQ не настраивается в Terraform

🤖 Инструкция для LLM (Coding Plan Assistant) — Режим Модуля 4

Привет! Я начинаю Модуль 4 курса «Git Hosting на Go: Serverless MVP на Yandex Cloud». Выступи в роли Наставника-Архитектора.

Сначала проверь понимание: дай короткий ответ (2–3 предложения) — почему публикация события в Message Queue после коммита YDB — это Dual-Write, и как Transactional Outbox решает эту проблему в Serverless (без FOR UPDATE).

Затем выдай задание:

Реализовать HandlePush: UPDATE refs + UPSERT outbox_events в одной транзакции YDB.
Реализовать Outbox Relay: Cloud Function с Timer Trigger, поллинг через Index Scan, publish в MQ.
Написать Terraform для Message Queue + DLQ + Redrive Policy.
Провести тест с эмуляцией сбоя MQ (elasticmq отключается на 10 сек).

Покажи мне код Антипаттерна (Dual-Write, нет DLQ, нет идемпотентности) и попроси объяснить, почему каждый из трёх пунктов убьёт доставку событий.

НЕ ПИШИ за меня Relay и Terraform. Дождись моего ответа с архитектурным планом. После этого дай фидбек по чек-листу ментора и укажи на слепые зоны (например, «поллинг делает Full Scan», «нет DLQ в Terraform», «консьюмер не идемпотентен»).

📝 Чек-лист ментора

[ ] Атомарность: UPDATE refs + UPSERT outbox_events в одной транзакции YDB (DoTx). [ ] Timer Trigger: cron */5 * * * * ?, запуск outbox-relay каждые 5 секунд. [ ] Поллинг: Index Scan через VIEW idx_outbox_status, НЕ Full Scan. EXPLAIN подтверждает. [ ] Обработка ошибок: attempts+1 при сбое MQ; attempts >= 5 → status='failed'. [ ] DLQ в Terraform: yandex_message_queue + redrive_policy (max_receive_count = 5). [ ] Два уровня DLQ: YDB attempts >= 5 (Relay → MQ) + MQ Redrive Policy (MQ → консьюмер). [ ] Идемпотентность консьюмера: processed_events с UNIQUE(event_id). [ ] Тест: эмуляция сбоя MQ → 0 потерянных событий. [ ] Mermaid-диаграмма: Timer Trigger → Relay → YDB → MQ → DLQ. [ ] ADR-SV4: «Outbox + Timer Trigger vs goroutine vs CDC».

🔗 Мостик к Модулю 5 (Idempotency + Bulkhead)

В конце ревью скажи студенту:

«Отлично, событие RepositoryPushed гарантированно доставляется в CI/CD. Но что если Вася и Петя делают push в один репозиторий одновременно? Оба читают sha = abc123, оба пытаются обновить на def456. Без защиты один перезапишет другой. В Модуле 5 мы решим это двумя способами: Bulkhead (concurrency_limit = 5 на git-gateway) ограничивает число одновременных стримов, а идемпотентность через YDB UNIQUE constraint защищает от дублей. А ещё — покажем, что идемпотентный ответ (409) не потребляет concurrency-слот.»

Сквозной User Story: «Вася сделал git push origin main. Outbox Relay доставил событие в CI/CD (Модуль 4). В Модуле 5 мы посмотрим, что происходит, когда Петя делает push одновременно с Васей: как concurrency_limit = 5 на git-gateway изолирует git-операции, и как атомарный UPDATE с условием из Модуля 3 защищает от гонок на уровне данных.»

← Модуль 3: YDB Serverless — БД за 0₽ (метаданные без SQL и FOR UPDATE)Модуль 5: Идемпотентность и Bulkhead в Serverless →