Outbox, Sagas e CQRS

Uma compra no BoxOffice envolve três serviços, e cada um é dono dos seus dados: order (o pedido), inventory (os lugares) e payment (a cobrança, por meio de um provedor externo). Em um monólito com um só banco de dados, a compra seria uma transação ACID: reservar os lugares, cobrar, confirmar — tudo ou nada. Dividida em serviços, não há um banco único nem uma transação única. A resposta clássica, uma transação distribuída com two-phase commit (2PC), também não serve: o provedor de pagamento não vai participar da sua transação, e o 2PC bloqueia todos os participantes enquanto um coordenador estiver inacessível.

Esta página reúne os padrões que substituem a transação distribuída em sistemas orientados a eventos. Cada um responde a uma pergunta precisa:

O problema da escrita dupla

O jeito mais simples de publicar OrderConfirmed é salvar o pedido e depois enviar o evento. São duas escritas em dois sistemas diferentes — uma escrita dupla (dual write) — e nenhuma transação cobre as duas:

sequenceDiagram
    participant O as order
    participant DB as banco do order
    participant K as Kafka
    O->>DB: BEGIN · UPDATE orders SET status = 'CONFIRMED' · COMMIT
    Note over O: pod derrubado por um deploy
    O--xK: publica OrderConfirmed (nunca enviado)
    Note over DB,K: o pedido está confirmado, mas ninguém fica sabendo: sem e-mail, sem ingresso, sem analytics

Inverter a ordem das escritas não ajuda:

Transactional outbox

O transactional outbox¹ elimina a escrita dupla gravando o evento no mesmo banco de dados, na mesma transação local, que a mudança de negócio. Um relay separado lê depois a tabela de outbox e publica no broker.

sequenceDiagram
    participant O as order
    participant DB as banco do order
    participant R as relay do outbox
    participant K as Kafka
    O->>DB: BEGIN
    O->>DB: UPDATE orders SET status = 'CONFIRMED'
    O->>DB: INSERT INTO outbox (OrderConfirmed)
    O->>DB: COMMIT (as duas linhas ou nenhuma)
    R->>DB: SELECT linhas não publicadas
    R->>K: publica OrderConfirmed (chave o-7731)
    K-->>R: ack
    R->>DB: marca a linha como publicada

A transação local garante que o pedido e seu evento existam juntos ou não existam. O relay garante que todo evento confirmado seja publicado em algum momento — pelo menos uma vez: se ele cai depois de publicar e antes de marcar a linha, publica a linha de novo ao reiniciar. Os consumidores precisam ser idempotentes de qualquer forma (Semântica de entrega).

A tabela de outbox:

CREATE TABLE outbox (
    id              BIGSERIAL   PRIMARY KEY,           -- publication order
    event_id        UUID        NOT NULL UNIQUE,       -- consumers deduplicate on it
    aggregate_type  TEXT        NOT NULL,              -- 'order'
    aggregate_id    TEXT        NOT NULL,              -- 'o-7731', becomes the Kafka key
    type            TEXT        NOT NULL,              -- 'OrderConfirmed'
    payload         JSONB       NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    published_at    TIMESTAMPTZ
);
CREATE INDEX outbox_unpublished ON outbox (id) WHERE published_at IS NULL;

O código de negócio grava as duas linhas em uma transação:

@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderRepository orders;
    private final OutboxRepository outbox;

    @Transactional
    public void confirm(String orderId) {
        Order order = orders.findByIdForUpdate(orderId);
        order.confirm();                                   // PENDING -> CONFIRMED
        orders.save(order);
        outbox.save(OutboxEvent.of("order", order.id(), "OrderConfirmed",
                                   OrderConfirmed.from(order)));   // same transaction
    }
}

O relay tem duas implementações comuns:

@Scheduled(fixedDelay = 200)
@Transactional
public void relay() {
    List<OutboxRow> rows = jdbc.query("""
        SELECT id, aggregate_id, payload FROM outbox
         WHERE published_at IS NULL
         ORDER BY id
         LIMIT 100
         FOR UPDATE SKIP LOCKED
        """, OutboxRow.MAPPER);
    for (OutboxRow row : rows) {
        kafka.send("order-events", row.aggregateId(), row.payload()).join();  // wait for the ack
        jdbc.update("UPDATE outbox SET published_at = now() WHERE id = ?", row.id());
    }
}

{
  "name": "order-outbox",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "order-db",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "debezium",
    "database.dbname": "order",
    "topic.prefix": "order",
    "plugin.name": "pgoutput",
    "table.include.list": "public.outbox",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.field.event.id": "event_id",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.route.topic.replacement": "${routedByValue}-events"
  }
}

@Transactional
public PlaceResult place(OrderRequest req) {
    int inserted = jdbc.sql("""
            INSERT INTO orders (id, show_id, hold_id, customer_email, amount, status)
            VALUES (?, ?, ?, ?, ?, 'PENDING_PAYMENT')
            ON CONFLICT (hold_id) DO NOTHING
            """)
            .params(UUID.randomUUID(), req.showId(), req.holdId(), req.customerEmail(), req.amount())
            .update();

    OrderView order = jdbc.sql(SELECT + " WHERE hold_id = ?")
            .param(req.holdId())
            .query(OrderView.class)
            .single();

    if (inserted == 0) {
        // same hold already ordered: idempotent replay, no second event
        return new PlaceResult(order, false);
    }

    OrderEvent event = OrderEvent.placed(order);
    if (props.notifyMode() == NotifyMode.SYNC) {
        notifications.send(event); // BEFORE: the fan waits for the e-mail
    } else {
        outbox.append(event.eventId(), "order", order.id(), event.eventType(), event);
    }
    return new PlaceResult(order, true);
}

@Scheduled(fixedDelayString = "${boxoffice.order.relay-interval-ms:200}")
@Transactional
public void relay() throws Exception {
    // SKIP LOCKED: several order replicas can relay in parallel without taking the same rows
    List<Pending> batch = jdbc.sql("""
            SELECT id, aggregate_id, event_type, payload::text AS payload
            FROM outbox
            WHERE published_at IS NULL
            ORDER BY created_at
            LIMIT 100
            FOR UPDATE SKIP LOCKED
            """)
            .query(Pending.class)
            .list();
    if (batch.isEmpty()) {
        return;
    }

    List<CompletableFuture<SendResult<String, String>>> acks = new ArrayList<>();
    for (Pending p : batch) {
        ProducerRecord<String, String> record =
                new ProducerRecord<>(topic, p.aggregateId().toString(), p.payload());
        record.headers().add("eventId", p.id().toString().getBytes(StandardCharsets.UTF_8));
        record.headers().add("eventType", p.eventType().getBytes(StandardCharsets.UTF_8));
        acks.add(kafka.send(record));
    }
    // wait for every ack; on failure the transaction rolls back and the rows are retried
    CompletableFuture.allOf(acks.toArray(new CompletableFuture<?>[0])).get(10, TimeUnit.SECONDS);

    jdbcTemplate.batchUpdate("UPDATE outbox SET published_at = clock_timestamp() WHERE id = ?",
            batch.stream().map(p -> new Object[] {p.id()}).toList());
    log.info("relayed {} event(s) to {}", batch.size(), topic);
}

Change data capture

Todo banco relacional mantém um log de replicação — o write-ahead log (WAL) do PostgreSQL, o binlog do MySQL — que registra cada mudança confirmada na ordem dos commits; é assim que as réplicas ficam sincronizadas (veja Replicação). O change data capture (CDC) lê esse log e transforma cada mudança em um evento². O relay do outbox acima é um uso do CDC; há muitos outros.

flowchart LR
    cat["catalog"] -->|"INSERT / UPDATE de eventos"| db[("banco do catalog<br/>PostgreSQL")]
    db -->|"WAL (logical decoding)"| dbz["Debezium"]
    dbz -->|"catalog.public.events"| k{{"Kafka"}}
    k --> s["indexador da busca"]
    k --> c["invalidador de cache"]
    k --> w["carregador do warehouse"]

No BoxOffice, o CDC sobre o banco do catalog mantém três sistemas derivados sincronizados sem mudar o código do catalog: o índice de busca (shows novos aparecem na busca segundos depois de criados), o cache (entradas são invalidadas quando um evento muda; veja Caching) e o data warehouse.

Sagas

Uma saga é uma transação de negócio implementada como uma sequência de transações locais, uma por serviço; cada passo publica um evento ou uma resposta que dispara o próximo. Se um passo falha, a saga executa transações de compensação que desfazem semanticamente os passos já concluídos³¹. A ideia vem de um artigo de bancos de dados de 1987 sobre transações de longa duração — décadas mais velha que os microsserviços.

A saga de compra do BoxOffice

O pivô é o ponto de não retorno. Antes dele, todo passo pode ser compensado; depois dele, a saga nunca volta atrás — os passos restantes são repetíveis (retriable): precisam ter sucesso em algum momento, com quantos retries forem necessários. É por isso que a reserva dos lugares precisa durar mais que o pagamento: se a reserva pudesse expirar enquanto o cartão é cobrado, o passo 4 poderia falhar de vez depois do pivô e forçar um estorno — uma experiência ruim que o projeto deveria tornar impossível (ou pelo menos rara e monitorada).

Há duas formas de coordenar os passos.

sequenceDiagram
    participant O as order
    participant I as inventory
    participant P as payment
    O->>I: OrderPlaced
    I->>I: reserva os lugares
    I->>P: SeatsHeld
    P->>P: cobra o cartão
    alt pagamento aprovado
        P->>I: PaymentCaptured
        I->>I: marca os lugares como SOLD
        I->>O: SeatsSold
        O->>O: CONFIRMED, publica OrderConfirmed
    else pagamento recusado
        P->>I: PaymentFailed
        I->>I: libera os lugares (compensação)
        I->>O: SeatsReleased
        O->>O: CANCELLED (compensação)
    end

sequenceDiagram
    participant S as saga de compra (order)
    participant I as inventory
    participant P as payment
    S->>I: HoldSeats
    I-->>S: SeatsHeld
    S->>P: ChargeCard (chave o-7731)
    alt cobrado
        P-->>S: PaymentCaptured
        S->>I: MarkSeatsSold
        I-->>S: SeatsSold
        S->>S: CONFIRMED, publica OrderConfirmed
    else recusado
        P-->>S: PaymentFailed
        S->>I: ReleaseSeats (compensação)
        I-->>S: SeatsReleased
        S->>S: CANCELLED
    end

O status do pedido é o estado da saga, então a saga é uma máquina de estados:

stateDiagram-v2
    [*] --> PENDING
    PENDING --> SEATS_HELD : SeatsHeld
    PENDING --> CANCELLED : NoSeats
    SEATS_HELD --> PAID : PaymentCaptured (pivô)
    SEATS_HELD --> RELEASING : PaymentFailed / timeout
    RELEASING --> CANCELLED : SeatsReleased
    PAID --> CONFIRMED : SeatsSold (repetido até dar certo)
    CONFIRMED --> [*]
    CANCELLED --> [*]

Um orquestrador mínimo trata cada resposta em uma transação local, avança a máquina de estados e envia o próximo comando pelo outbox:

@Component
@RequiredArgsConstructor
public class PurchaseSaga {

    private final SagaRepository sagas;
    private final CommandPublisher commands;          // writes commands to the outbox

    @Transactional
    public void handle(SagaReply reply) {
        PurchaseState s = sagas.lock(reply.orderId());  // SELECT ... FOR UPDATE
        if (s.alreadyApplied(reply.replyId())) return;  // duplicate reply: ignore
        switch (s.step()) {
            case HOLDING_SEATS -> {
                if (reply.ok()) { s.moveTo(CHARGING);  commands.send(new ChargeCard(s.orderId(), s.total())); }
                else            { s.moveTo(CANCELLED); commands.send(new CancelOrder(s.orderId())); }
            }
            case CHARGING -> {
                if (reply.ok()) { s.moveTo(CONFIRMING); commands.send(new MarkSeatsSold(s.orderId())); }
                else            { s.moveTo(RELEASING);  commands.send(new ReleaseSeats(s.orderId())); }
            }
            case CONFIRMING -> { s.moveTo(CONFIRMED); commands.send(new ConfirmOrder(s.orderId())); }
            case RELEASING  -> { s.moveTo(CANCELLED); commands.send(new CancelOrder(s.orderId())); }
            default -> throw new IllegalStateException("unexpected reply in " + s.step());
        }
        sagas.save(s.applied(reply.replyId()));
    }
}

Em produção, sagas orquestradas costumam ser escritas sobre um motor de execução durável / workflow — Temporal, AWS Step Functions, Azure Durable Functions, Camunda — que persiste o estado depois de cada passo, cuida de timers ("libere a reserva se não houver pagamento em 10 minutos") e de retries, e sobrevive a reinícios do processo que o executa.

Sagas não são transações

Uma saga é ACD sem o I: ela é atômica (no fim, todos os passos ou todas as compensações), consistente e durável, mas não é isolada. Outras requisições veem estados intermediários — lugares reservados mas não vendidos, um pedido pago mas ainda não confirmado. Projete para isso:

• Lock semântico: o estado intermediário é explícito e respeitado por todos. Lugares em HELD não são oferecidos a outros fãs; a página do pedido diz "processando pagamento".
• Compensações são ações de negócio, não rollbacks. Um estorno é uma nova transação com seu próprio registro na contabilidade (e uma notificação ao fã), não um apagar.
• Todo passo e toda compensação são idempotentes, porque comandos e eventos são entregues pelo menos uma vez.
• Todo passo tem um prazo. Se o provedor de pagamento não responde, não compense às cegas — a cobrança pode ter sido feita. Consulte o provedor (pela chave de idempotência) e reconcilie antes (veja Integrações).

CQRS

Command Query Responsibility Segregation (CQRS) separa o modelo que trata as escritas (comandos) dos modelos que atendem as leituras (consultas)⁴. O modelo de escrita é otimizado para correção e invariantes (tabelas normalizadas, transações, locks); cada modelo de leitura é uma projeção desnormalizada, no formato de uma tela ou consulta, construída a partir dos eventos do lado de escrita.

flowchart LR
    fan["fã"] -->|"POST /orders (comando)"| order["order<br/>modelo de escrita"]
    order --> odb[("banco do order<br/>normalizado")]
    order -->|"outbox: eventos de pedido"| k{{"Kafka"}}
    k --> p1["projetor de meus ingressos"]
    k --> p2["projetor da busca"]
    p1 --> v1[("visão my_tickets<br/>uma linha por ingresso")]
    p2 --> v2[("índice de busca")]
    fan -->|"GET /me/tickets (consulta)"| v1

A página "Meus ingressos" precisa, para um fã, de todos os ingressos com nome do show, data, local, lugar e QR code — dados que pertencem ao order, ao catalog e ao inventory. Fazer join entre três serviços a cada visualização da página é lento e frágil. Em vez disso, um projetor consome os eventos deles e mantém uma tabela my_tickets com exatamente as colunas de que a página precisa. As leituras viram uma única consulta indexada, e o lado de leitura escala de forma independente (réplicas, cache, até um banco de dados diferente).

O preço é a consistência eventual entre escrita e leitura: logo depois de comprar, o fã abre "Meus ingressos" e o ingresso novo ainda não está lá, porque o projetor está algumas centenas de milissegundos atrás. Soluções comuns para esse problema de read-your-writes (veja Consistência):

• Devolver o novo estado na resposta do comando e deixar o cliente exibi-lo imediatamente.
• Passar um token de versão: o cliente envia "preciso de pelo menos a versão 42" e o lado de leitura espera um pouco (ou recorre ao modelo de escrita) até tê-la.
• Ler as escritas recentes do próprio usuário do modelo de escrita e todo o resto da projeção.

Event sourcing

Na maioria dos sistemas, o banco guarda o estado atual e os eventos são um subproduto. O event sourcing inverte isso: a sequência de eventos é a fonte da verdade, e o estado atual é calculado reproduzindo-os⁵.

CREATE TABLE order_events (
    order_id    TEXT        NOT NULL,
    version     INT         NOT NULL,      -- 1, 2, 3... per order
    type        TEXT        NOT NULL,      -- 'OrderPlaced', 'SeatsHeld', ...
    payload     JSONB       NOT NULL,
    occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (order_id, version)        -- two writers cannot both append version 4
);

public record Order(String id, OrderStatus status, List<String> seats, long totalCents) {

    public static Order replay(List<OrderEvent> history) {
        Order order = null;
        for (OrderEvent e : history) {                // oldest first
            order = switch (e) {
                case OrderPlaced p     -> new Order(p.orderId(), PENDING, p.seats(), p.totalCents());
                case SeatsHeld h       -> order.with(SEATS_HELD);
                case PaymentCaptured c -> order.with(PAID);
                case OrderConfirmed c  -> order.with(CONFIRMED);
                case OrderCancelled c  -> order.with(CANCELLED);
            };
        }
        return order;
    }

    Order with(OrderStatus s) { return new Order(id, s, seats, totalCents); }
}

Para mudar um pedido, o serviço carrega seus eventos, reconstrói o estado, verifica as regras de negócio e acrescenta novos eventos com a próxima versão; a chave primária transforma um acréscimo concorrente em conflito (concorrência otimista). Históricos longos são carregados a partir de um snapshot mais os eventos posteriores a ele. Os modelos de leitura são projeções, como no CQRS — por isso os dois padrões costumam aparecer juntos.

Escolhendo os padrões

Como se conecta

flowchart LR
    DW["escrita dupla:<br/>banco + broker"] --> OUT["transactional outbox<br/>uma transação local"]
    OUT --> RELAY["relay: polling ou<br/>Debezium"]
    RELAY --> K["tópico kafka"]
    WAL["WAL / binlog<br/>→ Replicação"] --> CDC["change data capture"]
    CDC --> RELAY
    CDC --> DER["índice de busca, cache,<br/>warehouse"]
    K --> SAGA["saga: transações locais<br/>+ compensações"]
    SAGA --> PIVOT["pivô = pagamento<br/>→ Integrações"]
    SAGA --> ISO["sem isolamento:<br/>semantic locks, prazos"]
    SAGA --> TPC["2PC vs sagas<br/>→ Transações distribuídas"]
    K --> CQRS["projeções CQRS<br/>read-your-writes"]
    ES["event sourcing:<br/>eventos são a verdade"] --> CQRS

Componentes da paleta: sql (linhas de negócio + outbox, WAL para CDC) · kafka · stream · search (índice alimentado por CDC) · cache (invalidação por CDC) · dwh (warehouse alimentado por CDC) · payment (o pivô da saga). Cenários: payment-system, flash-sale. Continua em: Distribuídos — transações (2PC vs sagas), Bancos de dados — replicação (o log que o CDC lê), Caching — invalidação por CDC, Compute — integrações, Estudo de caso: sistema de pagamentos, Hands-on 7 — outbox, Exercício — event-driven.