Модуль 8. Background Processing

Цели модуля

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

• Понимать когда нужна фоновая обработка и какие задачи выносить в фон
• Реализовывать воркеры с планировщиком robfig/cron
• Применять outbox pattern для надёжной доставки событий
• Реализовывать retries с exponential backoff и jitter
• Настраивать Dead Letter Queue для необработанных сообщений
• Использовать idempotency keys для защиты от дублирования
• Выбирать между очередями сообщений (RMQ, Kafka, YMQ)

8.1 Зачем нужна фоновая обработка

Что выносить в фон

Не все операции должны выполняться синхронно в HTTP-запросе.

В фон выносим:

• Долгие операции (отправка email, генерация отчётов)
• Операции с внешними сервисами (вебхуки, уведомления)
• Периодические задачи (очистка данных, ротация ключей)
• Обработка очередей событий
• Тяжёлые вычисления

Синхронно vs Асинхронно

Паттерны фоновой обработки

1. In-process workers: горутины внутри приложения
2. Cron jobs: задачи по расписанию
3. Queue consumers: обработка из очереди сообщений
4. Outbox relay: доставка событий из outbox таблицы

8.2 Worker concept

Что такое воркер

Воркер — процесс или горутина которая выполняет фоновые задачи.

Характеристики воркера:

• Работает непрерывно (long-running)
• Обрабатывает задачи из источника (очередь, таблица, расписание)
• Обрабатывает ошибки (retry, DLQ)
• Поддерживает graceful shutdown

Пример простого воркера

go
type Worker struct {
    jobsCh  <-chan Job
    done    chan struct{}
    logger  *slog.Logger
}

func NewWorker(jobsCh <-chan Job, logger *slog.Logger) *Worker {
    return &Worker{
        jobsCh: jobsCh,
        done:   make(chan struct{}),
        logger: logger,
    }
}

func (w *Worker) Run(ctx context.Context) {
    w.logger.Info("worker started")
    defer w.logger.Info("worker stopped")

    for {
        select {
        case <-ctx.Done():
            w.logger.Info("worker cancelled", "reason", ctx.Err())
            return
        case job, ok := <-w.jobsCh:
            if !ok {
                return  // канал закрыт
            }
            w.process(ctx, job)
        }
    }
}

func (w *Worker) process(ctx context.Context, job Job) {
    w.logger.Info("processing job", "job_id", job.ID)

    if err := job.Execute(ctx); err != nil {
        w.logger.Error("job failed", "job_id", job.ID, "error", err)
        // обработка ошибки: retry, DLQ
        return
    }

    w.logger.Info("job completed", "job_id", job.ID)
}

Graceful shutdown воркера

Воркер должен корректно завершаться при получении сигнала:

go
func (w *Worker) Shutdown(ctx context.Context) error {
    w.logger.Info("worker shutting down")

    // Ждём завершения текущей задачи
    select {
    case <-w.done:
        w.logger.Info("worker stopped gracefully")
        return nil
    case <-ctx.Done():
        w.logger.Warn("worker shutdown timeout")
        return ctx.Err()
    }
}

8.3 Jobs and schedulers (robfig/cron)

Что такое cron

Cron — планировщик задач по расписанию. Задачи выполняются периодически: каждую минуту, каждый час, каждый день.

robfig/cron в Go

Установка:

bash
go get github.com/robfig/cron/v3

Cron expressions

Формат: секунды минуты часы день месяц день_недели

text
* * * * * *  // каждую секунду
0 * * * * *  // каждую минуту
0 0 * * * *  // каждый час
0 0 0 * * *  // каждый день в полночь
0 0 0 * * 0  // каждое воскресенье в полночь
0 */5 * * * * // каждые 5 минут

Реализация планировщика

go
package scheduler

import (
    "context"
    "github.com/robfig/cron/v3"
)

type Job interface {
    Name() string
    Run(ctx context.Context) error
    Schedule() string  // cron expression
}

type Scheduler struct {
    cron   *cron.Cron
    jobs   []Job
    logger *slog.Logger
}

func NewScheduler(logger *slog.Logger) *Scheduler {
    return &Scheduler{
        cron:   cron.New(cron.WithSeconds()),
        logger: logger,
    }
}

func (s *Scheduler) Register(job Job) {
    s.jobs = append(s.jobs, job)
}

func (s *Scheduler) Start(ctx context.Context) {
    for _, job := range s.jobs {
        job := job  // capture
        _, err := s.cron.AddFunc(job.Schedule(), func() {
            s.logger.Info("job started", "job", job.Name())

            if err := job.Run(ctx); err != nil {
                s.logger.Error("job failed", "job", job.Name(), "error", err)
                return
            }

            s.logger.Info("job completed", "job", job.Name())
        })

        if err != nil {
            s.logger.Error("failed to register job", "job", job.Name(), "error", err)
        }
    }

    s.cron.Start()
    s.logger.Info("scheduler started", "jobs", len(s.jobs))
}

func (s *Scheduler) Stop() {
    ctx := s.cron.Stop()
    <-ctx.Done()
    s.logger.Info("scheduler stopped")
}

Пример job: Outbox Relay

go
package jobs

type OutboxRelayJob struct {
    outboxService *service.OutboxService
    publisher     repository.EventPublisher
    logger        *slog.Logger
}

func NewOutboxRelayJob(outboxSvc *service.OutboxService, publisher repository.EventPublisher, logger *slog.Logger) *OutboxRelayJob {
    return &OutboxRelayJob{
        outboxService: outboxSvc,
        publisher:     publisher,
        logger:        logger,
    }
}

func (j *OutboxRelayJob) Name() string {
    return "outbox_relay"
}

func (j *OutboxRelayJob) Schedule() string {
    return "*/5 * * * * *"  // каждые 5 секунд
}

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

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

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

        if err := j.outboxService.MarkSent(ctx, event.ID); err != nil {
            return fmt.Errorf("mark sent: %w", err)
        }
    }

    return nil
}

Single-flight защита

Защита от одновременного запуска job:

go
type SingleFlightJob struct {
    job     Job
    running atomic.Bool
}

func (j *SingleFlightJob) Run(ctx context.Context) error {
    if j.running.Load() {
        return nil  // уже выполняется
    }

    j.running.Store(true)
    defer j.running.Store(false)

    return j.job.Run(ctx)
}

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

Проблема

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

1. Записать данные в БД
2. Отправить событие в очередь

Если отправить событие до записи — возможна отправка без записи. Если после — возможна запись без события (сбой между операциями).

Решение: Outbox Pattern

Событие записывается в ту же транзакцию что и данные. Отдельный воркер (outbox relay) читает outbox и доставляет события.

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

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

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

Статусы событий

Outbox Relay воркер

go
func (j *OutboxRelayJob) Run(ctx context.Context) error {
    // Получаем pending события
    events, err := j.outboxService.GetPending(ctx, 100)
    if err != nil {
        return fmt.Errorf("get pending: %w", err)
    }

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

    return nil
}

func (j *OutboxRelayJob) processEvent(ctx context.Context, event *models.OutboxEvent) error {
    // Отправка события
    if err := j.publisher.Publish(ctx, event); err != nil {
        // Увеличиваем retry count
        if err := j.outboxService.IncrementRetry(ctx, event.ID); err != nil {
            return fmt.Errorf("increment retry: %w", err)
        }

        // Проверяем лимит попыток
        if event.RetryCount >= MaxRetries {
            return j.outboxService.MarkDead(ctx, event.ID, err)
        }

        return fmt.Errorf("publish: %w", err)
    }

    // Помечаем как отправленное
    return j.outboxService.MarkSent(ctx, event.ID)
}

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

Outbox pattern обеспечивает at-least-once доставку:

• Событие не потеряется (записано в БД)
• Событие может быть доставлено повторно (при retry)

Для exactly-once нужна идемпотентность на стороне потребителя.

8.5 Retries и exponential backoff

Зачем нужны retries

Внешние сервисы могут быть временно недоступны:

• Сетевые сбои
• Таймауты
• Rate limiting
• Временные ошибки (503)

Retry позволяет повторить операцию после временного сбоя.

Exponential backoff

Задержка растёт экспоненциально: 1s, 2s, 4s, 8s, 16s...

go
func ExponentialBackoff(attempt int, baseDelay, maxDelay time.Duration) time.Duration {
    delay := baseDelay * time.Duration(1<<uint(attempt))  // 2^attempt
    if delay > maxDelay {
        delay = maxDelay
    }
    return delay
}

// Пример: baseDelay=1s, maxDelay=60s
// attempt 0: 1s
// attempt 1: 2s
// attempt 2: 4s
// attempt 3: 8s
// attempt 4: 16s
// attempt 5: 32s
// attempt 6: 60s (cap)

Jitter

Без jitter все клиенты повторяют запрос одновременно (thundering herd). Jitter добавляет случайность:

go
func WithJitter(delay time.Duration) time.Duration {
    jitter := time.Duration(rand.Int63n(int64(delay / 2)))
    return delay + jitter
}

Реализация retry

go
type RetryConfig struct {
    MaxAttempts int
    BaseDelay   time.Duration
    MaxDelay    time.Duration
}

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

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

        lastErr = fn()
        if lastErr == nil {
            return nil
        }

        // Не повторяем для постоянных ошибок
        if isPermanentError(lastErr) {
            return lastErr
        }

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

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

    return fmt.Errorf("max retries exceeded: %w", lastErr)
}

Использование

go
err := Retry(ctx, RetryConfig{
    MaxAttempts: 5,
    BaseDelay:   time.Second,
    MaxDelay:    30 * time.Second,
}, func() error {
    return publisher.Publish(ctx, event)
})

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

• Ошибки валидации (400)
• Ошибки авторизации (401, 403)
• Постоянные ошибки (404 для отсутствующего ресурса)
• Ошибки идемпотентности

8.6 Dead Letter Queue (DLQ)

Что такое DLQ

Dead Letter Queue — очередь для сообщений которые не удалось обработать после N попыток.

Зачем:

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

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

1. Превышен лимит retry (например, 5 попыток)
2. Постоянная ошибка (невалидные данные)
3. Таймаут обработки

Реализация DLQ

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

    // Увеличиваем retry count
    newCount := event.RetryCount + 1
    if err := j.outboxService.UpdateRetryCount(ctx, event.ID, newCount); err != nil {
        return fmt.Errorf("update retry: %w", err)
    }

    // Проверяем лимит
    if newCount >= MaxRetries {
        j.logger.Warn("event moved to DLQ",
            "event_id", event.ID,
            "retry_count", newCount,
            "last_error", err,
        )

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

    return err
}

Обработка DLQ

Сообщения в DLQ требуют ручного вмешательства:

1. Мониторинг: алерт на количество сообщений в DLQ
2. Анализ: почему сообщение не обработалось
3. Исправление: исправить данные или код
4. Повторная обработка: переотправить сообщение

go
// Эндпоинт для повторной обработки DLQ
func (h *Handler) RetryDLQ(w http.ResponseWriter, r *http.Request) {
    events, err := h.outboxService.GetDead(r.Context(), 100)
    if err != nil {
        h.handleError(w, err)
        return
    }

    for _, event := range events {
        // Сброс статуса в pending
        if err := h.outboxService.ResetToPending(r.Context(), event.ID); err != nil {
            h.logger.Error("reset failed", "event_id", event.ID, "error", err)
        }
    }

    h.writeJSON(w, http.StatusOK, map[string]int{"retried": len(events)})
}

8.7 Idempotency keys

Проблема дублирования

At-least-once доставка может привести к дублированию:

• Retry после таймаута (событие доставлено, но ACK потерян)
• Повторная отправка клиентом
• Сбой в outbox relay

Решение: Idempotency keys

Idempotency key — уникальный идентификатор операции. Сервис проверяет ключ и не обрабатывает повторно.

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

sql
CREATE TABLE idempotency_keys (
    key VARCHAR(255) PRIMARY KEY,
    response JSONB,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    expires_at TIMESTAMPTZ NOT NULL
);

CREATE INDEX idx_idempotency_expires ON idempotency_keys(expires_at);

Генерация ключей

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

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

{
    "name": "my-project",
    "owner_id": "user-123"
}

Проверка idempotency

go
func (uc *CreateRepoUseCase) Execute(ctx context.Context, req CreateRepoRequest, idempotencyKey string) (*models.Repo, error) {
    // Проверяем существующий ключ
    existing, err := uc.idempotencyRepo.Get(ctx, idempotencyKey)
    if err != nil {
        return nil, fmt.Errorf("check idempotency: %w", err)
    }
    if existing != nil {
        // Возвращаем кешированный ответ
        return existing.Response, nil
    }

    // Выполняем операцию
    repo, err := uc.createRepo(ctx, req)
    if err != nil {
        return nil, err
    }

    // Сохраняем ключ и ответ
    if err := uc.idempotencyRepo.Save(ctx, idempotencyKey, repo); err != nil {
        return nil, fmt.Errorf("save idempotency: %w", err)
    }

    return repo, nil
}

Срок жизни ключей

Idempotency keys не должны жить вечно:

go
// Очистка старых ключей (cron job)
func (j *CleanupIdempotencyKeysJob) Run(ctx context.Context) error {
    deleted, err := j.idempotencyRepo.DeleteExpired(ctx, time.Now())
    if err != nil {
        return err
    }

    j.logger.Info("cleanup completed", "deleted", deleted)
    return nil
}

Обычно срок жизни: 24 часа — 7 дней.

8.8 Очереди сообщений (обзор)

Когда использовать очереди

• Декомпозиция сервисов
• Асинхронная обработка
• Буферизация при пиковых нагрузках
• Гарантия доставки

Сравнение очередей

RabbitMQ

Классический message broker. Хорошо для:

• Задач (task queues)
• RPC между сервисами
• Pub/Sub с маршрутизацией

go
// Публикация
ch.PublishWithContext(ctx,
    "",           // exchange
    queueName,    // routing key
    false,        // mandatory
    false,        // immediate
    amqp.Publishing{
        ContentType: "application/json",
        Body:        payload,
    },
)

Kafka

Distributed commit log. Хорошо для:

• Event streaming
• Логирование
• Аналитика
• Высокий throughput

go
// Публикация
producer.Produce(&kafka.Message{
    TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
    Value:          payload,
}, nil)

Yandex Message Queue

Managed очередь в Yandex Cloud. Используется в шаблоне проекта.

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

Где background processing в шаблоне

text
internal/handlers/jobs/
├── outbox_relay/handler.go       # Outbox relay job
├── saga_orchestrator/handler.go  # Saga orchestrator job
└── hmac_rotator/handler.go       # HMAC rotator job

internal/adapters/scheduler/
├── cron.go                       # robfig/cron wrapper
└── job.go                        # Интерфейс Job

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

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

1. Jobs реализуют интерфейс Job
2. Scheduler регистрирует jobs по расписанию
3. Outbox relay доставляет события из outbox
4. Retries с exponential backoff
5. Idempotency keys для защиты от дублирования

Задание: изучите internal/handlers/jobs/

• Какие jobs реализованы?
• Какое расписание у каждого?
• Как обрабатываются ошибки?
• Есть ли retries?

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

• Синхронно vs асинхронно: операция в HTTP-запросе даёт мгновенный результат и простую обработку ошибок, но блокирует при > 100ms; фоновый воркер разгружает запрос, однако платит задержкой, retries и DLQ.
• At-least-once vs exactly-once: полный exactly-once дорог и хрупок; outbox relay даёт at-least-once с повторами после таймаута, а дубли отсекаем idempotency keys на стороне потребителя.
• Retries vs DLQ: exponential backoff с jitter лечит временные сбои (503, таймауты), но бесполезен для постоянных ошибок; MarkDead после MaxRetries не блокирует relay, однако требует ручного разбора.
• atomic.Bool vs распределённый лок: single-flight на атомике прост и без внешних зависимостей, но защищает только один процесс; нескольким репликам воркера нужен distributed lock — он платит внешним хранилищем.
• RabbitMQ vs Kafka: RMQ хорош для task queues и маршрутизации, но держит сообщение до прочтения и не тянет streaming; Kafka — distributed log с огромным throughput и хранением по времени, но дороже в эксплуатации.

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

Задание 1: Worker с graceful shutdown

Реализуйте воркер который:

• Работает непрерывно
• Обрабатывает задачи из канала
• Корректно завершается при отмене контекста

go
type Worker struct {
    jobsCh <-chan Job
    logger *slog.Logger
}

func (w *Worker) Run(ctx context.Context)
func (w *Worker) Shutdown(ctx context.Context) error

Задание 2: Scheduler с robfig/cron

Реализуйте планировщик:

go
type Scheduler struct {
    cron *cron.Cron
    jobs []Job
}

func (s *Scheduler) Register(job Job)
func (s *Scheduler) Start(ctx context.Context)
func (s *Scheduler) Stop()

Зарегистрируйте 3 jobs:

• Outbox relay: каждые 5 секунд
• Cleanup: каждый час
• Metrics export: каждую минуту

Задание 3: Outbox relay

Реализуйте outbox relay job:

go
func (j *OutboxRelayJob) Run(ctx context.Context) error {
    events, err := j.outboxService.GetPending(ctx, 100)
    // обработка событий
}

Добавьте:

• Retry count
• Перемещение в DLQ после 5 попыток
• Логирование ошибок

Задание 4: Retry с exponential backoff

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

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

С:

• Exponential backoff
• Jitter
• Отмена через context
• Максимум попыток

Задание 5: Idempotency keys

Реализуйте проверку idempotency:

go
func (uc *CreateRepoUseCase) Execute(ctx context.Context, req CreateRepoRequest, idempotencyKey string) (*models.Repo, error)

Добавьте:

• Проверку существующего ключа
• Возврат кешированного ответа
• Сохранение ключа после операции
• Cleanup job для удаления старых ключей

Задание 6: DLQ monitoring

Добавьте метрику для DLQ:

go
var dlqEventsTotal = promauto.NewGauge(
    prometheus.GaugeOpts{
        Name: "dlq_events_total",
        Help: "Number of events in DLQ",
    },
)

Обновляйте метрику в outbox relay. Добавьте алерт если DLQ > 100.

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

delay := time.Duration(1<<attempt) * time.Second
// attempt: 0, 1, 2, 3

<-ctx.Done()
os.Exit(0)

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

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

• Понимать когда нужна фоновая обработка
• Реализовывать воркеры с планировщиком
• Применять outbox pattern
• Реализовывать retries с exponential backoff
• Настраивать DLQ
• Использовать idempotency keys
• Выбирать очередь сообщений под задачу
• Ответить на все вопросы Quiz

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

Следующий модуль: Модуль 9. API Security - OWASP Top 10 на практике, authentication vs authorization, HMAC signing, secrets management, input validation, safe logging. Привязка к internal/security/ шаблона.