Padrão Transactional Outbox em Go com PostgreSQL

Dois writes que deveriam ser bem-sucedidos juntos acabarão falhando separadamente. Seu serviço de pedidos salva o pedido no banco de dados e, em seguida, publica um evento order.created em um message broker.

Essas duas operações são executadas uma após a outra.

Entre elas, algo dá errado: o broker está fora do ar, o tempo limite da rede expira, o processo é reiniciado ou o container é removido. A gravação no banco de dados foi bem-sucedida. A publicação não foi. O serviço downstream que precisa saber sobre o novo pedido nunca fica ciente. Ninguém percebeu até um cliente ligar.

Este é o problema da escrita dupla (dual-write) e é uma das fontes mais comuns de perda silenciosa de dados em sistemas distribuídos. O padrão Transactional Outbox é a correção padrão.

O problema da escrita dupla

O modo de falha é fácil de analisar assim que você o identifica:

BEGIN;
  INSERT INTO orders ...   -- succeeds
COMMIT;

PUBLISH order.created ...  -- fails, crashes, or is never reached

O banco de dados e o message broker não compartilham um limite de transação. Não há rollback que cubra ambos. Todo serviço que executa save -> publish em sequência tem essa lacuna. O padrão aparece em muitas formas:

• db.Save(order) seguido por events.Publish(OrderCreated{...})
• Handler HTTP que confirma uma transação e, em seguida, chama um webhook externo
• Worker que processa um registro de uma fila e grava resultados em outra

O resultado em todos os casos é o mesmo: um lado tem sucesso enquanto o outro falha, e o sistema termina em um estado invisível para a monitoração porque ambas as operações individuais retornaram sucesso em algum momento.

Um loop de repetição não corrige isso. Repetir a publicação após o commit do banco de dados só funciona se a própria repetição for confiável – o que requer a garantia de durabilidade exata que você não possui.

O que o padrão Transactional Outbox faz

O padrão outbox elimina a lacuna removendo a publicação direta inteiramente. Em vez de chamar o broker dentro da sua lógica de negócios, você grava um registro de evento em uma tabela outbox na mesma transação do banco de dados dos dados de negócios. Um processo em segundo plano separado – o relay – lê da tabela outbox e publica no broker.

BEGIN;
  INSERT INTO orders ...         -- business data
  INSERT INTO outbox_events ...  -- event record
COMMIT;

-- Relay process (separately):
SELECT ... FROM outbox_events FOR UPDATE SKIP LOCKED;
PUBLISH order.created ...
UPDATE outbox_events SET processed_at = NOW() WHERE id = $1;

Ambas as gravações têm sucesso ou ambas falham. A garantia de transação que você já possui do PostgreSQL agora também cobre o registro de evento. O relay pode repetir a publicação quantas vezes for necessário, pois o evento está em armazenamento durável. Se o relay falhar durante o processamento, ele reinicia e repete. O pior resultado é que o evento seja publicado mais de uma vez – o que é tratado tornando os consumidores idempotentes (veja Idempotência em Sistemas Distribuídos).

Esquema do PostgreSQL para a tabela outbox

O esquema é intencionalmente simples:

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
);

-- Partial index: only indexes unprocessed rows, stays small as rows are marked done
CREATE INDEX idx_outbox_unprocessed
    ON outbox_events (created_at)
    WHERE processed_at IS NULL;

O índice parcial em created_at WHERE processed_at IS NULL é importante. Sem ele, o índice cresce com cada evento já gravado e a consulta de polling do relay fica mais lenta com o tempo. Com ele, o índice cobre apenas as linhas pendentes, que, em estado estável, é um conjunto pequeno e limitado, independentemente de quantos eventos foram publicados.

Principais escolhas de campos:

• aggregate_type e aggregate_id descrevem a qual entidade o evento pertence. Útil para garantias de ordenação e roteamento.
• event_type é o nome do evento que seus consumidores esperam.
• payload JSONB armazena o corpo do evento. Use JSONB em vez de TEXT para que você possa consultá-lo se necessário.
• attempts rastreia quantas vezes o relay tentou publicar esta linha. Usado para limites de repetição e tratamento de mensagens envenenadas.
• processed_at é NULL para linhas pendentes e definido quando o relay publica com sucesso.

Gravando dados de negócios e evento outbox em uma única transação

A lógica de negócios grava ambos os registros dentro de uma única chamada BeginTx / Commit. Não há chamada de publicação aqui – apenas gravações no banco de dados.

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()
}

Se tx.Commit() falhar, nem a linha do pedido nem a linha do outbox são persistidas. Se tiver sucesso, ambos estão garantidos no banco de dados. O relay pode publicar o evento em qualquer ponto após isso – imediatamente, em um segundo ou após o relay reiniciar após uma falha.

Esta é a única alteração de código necessária na sua camada de negócios. O restante do padrão reside no relay.

Implementação do relay em Go

O relay é um worker em segundo plano que faz polling da tabela outbox em um timer. Ele busca um lote de linhas não processadas, publica cada uma delas e as marca como concluídas. Mantenha-o no mesmo binário que sua aplicação ou execute-o como um processo separado – ambos funcionam, mas o mesmo binário é mais simples de operar.

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)
            }
        }
    }
}

O relay respeita o cancelamento do contexto, o que facilita a integração com o encerramento gracioso. Para um tratamento detalhado sobre ciclo de vida do contexto e padrões de cancelamento, veja Go context.Context Done Right.

FOR UPDATE SKIP LOCKED: o padrão de worker concorrente

A função processBatch usa FOR UPDATE SKIP LOCKED para lidar com segurança com workers de relay concurrentes:

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 faz duas coisas. Primeiro, FOR UPDATE bloqueia as linhas selecionadas durante a duração da transação, impedindo que qualquer outra transação as selecione. Segundo, SKIP LOCKED significa que, se uma linha já estiver bloqueada por outra transação, a consulta a ignora em vez de esperar. O resultado é que vários workers de relay podem executar em paralelo e cada um pegará um subconjunto não sobreposto de linhas.

Sem SKIP LOCKED, um segundo worker bloquearia até que a primeira transação fosse confirmada antes de ver as mesmas linhas – momento em que elas já estariam marcadas como concluídas. Com SKIP LOCKED, o segundo worker imediatamente pega linhas diferentes em vez de esperar, proporcionando escalonamento horizontal seguro.

Observe a separação entre leitura e publicação no código acima: todas as linhas são lidas para um slice antes do loop de publicação começar. Isso evita manter um cursor *sql.Rows aberto durante chamadas de rede ao broker, o que manteria a transação aberta por mais tempo do que o necessário.

Idempotência e deduplicação

O relay publica pelo menos uma vez. Se ele publicar um evento e, em seguida, falhar antes de confirmar a atualização processed_at, ele publicará o mesmo evento novamente na reinicialização. Isso é inevitável – a entrega exatamente uma vez entre um banco de dados e um message broker sem um coordenador de transações distribuídas requer essa compensação.

Os consumidores devem ser idempotentes. A abordagem mais simples é rastrear os IDs de eventos processados em uma tabela 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 {
    // Deduplicate using the event ID as the natural key
    _, 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)
    }

    // Check if the insert actually happened (1 row) or was a no-op (0 rows)
    // A simpler approach: use RETURNING or check rows affected
    // If 0 rows affected, this is a duplicate -- skip it
    ...
}

Na prática, muitas equipes confiam nos próprios cabeçalhos de deduplicação do broker (como o campo key do Kafka para tópicos com compactação de log, ou o cabeçalho message-id do RabbitMQ) e tratam a deduplicação em nível de banco de dados como um fallback. Ambas são camadas válidas para aplicar.

Inclua o id do evento outbox (um UUID) na mensagem publicada como chave de deduplicação. Os consumidores podem então usá-lo independentemente do mecanismo de deduplicação que preferirem.

Política de repetição e mensagens envenenadas

A coluna attempts direciona a política de repetição. O relay ignora linhas onde attempts >= maxAttempts e trata essas linhas como dead letters. Um processo separado ou alerta de operador as trata.

Uma visualização simples de dead-letter:

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

Uma boa política de repetição para produção:

• Defina maxAttempts para 5-10 dependendo de quão custosas são as repetições.
• Considere backoff exponencial: inclua uma coluna retry_after e ignore linhas onde retry_after > NOW().
• Alertar quando COUNT(*) FROM outbox_dead_letters exceder um limite.
• Fornecer um caminho de repetição manual: um endpoint de administrador ou script que redefine attempts = 0 e retry_after = NULL para linhas específicas.

Mensagens envenenadas – linhas que falham consistentemente devido a um bug no consumidor ou incompatibilidade de schema – não devem bloquear mensagens saudáveis. Como o relay processa um lote por tick e marca falhas com um incremento de tentativa em vez de removê-las da fila, as linhas saudáveis prosseguem normalmente enquanto as envenenadas acumulam tentativas até atingirem o limite de dead-letter. Esta visualização outbox_dead_letters é uma versão do lado do banco de dados do mesmo padrão que dead-letter queues nativas do broker implementam – quarentena após um limite, alerta sobre volume e exigência de uma decisão deliberada antes da repetição.

Ordenação de eventos e particionamento

A consulta de polling ordena por created_at, o que oferece ordenação first-in-first-out dentro de um lote. Para a maioria dos casos de uso, isso é suficiente. Quando a ordenação estrita por entidade é importante – por exemplo, garantindo que order.updated nunca seja publicado antes de order.created para o mesmo pedido – você precisa de ordenação por agregado.

Adicione aggregate_id à cláusula ORDER BY e use-o como chave de mensagem ao publicar em um tópico particionado como Apache Kafka. O Kafka roteia todas as mensagens com a mesma chave para a mesma partição, e as partições são consumidas em ordem. Isso oferece garantias de ordenação por agregado sem ordenação global, o que exigiria uma única instância de relay.

ORDER BY aggregate_id, created_at

Para brokers que não suportam ordenação particionada (como filas AMQP básicas), a instância única de relay ou verificações de ordenação em nível de aplicativo no consumidor são as alternativas práticas.

Reduzir a latência do polling com LISTEN/NOTIFY

Um intervalo de polling de um segundo significa uma latência média de evento de 500 milissegundos. Para a maioria das cargas de trabalho, isso é aceitável. Para casos onde você precisa de latência próxima de zero, o mecanismo LISTEN/NOTIFY do PostgreSQL permite que o relay desperte imediatamente quando uma nova linha outbox é inserida.

Adicione um trigger à tabela outbox:

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();

No relay, escute no canal e desperte com notificações, mantendo ainda o fallback para polling periódico:

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) // fallback poll
    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)
            }
        }
    }
}

O ticker de fallback lida com quaisquer notificações perdidas durante uma reinicialização do relay ou falha de rede. Mantenha o intervalo de fallback em alguns segundos, não em milissegundos – seu trabalho é recuperação, não baixa latência.

Observabilidade: métricas, logs e alertas

O outbox é infraestrutura. Trate-o como infraestrutura e instrumente-o conforme o apropriado.

Métricas principais:

var (
    outboxPublished = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "outbox_events_published_total",
        Help: "Total outbox events successfully published.",
    })
    outboxFailed = prometheus.NewCounterVec(prometheus.CounterOpts{
        Name: "outbox_events_failed_total",
        Help: "Total outbox publish failures by event type.",
    }, []string{"event_type"})
    outboxPending = prometheus.NewGauge(prometheus.GaugeOpts{
        Name: "outbox_events_pending",
        Help: "Current number of unprocessed outbox events.",
    })
    outboxBatchDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
        Name:    "outbox_batch_duration_seconds",
        Help:    "Duration of each outbox processing batch.",
        Buckets: prometheus.DefBuckets,
    })
)

Atualização de Gauge: execute uma consulta periódica para manter outbox_events_pending preciso:

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

Limites de alerta para considerar:

• outbox_events_pending > 1000 por mais de dois minutos: o relay está ficando para trás ou travado.
• outbox_events_pending crescendo monotonamente: o broker está fora do ar ou o relay falhou.
• Contagem de dead-letter não zero: há um bug de schema ou consumidor que precisa de investigação.
• outbox_batch_duration_seconds p95 > 5s: o banco de dados está lento ou o tamanho do lote é muito grande.

Campos de log estruturados: inclua event_id, event_type, aggregate_id e attempt em cada linha de log do relay. Esses campos permitem correlacionar uma publicação falhada com a linha específica do outbox e o rastreamento do consumidor downstream.

Outbox vs. fila direta vs. saga

O padrão outbox não é a ferramenta certa para todos os problemas de coordenação. Aqui está a comparação:

O padrão outbox garante que um único serviço emita eventos confiáveis que refletem suas próprias mudanças de estado. Ele não coordena mudanças de estado entre múltiplos serviços – isso é para o Padrão Saga. A escolha do broker – seja RabbitMQ, SQS, ou Kafka – é independente do próprio padrão outbox; o relay publica para qualquer broker que seu sistema use.

Se você está construindo uma saga, o padrão outbox ainda é útil: cada participante da saga grava sua mudança de estado local e seu evento de saga em uma transação usando o outbox, então o orchestrator ou choreography da saga lê esses eventos confiavelmente.

CDC baseado em WAL como alternativa de relay

Em vez de polling, você pode acompanhar o Write-Ahead Log (WAL) do PostgreSQL e ler as inserções do outbox diretamente do stream de replicação. Ferramentas como o Debezium fazem isso. As vantagens são menor latência e nenhuma pressão de bloqueio na tabela outbox. As desvantagens são complexidade operacional, um slot de replicação PostgreSQL dedicado e um serviço externo para executar e monitorar.

Para a maioria das equipes, o relay de polling descrito acima é o ponto de partida certo. O acompanhamento do WAL faz sentido quando você tem altas taxas de inserção no outbox (dezenas de milhares por segundo), precisa de latência de evento sub-100ms ou já está executando o Debezium para outras necessidades de captura de alterações.

Integração com sqlc

Se você usa sqlc para código de banco de dados Go com tipagem segura, as consultas do outbox se encaixam naturalmente:

-- 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;

O sqlc gera funções com tipagem segura para cada consulta, o que evita erros de interpolação de strings e mantém a lógica da consulta do outbox localizada junto com o restante da sua camada de acesso ao banco de dados.

Checklist de produção

Use isso antes de enviar uma implementação de outbox:

Banco de Dados

• A tabela outbox tem o índice parcial em created_at WHERE processed_at IS NULL
• Coluna attempts presente com padrão 0
• Visualização ou consulta de dead-letter definida
• Linhas processadas antigas são arquivadas ou excluídas periodicamente (uma tarefa de limpeza noturna é suficiente)

Relay

• FOR UPDATE SKIP LOCKED usado na consulta de polling
• Relay executa dentro de uma transação (begin antes da consulta, commit após todas as atualizações)
• Tamanho do lote é limitado (50-200 linhas é típico)
• Relay respeita o cancelamento de contexto para encerramento gracioso
• Publicações falhas incrementam attempts em vez de causar o abort do lote

Idempotência

• A mensagem publicada inclui o id do outbox como chave de deduplicação
• Consumidores são idempotentes ou o broker fornece deduplicação
• Veja Idempotência em Sistemas Distribuídos para padrões de deduplicação

Observabilidade

• Gauge outbox_events_pending é monitorado e alertado
• Contagem de dead-letter é alertada
• Duração do lote do relay é rastreada
• Logs estruturados incluem event_id, event_type e aggregate_id

Operações

• Caminho de repetição manual existe para linhas de dead-letter
• Comportamento de reinicialização do relay é testado (ele repubblica corretamente?)
• Comportamento de falha do broker é testado (o outbox cresce e drena corretamente?)

Considerações finais

O problema da escrita dupla é fácil de descartar como um caso de borda até que cause um incidente. O padrão transactional outbox o resolve com ferramentas que você já possui: uma transação PostgreSQL, um goroutine em segundo plano e uma tabela extra. O relay é simples de construir, simples de operar e simples de raciocinar.

O custo é que os consumidores devem ser projetados para entrega pelo menos uma vez. Essa é uma compensação razoável. A entrega exatamente uma vez entre um banco de dados e um broker sem transações distribuídas não é alcançável na prática – e fingir o contrário leva a sistemas que silenciosamente perdem ou processam eventos duplicados sob condições de falha.

Grave o evento com os dados. Relay-o confiavelmente. Torne os consumidores idempotentes. Esse é todo o padrão.

Este artigo faz parte do cluster App Architecture in Production.

Fontes

• PostgreSQL FOR UPDATE SKIP LOCKED
• PostgreSQL LISTEN/NOTIFY
• Debezium outbox event router
• lib/pq LISTEN support
• database/sql package