Модуль 6. PR Saga: merge flow с компенсацией на Go

Цель модуля: Реализовать распределённый merge pull request как стейт-машину Saga (arch m4) на Go: VALIDATING → MERGING → NOTIFYING → INDEXING → MERGED. Понять, что для Git компенсация ≠ ROLLBACK (git merge не отменяем), и реализовать компенсацию в виде «пометить PR как merge-failed + ручной retry». Защитить оркестратор от гонок через FOR UPDATE на pgx, команды гонять через Outbox (модуль 4). Главное отличие от Symfony-модуля — нет Doctrine flush, есть явные tx.BeginTx + tx.Exec + tx.Commit, что даёт больше контроля и меньше «магии».

Перед началом: Arch m3 (Outbox), m4 (Saga) · Git-hosting-go модули 1–5 Инструменты: Go 1.23+, github.com/jackc/pgx/v5, PostgreSQL, Toxiproxy Трудоёмкость: ~7 ч (теория 2 ч · практика 4.5 ч · ADR 0.5 ч)

🔥 Инцидент: «просто перезапустите» — но merge уже прошёл

Пользователь нажимает Merge PR. Шаги: validate CI checks → git merge → notify watchers → update search index. CI проверку прошли, git merge сделали, но индексация упала (поисковый сервис 500-ил). PR остаётся в статусе merging бесконечно — вотчеры не уведомлены, код в master есть, а в поиске его нет.

Команда предлагает «просто перезапустить Saga с начала». Но git merge уже прошёл: повторный merge даст конфликт или Already up to date, а откатить коммит в master значит переписать историю всех, кто успел git pull. Это не платёжная операция, где компенсация = RefundWithdraw. Git merge нельзя отменить — значит, компенсация должна быть другой.

Saga из arch m4, но с Git-специфичной компенсацией: вместо ROLLBACK мы помечаем PR как merge-failed, пишем в лог и даём ручной retry только упавшего шага. На Go — стейт-машина на switch, без workflow-библиотеки.

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

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

• Saga (Orchestration) — последовательность шагов с компенсирующими действиями; оркестратор ведёт стейт-машину.
• Компенсация (Git-специфичная) — НЕ ROLLBACK: git merge не отменяем; компенсация = пометить PR merge-failed + лог + retry упавшего шага.
• Стейт-машина PR merge — VALIDATING → MERGING → NOTIFYING → INDEXING → MERGED.
• saga_instances — таблица состояния Saga (как в arch m4), FOR UPDATE от гонок между горутинами/подами.
• Команда через Outbox — MergeCommand → outbox_events → горутина-консьюмер → git merge → MergeCompleted.
• **pgx BeginTx](https://pkg.go.dev/github.com/jackc/pgx/v5) — явная транзакция: conn.BeginTx(ctx, pgx.TxOptions{}) возвращает pgx.Tx, на нём вызываем QueryRow/Exec, затем tx.Commit(ctx) или tx.Rollback(ctx).

6.1. Почему 2PC/XA не подходят для merge

Merge затрагивает Git-репозиторий (filesystem), сервис уведомлений и поисковый индекс — три независимых хранилища без общего XA. 2PC заблокировал бы refs на этапе Prepare и завис при сбое поисковика. Только Saga с компенсациями (arch m4).

6.2. Стейт-машина PR merge (на switch)

VALIDATING ──ok──> MERGING ──ok──> NOTIFYING ──ok──> INDEXING ──ok──> MERGED
    │                 │                 │                 │
    └──fail──> MERGE_FAILED <──fail─────┴──fail───────────┘
                       │
            (git merge НЕ отменяем; помечаем PR + лог + retry шага)

На Go стейт-машина — это обычный switch (без FSM-библиотеки):

go
func (o *MergeOrchestrator) advance(ctx context.Context, saga *Saga, event Event) error {
    switch saga.State {
    case StateValidating:
        if !event.Ok { return o.compensate(ctx, saga, "VALIDATING", event.Err) }
        return o.to(ctx, saga, StateMerging, "MergeCommand")
    case StateMerging:
        if !event.Ok { return o.compensate(ctx, saga, "MERGING", event.Err) }
        return o.to(ctx, saga, StateNotifying, "NotifyWatchers")
    case StateNotifying:
        if !event.Ok { return o.compensate(ctx, saga, "NOTIFYING", event.Err) }
        return o.to(ctx, saga, StateIndexing, "IndexRepo")
    case StateIndexing:
        if !event.Ok { return o.compensate(ctx, saga, "INDEXING", event.Err) }
        return o.to(ctx, saga, StateMerged, "")
    }
    return ErrUnknownState
}

Каждый шаг: оркестратор обновляет стейт + пишет следующую команду в Outbox в одной транзакции (arch m4). Команды обрабатываются горутинами-консьюмерами (модуль 4 — FOR UPDATE SKIP LOCKED), ответы возвращаются событиями (MergeCompleted, IndexingFailed).

6.3. Компенсация ≠ ROLLBACK (связь с arch m4)

В финтехе (arch m4) компенсация = RefundWithdraw — отмена эффекта. Для Git merge это невозможно: коммит уже в master, его нельзя «вернуть» без переписывания истории. Поэтому компенсация Git-Saga:

1. Перевести Saga в MERGE_FAILED.
2. Пометить PR как merge-failed + записать failed_step и last_error в лог.
3. Не трогать git (merge уже прошёл и валиден).
4. Дать ручной retry только упавшего шага (например, переиндексацию) без повтора git merge.

Принцип тот же (arch m4: компенсация — новая бизнес-операция, не SQL ROLLBACK), но здесь это «зафиксировать провал + точечный retry», а не «отменить действие».

6.4. Защита от гонок: pgx + FOR UPDATE

Несколько подов/горутин оркестратора могут одновременно взять одну Saga: SELECT ... FOR UPDATE по saga_id внутри pgx.Tx, плюс идемпотентная проверка текущего state (если уже сменился — выходим). Таймаут-джоб (отдельная горутина с time.Ticker) ищет Saga в *ING с updated_at < NOW() - INTERVAL '2 minutes' и запускает компенсацию (по индексу idx_saga_state).

go
func (o *MergeOrchestrator) transition(ctx context.Context, sagaID int64, to State, cmd string) error {
    tx, err := o.pool.BeginTx(ctx, pgx.TxOptions{}) // явная транзакция
    if err != nil { return err }
    defer tx.Rollback(ctx) // безопасный rollback, если Commit не дошёл

    var s Saga
    err = tx.QueryRow(ctx,
        `SELECT saga_id, state, aggregate_id FROM saga_instances
         WHERE saga_id = $1 FOR UPDATE`, sagaID).Scan(&s.ID, &s.State, &s.AggregateID)
    if err != nil { return err }
    if s.State != stateExpectedFor(to) { return ErrAlreadyAdvanced } // идемпотентность

    if _, err = tx.Exec(ctx,
        `UPDATE saga_instances SET state = $2, updated_at = NOW() WHERE saga_id = $1`,
        sagaID, to); err != nil { return err }

    if cmd != "" {
        if _, err = tx.Exec(ctx,
            `INSERT INTO outbox_events(aggregate_id, type, payload)
             VALUES ($1, $2, $3)`,
            s.AggregateID, cmd, []byte("{}")); err != nil { return err }
    }
    return tx.Commit(ctx) // одна транзакция — стейт + команда
}

Главное отличие от Symfony: нет Doctrine flush — мы явно видим каждый Exec, контролируем порядок, и defer tx.Rollback(ctx) безопасен (после Commit он no-op). Больше кода, но меньше сюрпризов: никаких «magic flush» в подписчиках событий.

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

Вывод: Orchestration Saga на switch; pgx с явными BeginTx/Commit/Rollback; компенсация для необратимого git merge = пометить merge-failed + лог + ручной retry упавшего шага. См. ADR-GG6.

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

Сценарий: Merge PR падает на шаге INDEXING. Цель: PR → merge-failed, git merge не повторяется, ручной retry переиндексации.

План:

1. Таблица saga_instances (arch m4 схема) + индексы state, updated_at.
2. MergeOrchestrator — стейт-машина на switch + переход в pgx-транзакции.
3. Компенсация: IndexingFailed → MERGE_FAILED (без ROLLBACK git).
4. Таймаут-горутина: *ING старше 2 мин → компенсация.
5. Тест: индексация падает → PR merge-failed, ручной retry → MERGED.
6. ADR-GG6 «Компенсация Git merge: mark-failed vs ROLLBACK; pgx vs Doctrine».

Задача 1: Схема saga_instances

sql
CREATE TABLE saga_instances (
    saga_id        UUID PRIMARY KEY,
    aggregate_id   UUID NOT NULL,        -- PR id
    state          VARCHAR(50) NOT NULL, -- VALIDATING, MERGING, ..., MERGE_FAILED, MERGED
    failed_step    VARCHAR(50),
    last_error     TEXT,
    updated_at     TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    created_at     TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_saga_state   ON saga_instances(state);
CREATE INDEX idx_saga_updated ON saga_instances(updated_at);

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

• Поля saga_id, aggregate_id (PR), state, failed_step, last_error, updated_at
• Индексы state (таймаут-горутина) и updated_at

Задача 2: MergeOrchestrator — переход + команда в Outbox

Реализуйте transition(ctx, sagaID, to State, cmd string) error из §6.4: BeginTx → SELECT ... FOR UPDATE → идемпотентная проверка state → UPDATE → INSERT outbox_events → Commit. Используйте defer tx.Rollback(ctx) (no-op после Commit).

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

• SELECT ... FOR UPDATE при поиске Saga (защита от гонок между горутинами)
• Идемпотентность: если state уже сменился — ErrAlreadyAdvanced, без повторного перехода
• Переход стейта + команда в Outbox одним tx.Commit(ctx)

Задача 3: Компенсация (mark merge-failed, без ROLLBACK git)

При IndexingFailed Saga → MERGE_FAILED. Компенсация: пометить PR merge-failed, записать failed_step='INDEXING' + last_error. Git не трогаем (merge уже в master). Ручной retry вызывает только переиндексацию.

go
func (o *MergeOrchestrator) compensate(ctx context.Context, saga *Saga, step, lastErr string) error {
    tx, err := o.pool.BeginTx(ctx, pgx.TxOptions{})
    if err != nil { return err }
    defer tx.Rollback(ctx)

    if _, err = tx.Exec(ctx,
        `UPDATE saga_instances
            SET state = $2, failed_step = $3, last_error = $4, updated_at = NOW()
          WHERE saga_id = $1`,
        saga.ID, StateMergeFailed, step, lastErr); err != nil { return err }
    if _, err = tx.Exec(ctx,
        `UPDATE pull_requests SET status = 'merge-failed' WHERE id = $1`,
        saga.AggregateID); err != nil { return err }
    return tx.Commit(ctx)
}

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

• IndexingFailed → MergeFailed; git merge НЕ повторяется и НЕ откатывается
• PR помечен merge-failed; записаны failed_step, last_error
• Ручной retry запускает только упавший шаг (INDEXING), а не весь флоу

Задача 4: Таймаут-горутина

time.Ticker (каждую минуту) ищет Saga в *ING с updated_at < NOW() - INTERVAL '2 minutes' (по индексу) и запускает компенсацию. Используйте ctx с WithTimeout, чтобы одна зависшая Saga не блокировала тикер.

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

• time.Ticker находит зависшие *ING Saga и запускает компенсацию (arch m4)
• Тикер не стопорится на одной Saga (per-итерация context.WithTimeout)

Задача 5: Тест

Индексация падает (Toxiproxy/stub 500). SLO: Saga → MERGE_FAILED, git merge выполнен ровно 1 раз, ручной retry переиндексации → MERGED, RTO < 1 мин.

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

• При падении INDEXING: PR → merge-failed, merge не повторяется
• Ручной retry → MERGED

⚠️ Антипаттерн

go
func (h *Handler) MergePR(w http.ResponseWriter, r *http.Request) {
    h.validator.AssertCIGreen(pr)
    h.git.Merge(pr)              // git merge в синхронном запросе
    h.notifier.NotifyWatchers(pr)
    h.search.Index(pr)           // ❌ упало → PR завис в 'merging', merge уже в master
    // «просто перезапустить» → повторный merge конфликтует или ломает историю
}

Вопрос: Почему «перезапустить Saga с начала» недопустимо, если git merge уже прошёл?

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

🔗 Мостик к Модулю 7 (Scale to 100k)

Saga работает на 10k RPS, pgx держит транзакции, git merge идемпотентно завершается через компенсацию. Но что при 100k RPS? На Symfony-курсе первым узким местом был PHP-FPM → RoadRunner. На Go этот шаг отсутствует: рантайм уже long-lived, горутины дёшевы, бинарник один. Узким местом становятся диск (Storage Nodes), брокер (Redis Streams → Kafka) и PostgreSQL (Citus). В Модуле 7 эволюционируем до Kafka + Citus + CDN — и поймём, почему Go «не нужен RoadRunner» — но нужна остальная инфраструктура.