Модуль 3. Distributed Systems Patterns

Цели модуля

После модуля студент сможет:

• Понимать фундаментальные ограничения распределённых систем (CAP, FLP)
• Проектировать системы с eventual consistency
• Реализовывать retries с exponential backoff и jitter
• Обеспечивать идемпотентность операций через ключи
• Применять outbox pattern для надёжной публикации событий
• Проектировать Saga оркестрацию для распределённых транзакций
• Реализовывать компенсационные транзакции
• Настраивать Dead Letter Queue и обработку failed events

3.1 Фундаментальные ограничения

CAP теорема

Распределённая система может гарантировать только 2 из 3 свойств:

• Consistency (консистентность): все узлы видят одни данные
• Availability (доступность): каждый запрос получает ответ
• Partition tolerance (устойчивость к разделению): система работает при разрыве сети

На практике Partition tolerance обязательна (сети ненадёжны). Выбор между C и A.

PACELC расширение

Когда система работает нормально (no Partition):

• Латентность (Latency) vs Консистентность (Consistency)

Пример:

• DynamoDB: PA/EL (availability при partition, low latency обычно)
• PostgreSQL: CA/EC (консистентность всегда)

FLP Impossibility

В асинхронной системе с хотя бы одним сбойным узлом невозможно достичь консенсуса за конечное время.

Практическое следствие:

• Всегда будут таймауты
• Всегда будут частичные отказы
• Нужны механизмы компенсации

8 Fallacies of Distributed Computing

Ложные предположения которые делают разработчики:

1. Сеть надёжна
2. Latency равен нулю
3. Bandwidth бесконечен
4. Сеть безопасна
5. Топология не меняется
6. Есть один администратор
7. Стоимость транспорта равна нулю
8. Сеть однородна

Каждое из этих предположений ложно. Архитектор должен это учитывать.

Следствия для проектирования

1. Сетевые вызовы могут падать — нужны retries и таймауты
2. Сообщения могут дублироваться — нужна идемпотентность
3. Сообщения могут приходить в неправильном порядке — нужны sequence numbers
4. Узлы могут падать — нужны health checks и failover
5. Данные могут быть неконсистентны — eventual consistency + reconciliation

3.2 Eventual consistency

Что это

Eventual consistency: если прекратить обновления, все реплики в конечном итоге придут к одному состоянию.

Промежуточно: разные клиенты могут видеть разные версии данных.

Когда приемлема

Eventual consistency приемлема когда:

• Задержка в несколько секунд не критична
• Пользователи работают с разными данными
• Можно полагаться на reconciliation
• UI может показывать "обновляется..."

Не приемлема когда:

• Финансовые операции (нужна strong consistency)
• Блокировки ресурсов
• Аутентификация (свежие credentials)

Паттерны eventual consistency

Read-your-writes

Клиент после записи должен видеть свои изменения:

go
func (s *Service) CreateRepo(ctx context.Context, repo *Repo) error {
    // Записываем в primary
    if err := s.repo.Create(ctx, repo); err != nil {
        return err
    }

    // Устанавливаем session sticky: читать из primary 5 секунд
    ctx = context.WithValue(ctx, ReadFromPrimaryKey, time.Now().Add(5*time.Second))
    return nil
}

Monotonic reads

Клиент никогда не видит более старые данные чем уже видел:

go
type ClientSession struct {
    LastSeenVersion int64
}

func (s *Service) GetRepo(ctx context.Context, id string, session *ClientSession) (*Repo, error) {
    repo, err := s.repo.GetByID(ctx, id)
    if err != nil {
        return nil, err
    }
    // Если версия меньше чем клиент уже видел — ждём или читаем из primary
    if repo.Version < session.LastSeenVersion {
        return s.repo.GetByIDFromPrimary(ctx, id)
    }

    return repo, nil
}

Causal consistency

Если A causally предшествует B, все клиенты видят A перед B:

Пример: комментарий должен быть виден после поста

Решение: vector clocks или logical timestamps

Reconciliation

Когда данные расходятся, запускается reconciliation job:

go
func (j *ReconciliationJob) Run(ctx context.Context) error {
    // Находим расхождения
    diffs, err := j.findDiffs(ctx)
    if err != nil {
        return err
    }

    for _, diff := range diffs {
        // Применяем последнюю версию (LWW) или мерджим
        if err := j.reconcile(ctx, diff); err != nil {
            j.logger.Error("reconcile failed", "key", diff.Key, "error", err)
        }
    }

    return nil
}

Last-Writer-Wins (LWW) — простая стратегия: последняя запись побеждает. Для сложных случаев — CRDT или custom merge функции.

3.3 Retries и exponential backoff (углублённо)

Показать кодКопировать код

На схеме — порядок, в котором паттерны этого модуля оборачивают один вызов: сначала bulkhead (свой лимит на зависимость), затем circuit breaker (fail fast, когда зависимость лежит), внутри — retry с джиттером, и у каждой попытки свой timeout; любой отказ сходится в fallback. Порядок не косметический: если поставить retry снаружи CB, открытый breaker будет умножаться на ретраи, а bulkhead перестанет защищать ресурсы.

Зачем нужны retries

Временные сбои в распределённых системах:

• Сетевые таймауты
• Перезапуск сервиса
• Rate limiting (429)
• Временная недоступность (503)
• Database connection pool exhaustion

Retry позволяет дождаться восстановления и повторить операцию.

Когда делать retry

Делаем retry для:

• 5xx (кроме 501 Not Implemented, 505)
• Таймаутов
• Сетевых ошибок
• 429 Too Many Requests (после Retry-After)

НЕ делаем retry для:

• 400 Bad Request
• 401 Unauthorized
• 403 Forbidden
• 404 Not Found
• 409 Conflict
• 422 Unprocessable Entity

Exponential backoff с jitter

go
func ExponentialBackoff(attempt int, base, max time.Duration, jitter bool) time.Duration {
    // 2^attempt * base
    exp := math.Min(float64(attempt), 10)  // ограничиваем степень
    delay := time.Duration(math.Pow(2, exp)) * base

    if delay > max {
        delay = max
    }

    if jitter {
        // Full jitter: случайная задержка от 0 до delay
        delay = time.Duration(rand.Int63n(int64(delay)))
    }

    return delay
}

Пример с base=100ms, max=30s, jitter=true:

text
attempt 0: 0-100ms
attempt 1: 0-200ms
attempt 2: 0-400ms
attempt 3: 0-800ms
attempt 4: 0-1.6s
attempt 5: 0-3.2s
...
attempt 9+: 0-30s (cap)

Зачем jitter

Без jitter все клиенты retry-ят одновременно:

text
t=0: все 1000 клиентов получили ошибку
t=1s: все 1000 retry-ят одновременно -> overload
t=3s: все 1000 retry-ят одновременно -> overload

С jitter:

text
t=0: 1000 клиентов получили ошибку
t=0-1s: retry распределены случайно
t=0-2s: retry распределены случайно

Нет thundering herd.

Реализация retry

go
type RetryConfig struct {
    MaxAttempts int
    BaseDelay   time.Duration
    MaxDelay    time.Duration
    Jitter      bool
    Retryable   func(error) bool
}

func DefaultRetryConfig() RetryConfig {
    return RetryConfig{
        MaxAttempts: 5,
        BaseDelay:   100 * time.Millisecond,
        MaxDelay:    30 * time.Second,
        Jitter:      true,
        Retryable:   isRetryableError,
    }
}

func Retry(ctx context.Context, cfg RetryConfig, operation string, fn func(ctx context.Context) error) error {
    var lastErr error

    for attempt := 0; attempt < cfg.MaxAttempts; attempt++ {
        if err := ctx.Err(); err != nil {
            return fmt.Errorf("%s: cancelled: %w", operation, err)
        }

        lastErr = fn(ctx)
        if lastErr == nil {
            if attempt > 0 {
                metrics.RetriesTotal.WithLabelValues(operation, "success").Inc()
            }
            return nil
        }

        if !cfg.Retryable(lastErr) {
            metrics.RetriesTotal.WithLabelValues(operation, "permanent").Inc()
            return fmt.Errorf("%s: permanent error: %w", operation, lastErr)
        }

        if attempt < cfg.MaxAttempts-1 {
            delay := ExponentialBackoff(attempt, cfg.BaseDelay, cfg.MaxDelay, cfg.Jitter)

            metrics.RetriesTotal.WithLabelValues(operation, "attempt").Inc()

            select {
            case <-ctx.Done():
                return ctx.Err()
            case <-time.After(delay):
            }
        }
    }

    metrics.RetriesTotal.WithLabelValues(operation, "exhausted").Inc()
    return fmt.Errorf("%s: max retries (%d) exceeded: %w", operation, cfg.MaxAttempts, lastErr)
}

func isRetryableError(err error) bool {
    // HTTP статусы
    var httpErr *HTTPError
    if errors.As(err, &httpErr) {
        return httpErr.StatusCode == 429 || httpErr.StatusCode >= 500
    }

    // Сетевые ошибки
    var netErr net.Error
    if errors.As(err, &netErr) {
        return netErr.Timeout() || netErr.Temporary()
    }

    // Database ошибки
    var pgErr *pgconn.PgError
    if errors.As(err, &pgErr) {
        // Class 53: Insufficient Resources
        // Class 57: Operator Intervention
        // Class 08: Connection Exception
        return strings.HasPrefix(pgErr.Code, "53") ||
            strings.HasPrefix(pgErr.Code, "57") ||
            strings.HasPrefix(pgErr.Code, "08")
    }

    return false
}

Circuit Breaker поверх retry

Если сервис постоянно падает, retries только усугубляют проблему. Circuit Breaker прекращает запросы к упавшему сервису:

go
type State int

const (
    StateClosed State = iota    // пропускаем запросы
    StateOpen                    // блокируем запросы
    StateHalfOpen                // тестируем одним запросом
)

type CircuitBreaker struct {
    mu              sync.Mutex
    state           State
    failures        int
    successes       int
    lastStateChange time.Time
    config          CircuitBreakerConfig
}

func (cb *CircuitBreaker) Allow() bool {
    cb.mu.Lock()
    defer cb.mu.Unlock()

    switch cb.state {
    case StateClosed:
        return true
    case StateOpen:
        if time.Since(cb.lastStateChange) > cb.config.OpenTimeout {
            cb.state = StateHalfOpen
            cb.lastStateChange = time.Now()
            return true
        }
        return false
    case StateHalfOpen:
        return true
    }
    return false
}

func (cb *CircuitBreaker) RecordSuccess() {
    cb.mu.Lock()
    defer cb.mu.Unlock()

    if cb.state == StateHalfOpen {
        cb.successes++
        if cb.successes >= cb.config.RequiredSuccesses {
            cb.state = StateClosed
            cb.failures = 0
            cb.successes = 0
            cb.lastStateChange = time.Now()
        }
    } else {
        cb.failures = 0
    }
}

func (cb *CircuitBreaker) RecordFailure() {
    cb.mu.Lock()
    defer cb.mu.Unlock()

    cb.failures++
    if cb.failures >= cb.config.FailureThreshold {
        cb.state = StateOpen
        cb.lastStateChange = time.Now()
    }
}

Bulkhead

Изоляция ресурсов чтобы сбой одного компонента не уронил всю систему:

go
type Bulkhead struct {
    sem chan struct{}
}

func NewBulkhead(maxConcurrent int) *Bulkhead {
    return &Bulkhead{
        sem: make(chan struct{}, maxConcurrent),
    }
}

func (b *Bulkhead) Acquire(ctx context.Context) error {
    select {
    case b.sem <- struct{}{}:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    default:
        return ErrBulkheadFull
    }
}

func (b *Bulkhead) Release() {
    <-b.sem
}

Применение: разные bulkhead'ы для разных внешних сервисов.

go
paymentBulkhead := NewBulkhead(50)
notificationBulkhead := NewBulkhead(20)

Сбой notification не влияет на payment.

3.4 Idempotency (углублённо)

Что такое идемпотентность

Операция идемпотентна если многократное выполнение даёт тот же результат что и однократное.

Идемпотентные HTTP методы: GET, PUT, DELETE, HEAD, OPTIONS Не идемпотентные: POST, PATCH (обычно)

Зачем нужна идемпотентность

В распределённых системах:

• Клиент не получил ответ (таймаут)
• Сервис выполнил операцию, но упал перед ответом
• Retry от клиента приводит к дублированию

Идемпотентность защищает от этих проблем.

Idempotency keys

Клиент генерирует уникальный ключ для каждой операции:

POST /api/v1/payments
Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000
Content-Type: application/json

{
    "amount": 100,
    "currency": "USD",
    "recipient": "user-123"
}

Схема таблицы

sql
CREATE TABLE idempotency_keys (
    key VARCHAR(255) PRIMARY KEY,
    user_id UUID NOT NULL,
    request_hash VARCHAR(64) NOT NULL,
    status VARCHAR(20) NOT NULL,
    response_code INT,
    response_body JSONB,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    completed_at TIMESTAMPTZ,
    expires_at TIMESTAMPTZ NOT NULL
);

CREATE INDEX idx_idempotency_user ON idempotency_keys(user_id);
CREATE INDEX idx_idempotency_expires ON idempotency_keys(expires_at);

Статусы ключа

Реализация

go
type IdempotencyService struct {
    repo repository.IdempotencyRepository
}

func (s *IdempotencyService) Execute(
    ctx context.Context,
    key string,
    userID string,
    requestHash string,
    fn func(ctx context.Context) (*Response, error),
) (*Response, error) {
    // Проверяем существующий ключ
    existing, err := s.repo.Get(ctx, key)
    if err != nil {
        return nil, fmt.Errorf("get idempotency key: %w", err)
    }

    if existing != nil {
        // Проверяем что request hash совпадает
        if existing.RequestHash != requestHash {
            return nil, ErrIdempotencyKeyConflict
        }

        switch existing.Status {
        case StatusCompleted:
            return existing.Response, nil  // возвращаем кешированный ответ
        case StatusInProgress:
            // Другой запрос выполняется с тем же ключом
            return nil, ErrIdempotencyKeyInProgress
        case StatusFailed:
            // Можно retry
        }
    }

    // Создаём in_progress запись
    record := &IdempotencyRecord{
        Key:         key,
        UserID:      userID,
        RequestHash: requestHash,
        Status:      StatusInProgress,
        ExpiresAt:   time.Now().Add(24 * time.Hour),
    }

    if err := s.repo.Create(ctx, record); err != nil {
        // Конфликт — другой запрос уже начался
        if errors.Is(err, ErrDuplicateKey) {
            return nil, ErrIdempotencyKeyInProgress
        }
        return nil, fmt.Errorf("create idempotency record: %w", err)
    }

    // Выполняем операцию
    response, err := fn(ctx)
    if err != nil {
        // Помечаем как failed
        s.repo.UpdateStatus(ctx, key, StatusFailed, nil, 0)
        return nil, err
    }

    // Сохраняем результат
    if err := s.repo.UpdateStatus(ctx, key, StatusCompleted, response, 200); err != nil {
        return nil, fmt.Errorf("update idempotency: %w", err)
    }

    return response, nil
}

Middleware для идемпотентности

go
func Idempotency(idempotencySvc *IdempotencyService) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            // Только для state-changing методов
            if r.Method == "GET" || r.Method == "HEAD" || r.Method == "OPTIONS" {
                next.ServeHTTP(w, r)
                return
            }

            key := r.Header.Get("Idempotency-Key")
            if key == "" {
                next.ServeHTTP(w, r)
                return
            }

            userID := GetUserID(r.Context())
            requestHash := hashRequest(r)

            // Выполняем через idempotency service
            response, err := idempotencySvc.Execute(r.Context(), key, userID, requestHash,
                func(ctx context.Context) (*Response, error) {
                    // Оборачиваем handler
                    recorder := httptest.NewRecorder()
                    next.ServeHTTP(recorder, r.WithContext(ctx))

                    return &Response{
                        StatusCode: recorder.Code,
                        Body:       recorder.Body.Bytes(),
                        Headers:    recorder.Header(),
                    }, nil
                })

            if err != nil {
                if errors.Is(err, ErrIdempotencyKeyConflict) {
                    writeProblem(w, 409, "Conflict", "idempotency key conflict")
                    return
                }
                // Ошибка уже записана в recorder
                return
            }

            // Записываем кешированный ответ
            for k, v := range response.Headers {
                w.Header()[k] = v
            }
            w.WriteHeader(response.StatusCode)
            w.Write(response.Body)
        })
    }
}

Конфликты ключей

Если клиент использует один ключ с разными запросами — это ошибка клиента:

POST /payments с ключом X и amount=100
POST /payments с ключом X и amount=200

Ответ: 409 Conflict — клиент должен использовать новый ключ.

TTL ключей

Idempotency keys не должны жить вечно. Обычно 24 часа.

Cleanup job:

go
func (j *CleanupIdempotencyKeysJob) Run(ctx context.Context) error {
    deleted, err := j.repo.DeleteExpired(ctx, time.Now())
    if err != nil {
        return err
    }
    j.logger.Info("cleanup completed", "deleted", deleted)
    return nil
}

3.5 Outbox pattern (углублённо)

Проблема dual write

Нужно атомарно:

1. Изменить состояние в БД
2. Опубликовать событие

Если сделать это отдельно:

go
// ПЛОХО: не атомарно
db.Exec("INSERT INTO orders ...")
kafka.Publish("order.created", event)  // может упасть

Сценарии сбоя:

• БД записала, событие не опубликовано — потребители не узнают
• Событие опубликовано, БД упала — потребители получили ghost event
• Всё ОК, но ack от Kafka потерян — retry опубликует дубликат

Решение: Outbox pattern

go
// АТОМАРНО: в одной транзакции
tx.Begin()
db.Exec("INSERT INTO orders ...")
db.Exec("INSERT INTO outbox_events ...")
tx.Commit()

Outbox таблица:

sql
CREATE TABLE outbox_events (
    id UUID PRIMARY KEY,
    aggregate_id UUID NOT NULL,
    aggregate_type VARCHAR(100) NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    metadata JSONB,
    status VARCHAR(20) NOT NULL DEFAULT 'pending',
    retry_count INT NOT NULL DEFAULT 0,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    published_at TIMESTAMPTZ,
    last_error TEXT
);

CREATE INDEX idx_outbox_pending ON outbox_events(status, created_at)
    WHERE status = 'pending';

Outbox relay worker

go
type OutboxRelayJob struct {
    outboxRepo repository.OutboxRepository
    publisher  repository.EventPublisher
    logger     *slog.Logger
    batchSize  int
}

func (j *OutboxRelayJob) Run(ctx context.Context) error {
    events, err := j.outboxRepo.GetPending(ctx, j.batchSize)
    if err != nil {
        return fmt.Errorf("get pending events: %w", err)
    }

    if len(events) == 0 {
        return nil
    }

    for _, event := range events {
        if err := j.publishWithRetry(ctx, event); err != nil {
            j.logger.Error("publish failed",
                "event_id", event.ID,
                "retry_count", event.RetryCount,
                "error", err,
            )
            continue
        }
    }

    metrics.OutboxEventsPending.Set(float64(countPending(ctx)))
    return nil
}

func (j *OutboxRelayJob) publishWithRetry(ctx context.Context, event *OutboxEvent) error {
    err := Retry(ctx, DefaultRetryConfig(), "publish_event",
        func(ctx context.Context) error {
            return j.publisher.Publish(ctx, event)
        })

    if err != nil {
        // Увеличиваем retry count
        newCount := event.RetryCount + 1
        if newCount >= MaxRetries {
            // В DLQ
            return j.outboxRepo.MarkDead(ctx, event.ID, err)
        }
        return j.outboxRepo.UpdateRetry(ctx, event.ID, newCount, err)
    }

    return j.outboxRepo.MarkPublished(ctx, event.ID)
}

Гарантии доставки

Outbox pattern обеспечивает:

• Durability: событие в БД, не потеряется при сбое приложения
• At-least-once: событие будет доставлено хотя бы раз
• Ordering: события публикуются в порядке создания (при правильной реализации)

НЕ обеспечивает:

• Exactly-once: потребитель должен быть идемпотентен
• Real-time: задержка между созданием и публикацией

Transactional Outbox с CDC

Альтернатива polling'у — CDC (Change Data Capture) через Debezium:

Outbox Table -> Debezium -> Kafka -> Consumers

Плюсы:

• Меньше нагрузка на БД
• Меньше latency
• Гарантии ordering per partition

Минусы:

• Сложнее инфраструктура
• Нужен Debezium + Kafka

Publishing Before DB (антипаттерн)

go
// ПЛОХО: publish before commit
kafka.Publish("order.created", event)
db.Exec("INSERT INTO orders ...")  // может упасть!

Результат: ghost events. Потребители получили событие, но данных нет.

Publishing After DB (тоже проблема)

go
// ПЛОХО: publish after commit
db.Exec("INSERT INTO orders ...")
db.Commit()
kafka.Publish("order.created", event)  // может упасть!

Результат: lost events. Данные есть, события нет.

Только Outbox решает обе проблемы.

3.6 Saga orchestration

Проблема распределённых транзакций

В микросервисах каждый сервис имеет свою БД. Локальные транзакции не работают кросс-сервисно.

Пример: создание заказа

Order Service: INSERT order
Inventory Service: UPDATE stock
Payment Service: INSERT payment

Нужна атомарность: либо все операции успешны, либо ни одной.

2PC (Two-Phase Commit) — почему не используем

2PC — классический протокол распределённых транзакций:

Phase 1 (Prepare): координатор спрашивает всех "готовы?"
Phase 2 (Commit): если все готовы — коммитим, иначе rollback

Проблемы 2PC:

• Блокировка ресурсов на всё время протокола
• Single point of failure (координатор)
• Низкая производительность
• Плохая доступность (все узлы должны быть online)
• Не работает с микросервисами

В реальных микросервисах 2PC не используется.

Saga pattern

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

T1 -> T2 -> T3 -> T4 (успех)
T1 -> T2 -> T3 (сбой) -> C2 -> C1 (компенсации)

Где:

• T — транзакция (forward)
• C — компенсация (backward)

Два варианта Saga

Saga Choreography

Сервисы общаются через события, без центрального координатора:

Order Created (Order Service)
    -> Inventory Reserved (Inventory Service)
    -> Payment Processed (Payment Service)
    -> Order Confirmed (Order Service)

При сбое:

Order Created
    -> Inventory Reserved
    -> Payment Failed
    -> Inventory Released (компенсация)
    -> Order Cancelled (компенсация)

Плюсы:

• Простая реализация для малых процессов
• Нет единой точки отказа
• Слабая связанность

Минусы:

• Сложно отследить состояние
• Циклические зависимости
• Сложно для больших процессов

Saga Orchestration

Центральный оркестратор управляет процессом:

Orchestrator:
    1. Создать заказ (Order Service)
    2. Зарезервировать товар (Inventory Service)
    3. Списать деньги (Payment Service)
    4. Подтвердить заказ (Order Service)

При сбое на шаге 3: Orchestrator:

1. Создать заказ ✓
2. Зарезервировать товар ✓
3. Списать деньги ✗
4. Освободить товар (компенсация 2)
5. Отменить заказ (компенсация 1)

Плюсы:

• Централизованный контроль
• Видимость состояния
• Легче отлаживать
• Подходит для сложных процессов

Минусы:

• Единая точка отказа (оркестратор)
• Более тесная связанность

Реализация Saga Orchestrator

State machine

Каждая saga — state machine с шагами:

go
type SagaState string

const (
    StatePending     SagaState = "pending"
    StateRunning     SagaState = "running"
    StateCompleted   SagaState = "completed"
    StateCompensating SagaState = "compensating"
    StateFailed      SagaState = "failed"
)

type SagaStep struct {
    Name        string
    Execute     func(ctx context.Context, data *SagaData) error
    Compensate  func(ctx context.Context, data *SagaData) error
}

type Saga struct {
    ID    string
    Type  string
    State SagaState
    Steps []SagaStep
    Data  *SagaData
    CurrentStep int
    LastError error
}

Схема таблицы

sql
CREATE TABLE saga_instances (
    id UUID PRIMARY KEY,
    type VARCHAR(100) NOT NULL,
    state VARCHAR(20) NOT NULL,
    data JSONB NOT NULL,
    current_step INT NOT NULL DEFAULT 0,
    last_error TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    completed_at TIMESTAMPTZ
);

CREATE TABLE saga_step_logs (
    id UUID PRIMARY KEY,
    saga_id UUID NOT NULL REFERENCES saga_instances(id),
    step_name VARCHAR(100) NOT NULL,
    status VARCHAR(20) NOT NULL,
    error TEXT,
    executed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_saga_state ON saga_instances(state) WHERE state IN ('running', 'compensating');

Определение Saga

go
type CreateOrderSaga struct {
    orderService     *OrderService
    inventoryService *InventoryService
    paymentService   *PaymentService
}

func (s *CreateOrderSaga) Steps() []SagaStep {
    return []SagaStep{
        {
            Name: "create_order",
            Execute: func(ctx context.Context, data *SagaData) error {
                order, err := s.orderService.Create(ctx, data.OrderRequest)
                if err != nil {
                    return err
                }
                data.OrderID = order.ID
                return nil
            },
            Compensate: func(ctx context.Context, data *SagaData) error {
                return s.orderService.Cancel(ctx, data.OrderID)
            },
        },
        {
            Name: "reserve_inventory",
            Execute: func(ctx context.Context, data *SagaData) error {
                reservation, err := s.inventoryService.Reserve(ctx, data.OrderRequest.Items)
                if err != nil {
                    return err
                }
                data.ReservationID = reservation.ID
                return nil
            },
            Compensate: func(ctx context.Context, data *SagaData) error {
                return s.inventoryService.Release(ctx, data.ReservationID)
            },
        },
        {
            Name: "process_payment",
            Execute: func(ctx context.Context, data *SagaData) error {
                payment, err := s.paymentService.Charge(ctx, data.OrderRequest.Amount)
                if err != nil {
                    return err
                }
                data.PaymentID = payment.ID
                return nil
            },
            Compensate: func(ctx context.Context, data *SagaData) error {
                return s.paymentService.Refund(ctx, data.PaymentID)
            },
        },
        {
            Name: "confirm_order",
            Execute: func(ctx context.Context, data *SagaData) error {
                return s.orderService.Confirm(ctx, data.OrderID)
            },
            Compensate: func(ctx context.Context, data *SagaData) error {
                return s.orderService.Cancel(ctx, data.OrderID)
            },
        },
    }
}

Выполнение Saga

go
type Orchestrator struct {
    sagaRepo repository.SagaRepository
    logger   *slog.Logger
}

func (o *Orchestrator) Execute(ctx context.Context, saga *Saga) error {
    // Сохраняем начальное состояние
    saga.State = StateRunning
    if err := o.sagaRepo.Save(ctx, saga); err != nil {
        return fmt.Errorf("save saga: %w", err)
    }

    // Выполняем forward шаги
    for i := saga.CurrentStep; i < len(saga.Steps); i++ {
        step := saga.Steps[i]

        o.logger.Info("executing step",
            "saga_id", saga.ID,
            "step", step.Name,
        )

        if err := step.Execute(ctx, saga.Data); err != nil {
            saga.LastError = err
            saga.CurrentStep = i

            // Переходим к компенсациям
            return o.compensate(ctx, saga, i-1)
        }

        saga.CurrentStep = i + 1
        if err := o.sagaRepo.Save(ctx, saga); err != nil {
            return fmt.Errorf("save saga progress: %w", err)
        }
    }

    saga.State = StateCompleted
    return o.sagaRepo.Save(ctx, saga)
}

func (o *Orchestrator) compensate(ctx context.Context, saga *Saga, fromStep int) error {
    saga.State = StateCompensating
    o.sagaRepo.Save(ctx, saga)

    // Компенсируем в обратном порядке
    for i := fromStep; i >= 0; i-- {
        step := saga.Steps[i]

        o.logger.Info("compensating step",
            "saga_id", saga.ID,
            "step", step.Name,
        )

        if err := step.Compensate(ctx, saga.Data); err != nil {
            // Компенсация упала — нужна ручная интервенция
            o.logger.Error("compensation failed, manual intervention required",
                "saga_id", saga.ID,
                "step", step.Name,
                "error", err,
            )
            saga.State = StateFailed
            o.sagaRepo.Save(ctx, saga)
            return fmt.Errorf("compensation failed at step %s: %w", step.Name, err)
        }
    }

    saga.State = StateFailed
    return o.sagaRepo.Save(ctx, saga)
}

3.7 Saga Orchestrator Job

Восстановление после сбоя

Orchestrator может упасть посередине saga. Нужен recovery job:

go
type SagaRecoveryJob struct {
    sagaRepo     repository.SagaRepository
    orchestrator *Orchestrator
    logger       *slog.Logger
}

func (j *SagaRecoveryJob) Run(ctx context.Context) error {
    // Находим незавершённые saga
    sagas, err := j.sagaRepo.FindStuck(ctx, StuckSagaThreshold)
    if err != nil {
        return err
    }

    for _, saga := range sagas {
        if err := j.orchestrator.Resume(ctx, saga); err != nil {
            j.logger.Error("saga recovery failed",
                "saga_id", saga.ID,
                "error", err,
            )
        }
    }

    return nil
}

Resume saga

go
func (o *Orchestrator) Resume(ctx context.Context, saga *Saga) error {
    switch saga.State {
    case StateRunning:
        // Продолжаем с текущего шага
        return o.Execute(ctx, saga)
    case StateCompensating:
        // Продолжаем компенсации
        return o.compensate(ctx, saga, saga.CurrentStep-1)
    default:
        return nil
    }
}

Идемпотентность шагов

Каждый шаг и компенсация должны быть идемпотентны:

go
// Компенсация может быть вызвана несколько раз
func (s *InventoryService) Release(ctx context.Context, reservationID string) error {
    // Проверяем что reservation ещё не освобождён
    reservation, err := s.repo.GetByID(ctx, reservationID)
    if err != nil {
        return err
    }
    if reservation == nil || reservation.Status == StatusReleased {
        return nil  // уже освобождено — идемпотентно
    }

    return s.repo.Release(ctx, reservationID)
}

Timeout для шагов

Каждый шаг должен иметь таймаут:

go
func (o *Orchestrator) executeWithTimeout(ctx context.Context, step SagaStep, data *SagaData) error {
    stepCtx, cancel := context.WithTimeout(ctx, step.Timeout)
    defer cancel()

    done := make(chan error, 1)
    go func() {
        done <- step.Execute(stepCtx, data)
    }()

    select {
    case err := <-done:
        return err
    case <-stepCtx.Done():
        return fmt.Errorf("step %s timeout", step.Name)
    }
}

3.8 Compensation transactions

Требования к компенсациям

Компенсации должны быть:

1. Идемпотентными: можно вызвать несколько раз
2. Коммутативными: порядок не важен (желательно)
3. Надёжными: не должны падать без причины
4. Быстрыми: не блокировать другие операции

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

Некомпенсируемые действия

Некоторые действия нельзя компенсировать:

• Отправленный email (уже в inbox пользователя)
• Физическая отправка товара
• SMS уведомление

Для них:

• Откладывать до финального шага saga
• Использовать "soft commit" (pending state)
• Предупреждать пользователя что действие необратимо

Semantic compensation

Иногда точная компенсация невозможна, используем семантическую:

Forward: зарядили батарею до 100%
Compensation: не можем "разрядить обратно", но можем отметить что действие отменено

Compensating vs Rollback

Rollback: отмена незафиксированной транзакции Compensation: новое действие которое отменяет эффект уже зафиксированного

Это разные концепции. Compensation создаёт новую запись, не удаляет старую. История сохраняется.

3.9 DLQ и обработка failed events

Что такое Dead Letter Queue

DLQ — место для сообщений которые не удалось обработать после всех попыток retry.

Когда сообщение попадает в DLQ

1. Превышен лимит retry
2. Невалидные данные (permanent error)
3. Исчерпан budget ошибок
4. Ручная отправка администратором

Схема DLQ таблицы

sql
CREATE TABLE dead_letter_queue (
    id UUID PRIMARY KEY,
    source_queue VARCHAR(100) NOT NULL,
    message_id VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    error_message TEXT,
    error_type VARCHAR(100),
    retry_count INT NOT NULL,
    first_failure_at TIMESTAMPTZ NOT NULL,
    last_failure_at TIMESTAMPTZ NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'pending',
    resolved_at TIMESTAMPTZ,
    resolved_by VARCHAR(255)
);

CREATE INDEX idx_dlq_status ON dead_letter_queue(status);

Перемещение в DLQ

go
func (j *OutboxRelayJob) processEvent(ctx context.Context, event *OutboxEvent) error {
    err := j.publisher.Publish(ctx, event)
    if err == nil {
        return j.outboxRepo.MarkPublished(ctx, event.ID)
    }

    newCount := event.RetryCount + 1

    if newCount >= MaxRetries || isPermanentError(err) {
        // В DLQ
        dlqEntry := &DLQEntry{
            SourceQueue:    "outbox",
            MessageID:      event.ID,
            Payload:        event.Payload,
            ErrorMessage:   err.Error(),
            ErrorType:      errorType(err),
            RetryCount:     newCount,
            FirstFailureAt: event.CreatedAt,
            LastFailureAt:  time.Now(),
        }

        if err := j.dlqRepo.Create(ctx, dlqEntry); err != nil {
            return fmt.Errorf("create dlq entry: %w", err)
        }

        return j.outboxRepo.MarkDead(ctx, event.ID, err)
    }

    return j.outboxRepo.UpdateRetry(ctx, event.ID, newCount, err)
}

Обработка DLQ

Автоматический retry (редко)

go
func (j *DLQRetryJob) Run(ctx context.Context) error {
    entries, err := j.dlqRepo.GetRetryable(ctx, time.Now().Add(-24*time.Hour))
    if err != nil {
        return err
    }

    for _, entry := range entries {
        if err := j.processor.Process(ctx, entry); err != nil {
            j.logger.Error("dlq retry failed", "entry_id", entry.ID, "error", err)
            continue
        }
        j.dlqRepo.MarkResolved(ctx, entry.ID, "auto-retry")
    }

    return nil
}

Ручная обработка (обычно)

API для администратора:

GET    /admin/dlq                    # список DLQ entries
GET    /admin/dlq/{id}               # детали entry
POST   /admin/dlq/{id}/retry         # повторная обработка
POST   /admin/dlq/{id}/discard       # отбросить
POST   /admin/dlq/bulk-retry         # массовый retry
DELETE /admin/dlq/{id}               # удалить

Мониторинг DLQ

go
var dlqSize = promauto.NewGaugeVec(
    prometheus.GaugeOpts{
        Name: "dlq_size",
        Help: "Number of messages in DLQ",
    },
    []string{"queue"},
)

var dlqAdditions = promauto.NewCounterVec(
    prometheus.CounterOpts{
        Name: "dlq_additions_total",
        Help: "Total messages added to DLQ",
    },
    []string{"queue", "error_type"},
)

Алерт:

yaml
- alert: DLQGrowing
  expr: increase(dlq_additions_total[1h]) > 100
  for: 10m
  labels:
    severity: critical

Root cause analysis

При росте DLQ:

1. Посмотреть на error_type распределение
2. Проверить последние деплои
3. Проверить внешние сервисы
4. Воспроизвести ошибку локально
5. Исправить и запустить bulk retry

Привязка к шаблону проекта

Где distributed patterns в шаблоне

internal/models/
├── outbox_event.go      # Модель outbox event
└── saga_instance.go     # Модель saga instance

internal/usecase/
├── relay_outbox_usecase.go      # Outbox relay use case
├── advance_saga_usecase.go      # Saga orchestrator use case
└── rotate_hmac_usecase.go

internal/service/
├── outbox_service.go
└── saga_service.go

internal/handlers/jobs/
├── outbox_relay/handler.go      # Outbox relay job
├── saga_orchestrator/handler.go # Saga recovery job
└── hmac_rotator/handler.go

internal/adapters/ymq/
└── publisher.go                 # Yandex MQ publisher

Что изучаем на примере шаблона

1. Outbox pattern с relay job
2. Saga orchestrator для распределённых транзакций
3. Компенсационные транзакции
4. Retry с exponential backoff
5. Cron jobs для фоновой обработки

Задание: изучите internal/usecase/advance_saga_usecase.go

• Какие шаги у saga?
• Как обрабатываются сбои?
• Есть ли компенсации?
• Как сохраняется состояние?

💱 Компромиссы

• CP vs AP: при partition etcd и ZooKeeper жертвуют доступностью ради консистентности; Cassandra и DynamoDB отвечают всегда, но реплики расходятся и нужен reconciliation. Разделение сети обязательно случится — между C и A выбирают заранее.
• Saga Orchestration vs Choreography: оркестратор централизует контроль и видит состояние в saga_instances, но сам становится единой точкой отказа; choreography без координатора слабее связана, но ход процесса уже не отследить.
• At-least-once vs exactly-once: outbox гарантирует только at-least-once, и потребитель обязан быть идемпотентным по message_id; честный exactly-once дорог и на практике редко оправдан.
• Outbox polling vs CDC: relay job проще по инфраструктуре, но опрашивает таблицу и добавляет задержку; Debezium с Kafka снижает latency и даёт ordering per partition, но тащит в стек ещё два компонента.
• Retries vs Circuit Breaker: retry с exponential backoff вытягивает временные сбои, но добивает упавший сервис; breaker прекращает штурм, но может ложно открыться на короткой деградации и ронять честные запросы.

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

Задание 1: Retry с jitter

Реализуйте универсальную Retry функцию:

go
func Retry(ctx context.Context, cfg RetryConfig, operation string, fn func(ctx context.Context) error) error

С:

• Exponential backoff
• Full jitter
• Конфигурируемый retryable predicate
• Метрики попыток

Задание 2: Circuit Breaker

Реализуйте Circuit Breaker:

go
type CircuitBreaker struct {
    // состояние, счётчики
}

func (cb *CircuitBreaker) Execute(ctx context.Context, fn func() error) error

С состояниями: Closed, Open, Half-Open.

Задание 3: Idempotency middleware

Реализуйте middleware для POST-запросов:

go
func Idempotency(service *IdempotencyService) func(http.Handler) http.Handler

Middleware:

• Извлекает Idempotency-Key из заголовка
• Проверяет существующий ключ
• Кеширует ответ при успехе
• Возвращает 409 при конфликте

Задание 4: Outbox pattern

Реализуйте атомарное создание заказа с событием:

go
func (uc *CreateOrderUseCase) Execute(ctx context.Context, req CreateOrderRequest) (*Order, error) {
    // В одной транзакции:
    // 1. INSERT order
    // 2. INSERT outbox_event
}

Задание 5: Saga orchestration

Реализуйте CreateOrderSaga с шагами:

1. create_order (compensation: cancel_order)
2. reserve_inventory (compensation: release_inventory)
3. process_payment (compensation: refund_payment)
4. confirm_order (compensation: cancel_order)

Протестируйте:

• Успешный сценарий
• Сбой на каждом шаге с правильными компенсациями
• Восстановление после падения оркестратора

Задание 6: Saga recovery job

Реализуйте recovery job:

go
func (j *SagaRecoveryJob) Run(ctx context.Context) error {
    // Находит stuck sagas
    // Восстанавливает их выполнение
}

Задание 7: DLQ

Реализуйте DLQ repository:

go
func (r *DLQRepository) Create(ctx context.Context, entry *DLQEntry) error
func (r *DLQRepository) GetRetryable(ctx context.Context, olderThan time.Time) ([]*DLQEntry, error)
func (r *DLQRepository) MarkResolved(ctx context.Context, id string, resolvedBy string) error

Добавьте admin API для просмотра и retry.

Задание 8: Метрики и алерты

Добавьте метрики:

text
saga_steps_total{status="success|failure"}
saga_duration_seconds{saga_type="..."}
compensations_total{step="..."}
outbox_lag_seconds
dlq_size{queue="..."}

Настройте алерты:

• Saga duration > 5min
• DLQ growing > 100/hour
• Compensation failure rate > 1%

Quiz "Проверь себя"

BEGIN;
INSERT INTO orders ...;
INSERT INTO outbox_events ...;
COMMIT;

POST /api/v1/payments
Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000

✅ Итоги модуля

Студент должен:

• Понимать CAP теорему и ограничения распределённых систем
• Проектировать eventual consistency
• Реализовывать retry с exponential backoff и jitter
• Обеспечивать идемпотентность через ключи
• Применять outbox pattern
• Проектировать Saga orchestration
• Реализовывать компенсационные транзакции
• Настраивать DLQ и обработку failed events
• Понимать Circuit Breaker и Bulkhead
• Ответить на все вопросы Quiz

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

Следующий модуль: Модуль 4. Platform and Deployment Thinking - Serverless vs VDS vs Kubernetes, single binary strategy, cron vs timer triggers, graceful shutdown, rollout and rollback, canary, blue-green, feature flags, GitOps. Привязка к deploy/ шаблона.