Transactional outbox no Laravel: como publicar eventos sem perder nenhum quando a fila cai

O padrão transactional outbox resolve um problema específico: garantir que, quando uma mudança de estado é gravada no banco, o evento correspondente seja publicado na fila em algum momento, mesmo que o Redis caia ou o processo morra logo depois do commit. A ideia é gravar o evento numa tabela do próprio banco, na mesma transação da regra de negócio, e deixar um processo separado (o relay) ler essa tabela e publicar na fila. Você troca "exatamente uma vez", que não existe, por "pelo menos uma vez" com consumidores idempotentes.

Em sistema financeiro isso não é detalhe. Um pagamento marcado como pago sem o evento payment.paid publicado significa lojista sem notificação, saldo sem atualização num serviço vizinho e conciliação acusando divergência dias depois. Vou mostrar como implemento esse padrão em Laravel com PostgreSQL e quais trade-offs aparecem no caminho.

Por que afterCommit não resolve a perda de eventos?

O primeiro passo que quase todo mundo dá é mover o dispatch para depois do commit. Isso resolve o bug clássico do job que roda antes de os dados existirem, que detalhei no post sobre jobs disparados dentro de transação e o afterCommit no Laravel. Mas não resolve o problema de fundo, que tem nome: dual write, ou escrita dupla, quando uma operação precisa alterar dois sistemas diferentes (PostgreSQL e Redis) sem uma transação que englobe os dois.

Com afterCommit, a sequência é:

1. A transação no PostgreSQL faz commit. O pagamento está pago.
2. O Laravel tenta empurrar o job para o Redis.
3. Se o Redis estiver fora, se a conexão der timeout ou se o worker PHP-FPM for morto nesse intervalo, o job some.

Não existe exceção para você tratar no passo 3 que traga o estado de volta: o commit já aconteceu. Você pode fazer try/catch e logar, mas aí depende de alguém ler o log e reprocessar na mão. O inverso (publicar antes do commit) é pior: o evento sai e a transação pode dar rollback, e aí você notifica um pagamento que não existe.

A saída é parar de escrever em dois lugares. Você escreve só no PostgreSQL, numa única transação, e deixa a publicação para depois.

O que é o padrão transactional outbox?

É uma tabela, a outbox, que funciona como caixa de saída de mensagens. A regra de negócio insere ali uma linha descrevendo o evento, dentro da mesma transação que altera o estado. Como é tudo no mesmo banco, ou as duas coisas são gravadas ou nenhuma é.

Um processo separado, o relay (retransmissor), varre a outbox, publica cada mensagem pendente na fila e marca como publicada. Se o relay cair no meio, ele recomeça de onde parou, porque o estado "pendente" está persistido no banco.

O custo é explícito: uma mensagem pode ser publicada mais de uma vez (o relay publica, morre antes de marcar, e publica de novo ao voltar). Por isso o padrão só fica completo com consumidores idempotentes, ou seja, que produzem o mesmo resultado se receberem a mesma mensagem duas vezes.

Como modelar a tabela de outbox no PostgreSQL?

Schema::create('outbox_messages', function (Blueprint $table) {
    $table->id();
    $table->uuid('message_id')->unique();
    $table->string('aggregate_type');
    $table->string('aggregate_id');
    $table->string('event_type');
    $table->jsonb('payload');
    $table->unsignedSmallInteger('attempts')->default(0);
    $table->text('last_error')->nullable();
    $table->timestampTz('created_at')->useCurrent();
    $table->timestampTz('published_at')->nullable();
});

DB::statement(
    'CREATE INDEX outbox_pending_idx ON outbox_messages (id) WHERE published_at IS NULL'
);

Algumas decisões aqui:

• message_id em UUID: é a identidade da mensagem que viaja até o consumidor e vira chave de idempotência. O id sequencial é só para ordenar e paginar internamente.
• aggregate_type e aggregate_id: identificam a entidade de origem (por exemplo, payment e o id do pagamento). Ajudam a depurar e permitem raciocinar sobre ordem por entidade.
• Índice parcial: o relay só consulta linhas com published_at IS NULL. Com o índice parcial, a consulta continua rápida mesmo com milhões de linhas já publicadas, porque o índice só contém as pendentes.
• payload com dados suficientes: prefiro mandar o necessário para o consumidor agir, em vez de só o id. Se o consumidor tiver que ler o banco de novo, ele pode ver um estado mais novo que o do evento, o que confunde quando eventos chegam fora de ordem.

Como gravar o evento na mesma transação da regra de negócio?

Uma classe pequena, com uma trava que impede o uso fora de transação, porque é exatamente aí que o padrão quebra sem ninguém perceber:

final class Outbox
{
    public static function record(
        string $aggregateType,
        string $aggregateId,
        string $eventType,
        array $payload,
    ): void {
        if (DB::transactionLevel() === 0) {
            throw new LogicException('Outbox::record precisa rodar dentro de uma transação.');
        }

        DB::table('outbox_messages')->insert([
            'message_id' => (string) Str::uuid(),
            'aggregate_type' => $aggregateType,
            'aggregate_id' => $aggregateId,
            'event_type' => $eventType,
            'payload' => json_encode($payload, JSON_THROW_ON_ERROR),
        ]);
    }
}

No serviço que confirma o pagamento, o evento entra junto com a mudança de status e os lançamentos contábeis. Se você usa um ledger como o que descrevi em ledger de partidas dobradas no Laravel e PostgreSQL, é aqui que as três coisas ficam atômicas:

DB::transaction(function () use ($payment, $paidAt) {
    $payment->markAsPaid($paidAt);

    $this->ledger->post(PaymentCaptured::entriesFor($payment));

    Outbox::record('payment', (string) $payment->id, 'payment.paid', [
        'payment_id' => $payment->id,
        'merchant_id' => $payment->merchant_id,
        'amount' => $payment->amount, // em centavos
        'paid_at' => $paidAt->toIso8601String(),
    ]);
});

Repare que não existe nenhum dispatch aqui. A regra de negócio não sabe que fila existe.

Como implementar o relay com FOR UPDATE SKIP LOCKED?

SKIP LOCKED é uma cláusula do PostgreSQL que faz o SELECT ... FOR UPDATE pular linhas já travadas por outra transação, em vez de esperar por elas. Com isso você pode rodar mais de uma instância do relay sem que duas peguem a mesma mensagem.

final class RelayOutbox extends Command
{
    protected $signature = 'outbox:relay {--batch=100} {--max-attempts=10}';

    private bool $stopping = false;

    public function handle(OutboxPublisher $publisher): int
    {
        $this->trap([SIGTERM, SIGINT], fn () => $this->stopping = true);
        $startedAt = now();

        while (! $this->stopping && $startedAt->diffInMinutes(now()) < 30) {
            $processed = DB::transaction(fn () => $this->relayBatch($publisher));

            if ($processed === 0) {
                usleep(500_000);
            }
        }

        return self::SUCCESS;
    }

    private function relayBatch(OutboxPublisher $publisher): int
    {
        $messages = DB::table('outbox_messages')
            ->whereNull('published_at')
            ->where('attempts', '<', (int) $this->option('max-attempts'))
            ->orderBy('id')
            ->limit((int) $this->option('batch'))
            ->lock('for update skip locked')
            ->get();

        foreach ($messages as $message) {
            try {
                $publisher->publish($message);

                DB::table('outbox_messages')->where('id', $message->id)
                    ->update(['published_at' => now()]);
            } catch (Throwable $e) {
                DB::table('outbox_messages')->where('id', $message->id)->update([
                    'attempts' => $message->attempts + 1,
                    'last_error' => mb_substr($e->getMessage(), 0, 1000),
                ]);
            }
        }

        return $messages->count();
    }
}

E o publisher, que mapeia tipo de evento para jobs:

final class OutboxPublisher
{
    private const HANDLERS = [
        'payment.paid' => [NotifyMerchantOfPayment::class, SyncPaymentToAnalytics::class],
    ];

    public function publish(object $message): void
    {
        $payload = json_decode($message->payload, true, flags: JSON_THROW_ON_ERROR);

        foreach (self::HANDLERS[$message->event_type] ?? [] as $job) {
            $job::dispatch($message->message_id, $payload)->beforeCommit();
        }
    }
}

Dois detalhes que mordem:

1. O beforeCommit() é proposital. Se a sua conexão de fila está com after_commit => true (e deveria estar, pelo motivo do post sobre afterCommit), o dispatch feito dentro da transação do relay seria adiado para depois do commit. Aí o relay marca published_at, faz commit, e só então tenta o Redis. Se o Redis falhar nesse momento, você recriou exatamente o problema que a outbox existe para resolver. No relay, o push tem que acontecer antes do commit, para que uma falha impeça a marcação.
2. O processo reinicia sozinho a cada 30 minutos. Comando PHP de longa duração acumula memória e conexões. Rodando sob Supervisor, ele sai limpo, respeita SIGTERM no deploy e volta em seguida.

O limite de attempts evita que uma mensagem envenenada (payload inválido, evento sem handler que lança exceção) trave o fluxo para sempre. Ela fica parada com last_error preenchido, e o alerta que mostro adiante avisa.

Como o consumidor lida com mensagens duplicadas?

O relay garante entrega pelo menos uma vez, então o consumidor precisa deduplicar. O mesmo raciocínio dos webhooks de pagamento duplicados e fora de ordem vale aqui, com a vantagem de que agora você controla o identificador:

public function handle(): void
{
    DB::transaction(function () {
        $inserted = DB::table('processed_messages')->insertOrIgnore([
            'message_id' => $this->messageId,
            'consumer' => static::class,
            'processed_at' => now(),
        ]);

        if ($inserted === 0) {
            return; // já processada por este consumidor
        }

        // efeito colateral no banco aqui, na mesma transação
    });
}

A tabela processed_messages tem chave única em (message_id, consumer). O par importa: a mesma mensagem é legítima para consumidores diferentes.

Quando o efeito colateral é uma chamada HTTP externa (notificar o lojista, mandar mensagem no WhatsApp), a transação local não protege nada. Nesse caso, repasse o message_id como chave de idempotência para o serviço de destino, se ele aceitar, ou registre a tentativa antes e o resultado depois, aceitando que a duplicata é possível no pior caso.

A outbox garante a ordem dos eventos?

Não de forma global, e é bom não prometer isso. Três motivos:

1. Sequência não é ordem de commit. Duas transações concorrentes pegam ids 10 e 11; a do 11 faz commit primeiro. Por isso o relay usa published_at IS NULL e não um cursor do tipo "último id visto": com cursor, o 10 seria pulado para sempre.
2. Vários relays e vários workers processam em paralelo.
3. Retries do Horizon recolocam um job falho no fim da fila.

Na prática, o que importa é a ordem por entidade. Duas saídas que uso:

• Mandar no payload uma versão ou timestamp do agregado e fazer o consumidor ignorar eventos mais velhos que o estado que ele já tem.
• Para casos que realmente exigem sequência estrita, rotear eventos de um mesmo agregado para uma fila dedicada com um único worker. Funciona, mas limita a vazão, então reservo para poucos fluxos.

Se você já tem uma máquina de estados para o status de pagamento, ela vira uma defesa natural: um payment.refunded processado antes do payment.paid vira uma transição inválida, rejeitada e reprocessada depois.

Polling, CDC ou afterCommit: qual escolher?

CDC (change data capture) é ler as mudanças direto do log de replicação do banco, com ferramentas como Debezium usando a replicação lógica do PostgreSQL, em vez de consultar a tabela.

Critério	afterCommit puro	Outbox com polling	Outbox com CDC
Perde evento se a fila cair	Sim	Não	Não
Latência típica	Imediata	Intervalo do polling	Baixa
Infra extra	Nenhuma	Um processo no Supervisor	Debezium, Kafka ou similar
Carga no banco	Nenhuma	Consultas frequentes (leves com índice parcial)	Leitura do WAL
Complexidade operacional	Baixa	Baixa	Alta

Para uma equipe pequena, polling com SKIP LOCKED cobre a grande maioria dos casos. CDC passa a fazer sentido quando o volume é alto o bastante para o polling pesar no banco ou quando você já opera Kafka por outros motivos.

Como monitorar e limpar a tabela de outbox?

A outbox cresce sem parar e esconde falhas se ninguém olhar. Eu agendo duas coisas:

Schedule::call(function () {
    $oldestPending = DB::table('outbox_messages')
        ->whereNull('published_at')
        ->min('created_at');

    if ($oldestPending && now()->diffInMinutes($oldestPending, true) > 5) {
        Log::critical('Outbox com mensagens pendentes há mais de 5 minutos');
    }
})->everyMinute();

Schedule::call(function () {
    DB::table('outbox_messages')
        ->whereNotNull('published_at')
        ->where('published_at', '<', now()->subDays(15))
        ->delete();
})->daily();

A métrica que importa é a idade da mensagem pendente mais antiga, não a quantidade. Mil pendentes há dois segundos é pico de tráfego; uma pendente há dez minutos é relay parado ou mensagem envenenada. Mensagens que atingiram o limite de tentativas entram nessa mesma conta e disparam o alerta.

Se o volume for grande, troque o delete por particionamento da tabela por data e descarte partições antigas, que é bem mais barato para o PostgreSQL do que apagar linhas em lote.

Quando não vale a pena usar outbox?

Nem todo job precisa disso. Gerar thumbnail, mandar e-mail de boas-vindas, recalcular um relatório: se perder, o usuário reclama ou um cron refaz. Para esses, afterCommit resolve. Eu reservo a outbox para eventos cuja perda gera inconsistência entre sistemas e que ninguém vai perceber até a conciliação: mudança de status de pagamento, movimentação de saldo, confirmação de estorno, liberação de recebíveis.

Conclusão prática

Se você tem um fluxo em que "gravou no banco mas o evento não saiu" significa dinheiro inconsistente, comece pequeno: crie a tabela outbox_messages com o índice parcial, troque o dispatch desse único fluxo por Outbox::record dentro da transação, suba o comando outbox:relay no Supervisor e adicione a tabela processed_messages nos consumidores dele. Coloque o alerta de idade da mensagem pendente no mesmo dia, porque é ele que transforma uma falha silenciosa em algo que você vê. Depois que esse fluxo rodar estável, migrar os outros eventos críticos é repetir o mesmo padrão.