Skip to Content
PatternsSaga

Saga

Deep-dive по паттерну Saga: как сделать cross-service операцию с компенсацией, когда изменения затрагивают N сервисов и любой шаг может упасть. Reference по outbox-событиям — ../conventions/events, по timeouts — ../conventions/context-and-timeouts. Эта страница — про координацию, компенсации, state persistence, deadline, failure modes и тестирование.

Saga применяется, когда одному бизнес-действию соответствуют записи в двух и более сервисах, каждая запись может упасть по бизнес- или инфраструктурной причине, и тогда уже сделанные шаги нужно откатить логически — не через XA-транзакцию, а через компенсирующие действия.

Содержание

Когда Saga, когда нет

Saga — единственный паттерн для координированной записи в несколько сервисов с откатом. Не путай с другими случаями cross-service взаимодействия.

  • Один сервис пишет и публикует событие, downstream реагирует независимо — не saga, обычный outbox + eventual consistency. Источник истины один, откат не нужен.
  • Нужна read-модель с денормализацией по событиям из нескольких сервисов — не saga, это CQRS projection. Проекция не пишет обратно в источники, откатывать нечего.
  • Нужно собрать данные из N сервисов для одного API-ответа — не saga, это api composition. Композиция делает только чтения, без мутаций.
  • Координированная запись в N сервисов с откатом — saga. Пример: публикация отзыва с прикреплёнными фото — медиа надо «зарезервировать» в media-service, создать отзыв в review-service, поставить в очередь уведомление подписчикам в notification-service. Если где-то падение — ранее сделанные шаги откатываются компенсирующими действиями.

Если можно обойтись одним outbox’ом + eventually-consistent reactions — обходись. Saga — это лишний код, лишняя state-машина, лишний мониторинг. Включай её, только когда инвариант «либо все шаги, либо ни одного» действительно нужен бизнесу.

Choreography vs Orchestration

Есть два способа построить саги.

Choreography

Каждый сервис подписан на события и сам решает, что делать дальше. Шаг 1 публикует событие через outbox → сервис-владелец шага 2 его читает, выполняет свою часть, публикует своё событие → и так далее. Компенсация — тоже через события (review.creation_failed → media-service читает и делает release).

Использовать для простых линейных цепочек (2–3 шага, одна ветка, понятный compensation). Плюсы — меньше кода, нет центрального компонента. Минус — логика саги размазана по сервисам, debug «где мы сейчас в саге» требует склейки логов по correlation-id.

Orchestration

В сервисе-инициаторе живёт отдельный компонент — saga-orchestrator. Он держит state-машину, отправляет команды на шаги, слушает результаты, решает, двигаться дальше или запускать компенсацию. Шаги и компенсации — явные вызовы (через Kafka-команды или HTTP в редких случаях), orchestrator владеет всем порядком.

Использовать для ветвлений, >3 шагов, сложной логики compensation (частичная компенсация, условные откаты). Orchestrator живёт как компонент внутри сервиса-инициатора саги; это не отдельный сервис. Состояние — в таблице saga_instances той же БД сервиса.

Правило: ветвлений/условий нет и шагов ≤ 3 — choreography. Иначе orchestration. В сомнении — orchestration, так проще расширять.

Правила компенсаций

Каждая операция шага должна иметь парную compensating action. Требования к компенсации:

  • Идемпотентна. Повторный вызов ReleaseMedia(saga_id) не должен падать и не должен «перекомпенсировать» — ключ дедупликации — saga_id + step_idx.
  • Не падает по бизнес-причинам. ReleaseMedia не может вернуть «нельзя освободить, потому что пользователь заблокирован» — компенсация должна быть технически возможна всегда. Только инфраструктурные ошибки (Kafka down, БД недоступна) — повод для retry.
  • Семантически откатывает шаг, не обязательно физически удаляет запись. Для audit log — soft-delete с пометкой saga_rollback, не hard-delete.

Примеры пар:

ШагКомпенсация
media.Reserve(media_id, review_id)media.Release(media_id, saga_id) — снять reservation, освободить
review.Create(payload)review.Delete(review_id, saga_id) — soft-delete с rollback_reason='saga_rollback'
notification.Enqueue(payload)notification.Cancel(notification_id, saga_id) — пометить как cancelled до отправки

Правило: если компенсацию нельзя написать идемпотентной и без бизнес- проверок — шаг не годится для саги, пересмотри границы операции.

State persistence

Состояние саги хранится в таблице saga_instances в БД сервиса- инициатора (в схеме public, без префикса):

BEGIN;
SELECT pg_advisory_xact_lock(hashtext('review_migrations'));
 
CREATE TABLE saga_instances (
    id              UUID        PRIMARY KEY,
    saga_type       TEXT        NOT NULL,
    state           JSONB       NOT NULL,
    status          TEXT        NOT NULL,
    current_step    INT         NOT NULL DEFAULT 0,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    completed_at    TIMESTAMPTZ,
    CONSTRAINT saga_status_check CHECK (
        status IN ('running', 'compensating', 'completed', 'failed_compensation')
    )
);
 
CREATE INDEX idx_saga_instances_status
    ON saga_instances (status)
    WHERE status IN ('running', 'compensating', 'failed_compensation');
 
COMMIT;

Поля:

  • id — UUID саги, он же saga_id во всех командах и событиях шагов.
  • saga_type — publish_review, ban_user_with_reviews_cleanup и т.п.
  • state — JSONB с полным контекстом (входные параметры, accumulated результаты шагов: media_ids, review_id, notification_id). Каждое успешное завершение шага update’ит state.
  • status — одно из running, compensating, completed, failed_compensation. Переходы: running → completed (happy), running → compensating → failed (rollback успешен) или compensating → failed_compensation (5 попыток retry исчерпаны).
  • current_step — номер последнего успешно завершённого шага (в running) или шага, с которого пошла компенсация (в compensating).

Все update’ы state/status/current_step делает orchestrator в той же транзакции, что и запись в outbox (следующая команда для шага): InTx { UPDATE saga_instances ... + PublishTx(outbox, next-step-command) }. Это даёт at-least-once доставку команды шагу и атомарное продвижение state-машины.

Partial-index по незавершённым статусам нужен для быстрого скана «повисших» саг при рестарте orchestrator’а (см. §Failed compensation).

Deadline и timeout

У каждой саги есть общий deadline. Правила из ../conventions/context-and-timeouts:

  • User-facing saga (HTTP-endpoint ждёт ответ) — deadline 60 секунд.
  • Backoffice / background saga — deadline 10 минут. Дольше — только обоснованно и явным решением owner’а сервиса.

Deadline хранится в state.deadline_at (TIMESTAMPTZ). При старте каждого шага orchestrator проверяет:

if time.Now().After(state.DeadlineAt) {
    return orchestrator.transitionToCompensating(ctx, saga, "deadline_exceeded")
}

Если deadline истёк — сага уходит в compensating, компенсации запускаются по уже завершённым шагам. Это работает даже для «застрявшей» саги, у которой downstream не отвечает: watchdog-процесс раз в минуту сканит saga_instances WHERE status = 'running' AND state->>'deadline_at' < now() и форсит перевод в compensation.

Правило: deadline саги всегда меньше, чем retention outbox (7 дней), иначе watchdog может не успеть «поймать» повисшую сагу до cleanup команд-событий.

Idempotency шагов

Каждый шаг принимает saga_id + step_idx как idempotency-key. При ретрае команды (выполненной шагом ранее) сервис-исполнитель не должен делать работу повторно. Механика — через idempotent consumer на consumer-side.

// media-service handler на команду saga-шага
func (h *MediaHandler) HandleReserve(ctx context.Context, cmd ReserveMediaCmd) error {
    key := fmt.Sprintf("saga:%s:step:%d", cmd.SagaID, cmd.StepIdx)
    // Deduplicator + UPSERT по natural key
    _, err := h.db.Exec(ctx, `
        INSERT INTO media.reservations (saga_id, step_idx, media_id, status, created_at)
        VALUES ($1, $2, $3, 'reserved', NOW())
        ON CONFLICT (saga_id, step_idx) DO NOTHING`,
        cmd.SagaID, cmd.StepIdx, cmd.MediaID,
    )
    return err
}

Правило: natural key на стороне каждого сервиса — (saga_id, step_idx). Это и идемпотентность, и аудит (кто, когда, в рамках какой саги).

Versioning саг и шагов

State-машина саги — живой код, она эволюционирует: появляются новые шаги, существующие переименовываются, меняется порядок. При этом в БД в любой момент лежат in-flight саги — записи в saga_instances со status='running' или 'compensating', которые были начаты под старой версией оркестратора. Если деплой просто поменяет current_step ↔ имя шага — in-flight саги либо пропустят шаг, либо сломают компенсацию.

Правило: никакого in-place переопределения шагов. Любое изменение порядка / набора шагов оформляется как новая версия саги, старая доживает свои in-flight экземпляры.

saga_type_version

В таблице saga_instances поле saga_type хранит имя + номер версии: publish_review.v1, publish_review.v2. Новый orchestrator знает оба набора шагов; HandleStepResult диспатчит на нужный handler по saga_type:

func (o *Orchestrator) handle(saga *Saga, res StepResult) error {
    switch saga.Type {
    case "publish_review.v1":
        return o.v1.Handle(saga, res)
    case "publish_review.v2":
        return o.v2.Handle(saga, res)
    default:
        return fmt.Errorf("unknown saga_type %q", saga.Type)
    }
}
 
// Start всегда создаёт сагу в текущей активной версии.
func (o *Orchestrator) Start(ctx context.Context, in StartInput) (uuid.UUID, error) {
    return o.v2.Start(ctx, in) // активная версия — v2
}

Старт новых саг идёт только под активной версией (v2). Регистрация шагов старой версии в оркестраторе остаётся на время жизни in-flight саг.

Миграция in-flight

Когда меняется порядок шагов, in-flight саги версии v1 не мигрируются на v2 автоматически. Они доживают свой цикл под v1: либо доходят до completed, либо компенсируются под v1-схему, либо ловятся watchdog’ом по deadline и уходят в failed_compensation.

Это нормально: сагу нельзя «переложить» на новую схему без знания того, сколько её шагов уже выполнено в downstream и какие компенсации применимы к чему. Лучше дождаться окончания, чем автоматически «дорисовывать» state.

Проверка перед деплоем новой версии — всегда:

SELECT saga_type, status, count(*)
  FROM saga_instances
 WHERE status IN ('running', 'compensating', 'failed_compensation')
 GROUP BY saga_type, status;

Если в running/compensating есть старые saga_type — деплой новой версии их не затронет (v1-handler’ы остаются в коде). Если накопились failed_compensation — разобрать до деплоя v2, иначе смешиваются инциденты версий.

Когда удалять старую версию

Handler’ы версии v1 удаляются из оркестратора, когда:

  1. count(*) WHERE saga_type = 'publish_review.v1' AND status IN ('running', 'compensating') = 0 — ни одной in-flight.
  2. count(*) WHERE saga_type = 'publish_review.v1' AND status = 'failed_compensation' — все разобраны (или физически archived в отдельную таблицу).
  3. Прошёл как минимум один SLO-recovery-период (см. ../conventions/slo-and-budget) — чтобы свежий Kafka-replay не воскресил сагу старой версии.

Удаление — отдельным PR, с чеклистом выше в описании.

Регистр саг

Каждый сервис-оркестратор держит internal/saga/registry.go, который перечисляет все известные версии:

var ActiveSagas = []SagaDescriptor{
    {Type: "publish_review.v2", Active: true,  Handler: v2.Handle},
    {Type: "publish_review.v1", Active: false, Handler: v1.Handle}, // drain only
}
  • Active: true — новая сага этого типа запускается именно в этой версии.
  • Active: false — версия доживает in-flight, Start для неё запрещён. Handler остаётся, пока таблица пуста от этой версии (см. выше).

Конвенция: активная версия на saga_type всегда ровно одна. Параллельная активность v1 и v2 = расходящаяся state-машина без понятного rollback. Если нужно одновременно гонять две схемы — вводи новый saga_type (publish_review и, например, publish_review_fast), а не две Active=true версии одного типа.

Failed compensation

Если компенсация падает 5 попыток подряд (backoff по retry-and-circuit-breaker) — сага переходит в failed_compensation:

  • Запись остаётся в saga_instances.
  • Alert lead-инженеру backend-команды: saga_failed_compensation_total > 0 for 5m.
  • Запись попадает в DLQ-таблицу (или её роль играет сам статус failed_compensation — в зависимости от сервиса).
  • Ручной разбор обязателен. Автоматический ретрай после 5 попыток не делаем: если компенсация падает так долго, у системы что-то не так на уровне инвариантов.
  • Runbook — ../troubleshooting/saga-failed-compensation. Пошаговый разбор: найти сагу, проверить state шагов, выбрать путь (закрыть вручную / починить и replay компенсацию).

Запрос для runbook’а:

SELECT id, saga_type, status, current_step, state, updated_at
  FROM saga_instances
 WHERE status = 'failed_compensation'
 ORDER BY updated_at DESC;

Владелец сервиса (определяется по saga_type) разбирает каждую строку вручную: что за шаг, почему компенсация падает, какие побочные эффекты остались в downstream, как их руками привести в согласованное состояние. После ручного исправления — UPDATE status на completed или failed с комментарием в state.manual_resolution.

Observability

Метрики orchestrator’а (Prometheus, см. ../conventions/observability):

// registrar
sagaDuration := prometheus.NewHistogramVec(
    prometheus.HistogramOpts{
        Name:    "saga_duration_seconds",
        Help:    "End-to-end длительность саги от start до completed/failed.",
        Buckets: []float64{0.5, 1, 2, 5, 10, 30, 60, 300, 600},
    },
    []string{"saga_type", "outcome"}, // outcome = completed|failed|failed_compensation
)
 
sagaActive := prometheus.NewGaugeVec(
    prometheus.GaugeOpts{
        Name: "saga_active",
        Help: "Количество саг в running/compensating.",
    },
    []string{"saga_type"},
)
 
sagaCompensationTriggered := prometheus.NewCounterVec(
    prometheus.CounterOpts{
        Name: "saga_compensation_triggered_total",
        Help: "Сколько раз сага перешла в compensating; reason = step_error|deadline|manual.",
    },
    []string{"saga_type", "reason"},
)
 
sagaFailedCompensation := prometheus.NewCounterVec(
    prometheus.CounterOpts{
        Name: "saga_failed_compensation_total",
        Help: "Компенсации, исчерпавшие retry-budget.",
    },
    []string{"saga_type"},
)

Подключение — в cmd/server/main.go сервиса-инициатора. Gauge saga_active обновляется раз в 10 секунд фоном по SELECT count(*) ... GROUP BY saga_type, status.

Alerts:

  • saga_failed_compensation_total > 0 for 5m → page, разбирайся вручную.
  • histogram_quantile(0.99, saga_duration_seconds{outcome="completed"}) > 30s for 15m → ticket, что-то замедляет шаги.
  • rate(saga_compensation_triggered_total{reason="deadline"}[15m]) > 0 → ticket, deadline не покрывает реальную длительность.

Реализация на Go

Orchestration для саги publish_review. Шаги: media.Reserve → review.Create → notification.Enqueue; компенсации в обратном порядке. Orchestrator живёт в review-service, читает success/failure-события каждого шага через consumer на Kafka, двигает state-машину через outbox.

package sagareview
 
import (
    "context"
    "encoding/json"
    "fmt"
    "time"
 
    "github.com/google/uuid"
    "github.com/jackc/pgx/v5"
)
 
type Status string
 
const (
    StatusRunning             Status = "running"
    StatusCompensating        Status = "compensating"
    StatusCompleted           Status = "completed"
    StatusFailedCompensation  Status = "failed_compensation"
)
 
type State struct {
    UserID       int64     `json:"user_id"`
    PlaceID      int64     `json:"place_id"`
    MediaIDs     []int64   `json:"media_ids"`
    ReviewID     int64     `json:"review_id,omitempty"`
    NotificationID int64   `json:"notification_id,omitempty"`
    DeadlineAt   time.Time `json:"deadline_at"`
    LastError    string    `json:"last_error,omitempty"`
}
 
type Orchestrator struct {
    db       DB
    outbox   OutboxPublisher
    logger   Logger
}
 
// Start создаёт новую сагу, публикует команду на первый шаг.
func (o *Orchestrator) Start(ctx context.Context, input StartInput) (uuid.UUID, error) {
    sagaID := uuid.New()
    state := State{
        UserID:     input.UserID,
        PlaceID:    input.PlaceID,
        MediaIDs:   input.MediaIDs,
        DeadlineAt: time.Now().Add(60 * time.Second),
    }
 
    err := o.db.InTx(ctx, func(tx pgx.Tx) error {
        stateJSON, _ := json.Marshal(state)
        _, err := tx.Exec(ctx, `
            INSERT INTO saga_instances (id, saga_type, state, status, current_step)
            VALUES ($1, 'publish_review', $2, 'running', 0)`,
            sagaID, stateJSON,
        )
        if err != nil {
            return fmt.Errorf("insert saga: %w", err)
        }
        cmd := ReserveMediaCmd{SagaID: sagaID, StepIdx: 1, MediaIDs: state.MediaIDs}
        return o.outbox.PublishTx(ctx, tx, "outbox", newSagaCmd("media.reserve", cmd))
    })
    if err != nil {
        return uuid.Nil, err
    }
    return sagaID, nil
}
 
// HandleStepResult — вызывается consumer'ом orchestrator'а на success/failure
// события любого шага. Двигает state-машину.
func (o *Orchestrator) HandleStepResult(ctx context.Context, res StepResult) error {
    return o.db.InTx(ctx, func(tx pgx.Tx) error {
        saga, err := o.loadForUpdate(ctx, tx, res.SagaID)
        if err != nil {
            return err
        }
 
        if time.Now().After(saga.State.DeadlineAt) && saga.Status == StatusRunning {
            return o.transitionToCompensating(ctx, tx, saga, "deadline_exceeded")
        }
 
        switch saga.Status {
        case StatusRunning:
            if !res.Success {
                saga.State.LastError = res.Error
                return o.transitionToCompensating(ctx, tx, saga, "step_error")
            }
            return o.advanceForward(ctx, tx, saga, res)
        case StatusCompensating:
            return o.advanceCompensation(ctx, tx, saga, res)
        default:
            return nil // терминальный статус — игнор
        }
    })
}
 
func (o *Orchestrator) advanceForward(ctx context.Context, tx pgx.Tx, s *Saga, res StepResult) error {
    switch res.StepIdx {
    case 1: // media.Reserve done → review.Create
        s.State.MediaIDs = res.Payload.(ReservedMedia).IDs
        return o.emitStep(ctx, tx, s, 2, "review.create",
            CreateReviewCmd{SagaID: s.ID, StepIdx: 2, UserID: s.State.UserID, PlaceID: s.State.PlaceID})
    case 2: // review.Create done → notification.Enqueue
        s.State.ReviewID = res.Payload.(CreatedReview).ID
        return o.emitStep(ctx, tx, s, 3, "notification.enqueue",
            EnqueueCmd{SagaID: s.ID, StepIdx: 3, ReviewID: s.State.ReviewID})
    case 3: // notification.Enqueue done → completed
        s.State.NotificationID = res.Payload.(EnqueuedNotification).ID
        return o.finish(ctx, tx, s, StatusCompleted)
    }
    return fmt.Errorf("unknown forward step %d", res.StepIdx)
}
 
func (o *Orchestrator) transitionToCompensating(ctx context.Context, tx pgx.Tx, s *Saga, reason string) error {
    s.Status = StatusCompensating
    o.metrics.CompensationTriggered(s.Type, reason)
    // Запускаем компенсацию последнего завершённого шага.
    return o.emitCompensation(ctx, tx, s)
}
 
func (o *Orchestrator) advanceCompensation(ctx context.Context, tx pgx.Tx, s *Saga, res StepResult) error {
    if !res.Success {
        s.CompensationAttempts[res.StepIdx]++
        if s.CompensationAttempts[res.StepIdx] >= 5 {
            return o.finish(ctx, tx, s, StatusFailedCompensation)
        }
        // retry через tombstone: outbox публикует ту же команду ещё раз.
        return o.emitCompensation(ctx, tx, s)
    }
    s.CurrentStep--
    if s.CurrentStep == 0 {
        return o.finish(ctx, tx, s, StatusFailed)
    }
    return o.emitCompensation(ctx, tx, s)
}
 
// emitStep / emitCompensation / finish — обёртки над
// update saga_instances + outbox.PublishTx с правильными envelope'ами.
// Детали опущены.

Важные моменты:

  • loadForUpdate использует SELECT ... FOR UPDATE — orchestrator может иметь несколько реплик, lock на строке saga_instances защищает от параллельной обработки события одной саги двумя репликами.
  • Все переходы state-машины — через InTx + PublishTx в outbox. Без outbox’а (см. outbox) при падении orchestrator’а между update БД и публикацией команды сага «застынет».
  • Consumer orchestrator’а (на стороне review-service) слушает топики kazmaps.media.reservation_succeeded, kazmaps.review.created, kazmaps.notification.enqueued и их *_failed варианты, вызывает HandleStepResult.

Anti-patterns

XA / 2PC

Попытка сделать «распределённую транзакцию» через two-phase commit между Postgres / Kafka / Redis. На практике не работает: Kafka не поддерживает XA, Redis — тоже, а даже между Postgres-инстансами доступность резко падает. Наш стек — гетерогенный, 2PC запрещены (см. ../architecture-overview §Что мы ЯВНО не делаем). Используй Saga с compensation-based rollback.

Синхронный оркестратор, ждущий HTTP от всех шагов в одной горутине

// ПЛОХО
func (o *Orchestrator) Run(ctx context.Context, input Input) error {
    media, err := o.mediaHTTP.Reserve(ctx, input)  // 2s timeout
    if err != nil { return err }
    review, err := o.reviewHTTP.Create(ctx, input) // 2s timeout
    if err != nil {
        _ = o.mediaHTTP.Release(ctx, media)
        return err
    }
    return o.notifHTTP.Enqueue(ctx, review)
}

Три проблемы:

  1. Если orchestrator упадёт между Create и Enqueue — сага застряла, state нигде не зафиксирован.
  2. mediaHTTP.Release может упасть — компенсация потеряна.
  3. Deadline handler’а 2s × 3 = 6s — на user-facing endpoint’е это много.

Правильно: state в таблице, команды через outbox+Kafka, consumer слушает результаты.

Компенсации без идемпотентности

Повторный вызов компенсации → повторный эффект. Классический баг: при retry release’а media-service делает UPDATE quota = quota + 1 дважды. Используй saga_id + step_idx natural key + UPSERT / ON CONFLICT DO NOTHING.

Saga без state-persistence

«State saga — в памяти orchestrator’а». Pod упал → state потерян → висящие reservations в media-service, непонятно, что откатывать. Всегда — таблица saga_instances в БД сервиса-инициатора.

Orchestrator как отдельный сервис

Поначалу кажется чистым разделением. На практике — новый deployment, новый жизненный цикл, дублирование saga_instances в своей БД, отдельные dashboard’ы. Пока у нас нет саг, переживающих границу одного инициатора, orchestrator живёт внутри сервиса-инициатора как компонент (internal/saga/).

Тестирование

Integration через testcontainers

Поднимаем Postgres + Kafka через testcontainers, orchestrator с реальным saga_instances, fake handlers для шагов с возможностью инжектировать отказ:

func TestSaga_PublishReview_CompensatesOnNotificationFailure(t *testing.T) {
    env := testenv.Setup(t) // testcontainers pg + kafka + migrations
    defer env.Close()
 
    fakeMedia := env.StartFakeMediaService(t,
        fakeMediaConfig{ReserveOK: true, ReleaseOK: true})
    fakeReview := env.StartFakeReviewService(t,
        fakeReviewConfig{CreateOK: true, DeleteOK: true})
    fakeNotif := env.StartFakeNotificationService(t,
        fakeNotifConfig{EnqueueOK: false}) // отказ на шаге 3
 
    orch := sagareview.NewOrchestrator(env.DB, env.Outbox, env.Logger)
    sagaID, err := orch.Start(ctx, sagareview.StartInput{
        UserID: 42, PlaceID: 7, MediaIDs: []int64{100, 101},
    })
    if err != nil {
        t.Fatalf("start: %v", err)
    }
 
    testing.Eventually(t, 5*time.Second, func() bool {
        s, _ := env.LoadSaga(sagaID)
        return s.Status == sagareview.StatusFailed // успешно компенсировали
    })
 
    // Верификация компенсаций
    testing.Eventually(t, 2*time.Second, func() bool {
        return fakeReview.DeleteCalls(sagaID) == 1 &&
               fakeMedia.ReleaseCalls(sagaID) == 1
    })
}

Helper testing.Eventually — см. ../conventions/testing.

Важные сценарии

Минимальный набор на сагу:

  • Happy path — все шаги зелёные, completed.
  • Отказ на каждом шаге по очереди — проверяем, что компенсации идут в обратном порядке только по уже завершённым шагам.
  • Отказ компенсации — 5 попыток, переход в failed_compensation, метрика инкрементится.
  • Deadline истёк на running — переход в compensating с reason=deadline_exceeded.
  • Ретрай команды шага (дубликат) — шаг-сервис не выполняет работу дважды (natural key saga_id + step_idx).

FAQ

А если orchestrator упал посередине саги? State в saga_instances, команды — в outbox. При рестарте consumer orchestrator’а подхватит недоставленные события шагов (at-least-once от Kafka), HandleStepResult прочитает текущий state из БД и продолжит с той же точки. Watchdog-процесс раз в минуту сканит saga с истёкшим deadline.

Как отличить retry команды шага от нового вызова? По (saga_id, step_idx). Это natural key, исполнитель шага делает UPSERT или INSERT ON CONFLICT DO NOTHING. Retry не создаёт дубль, новый вызов (другой saga_id) — создаёт новую запись.

Можно ли смешивать choreography и orchestration? В пределах одной саги — нет. Либо все шаги управляются orchestrator’ом через команды, либо все реагируют на события соседей. Смешение даёт непонятное состояние: часть шагов знает про state-машину, часть — нет. В пределах разных саг в одном сервисе — можно: одна сага orchestration, другая простая linear — choreography.

Зачем столько церемонии, если большинство саг завершаются успешно? Потому что когда не завершаются — поломки тяжёлые и безмолвные. «Оторванные» reservations в media-service, висящие review без notifications. Стоимость отладки одного такого инцидента намного больше стоимости поддержки saga_instances + метрик.

Нужен ли timeout на каждом шаге отдельно? Да, per-step timeout через context — в дополнение к общему deadline саги. Per-step timeout короче (5–15 секунд для HTTP-шага), общий deadline — cap на всю сагу. Per-step истёк — шаг возвращает failure, orchestrator решает: retry шага (для transient ошибок) или переход в compensation.

Связанные разделы

Last updated on
Transactional outboxCQRS