Salvar e Depois Publicar São Duas Escritas: O Transactional Outbox Pattern

O command handler da versão em inglês do meu post de CQRS e Event Sourcing termina assim:

// Save events
if err := h.repo.Save(ctx, payment); err != nil {
    return fmt.Errorf("saving payment: %w", err)
}

// Publish for read model updates and integration
for _, event := range events {
    if err := h.eventBus.Publish(ctx, event); err != nil {
        // Log but don't fail - events are persisted
        log.Printf("failed to publish event: %v", err)
    }
}

O comentário é verdadeiro, e também é o bug. Os eventos estão persistidos. Só que nada volta a lê-los para descobrir quais foram publicados, então uma publicação que falhou fica falhada para sempre e o modelo de leitura nunca fica sabendo que o pagamento aconteceu.

O post sobre os custos do CQRS tem uma versão pior, só que mais silenciosa:

go s.eventBus.Publish(context.Background(), OrderCreatedEvent{OrderID: order.ID})

Uma goroutine que ninguém espera. Um deploy manda SIGTERM, o processo sai, e a publicação nunca roda. Escrevi um post inteiro sobre por que graceful shutdown é mais difícil do que parece e mesmo assim publiquei isso.

Os dois são o mesmo erro, chamado de dual write: uma operação que precisa alterar dois sistemas sem nenhuma transação cobrindo os dois. Este post é sobre a correção padrão, o Transactional Outbox Pattern, e sobre quanto ela custa, porque não sai de graça.

Quatro jeitos de duas escritas discordarem

O banco e o broker são sistemas separados. Cada escrita pode dar certo ou falhar de forma independente, e o processo pode morrer entre as duas. Colocando os casos na mesa:

O post de custos ocultos trata consistência eventual como o intervalo entre uma escrita e o modelo de leitura alcançá-la. As três primeiras linhas não são isso. São um modelo de leitura que nunca alcança, porque o evento que o atualizaria não existe em nenhum lugar visível para um consumidor. Consistência eventual pressupõe que o "eventual" chega.

Por que as correções óbvias não funcionam

Publicar primeiro. Isso troca um evento perdido por um evento fantasma (a quarta linha). O fantasma costuma ser pior: um consumidor que manda um e-mail ou captura um pagamento não consegue desfazer isso quando a escrita no banco falha.

Fazer retry do publish. Retries resolvem o broker indisponível por alguns instantes. Não resolvem o processo morrendo, porque o estado do retry vive na memória do processo que acabou de morrer. Mesma história para uma goroutine em background com backoff.

Uma transação distribuída. Two-phase commit entre o Postgres e o broker tornaria as escritas atômicas, mas SNS, SQS, EventBridge e Kafka não participam de transações XA, e quando 2PC está disponível ele segura locks durante um round trip de rede até um coordenador. Não é uma opção realista para os brokers que as pessoas usam de fato.

As três têm em comum a tentativa de coordenar dois sistemas no momento da escrita. O outbox desiste de tentar.

O Transactional Outbox Pattern

Existe um sistema capaz de tornar várias escritas atômicas: o banco em que você já está escrevendo. Então escreva o evento lá também, na mesma transação da mudança de estado, e leve-o para o broker depois.

  command handler                          relay                     broker
 ┌────────────────────────────┐      ┌────────────────────┐      ┌─────────────┐
 │ BEGIN                      │      │ busca linhas não   │      │             │
 │   UPDATE payments ...      │      │ publicadas         │ ───► │ SNS / Kafka │ ──► consumidores
 │   INSERT INTO outbox ...   │ ───► │ publica            │      │ / SQS       │
 │ COMMIT                     │      │ marca publicada    │      │             │
 └────────────────────────────┘      └────────────────────┘      └─────────────┘
       uma escrita atômica              at-least-once

Agora a mudança de estado e o registro do evento fazem commit juntos ou rollback juntos. Publicar vira um job separado que pode falhar, cair e tentar de novo quantas vezes precisar, porque aquilo que ele está tentando de novo é durável.

A tabela:

CREATE TABLE outbox (
    id            uuid        PRIMARY KEY,
    aggregate_id  uuid        NOT NULL,
    event_type    text        NOT NULL,
    payload       jsonb       NOT NULL,
    created_at    timestamptz NOT NULL DEFAULT now(),
    published_at  timestamptz
);

CREATE INDEX outbox_unpublished
    ON outbox (created_at)
    WHERE published_at IS NULL;

O índice parcial importa. O relay só pede linhas não publicadas, que deveriam ser uma fração mínima da tabela, então o índice continua pequeno não importa quanto histórico acumule.

O handler do primeiro post, reescrito:

func (h *ProcessPaymentHandler) Handle(ctx context.Context, cmd ProcessPayment) error {
    tx, err := h.db.Begin(ctx)
    if err != nil {
        return fmt.Errorf("begin: %w", err)
    }
    defer tx.Rollback(ctx)

    payment, err := h.repo.Load(ctx, tx, cmd.PaymentID)
    if err != nil {
        return fmt.Errorf("loading payment: %w", err)
    }

    risk := h.riskEngine.Assess(cmd.CustomerID, cmd.Amount)
    events, err := payment.Process(cmd.Amount, cmd.Method, risk, cmd.IdempotencyKey)
    if err != nil {
        return fmt.Errorf("processing payment: %w", err)
    }

    if err := h.repo.Save(ctx, tx, payment); err != nil {
        return fmt.Errorf("saving payment: %w", err)
    }
    if err := outbox.Add(ctx, tx, events...); err != nil {
        return fmt.Errorf("recording events: %w", err)
    }

    return tx.Commit(ctx)
}

Não existe mais event bus no handler. Ele não consegue publicar, então não consegue publicar pela metade.

func Add(ctx context.Context, tx pgx.Tx, events ...Event) error {
    for _, e := range events {
        payload, err := json.Marshal(e)
        if err != nil {
            return fmt.Errorf("marshal %s: %w", e.EventType(), err)
        }
        _, err = tx.Exec(ctx, `
            INSERT INTO outbox (id, aggregate_id, event_type, payload)
            VALUES ($1, $2, $3, $4)`,
            e.EventID(), e.AggregateID(), e.EventType(), payload)
        if err != nil {
            return fmt.Errorf("insert outbox %s: %w", e.EventType(), err)
        }
    }
    return nil
}

O ID do evento é criado pelo domínio quando o evento nasce, não pelo relay na hora de publicar. Esse detalhe volta mais adiante: é ele que permite aos consumidores reconhecer uma duplicata.

O relay

func (r *Relay) Run(ctx context.Context) error {
    for {
        n, err := r.publishBatch(ctx)
        if err != nil {
            r.log.Error("outbox relay", "err", err)
        }
        if n == r.batchSize {
            continue // lote cheio indica backlog; drena sem esperar
        }
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(r.interval):
        }
    }
}

func (r *Relay) publishBatch(ctx context.Context) (int, error) {
    tx, err := r.db.Begin(ctx)
    if err != nil {
        return 0, err
    }
    defer tx.Rollback(ctx)

    rows, err := tx.Query(ctx, `
        SELECT id::text, event_type, aggregate_id::text, payload
        FROM outbox
        WHERE published_at IS NULL
        ORDER BY created_at
        LIMIT $1
        FOR UPDATE SKIP LOCKED`, r.batchSize)
    if err != nil {
        return 0, err
    }
    batch, err := pgx.CollectRows(rows, pgx.RowToStructByPos[Message])
    if err != nil {
        return 0, err
    }

    published := make([]string, 0, len(batch))
    for _, m := range batch {
        if err := r.publisher.Publish(ctx, m); err != nil {
            // Para em vez de pular: publicar eventos posteriores a um que falhou os reordena.
            r.log.Warn("publish failed, will retry", "event_id", m.EventID, "err", err)
            break
        }
        published = append(published, m.EventID)
    }

    if _, err := tx.Exec(ctx, `
        UPDATE outbox SET published_at = now()
        WHERE id = ANY($1::uuid[])`, published); err != nil {
        return 0, err
    }
    return len(published), tx.Commit(ctx)
}

publishBatch retorna quantos publicou, não quantos buscou. Se retornasse o tamanho do lote com o broker fora do ar, o Run veria um lote cheio, pularia a espera e entraria em loop.

FOR UPDATE SKIP LOCKED permite rodar mais de um relay sem que eles publiquem as mesmas linhas. Como explico mais abaixo, rodar mais de um relay tem um custo.

Uma armadilha merece destaque porque parece otimização. Não faça o relay lembrar "o último ID ou timestamp que publiquei" e buscar tudo depois disso. Valores de sequence e now() são atribuídos quando a linha é inserida, não quando a transação faz commit. A transação A insere a linha 10, a transação B insere a 11, B faz commit primeiro. Um relay que consulta nessa janela vê a 11, registra 11 como high-water mark, e a linha 10 faz commit atrás dele e nunca é publicada. Controlar published_at por linha, como acima, não tem esse problema. A linha 10 simplesmente aparece na próxima consulta.

O que o outbox de fato garante

Ele transforma "talvez zero vezes" em "pelo menos uma vez". Não dá exactly once, e o relay acima mostra por quê: se o Publish dá certo e depois o UPDATE ou o COMMIT falha, ou o relay morre entre eles, a linha continua não publicada e sai de novo na próxima consulta.

É o lado certo para errar, já que uma duplicata pode ser detectada e um evento ausente não. Mas significa que todo consumidor precisa ser idempotente, e isso é um requisito de verdade, não um bônus.

A abordagem em que confio é a mesma de idempotência pertence ao banco de dados: uma constraint, com a colisão tratada como "já foi feito".

func (p *PaymentProjection) Handle(ctx context.Context, msg Message) error {
    tx, err := p.db.Begin(ctx)
    if err != nil {
        return err
    }
    defer tx.Rollback(ctx)

    tag, err := tx.Exec(ctx, `
        INSERT INTO processed_events (consumer, event_id)
        VALUES ('payment_summary', $1)
        ON CONFLICT DO NOTHING`, msg.EventID)
    if err != nil {
        return err
    }
    if tag.RowsAffected() == 0 {
        return nil // reentrega; a primeira cópia já aplicou isto
    }

    if err := p.apply(ctx, tx, msg); err != nil {
        return err
    }
    return tx.Commit(ctx)
}

Às vezes isso é chamado de inbox, e espelha o outbox: o registro de deduplicação e o efeito fazem commit juntos. Exige que o modelo de leitura esteja num armazenamento transacional. Se a projeção escreve no Elasticsearch ou num cache, use uma escrita protegida por versão (UPDATE ... WHERE version < $new_version, ou o versionamento externo do próprio armazenamento). Isso cobre duplicatas e reordenação em projeções que substituem estado. Não funciona para projeções que acumulam, como contadores, já que não há versão para comparar com um incremento.

Deduplicação no broker não substitui isso. SNS FIFO e SQS FIFO deduplicam numa janela de cinco minutos, e um relay que fica fora do ar por mais tempo que isso e depois republica está fora dela.

Os trade-offs

Esta seção é o motivo de eu ter escrito o post. O outbox costuma ser apresentado como a resposta, ponto final. É uma boa resposta, e isto é o que ela custa.

Latência

Os eventos agora esperam o relay. Com um intervalo de polling T e um relay ocioso, um evento espera entre zero e T, ou cerca de T/2 em média. Um intervalo de 500 ms adiciona algo como 250 ms entre o commit e a publicação. Isso é aritmética, não medição, mas mostra como o custo escala: cortar o intervalo pela metade corta o atraso pela metade e dobra as consultas vazias.

O Postgres consegue eliminar a maior parte disso. Chame pg_notify('outbox', '') dentro da transação do comando e deixe o relay fazer LISTEN. Notificações só são entregues no commit, então o relay acorda quando há algo commitado para publicar. Mantenha o polling mesmo assim: notificações não são duráveis, e uma enviada enquanto o relay está reconectando simplesmente se perde. O polling vira a rede de segurança e o NOTIFY vira o caminho rápido.

Carga no banco que você queria proteger

Todo comando agora faz um insert a mais, o relay faz um update por evento, e algo precisa apagar as linhas antigas. No Postgres isso é uma tabela com inserts, updates e deletes constantes, exatamente a carga que gera bloat e mantém o autovacuum ocupado. Em volume baixo você nem percebe. Em volume alto o outbox pode virar uma das tabelas mais movimentadas do banco, e ela está dentro das mesmas transações que as suas escritas de negócio.

Retenção é uma decisão por si só. Apagar as linhas assim que são publicadas mantém a tabela pequena. Mantê-las por alguns dias (DELETE ... WHERE published_at < now() - interval '7 days', em lotes) permite responder "chegamos a publicar o evento X, e quando?", que é a pergunta que você vai estar fazendo durante um incidente. Eu manteria por um tempo, a menos que o volume torne isso doloroso.

Uma transação aberta durante uma chamada de rede

O relay acima segura locks de linha enquanto chama o broker. Se o broker estiver lento, esses locks ficam presos o tempo todo, e uma transação longa também impede o vacuum de limpar o que ficou para trás. A alternativa é reivindicar as linhas com um lease (SET locked_until = now() + interval '30 seconds'), fazer commit, publicar fora de qualquer transação, e depois marcá-las como publicadas. Isso elimina a transação longa e adiciona lógica de expiração de lease e um segundo round trip. Comece pela versão simples e troque se a latência do broker aparecer nas métricas do banco.

Ordenação compete com throughput

Um único relay publicando em ordem de created_at dá aproximadamente a ordem de commit, e geralmente é disso que os consumidores precisam. Não exatamente a ordem de commit, pelo mesmo motivo de horário de insert versus horário de commit da armadilha do high-water mark, então números de versão por agregado no evento continuam valendo a pena.

Adicione relays com SKIP LOCKED e você ganha throughput, mas o relay 1 pode publicar o evento 2 de um agregado enquanto o relay 2 ainda está publicando o evento 1. Se os consumidores se importam com a ordem por agregado, particione: cada relay é dono de hash(aggregate_id) % N e publica com uma chave de partição (uma key do Kafka, um message group ID do SQS FIFO) para que o broker preserve a ordem. São mais peças móveis, e é por isso que "é só adicionar outro relay" é uma decisão de design, não um botão de escala.

Mais um processo para rodar e observar

O relay é um novo modo de falha: se ele para, nada dá erro. Os comandos continuam funcionando, o outbox continua crescendo e o modelo de leitura fica velho. O lado bom é que o outbox agora funciona como buffer: quando o broker cai por uma hora, nada se perde, só enfileira. O lado ruim é que alguém precisa perceber. Uma consulta cobre quase tudo:

SELECT now() - min(created_at) AS oldest_unpublished
FROM outbox
WHERE published_at IS NULL;

Ela usa o índice parcial e é barata de rodar a cada poucos segundos. Alerte pela idade, não pela contagem de linhas. Mil linhas não publicadas do último segundo são um minuto movimentado. Uma linha de vinte minutos atrás é um relay travado.

Alternativas, e o que cada uma troca

Change data capture merece um olhar mais atento porque elimina o polling e dá a ordem real de commit, direto do log. O custo é operacional: um replication slot lógico do Postgres que para de ser consumido segura WAL até o disco encher, o que transforma um conector travado numa queda do banco. Capturar as tabelas de negócio direto é tentador porque dispensa o outbox, mas aí todo rename de coluna vira uma quebra no contrato dos seus eventos. CDC numa tabela de outbox mantém o contrato explícito e ainda elimina o relay.

O handler com event sourcing da versão em inglês do meu primeiro post estava mais perto do certo do que parecia. A tabela events dele já é escrita na mesma transação que o estado, porque ela é o estado. Ela já é um outbox. O que faltava era um leitor que controlasse quais eventos foram entregues, e a armadilha do high-water mark vale para uma sequência global de eventos exatamente como vale acima.

No DynamoDB, o equivalente é um TransactWriteItems gravando o item e um item de outbox juntos, ou simplesmente DynamoDB Streams na tabela, que entrega CDC como recurso gerenciado.

Quando você não precisa disso

Se o modelo de leitura vive no mesmo Postgres que o modelo de escrita, a correção mais simples não é um outbox. É atualizar a projeção dentro da transação do comando. Sem broker, sem relay, sem consistência eventual, sem duplicatas. Você abre mão de escalar leituras independentemente das escritas e de manter falhas de projeção fora do caminho do comando, que eram dois dos motivos do CQRS para começo de conversa. Muitos sistemas ainda não precisam de nenhum dos dois, e o post de custos ocultos defendia começar por aí.

Se o evento é só uma dica, como a invalidação de um cache que também tem TTL, perder um de vez em quando pode ser aceitável, e salvar e depois publicar é uma escolha honesta desde que seja deliberada.

E se um job noturno de reconciliação já compara os dois lados e corrige a divergência, você tem um outbox lento e de granularidade grossa. Pode ser suficiente.

O que eu levaria disso

Escrever num banco e depois publicar num broker são duas escritas, e nenhuma quantidade de reordenação, retry ou log as transforma em uma. O comentário "events are persisted" era exatamente o instinto certo apontado para a pergunta errada. Estar persistido não é a questão. A questão é se alguma coisa vai voltar para buscá-los.

O Transactional Outbox Pattern funciona porque se recusa a coordenar dois sistemas: ele torna o evento parte da única escrita que o banco consegue fazer de forma atômica, e transforma publicar num job que pode tentar de novo. O preço é a entrega at-least-once, e isso empurra o problema das duplicatas para os consumidores, onde uma constraint precisa lidar com ele.

Os custos são reais e quase todos operacionais: latência limitada pelo intervalo de polling, carga de escrita numa tabela que incha, uma garantia de ordenação que enfraquece conforme você escala o relay, e um processo cuja falha parece exatamente sucesso até alguém conferir a idade da linha não publicada mais antiga. Antes de escolhê-lo, pergunte se o modelo de leitura não poderia simplesmente ser atualizado na mesma transação. Muitas vezes essa é a resposta sem graça, e sem graça está ótimo.