Паттерн транзакционного почтового ящика в Go с PostgreSQL

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

Ваш сервис заказов сохраняет заказ в базе данных, а затем публикует событие order.created в брокере сообщений.

Эти две операции выполняются последовательно.

Между ними что-то идёт не так: брокер недоступен, тайм-аут сети, процесс перезапускается или контейнер отменяется. Запись в базу данных выполнена успешно. Публикация не удалась. Сервис, которому нужно знать о новом заказе, так и не узнает об этом. Никто не заметит проблему, пока клиент не позвонит.

Это проблема двойной записи (dual-write), и она является одним из самых распространённых источников скрытой потери данных в распределённых системах. Стандартным решением является паттерн транзакционного почтового ящика (transactional outbox).

Проблема двойной записи (Dual-write problem)

Режим отказа легко проанализировать, как только вы его увидите:

BEGIN;
  INSERT INTO orders ...   -- успешно
COMMIT;

PUBLISH order.created ...  -- ошибка, сбой или никогда не выполняется

База данных и брокер сообщений не разделяют границу транзакции. Не существует отката, который охватывал бы обе операции. Любой сервис, выполняющий последовательность save -> publish, имеет этот разрыв. Этот паттерн проявляется во многих формах:

• db.Save(order) followed by events.Publish(OrderCreated{...})
• HTTP-обработчик, который фиксирует транзакцию, а затем вызывает внешний вебхук
• Воркер, который обрабатывает запись из одной очереди и записывает результаты в другую

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

Цикл повторных попыток не решает эту проблему. Повторная попытка публикации после фиксации транзакции базы данных сработает только если сама повторная попытка будет надёжной — а это требует гарантии точности, которой у вас нет.

Что делает паттерн транзакционного почтового ящика

Паттерн почтового ящика устраняет разрыв, полностью устраняя прямую публикацию. Вместо вызова брокера из вашей бизнес-логики вы записываете запись события в таблицу outbox в той же транзакции базы данных, что и бизнес-данные. Отдельный фоновый процесс — ретранслятор (relay) — читает из таблицы почтового ящика и публикует событие в брокере.

BEGIN;
  INSERT INTO orders ...         -- бизнес-данные
  INSERT INTO outbox_events ...  -- запись события
COMMIT;

-- Процесс ретрансляции (отдельно):
SELECT ... FROM outbox_events FOR UPDATE SKIP LOCKED;
PUBLISH order.created ...
UPDATE outbox_events SET processed_at = NOW() WHERE id = $1;

Обе записи либо завершаются успешно, либо оба fails. Гарантия транзакции, которую вы уже имеете от PostgreSQL, теперь также охватывает запись события. Ретранслятор может повторять публикацию столько раз, сколько потребуется, потому что событие находится в надёжном хранилище. Если ретранслятор аварийно завершит работу посередине, он перезапустится и повторит попытку. Худший исход — событие будет опубликовано более одного раза — это обрабатывается путём обеспечения идемпотентности потребителей (см. Идемпотентность в распределённых системах).

Схема PostgreSQL для таблицы почтового ящика

Схема намеренно проста:

CREATE TABLE outbox_events (
    id             UUID         PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id   VARCHAR(100) NOT NULL,
    event_type     VARCHAR(100) NOT NULL,
    payload        JSONB        NOT NULL,
    attempts       INT          NOT NULL DEFAULT 0,
    created_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    processed_at   TIMESTAMPTZ
);

-- Частичный индекс: индексирует только необработанные строки, остаётся небольшим по мере пометки строк как выполненных
CREATE INDEX idx_outbox_unprocessed
    ON outbox_events (created_at)
    WHERE processed_at IS NULL;

Частичный индекс по created_at WHERE processed_at IS NULL важен. Без него индекс будет расти с каждым записанным событием, и запрос выборки ретранслятора со временем станет медленнее. С ним индекс покрывает только ожидающие строки, что в стабильном состоянии является небольшим, ограниченным набором, независимо от того, сколько событий было опубликовано.

Ключевые выборы полей:

• aggregate_type и aggregate_id описывают, какому сущности принадлежит событие. Полезно для гарантий порядка и маршрутизации.
• event_type — это имя события, которое ожидают ваши потребители.
• payload JSONB хранит тело события. Используйте JSONB, а не TEXT, чтобы при необходимости можно было выполнять запросы к нему.
• attempts отслеживает, сколько раз ретранслятор пытался опубликовать эту строку. Используется для ограничений повторных попыток и обработки мёртвых писем.
• processed_at равен NULL для ожидающих строк и устанавливается, когда ретранслятор успешно публикует.

Запись бизнес-данных и события почтового ящика в одной транзакции

Бизнес-логика записывает обе записи внутри одного вызова BeginTx / Commit. Здесь нет вызова публикации — только записи в базу данных.

type OrderService struct {
    db *sql.DB
}

func (s *OrderService) CreateOrder(ctx context.Context, order Order) error {
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback()

    if _, err := tx.ExecContext(ctx, `
        INSERT INTO orders (id, customer_id, total, created_at)
        VALUES ($1, $2, $3, NOW())
    `, order.ID, order.CustomerID, order.Total); err != nil {
        return fmt.Errorf("insert order: %w", err)
    }

    payload, err := json.Marshal(map[string]any{
        "order_id":    order.ID,
        "customer_id": order.CustomerID,
        "total":       order.Total,
    })
    if err != nil {
        return fmt.Errorf("marshal payload: %w", err)
    }

    if _, err := tx.ExecContext(ctx, `
        INSERT INTO outbox_events
            (aggregate_type, aggregate_id, event_type, payload)
        VALUES ($1, $2, $3, $4)
    `, "order", order.ID, "order.created", payload); err != nil {
        return fmt.Errorf("insert outbox event: %w", err)
    }

    return tx.Commit()
}

Если tx.Commit() завершится ошибкой, ни строка заказа, ни строка почтового ящика не будут сохранены. Если она завершится успешно, обе гарантированно окажутся в базе данных. Ретранслятор может опубликовать событие в любой момент после этого — немедленно, через секунду или после перезапуска ретранслятора после сбоя.

Это единственное изменение кода, требующееся в вашем бизнес-слое. Остальная часть паттерна живёт в ретрансляторе.

Реализация ретранслятора на Go

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

type OutboxRelay struct {
    db          *sql.DB
    publisher   Publisher
    logger      *slog.Logger
    batchSize   int
    pollInterval time.Duration
    maxAttempts  int
}

func (r *OutboxRelay) Run(ctx context.Context) error {
    ticker := time.NewTicker(r.pollInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-ticker.C:
            if err := r.processBatch(ctx); err != nil {
                r.logger.Error("outbox relay batch failed", "err", err)
            }
        }
    }
}

Ретранслятор уважает отмену контекста, что упрощает интеграцию с корректным завершением работы. Для подробного рассмотрения жизненного цикла контекста и паттернов отмены см. Go context.Done: Правильное использование context.Context в Go.

FOR UPDATE SKIP LOCKED: паттерн конкурентного воркера

Функция processBatch использует FOR UPDATE SKIP LOCKED для безопасной обработки конкурентных воркеров ретранслятора:

func (r *OutboxRelay) processBatch(ctx context.Context) error {
    tx, err := r.db.BeginTx(ctx, nil)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback()

    rows, err := tx.QueryContext(ctx, `
        SELECT id, aggregate_type, aggregate_id, event_type, payload
        FROM outbox_events
        WHERE processed_at IS NULL
          AND attempts < $1
        ORDER BY created_at
        LIMIT $2
        FOR UPDATE SKIP LOCKED
    `, r.maxAttempts, r.batchSize)
    if err != nil {
        return fmt.Errorf("query outbox: %w", err)
    }
    defer rows.Close()

    type row struct {
        id            string
        aggregateType string
        aggregateID   string
        eventType     string
        payload       json.RawMessage
    }

    var batch []row
    for rows.Next() {
        var e row
        if err := rows.Scan(
            &e.id, &e.aggregateType, &e.aggregateID, &e.eventType, &e.payload,
        ); err != nil {
            return fmt.Errorf("scan row: %w", err)
        }
        batch = append(batch, e)
    }
    if err := rows.Err(); err != nil {
        return err
    }

    for _, e := range batch {
        if err := r.publisher.Publish(ctx, e.eventType, e.aggregateID, e.payload); err != nil {
            r.logger.Error("publish failed", "event_id", e.id, "err", err)
            if _, err := tx.ExecContext(ctx,
                `UPDATE outbox_events SET attempts = attempts + 1 WHERE id = $1`, e.id,
            ); err != nil {
                r.logger.Error("increment attempts failed", "event_id", e.id, "err", err)
            }
            continue
        }

        if _, err := tx.ExecContext(ctx,
            `UPDATE outbox_events SET processed_at = NOW() WHERE id = $1`, e.id,
        ); err != nil {
            return fmt.Errorf("mark processed: %w", err)
        }
    }

    return tx.Commit()
}

FOR UPDATE SKIP LOCKED делает две вещи. Во-первых, FOR UPDATE блокирует выбранные строки на время транзакции, предотвращая выбор этих строк любой другой транзакцией. Во-вторых, SKIP LOCKED означает, что если строка уже заблокирована другой транзакцией, запрос пропускает её вместо ожидания. В результате несколько воркеров ретранслятора могут работать параллельно, и каждый выберет непересекающийся подмножество строк.

Без SKIP LOCKED второй воркер заблокируется до завершения первой транзакции, прежде чем увидит те же строки — в этот момент они уже будут помечены как выполненные. С SKIP LOCKED второй воркер немедленно выбирает другие строки вместо ожидания, обеспечивая безопасное горизонтальное масштабирование.

Обратите внимание на разделение сканирования и публикации в коде выше: все строки сканируются в срез перед началом цикла публикации. Это избегает удержания открытого курсора *sql.Rows во время вызовов сети к брокеру, что продлило бы транзакцию дольше необходимого.

Идемпотентность и дедупликация

Ретранслятор публикует как минимум один раз. Если он публикует событие, а затем аварийно завершает работу до фиксации обновления processed_at, он опубликует то же событие снова при перезапуске. Это неизбежно — точная доставка (exactly-once) между базой данных и брокером сообщений без координатора распределённых транзакций требует такого компромисса.

Потребители должны быть идемпотентными. Самый простой подход — отслеживать обработанные идентификаторы событий в таблице processed_events:

CREATE TABLE processed_events (
    event_id   UUID PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

func (h *OrderHandler) HandleOrderCreated(ctx context.Context, eventID string, payload []byte) error {
    // Дедупликация с использованием идентификатора события как естественного ключа
    _, err := h.db.ExecContext(ctx, `
        INSERT INTO processed_events (event_id) VALUES ($1)
        ON CONFLICT (event_id) DO NOTHING
    `, eventID)
    if err != nil {
        return fmt.Errorf("dedup check: %w", err)
    }

    // Проверить, действительно ли вставка произошла (1 строка) или была бездействием (0 строк)
    // Более простой подход: использовать RETURNING или проверить затронутые строки
    // Если затронут 0 строк, это дубликат — пропустить его
    ...
}

На практике многие команды полагаются на собственную дедупликацию брокера (например, поле key в Kafka для тем с лог-сжатием или заголовок message-id в RabbitMQ) и рассматривают дедупликацию на уровне базы данных как резервный вариант. Оба являются допустимыми слоями для применения.

Включите идентификатор события почтового ящика id (UUID) в публикуемое сообщение в качестве ключа дедупликации. Потребители затем могут использовать его независимо от того, какой механизм дедупликации они предпочитают.

Политика повторных попыток и отравляющие сообщения

Столбец attempts управляет политикой повторных попыток. Ретранслятор пропускает строки, где attempts >= maxAttempts, и рассматривает эти строки как мёртвые письма. Отдельный процесс или оповещение оператора обрабатывает их.

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

CREATE VIEW outbox_dead_letters AS
SELECT *
FROM outbox_events
WHERE attempts >= 5
  AND processed_at IS NULL
ORDER BY created_at;

Хорошая производственная политика повторных попыток:

• Установите maxAttempts в 5-10 в зависимости от стоимости повторных попыток.
• Рассмотрите экспоненциальную задержку: включите столбец retry_after и пропускайте строки, где retry_after > NOW().
• Оповещайте, если COUNT(*) FROM outbox_dead_letters превышает порог.
• Предоставьте путь ручной повторной попытки: конечную точку администратора или скрипт, который сбрасывает attempts = 0 и retry_after = NULL для конкретных строк.

Отравляющие сообщения — строки, которые постоянно завершаются ошибкой из-за ошибки в потребителе или несоответствия схемы, — не должны блокировать здоровые сообщения. Поскольку ретранслятор обрабатывает пакет за тик и помечает ошибки увеличением попыток, а не удаляет их из очереди, здоровые строки продолжают нормально, а отравляющие накапливают попытки, пока не достигнут порога мёртвых писем. Это представление outbox_dead_letters является версией на стороне базы данных того же паттерна, который брокер-native очереди мёртвых писем реализуют — карантина после порога, оповещение о объёме и требующее сознательного решения перед воспроизведением.

Порядок событий и партиционирование

Запрос выборки упорядочивает по created_at, что даёт порядок FIFO (первым пришёл — первым ушёл) внутри пакета. Для большинства случаев этого достаточно. Когда важен строгий порядок для каждой сущности — например, обеспечение того, чтобы order.updated никогда не публиковался до order.created для одного и того же заказа — вам нужен порядок по агрегату.

Добавьте aggregate_id в предложение ORDER BY и используйте его как ключ сообщения при публикации в тематическую тему с партиционированием, такую как Apache Kafka. Kafka маршрутизирует все сообщения с одним ключом к одному и тому же разделу, а разделы потребляются по порядку. Это даёт вам гарантии порядка по агрегату без глобального порядка, который потребовал бы одного экземпляра ретранслятора.

ORDER BY aggregate_id, created_at

Для брокеров, которые не поддерживают партиционированный порядок (например, базовые очереди AMQP), единственная альтернатива — ретранслятор одного экземпляра или проверки порядка на уровне приложения в потребителе.

Снижение задержки опроса с помощью LISTEN/NOTIFY

Интервал опроса в одну секунду означает среднюю задержку события в 500 миллисекунд. Для большинства рабочих нагрузок это нормально. Для случаев, когда вам нужна почти нулевая задержка, механизм LISTEN/NOTIFY PostgreSQL позволяет ретранслятору просыпаться немедленно при вставке новой строки почтового ящика.

Добавьте триггер в таблицу почтового ящика:

CREATE OR REPLACE FUNCTION notify_outbox_insert() RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify('outbox_event', NEW.id::text);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER outbox_insert_notify
AFTER INSERT ON outbox_events
FOR EACH ROW EXECUTE FUNCTION notify_outbox_insert();

В ретрансляторе прослушивайте канал и просыпайтесь на уведомлениях, всё ещё продолжая использовать периодический опрос в качестве резервного варианта:

func (r *OutboxRelay) Run(ctx context.Context) error {
    listener := pq.NewListener(r.dsn, 10*time.Second, time.Minute, nil)
    defer listener.Close()

    if err := listener.Listen("outbox_event"); err != nil {
        return fmt.Errorf("listen: %w", err)
    }

    ticker := time.NewTicker(5 * time.Second) // резервный опрос
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-listener.Notify:
            if err := r.processBatch(ctx); err != nil {
                r.logger.Error("outbox batch failed (notify)", "err", err)
            }
        case <-ticker.C:
            if err := r.processBatch(ctx); err != nil {
                r.logger.Error("outbox batch failed (poll)", "err", err)
            }
        }
    }
}

Резервный тикер обрабатывает любые пропущенные уведомления во время перезапуска ретранслятора или сетевых неполадок. Держите интервал резервного варианта в несколько секунд, а не миллисекунд — его задача — восстановление, а не низкая задержка.

Наблюдаемость: метрики, журналы и оповещения

Почтовый ящик — это инфраструктура. Относитесь к нему как к инфраструктуре и соответствующим образом оснастите его.

Ключевые метрики:

var (
    outboxPublished = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "outbox_events_published_total",
        Help: "Всего событий почтового ящика успешно опубликовано.",
    })
    outboxFailed = prometheus.NewCounterVec(prometheus.CounterOpts{
        Name: "outbox_events_failed_total",
        Help: "Всего сбоев публикации почтового ящика по типу события.",
    }, []string{"event_type"})
    outboxPending = prometheus.NewGauge(prometheus.GaugeOpts{
        Name: "outbox_events_pending",
        Help: "Текущее число необработанных событий почтового ящика.",
    })
    outboxBatchDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
        Name:    "outbox_batch_duration_seconds",
        Help:    "Длительность каждого пакета обработки почтового ящика.",
        Buckets: prometheus.DefBuckets,
    })
)

Обновление гаужей: запустите периодический запрос для поддержания точности outbox_events_pending:

SELECT COUNT(*) FROM outbox_events WHERE processed_at IS NULL;

Пороги оповещений для рассмотрения:

• outbox_events_pending > 1000 более двух минут: ретранслятор отстаёт или завис.
• outbox_events_pending растёт монотонно: брокер недоступен или ретранслятор аварийно завершил работу.
• Счётчик мёртвых писем не равен нулю: требуется расследование ошибки схемы или потребителя.
• outbox_batch_duration_seconds p95 > 5s: база данных медленная или размер пакета слишком велик.

Структурированные поля журнала: включите event_id, event_type, aggregate_id и attempt в каждую строку журнала от ретранслятора. Эти поля позволяют коррелировать неудачную публикацию с конкретной строкой почтового ящика и отслеживанием потребителя.

Почтовый ящик против прямой очереди против саги

Паттерн почтового ящика — не правильный инструмент для каждой проблемы координации. Вот сравнение:

Паттерн почтового ящика гарантирует, что один сервис надёжно излучает события, отражающие его собственные изменения состояния. Он не координирует изменения состояния между несколькими сервисами — для этого предназначен Паттерн Саги. Выбор брокера — будь то RabbitMQ, SQS, или Kafka — не зависит от самого паттерна почтового ящика; ретранслятор публикует в любой брокер, который использует ваша система.

Если вы строите сагу, паттерн почтового ящика всё ещё полезен: каждый участник саги записывает локальное изменение состояния и событие саги в одной транзакции с помощью почтового ящика, затем оркестратор или хореография саги надёжно читает эти события.

CDC на основе WAL в качестве альтернативного ретранслятора

Вместо опроса вы можете отслеживать журнал предзаписи (WAL) PostgreSQL и читать вставки почтового ящика непосредственно из потока репликации. Такие инструменты, как Debezium, делают это. Преимущества — более низкая задержка и отсутствие давления блокировок на таблицу почтового ящика. Недостатки — операционная сложность, выделенный слот репликации PostgreSQL и внешний сервис для запуска и мониторинга.

Для большинства команд ретранслятор опроса, описанный выше, является правильной отправной точкой. Отслеживание WAL имеет смысл, когда у вас высокие скорости вставки почтового ящика (десятки тысяч в секунду), нужна задержка события менее 100 мс или вы уже используете Debezium для других потребностей захвата изменений.

Интеграция с sqlc

Если вы используете sqlc для типобезопасного кода Go базы данных, запросы почтового ящика вписываются естественно:

-- name: InsertOutboxEvent :exec
INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload)
VALUES (@aggregate_type, @aggregate_id, @event_type, @payload);

-- name: FetchOutboxBatch :many
SELECT id, aggregate_type, aggregate_id, event_type, payload
FROM outbox_events
WHERE processed_at IS NULL
  AND attempts < @max_attempts
ORDER BY created_at
LIMIT @batch_size
FOR UPDATE SKIP LOCKED;

-- name: MarkOutboxProcessed :exec
UPDATE outbox_events SET processed_at = NOW() WHERE id = @id;

-- name: IncrementOutboxAttempts :exec
UPDATE outbox_events SET attempts = attempts + 1 WHERE id = @id;

-- name: OutboxPendingCount :one
SELECT COUNT(*) FROM outbox_events WHERE processed_at IS NULL;

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

Чеклист производства

Используйте это перед выпуском реализации почтового ящика:

База данных

• Таблица почтового ящика имеет частичный индекс по created_at WHERE processed_at IS NULL
• Столбец attempts присутствует по умолчанию равным 0
• Определено представление мёртвых писем или запрос
• Старые обработанные строки периодически архивируются или удаляются (достаточно ночного задания очистки)

Ретранслятор

• FOR UPDATE SKIP LOCKED используется в запросе опроса
• Ретранслятор работает внутри транзакции (начинает перед запросом, фиксирует после всех обновлений)
• Размер пакета ограничен (обычно 50-200 строк)
• Ретранслятор уважает отмену контекста для корректного завершения работы
• Неудачные публикации увеличивают attempts, а не вызывают прерывание пакета

Идемпотентность

• Публикуемое сообщение включает идентификатор почтового ящика id в качестве ключа дедупликации
• Потребители идемпотентны или брокер предоставляет дедупликацию
• См. Идемпотентность в распределённых системах для паттернов дедупликации

Наблюдаемость

• Гауж outbox_events_pending мониторится и оповещается
• Счётчик мёртвых писем оповещается
• Длительность пакета ретранслятора отслеживается
• Структурированные журналы включают event_id, event_type и aggregate_id

Операции

• Существует путь ручной повторной попытки для строк мёртвых писем
• Поведение перезапуска ретранслятора протестировано (правильно ли он повторно публикует?)
• Поведение при отказе брокера протестировано (правильно ли растёт и опорожняется почтовый ящик?)

Последние мысли

Проблему двойной записи легко отклонить как крайний случай, пока она не вызовет инцидент. Паттерн транзакционного почтового ящика решает его с помощью инструментов, которые у вас уже есть: транзакция PostgreSQL, фоновая горутинка и одна дополнительная таблица. Ретранслятор прост в построении, прост в эксплуатации и прост для понимания.

Стоимость заключается в том, что потребители должны быть спроектированы для доставки как минимум один раз. Это разумный компромисс. Точная доставка между базой данных и брокером без распределённых транзакций практически достижима — и притворство обратным приводит к системам, которые молча пропускают или дважды обрабатывают события при условиях сбоя.

Записывайте событие вместе с данными. Надёжно ретранслируйте его. Делайте потребителей идемпотентными. Это весь паттерн.

Эта статья является частью кластера Архитектура приложений в производстве.

Источники

• PostgreSQL FOR UPDATE SKIP LOCKED
• PostgreSQL LISTEN/NOTIFY
• Маршрутизатор событий почтового ящика Debezium
• Поддержка уведомлений lib/pq
• Пакет database/sql