Модуль 6: PR Saga в YDB — стейт-машина merge без оркестратора-процесса — Git Hosting на Go: Serverless MVP на Yandex Cloud

Модуль 6. PR Saga в YDB: стейт-машина merge без оркестратора-процесса

Цель модуля: Реализовать merge Pull Request как распределённую транзакцию Saga на Cloud Function Timer Trigger. Понять, что git merge необратим — компенсация не ROLLBACK, а «зафиксировать провал + точечный retry упавшего шага». Научиться логировать каждый переход стейт-машины для отладки «зависших» PR.

Стадия роста проекта: Сложные процессы. Git Smart HTTP работает, Outbox доставляет события, Bulkhead защищает от гонок. Теперь пользователь нажимает Merge PR — и это многошаговый процесс: validate CI → git merge → notify → index. Если один шаг падает — PR не должен зависнуть бесконечно.

Перед началом: Модули 1–5 (Setup, Smart HTTP, YDB, Outbox, Idempotency/Bulkhead)

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

Трудоёмкость: ~8 ч (теория 2 ч · практика 5 ч · логирование 1 ч)

🔥 Инцидент: PR завис в статусе «merging» на три дня

Вася открывает PR #42, все CI-проверки зелёные, нажимает Merge. Система начинает процесс: validate → merge → notify → index. На шаге index (обновление поискового индекса) Cloud Function падает по таймауту. PR остаётся в статусе merging.

Проходит час. День. Три дня. Вася видит «merging» и не понимает: merge прошёл или нет? Можно ли пушить в ветку? Удалять ли PR? Поддержка не может ответить — в логах тишина, переходы стейтов не логируются.

Вторая проблема: в K8s-версии (Архитектура v2, Модуль 4) оркестратор Saga — это goroutine с бесконечным циклом. В Serverless нет долгоживущих процессов. Замена — Timer Trigger: функция запускается каждые 5 секунд, проверяет «зависшие» PR, продвигает стейт-машину.

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

Три задачи:

Стейт-машина — явные статусы PR: VALIDATING → MERGING → NOTIFYING → INDEXING → MERGED / MERGE_FAILED.
Компенсация ≠ ROLLBACK — для git merge компенсация = «пометить merge-failed + retry только упавшего шага».
Логирование переходов — каждый переход стейта логируется с pr_id, from_state, to_state, reason.

Сквозной User Story: «Вася нажимает Merge PR #42. В Модуле 5 push был защищён от гонок. В этом модуле saga-orchestrator ведёт стейт-машину: VALIDATING → MERGING → NOTIFYING → INDEXING → MERGED. Если индексация падает — PR не зависает, а помечается MERGE_FAILED с retry. В Модуле 7 — как OpenTelemetry отслеживает каждый переход.»

📖 Теория: Saga в Serverless

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

Saga — последовательность локальных транзакций, где каждый шаг имеет компенсирующее действие. В Serverless — стейт-машина в YDB + Timer Trigger.

Стейт-машина — модель с конечным набором состояний и переходов между ними по событиям.

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

Компенсация — бизнес-операция, отменяющая эффект уже выполненного шага. Для git merge — НЕ ROLLBACK (коммит уже в master), а «пометить merge-failed + retry упавшего шага».

Идемпотентность компенсации — компенсация обязана быть идемпотентной через saga_id + step, чтобы при ретраях не выполнить шаг дважды.

MERGE_FAILED — терминальный статус: git merge прошёл, но последующий шаг упал. Git не трогаем. Ручной retry только упавшего шага.

Ментальная модель: «Представьте конвейер на заводе. Деталь проходит 4 станции: покраска → сборка → упаковка → отгрузка. Если упаковка сломалась — деталь НЕ возвращается на покраску (краска уже высохла). Вместо этого: деталь помечается "упаковка не завершена", конвейер останавливается, мастер подходит и вручную завершает упаковку. Git merge — это покраска: высохла, не отменишь. Компенсация — это вызов мастера, а не переделка.»

6.1. Стейт-машина PR merge

100%

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

Ключевые переходы:

Из	В	Условие	Действие
VALIDATING	MERGING	CI checks OK	Запуск git merge
VALIDATING	MERGE_FAILED	CI checks failed	PR не merge'ится
MERGING	NOTIFYING	git merge OK	Коммит в master, уведомление watchers
MERGING	MERGE_FAILED	git merge failed	Конфликт или timeout
NOTIFYING	INDEXING	notify OK	Обновление поискового индекса
INDEXING	MERGED	index OK	PR закрыт, успех
Любое	MERGE_FAILED	timeout / ошибка	failed_step записан, retry возможен

6.2. Saga Orchestrator: Cloud Function Timer Trigger

В K8s-версии оркестратор — goroutine с бесконечным циклом. В Serverless — Timer Trigger каждые 5 секунд:

go
// cmd/saga-orchestrator/main.go
func handleSagaTick(ctx context.Context) error {
    // (1) Находим PR в «промежуточных» статусах
    stuckPRs, err := findStuckPRs(ctx)
    if err != nil {
        return err
    }

    for _, pr := range stuckPRs {
        // (2) Продвигаем стейт-машину
        if err := advancePR(ctx, pr); err != nil {
            slog.Error("saga advance failed",
                "pr_id", pr.ID,
                "state", pr.State,
                "err", err,
            )
            // Не прерываем цикл — другие PR могут продвинуться
        }
    }
    return nil
}

func findStuckPRs(ctx context.Context) ([]PR, error) {
    // Index Scan: ищем PR в промежуточных статусах
    res, err := ydbClient.Execute(ctx, `
        SELECT pr_id, repo_id, state, failed_step, attempts, updated_at
        FROM pr_saga
        VIEW idx_saga_state
        WHERE state IN ('VALIDATING', 'MERGING', 'NOTIFYING', 'INDEXING')
          AND updated_at < CurrentUtcTimestamp() - Interval('PT2M')
        LIMIT 10
    `, nil)
    // ...
}

Критично: updated_at < NOW() - 2 min — не трогаем PR, которые обрабатываются прямо сейчас. Только «зависшие».

6.3. Атомарный переход стейта в YDB

Нет FOR UPDATE — используем атомарный UPDATE с условием (из Модуля 3):

go
func advancePR(ctx context.Context, pr PR) error {
    nextState := getNextState(pr.State)

    // Атомарный переход: UPDATE с условием state = $current_state
    res, err := ydbClient.Execute(ctx, `
        UPDATE pr_saga
        SET state = $next_state,
            updated_at = CurrentUtcTimestamp(),
            attempts = 0
        WHERE pr_id = $pr_id
          AND state = $current_state
    `, map[string]any{
        "$pr_id":         pr.ID,
        "$current_state": pr.State,
        "$next_state":    nextState,
    })

    if res.RowsAffected() == 0 {
        // Конкурентный оркестратор уже продвинул этот PR
        return nil
    }

    // Логируем переход
    slog.Info("saga transition",
        "pr_id", pr.ID,
        "from", pr.State,
        "to", nextState,
    )

    // Выполняем действие нового состояния
    return executeStep(ctx, pr, nextState)
}

6.4. Компенсация: git merge необратим

Ключевой инсайт из ревью: компенсация для git merge — это НЕ ROLLBACK.

go
func executeStep(ctx context.Context, pr PR, state string) error {
    switch state {
    case "MERGING":
        err := runGitMerge(ctx, pr)
        if err != nil {
            // git merge не прошёл → MERGE_FAILED, git не трогаем
            return markFailed(ctx, pr, "MERGING", err.Error())
        }
        return nil

    case "NOTIFYING":
        err := notifyWatchers(ctx, pr)
        if err != nil {
            // git merge УЖЕ прошёл → НЕ откатываем коммит
            // Компенсация: MERGE_FAILED + failed_step = "NOTIFYING"
            return markFailed(ctx, pr, "NOTIFYING", err.Error())
        }
        return nil

    case "INDEXING":
        err := updateSearchIndex(ctx, pr)
        if err != nil {
            // git merge УЖЕ прошёл → НЕ откатываем коммит
            // Компенсация: MERGE_FAILED + failed_step = "INDEXING"
            return markFailed(ctx, pr, "INDEXING", err.Error())
        }
        return nil
    }
    return nil
}

func markFailed(ctx context.Context, pr PR, failedStep string, reason string) error {
    _, err := ydbClient.Execute(ctx, `
        UPDATE pr_saga
        SET state = 'MERGE_FAILED',
            failed_step = $failed_step,
            last_error = $reason,
            updated_at = CurrentUtcTimestamp()
        WHERE pr_id = $pr_id
    `, map[string]any{
        "$pr_id":       pr.ID,
        "$failed_step": failedStep,
        "$reason":      reason,
    })

    // Логируем переход в MERGE_FAILED
    slog.Warn("saga merge_failed",
        "pr_id", pr.ID,
        "failed_step", failedStep,
        "reason", reason,
    )
    return err
}

Ручной retry: если PR в статусе MERGE_FAILED с failed_step = "INDEXING" — администратор запускает retry только индексации, не трогая git:

go
func retryFailedStep(ctx context.Context, prID string) error {
    pr, _ := getPR(ctx, prID)
    if pr.State != "MERGE_FAILED" {
        return fmt.Errorf("PR not in MERGE_FAILED state")
    }

    // Возвращаемся в состояние ПЕРЕД упавшим шагом
    retryState := getStateBefore(pr.FailedStep)

    _, err := ydbClient.Execute(ctx, `
        UPDATE pr_saga
        SET state = $retry_state, attempts = 0, updated_at = CurrentUtcTimestamp()
        WHERE pr_id = $pr_id AND state = 'MERGE_FAILED'
    `, map[string]any{
        "$pr_id":       prID,
        "$retry_state": retryState,
    })
    return err
}

6.5. Логирование переходов (рекомендация из ревью)

Каждый переход стейта логируется с полной информацией:

go
// Логирование каждого перехода
func logTransition(pr PR, from, to string, reason string) {
    slog.Info("saga_transition",
        "pr_id", pr.ID,
        "repo_id", pr.RepoID,
        "from_state", from,
        "to_state", to,
        "reason", reason,
        "attempts", pr.Attempts,
        "failed_step", pr.FailedStep,
        "timestamp", time.Now().UTC().Format(time.RFC3339),
    )
}

Это позволяет отлаживать «зависшие» PR:

bash
# Поиск всех переходов PR #42
yc logging read --folder-id=... --filter='pr_id="42"' --since="3d"

# Результат:
# 2024-01-15T10:00:00Z saga_transition pr_id=42 from=VALIDATING to=MERGING reason="ci_ok"
# 2024-01-15T10:00:05Z saga_transition pr_id=42 from=MERGING to=NOTIFYING reason="merge_ok"
# 2024-01-15T10:00:10Z saga_transition pr_id=42 from=NOTIFYING to=INDEXING reason="notify_ok"
# 2024-01-15T10:00:15Z saga_merge_failed pr_id=42 failed_step=INDEXING reason="timeout"

6.6. Схема YDB для PR Saga

sql
-- migrations/005_pr_saga.yql

CREATE TABLE pr_saga (
    pr_id       UUID NOT NULL,
    repo_id     UUID NOT NULL,
    state       UTF8 NOT NULL DEFAULT 'VALIDATING',
    failed_step UTF8,
    last_error  UTF8,
    attempts    INT64 NOT NULL DEFAULT 0,
    updated_at  TIMESTAMP NOT NULL,
    created_at  TIMESTAMP NOT NULL,
    PRIMARY KEY (pr_id)
);

-- Индекс для поиска «зависших» PR
CREATE INDEX idx_saga_state ON pr_saga (state, updated_at);

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

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

Архетип	Шаги Saga	Компенсация	Обратимость
Финтех	Reserve → Withdraw → Notify	RefundWithdraw (отмена эффекта)	Обратимая
Git-хостинг	Validate → Merge → Notify → Index	Mark merge-failed + retry шага	Необратимая (git merge)
Мессенджер	Send → Deliver → Read	Mark delivery-failed + retry	Частично обратимая
Лента	N/A (нет многошаговых процессов)	N/A	N/A

Общий принцип: компенсация — новая бизнес-операция, не SQL ROLLBACK. Для обратимых операций (финтех) — отмена эффекта. Для необратимых (git merge) — фиксация провала + точечный retry.

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

Решение	Плюс	Минус	Когда выбирать
Saga + Timer Trigger (Serverless)	Нет долгоживущих процессов; 0₽ при малой нагрузке	Задержка до 5 сек; нет FOR UPDATE	Serverless MVP
Saga + goroutine (K8s)	Минимальная задержка; FOR UPDATE	Нужен долгоживущий процесс	K8s / VPS
Компенсация = mark failed + retry	Git не трогаем; точечный retry	Ручной шаг для администратора	Необратимые операции (git merge)
Компенсация = отмена эффекта	Полный откат	Невозможно для git merge	Обратимые операции (финтех)
Логирование каждого перехода	Отладка «зависших» PR	Больше логов (стоимость)	Всегда для production

Вывод: для Serverless Git-хостинга — Saga + Timer Trigger + mark failed + логирование переходов. См. ADR-SV6.

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

Сценарий: Вася нажимает Merge PR #42. CI-проверки зелёные. Цель: merge проходит за < 30 сек, при падении индексации PR не зависает, а помечается MERGE_FAILED с retry.

Задача 1: Стейт-машина в YDB

Создать таблицу pr_saga с индексами. Реализовать атомарный переход стейта (UPDATE с условием state = $current_state).

✅ Готово, когда: [ ] Таблица pr_saga: pr_id, repo_id, state, failed_step, last_error, attempts, updated_at [ ] Индекс idx_saga_state на (state, updated_at) [ ] Атомарный UPDATE: WHERE state = $current_state, rowsAffected = 0 → конфликт [ ] Тест: два конкурентных advancePR → один успех, один no-op

Задача 2: Saga Orchestrator (Timer Trigger)

Реализовать Cloud Function с Timer Trigger каждые 5 секунд: поиск «зависших» PR, продвижение стейт-машины.

✅ Готово, когда: [ ] Timer Trigger: cron */5 * * * * ? [ ] Поиск PR: state IN ('VALIDATING', 'MERGING', 'NOTIFYING', 'INDEXING') AND updated_at < NOW() - 2 min [ ] Для каждого PR: advancePR → executeStep [ ] Ошибка на одном PR не прерывает обработку остальных

Задача 3: Компенсация (mark failed + retry)

Реализовать markFailed: git merge не откатываем, записываем failed_step. Реализовать retryFailedStep: возврат в состояние перед упавшим шагом.

✅ Готово, когда: [ ] При падении NOTIFYING/INDEXING: state = MERGE_FAILED, failed_step записан [ ] Git merge НЕ откатывается (коммит остаётся в master) [ ] retryFailedStep: возвращает в состояние перед упавшим шагом [ ] Тест: merge прошёл, index упал → MERGE_FAILED → retry → MERGED

Задача 4: Логирование переходов

Добавить логирование каждого перехода: pr_id, from_state, to_state, reason, timestamp. Проверить через yc logging read.

✅ Готово, когда: [ ] Каждый переход логируется: slog.Info("saga_transition", ...) [ ] MERGE_FAILED логируется с slog.Warn + failed_step + reason [ ] yc logging read --filter='pr_id="42"' показывает полную историю [ ] Можно восстановить последовательность переходов из логов

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

go
// Go Cloud Function (Антипаттерн)
func (h *Handler) MergePR(w http.ResponseWriter, r *http.Request) {
    // ❌ 1. Все шаги в одном HTTP-запросе (нет стейт-машины)
    validateCI(pr)
    gitMerge(pr)     // Если упадёт на notify — PR зависнет
    notifyWatchers(pr)
    updateIndex(pr)

    // ❌ 2. При ошибке — «просто перезапустить с начала»
    // Но git merge уже прошёл! Повторный merge = конфликт

    // ❌ 3. Нет логирования переходов
    // Поддержка не может понять, где завис PR
}

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

Что произойдёт, если notifyWatchers упадёт после gitMerge? (PR зависнет, git merge уже прошёл)
Почему «перезапустить с начала» недопустимо после git merge? (Коммит уже в master, повторный merge = конфликт)
Как поддержка поймёт, где завис PR, без логирования переходов? (Никак — в логах тишина)

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

Студенту необходимо нарисовать State Diagram стейт-машины PR merge с ветками компенсации.

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

Все состояния: VALIDATING, MERGING, NOTIFYING, INDEXING, MERGED, MERGE_FAILED.
Переходы с условиями (CI OK, merge OK, timeout).
Ветка MERGE_FAILED → retry только упавшего шага.
Note: «git merge необратим, компенсация ≠ ROLLBACK».

Шаблон для ориентира: /serverless-mvp/diagrams/06-pr-saga-states.svg

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

Q: Почему компенсация git merge — это НЕ ROLLBACK?

Git не поддерживает транзакции
Коммит уже в master; откат перепишет историю тех, кто успел pull*
ROLLBACK медленнее компенсации

Q: Чем Saga в Serverless отличается от Saga в K8s?

Ничем
В Serverless оркестратор = Timer Trigger (каждые 5 сек); в K8s = goroutine с бесконечным циклом*
В Serverless нет компенсаций

Q: Что делает saga-orchestrator при падении INDEXING после успешного git merge?

Откатывает git merge (git reset --hard)
Помечает PR MERGE_FAILED с failed_step=INDEXING; git не трогает; retry только индексации*
Перезапускает всю Saga с начала

Q: Зачем нужен индекс idx_saga_state на (state, updated_at)?

Для красоты
Для быстрого поиска «зависших» PR: state IN (...) AND updated_at < NOW() - 2 min*
Для сортировки по дате

Q: Атомарный переход стейта в YDB — как реализуется?

SELECT ... FOR UPDATE
UPDATE ... WHERE state = $current_state; rowsAffected = 0 → конфликт*
BEGIN TRANSACTION ... COMMIT

Q: Зачем логировать каждый переход стейта?

Для красоты логов
Для отладки «зависших» PR: можно восстановить последовательность и найти, где упал*
Это не обязательно

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

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

Сначала проверь понимание: дай короткий ответ (2–3 предложения) — почему компенсация git merge — это НЕ ROLLBACK, и что значит «mark merge-failed + retry только упавшего шага».

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

Спроектировать стейт-машину PR merge (VALIDATING → MERGING → NOTIFYING → INDEXING → MERGED / MERGE_FAILED).
Реализовать Saga Orchestrator как Cloud Function Timer Trigger (каждые 5 сек).
Реализовать markFailed и retryFailedStep (git merge не откатываем).
Добавить логирование каждого перехода стейта.

Покажи мне код Антипаттерна (все шаги в одном HTTP-запросе, «перезапустить с начала» после git merge, нет логирования) и попроси объяснить, почему каждый из трёх пунктов убьёт систему.

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

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

[ ] Стейт-машина: все состояния (VALIDATING, MERGING, NOTIFYING, INDEXING, MERGED, MERGE_FAILED). [ ] Атомарный переход: UPDATE WHERE state = $current_state, rowsAffected = 0 → конфликт. [ ] Timer Trigger: cron каждые 5 секунд, поиск «зависших» PR (updated_at < NOW() - 2 min). [ ] Компенсация: git merge НЕ откатываем. failed_step записан. Retry только упавшего шага. [ ] Логирование: каждый переход (pr_id, from, to, reason). MERGE_FAILED с failed_step. [ ] Тест: merge прошёл, index упал → MERGE_FAILED → retry → MERGED. [ ] Тест: два конкурентных advancePR → один успех, один no-op. [ ] Mermaid State Diagram: все переходы + ветка retry + note про необратимость git merge. [ ] ADR-SV6: «Saga в Serverless: Timer Trigger vs goroutine. Компенсация: mark failed vs ROLLBACK».

🔗 Мостик к Модулю 7 (Observability)

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

«Отлично, PR merge проходит через стейт-машину Saga: каждый переход атомарен, git merge не откатывается, зависшие PR не теряются. Но как мы узнаем, что Saga работает правильно? Как увидим, что PR #42 прошёл VALIDATING → MERGING → NOTIFYING, а на INDEXING упал с таймаутом? В Модуле 7 мы внедрим OpenTelemetry: сквозной trace_id от API до saga-orchestrator, метрики (saga_transitions_total, saga_merge_failed_total), и cost-алерты на Cloud Monitoring. А ещё — как читать логи через yc logging read, когда в Cloud Functions логи — единственный спасательный круг.»

Сквозной User Story: «Вася нажал Merge PR #42. Saga прошла VALIDATING → MERGING → NOTIFYING → INDEXING → MERGED (Модуль 6). В Модуле 7 мы посмотрим, как OpenTelemetry связывает все эти переходы в один трейс, как Cloud Monitoring алертит при saga_merge_failed_total > 5 за 10 минут, и как cost-алерт срабатывает при превышении 10 RU/с в YDB.»

← Модуль 5: Идемпотентность и Bulkhead в Serverless Модуль 7: Observability на минимальном стеке — OpenTelemetry, Jaeger, Cloud Monitoring →