Модуль 3. Transactional Outbox и борьба с Dual‑Write

Цель модуля: Научить студента гарантировать, что бизнес-событие будет доставлено в брокер (Kafka) и не потеряется при сбоях БД или сети. Устранить проблему Dual-Write. Понять ограничения Outbox (At-least-once в брокер) и реализовать полноценный Production-ready воркер с DLQ и очисткой.

Стадия роста проекта: Интеграции. Проект пережил MVP, идемпотентность работает. Теперь система начинает взаимодействовать с внешними сервисами: уведомления, аналитика, антифрод. И первое, что ломается — события теряются.

Перед началом: Модули 1–2 (Trade-offs, Идемпотентность) Инструменты: PostgreSQL, Kafka, pgx v5, k6, Toxiproxy Трудоёмкость: ~7 ч (теория 1.5 ч · практика 5 ч · ADR 0.5 ч)

🔥 Инцидент: событие, которое исчезло

Идемпотентность работает (Модуль 2) — платежи не дублируются. Но новый инцидент: пользователь сделал перевод, деньги списаны, а уведомление не пришло. Антифрод не получил событие, аналитика не зафиксировала транзакцию. В логах Kafka — пусто. В базе — платёж есть, всё закоммичено.

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

go
// ❌ Антипаттерн: событие публикуется после коммита БД
func (h *PaymentHandler) Pay(w http.ResponseWriter, r *http.Request) {
    tx, err := h.db.BeginTx(r.Context(), pgx.TxOptions{})
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    defer tx.Rollback(r.Context())
    
    // (1) COMMIT платёж в БД
    _, err = tx.Exec(r.Context(), "INSERT INTO payments (...) VALUES (...)", args...)
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    if err := tx.Commit(r.Context()); err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    
    // (2) Публикация в Kafka — ПОСЛЕ коммита, если упадёт → потеряна
    if err := h.kafka.Produce(r.Context(), "PaymentCreated", payload); err != nil {
        slog.Error("kafka publish failed", "err", err) // всё: платёж остался, событие ушло в void
    }
    
    writeJSON(w, map[string]string{"status": "ok"})
}

Kafka была недоступна ровно 2 секунды — публикация упала после коммита. Откатывать уже принятый платёж бессмысленно. Событие потеряно, платёж остался. Антифрод не знает о транзакции, аналитика не зафиксировала, уведомление не ушло.

Знакомая история из Модуля 2: две независимые операции, одна прошла — вторая нет. Решение то же — атомарность, но на уровне «БД + брокер»: Transactional Outbox. Событие сохраняется в той же транзакции БД, что и платёж, а публикацией занимается отдельный воркер.

📖 Теория: Раскрытые термины

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

Transactional Outbox — событие сохраняется в той же транзакции БД, что и бизнес-данные; отдельный воркер гарантирует доставку.
Dual-Write — запись в два разных хранилища (например, БД + брокер) без общей транзакции: одно может зафиксироваться, а второе — нет.
FOR UPDATE SKIP LOCKED — Postgres-расширение: блокирует строку от конкурентных воркеров, а занятые пропускает — основа параллельного polling-worker'а.
DLQ (Dead-Letter Queue) — очередь для событий, не доставленных после всех попыток; разбираются вручную.
At-least-once Delivery — Outbox гарантирует, что событие не потеряется, но НЕ гарантирует однократную отправку. Консьюмер обязан быть идемпотентным (Модуль 2).
Polling vs CDC — Polling: воркер сам периодически опрашивает таблицу. CDC (Change Data Capture): БД сама стримит изменения (например, Debezium по WAL).

3.1. Проблема Dual‑Write

Dual‑Write — это попытка сделать две независимые операции в одной бизнес‑логике:

INSERT INTO payments (...) COMMIT;
Publish(event, Kafka)

Сбои:

БД записала платёж, брокер недоступен → событие потеряно.
Брокер принял событие, БД откатила транзакцию → событие есть, данных нет.

Итог: система неконсистентна. Это та же боль, что в Модуле 2 (Redis + БД), но на уровне «БД + брокер». Паттерн решения тот же — атомарность в одной транзакции.

3.2. Паттерн Transactional Outbox

Решение: обе операции происходят в рамках одной транзакции БД.

go
// Платёж и событие — в одной транзакции pgx
func (h *PaymentHandler) Pay(w http.ResponseWriter, r *http.Request) {
    tx, err := h.db.BeginTx(r.Context(), pgx.TxOptions{})
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    defer tx.Rollback(r.Context()) // no-op после Commit
    
    // (1) Платёж
    var paymentID string
    err = tx.QueryRow(r.Context(),
        "INSERT INTO payments (...) VALUES (...) RETURNING id").Scan(&paymentID)
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    
    // (2) Событие Outbox — в ТОЙ ЖЕ транзакции
    payload, _ := json.Marshal(map[string]any{
        "payment_id": paymentID,
        "amount":     amount,
        "user_id":    userID,
    })
    _, err = tx.Exec(r.Context(),
        `INSERT INTO outbox_events (event_type, aggregate_id, payload, status)
         VALUES ('PaymentCreated', $1, $2, 'pending')`, paymentID, payload)
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    
    if err := tx.Commit(r.Context()); err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    writeJSON(w, map[string]string{"payment_id": paymentID})
}

Публикацией события в Kafka занимается отдельный воркер — Dual-Write устранён, платёж и событие коммитятся атомарно.

3.3. Воркер доставки: `FOR UPDATE SKIP LOCKED`

Воркер — это goroutine с бесконечным циклом, запускаемая в main.go через go outboxWorker.Run(ctx). Каждую секунду: поллит таблицу, захватывает строки, публикует в Kafka, обновляет статус.

go
// internal/outbox/worker.go
type Worker struct {
    db       *pgxpool.Pool
    producer *kafka.Writer
}

func (w *Worker) Run(ctx context.Context) {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return // graceful shutdown
        case <-ticker.C:
            if err := w.poll(ctx); err != nil {
                slog.Error("outbox poll error", "err", err)
            }
        }
    }
}

func (w *Worker) poll(ctx context.Context) error {
    tx, err := w.db.BeginTx(ctx, pgx.TxOptions{})
    if err != nil {
        return err
    }
    defer tx.Rollback(ctx)
    
    // FOR UPDATE SKIP LOCKED — конкурентные воркеры пропускают занятые строки
    rows, err := tx.Query(ctx,
        `SELECT id, payload, attempts
         FROM outbox_events
         WHERE status = 'pending' AND attempts < 5
         ORDER BY created_at
         LIMIT 10
         FOR UPDATE SKIP LOCKED`)
    if err != nil {
        return err
    }
    defer rows.Close()
    
    for rows.Next() {
        var ev OutboxEvent
        if err := rows.Scan(&ev.ID, &ev.Payload, &ev.Attempts); err != nil {
            return err
        }
        if err := w.publish(ctx, &ev); err != nil {
            tx.Exec(ctx, `UPDATE outbox_events SET attempts = attempts + 1, last_error = $2 WHERE id = $1`,
                ev.ID, err.Error())
            continue
        }
        tx.Exec(ctx, `UPDATE outbox_events SET status = 'published', published_at = NOW() WHERE id = $1`, ev.ID)
    }
    return tx.Commit(ctx)
}

LIMIT 10 + FOR UPDATE SKIP LOCKED — пачка по 10 событий за одну транзакцию. Если запустить 3 воркера (go w.Run(ctx) × 3), каждый возьмёт свой набор — SKIP LOCKED обеспечивает это автоматически без согласованного блокирования.

3.4. Retry + DLQ

При ошибке сети: attempts = attempts + 1, last_error = err.Error(). При attempts >= 5 → статус failed, событие в DLQ (отдельная таблица или Kafka-топик), алерт в PagerDuty. Cron (или ещё одна goroutine) удаляет published старше 30 дней по индексу created_at.

go
if ev.Attempts >= 5 {
    tx.Exec(ctx, `UPDATE outbox_events SET status = 'failed', last_error = $2 WHERE id = $1`,
        ev.ID, "max attempts reached — DLQ")
    // + запись в DLQ: kafka.Produce("dlq-events", ev) или INSERT INTO outbox_dlq
}

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

Outbox гарантирует доставку At-least-once. Если воркер упал между шагом publish и UPDATE status='published' — строка останется pending, другой воркер возьмёт её и опубликует повторно. Консьюмер обязан быть идемпотентным через X-Event-Id (аналог Idempotency-Key из Модуля 2):

go
// Консьюмер Kafka: хранит обработанные event_id в UNIQUE-колонке
func (c *Consumer) handle(ctx context.Context, msg kafka.Message) error {
    eventID := extractHeader(msg.Headers, "X-Event-Id")
    
    // Идемпотентность: UNIQUE-индекс на event_id
    _, err := c.db.Exec(ctx,
        `INSERT INTO processed_events (event_id) VALUES ($1)`, eventID)
    if isUniqueViolation(err) {
        return nil // уже обработано — no-op
    }
    
    // Обработка события
    return c.process(ctx, msg)
}

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

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

Архетип	Событие	Aggregate ID	Консьюмер	Почему критично
Финтех (P2P)	`PaymentCreated`	`payment_id`	Уведомления, антифрод, аналитика	Деньги списаны, но система не знает
Git-хостинг	`RepositoryPushed`	`repo_id`	CI/CD вебхуки, поиск	Push прошёл, но сборка не запустилась
Мессенджер	`MessageSent`	`chat_id`	Push-уведомления, доставка	Сообщение отправлено, но получатель не уведомлён
Лента / чтение	`PostCreated`	`post_id`	Поисковый индекс, рекомендации	Пост опубликован, но не виден в поиске

Общий принцип: бизнес-операция и событие должны быть атомарны. Различается только консьюмер и критичность потери события.

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

Подход	Плюс	Минус	Когда выбирать
Outbox + polling (наш)	без брокера для записи, транзакции PG, at-least-once	таблица растёт → TTL; polling latency ~1s	всегда для event delivery
Прямой HTTP/Kafka после коммита	простота	Dual-Write: событие теряется при сбое брокера	антипаттерн (никогда)
CDC (Debezium) вместо polling	real-time delivery, без polling overhead	сложнее инфраструктура; требует WAL-доступ	high-load, тысячи событий/сек
LISTEN/NOTIFY вместо polling	минимальная latency	держит коннект, нет retain при рестарте	small-scale, тестируемо

Вывод для курса: Polling Outbox с FOR UPDATE SKIP LOCKED — вариант по умолчанию для Go-воркера. CDC — эволюция при росте нагрузки. См. ADR-3.

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

Сценарий: После успешного платежа нужно отправить событие PaymentCreated в Kafka для уведомлений и антифрода. Цель: 0 потерянных событий при 50% недоступности Kafka.

Примерный план работы:

Таблица outbox_events + индексы (status, created_at), TTL 30 дней.
PaymentHandler.Pay() внутри одной транзакции pgx сохраняет платёж + INSERT outbox_events.
outbox.Worker: goroutine с ticker.NewTicker(1s) + FOR UPDATE SKIP LOCKED + kafka.WriteMessages + UPDATE status='published'.
Retry (increment attempts) + DLQ при attempts >= 5 + алерт.
Идемпотентность: X-Event-Id + консьюмер с UNIQUE-колонкой.
Тест Toxiproxy: Kafka недоступна 50% времени → 0 потерянных, outbox_pending_count → 0.
ADR-3: «Polling vs CDC для Outbox».

Задача 1: Таблица `outbox_events` и индексы

sql
CREATE TABLE outbox_events (
    id            BIGSERIAL PRIMARY KEY,
    event_type    VARCHAR(100) NOT NULL,
    aggregate_id  VARCHAR(255) NOT NULL, -- ID сущности (использовать как Kafka Partition Key)
    payload       JSONB NOT NULL,
    status        VARCHAR(20) NOT NULL DEFAULT 'pending', -- pending, published, failed
    published_at  TIMESTAMP WITH TIME ZONE,
    attempts      INT NOT NULL DEFAULT 0,
    last_error    TEXT,
    created_at    TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_outbox_status_created ON outbox_events(status, created_at);

✅ Готово, когда:

Таблица outbox_events: id, event_type, aggregate_id, payload JSONB, status, attempts, last_error, created_at
Составной индекс idx_outbox_status_created на (status, created_at) — для эффективного поллинга
aggregate_id помечен как будущий Kafka Partition Key
TTL-политика: published старше 30 дней удаляются cron'ом

Задача 2: `PaymentHandler.Pay` в одной транзакции pgx

Платёж и событие сохраняются одним tx.Commit(). aggregate_id = payment_id — будущий partition key.

go
func (h *PaymentHandler) Pay(w http.ResponseWriter, r *http.Request) {
    tx, err := h.db.BeginTx(r.Context(), pgx.TxOptions{})
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    defer tx.Rollback(r.Context()) // no-op после Commit
    
    // (1) Платёж
    var paymentID string
    err = tx.QueryRow(r.Context(),
        "INSERT INTO payments (...) VALUES (...) RETURNING id").Scan(&paymentID)
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    
    // (2) Событие Outbox — в ТОЙ ЖЕ транзакции
    payload, _ := json.Marshal(map[string]any{
        "payment_id": paymentID,
        "amount":     amount,
        "user_id":    userID,
    })
    _, err = tx.Exec(r.Context(),
        `INSERT INTO outbox_events (event_type, aggregate_id, payload, status)
         VALUES ('PaymentCreated', $1, $2, 'pending')`, paymentID, payload)
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    
    if err := tx.Commit(r.Context()); err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    writeJSON(w, map[string]string{"payment_id": paymentID})
}

✅ Готово, когда:

INSERT INTO payments + INSERT INTO outbox_events в одной транзакции pgx
payload содержит {payment_id, amount, user_id}
aggregate_id = payment_id помечен как partition key
ADR-3 «Polling vs CDC» написан

Задача 3: `outbox.Worker` (goroutine + FOR UPDATE SKIP LOCKED)

Запустить в main.go:

go
outboxWorker := outbox.NewWorker(db, kafkaWriter)
go outboxWorker.Run(ctx) // запускаем goroutine до http.ListenAndServe

Воркер в одной транзакции: захват строки → kafka.WriteMessages → UPDATE status='published'.

✅ Готово, когда:

Захват через SELECT ... FOR UPDATE SKIP LOCKED
Успех → status='published', published_at=NOW(); ошибка → attempts+1, last_error
attempts >= 5 → status='failed' + DLQ + алерт
Graceful shutdown: ctx.Done() останавливает воркер, tx.Rollback сбрасывает незавершённую транзакцию

Задача 4: Идемпотентный консьюмер

Консьюмер Kafka хранит event_id в UNIQUE-колонке. Повтор → no-op.

✅ Готово, когда:

Консьюмер проверяет X-Event-Id по UNIQUE-индексу
Дубль не вызывает повторной обработки (Модуль 2)

Задача 5: Тест Toxiproxy + k6

Toxiproxy отбрасывает 50% запросов к Kafka. k6 шлёт 100 платежей. SLO: потерянных событий = 0, outbox_pending_count → 0 за 5 минут, консьюмер корректно гасит дубли через X-Event-Id.

✅ Готово, когда:

Toxiproxy: 50% timeout между воркером и Kafka
k6: 100 платежей параллельно
Потерянных событий = 0; дубли гасятся консьюмером

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

Студенту даётся код — нужно найти уязвимость (Dual-Write).

go
// Go HTTP-хендлер (Антипаттерн)
func (h *PaymentHandler) Pay(w http.ResponseWriter, r *http.Request) {
    tx, err := h.db.BeginTx(r.Context(), pgx.TxOptions{})
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    
    // (1) COMMIT платёж в БД
    _, err = tx.Exec(r.Context(), "INSERT INTO payments (...) VALUES (...)", args...)
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    if err := tx.Commit(r.Context()); err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    
    // (2) Публикация в Kafka — ПОСЛЕ коммита, если упадёт → потеряна
    if err := h.kafka.Produce(r.Context(), "PaymentCreated", payload); err != nil {
        slog.Error("kafka publish failed", "err", err) // всё: платёж остался, событие ушло в void
    }
    
    writeJSON(w, map[string]string{"status": "ok"})
}

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

Что произойдёт, если Kafka будет недоступна 2 секунды после коммита платежа?
Что произойдёт, если Kafka приняла событие, но БД откатила транзакцию?
Почему проверка в Redis не спасает от гонки?

✅ Правильное решение

Платёж и событие сохраняются в одной транзакции БД. Публикацией занимается отдельный воркер.

go
func (h *PaymentHandler) Pay(w http.ResponseWriter, r *http.Request) {
    tx, err := h.db.BeginTx(r.Context(), pgx.TxOptions{})
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    defer tx.Rollback(r.Context()) // no-op после Commit
    
    // (1) Платёж
    var paymentID string
    err = tx.QueryRow(r.Context(),
        "INSERT INTO payments (...) VALUES (...) RETURNING id").Scan(&paymentID)
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    
    // (2) Событие Outbox — в ТОЙ ЖЕ транзакции
    payload, _ := json.Marshal(map[string]any{
        "payment_id": paymentID,
        "amount":     amount,
        "user_id":    userID,
    })
    _, err = tx.Exec(r.Context(),
        `INSERT INTO outbox_events (event_type, aggregate_id, payload, status)
         VALUES ('PaymentCreated', $1, $2, 'pending')`, paymentID, payload)
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    
    if err := tx.Commit(r.Context()); err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    writeJSON(w, map[string]string{"payment_id": paymentID})
}

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

Студенту необходимо нарисовать Sequence Diagram, показывающий работу Outbox-воркера и обработку ошибок.

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

PaymentHandler делает INSERT payments + INSERT outbox_events в одной транзакции.
Outbox Worker поллит таблицу через FOR UPDATE SKIP LOCKED.
Worker публикует событие в Kafka.
При ошибке Kafka: attempts+1, last_error.
При attempts >= 5: status='failed' + DLQ.
Консьюмер Kafka гасит дубли через X-Event-Id.

Шаблон для ориентира:

Transactional Outbox: поллинг, ретрай, атомарный статус

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

1. Воркер Outbox захватывает строки через…

2. Какое ключевое отличие Outbox от прямой публикации в Kafka?

3. Зачем заголовок X-Event-Id в Kafka-сообщении?

4. Почему отправка события в Kafka после коммита платежа — это Dual-Write?

5. Что делает goroutine воркера при graceful shutdown?

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

Привет! Я начинаю Модуль 3 курса «Архитектура распределённых систем».

Дай короткий ответ (2–3 предложения): почему публикация события в Kafka после коммита БД — это Dual-Write, и какие 3 сценария сбоя это ломает?

Объясни, почему Outbox даёт гарантию At-least-once в брокер, а не Exactly-once. Что обязан делать консьюмер на стороне Kafka?

Предоставь требования к таблице outbox_events: поля, индексы, TTL‑политика, механизм очистки.

Покажи код Антипаттерна (публикация вне транзакции) и попроси меня объяснить, где именно произойдёт рассинхронизация.

Дай описание задания: реализовать Transactional Outbox для платежа, воркер с FOR UPDATE SKIP LOCKED, retry + DLQ, идемпотентный консьюмер.

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

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

🔗 Мостик к Модулю 4 (Saga)

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

«Отлично, мы добились того, что событие о платеже гарантированно попадёт в Kafka. Но что если для завершения бизнес-операции нужно выполнить несколько шагов в разных сервисах (зарезервировать товар, списать деньги, отправить уведомление)? Если один из шагов упадёт, у нас не будет единой транзакции. В Модуле 4 мы решим эту проблему с помощью паттерна Saga: разобьём операцию на шаги и реализуем компенсирующие транзакции. При этом мы будем использовать тот же Outbox, чтобы гарантировать доставку каждого шага Saga.»

Если ты прошёл курс «Git Hosting: Serverless MVP», ты уже видел этот паттерн на упрощённом стеке: YDB Serverless с атомарным UPDATE вместо PostgreSQL с FOR UPDATE SKIP LOCKED, Message Queue вместо Kafka. Паттерн тот же — инфраструктура разная. Здесь мы разбираем его на полном стеке, с глубиной и trade-offs.

Если нужно глубже в брокеров — семантику доставки и разбор инцидентов смотри в треке «Брокеры сообщений»: модуль 3 и модуль 4 (ровно-один-раз).

← Модуль 2: Идемпотентность и Exactly-once Processing Модуль 4: Распределённые транзакции: Saga →

Модуль 3. Transactional Outbox и борьба с Dual‑Write

🔥 Инцидент: событие, которое исчезло

📖 Теория: Раскрытые термины

3.1. Проблема Dual‑Write

3.2. Паттерн Transactional Outbox

3.3. Воркер доставки: FOR UPDATE SKIP LOCKED

3.4. Retry + DLQ

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

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

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

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

Задача 1: Таблица outbox_events и индексы

Задача 2: PaymentHandler.Pay в одной транзакции pgx

Задача 3: outbox.Worker (goroutine + FOR UPDATE SKIP LOCKED)

Задача 4: Идемпотентный консьюмер

Задача 5: Тест Toxiproxy + k6

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

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

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

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

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

🔗 Мостик к Модулю 4 (Saga)

3.3. Воркер доставки: `FOR UPDATE SKIP LOCKED`

Задача 1: Таблица `outbox_events` и индексы

Задача 2: `PaymentHandler.Pay` в одной транзакции pgx

Задача 3: `outbox.Worker` (goroutine + FOR UPDATE SKIP LOCKED)