Idempotência e at-least-once delivery

Em qualquer sistema distribuído, a rede é não confiável. Uma mensagem publicada pode ser entregue zero vezes (o broker caiu), uma vez (caso feliz), ou mais de uma vez (o ACK se perdeu e o broker reenviou). A escolha entre at-most-once, at-least-once e exactly-once não é uma configuração de broker — é um contrato arquitetural com consequências em cada camada do sistema. A grande maioria dos sistemas de produção opera em at-least-once porque exactly-once é proibitivo fora de ambientes controlados, e at-most-once é inaceitável quando perder mensagens tem custo de negócio. A consequência direta de at-least-once é que consumidores precisam ser idempotentes.

As três garantias de entrega

At-most-once: a mensagem é entregue zero ou uma vez. O produtor envia e esquece; se houver falha de rede, a mensagem é perdida. Simples de implementar — sem retry, sem ACK — e de baixíssima latência. Adequado apenas para dados verdadeiramente descartáveis: métricas de baixa prioridade, heartbeats, telemetria onde perder 0,1% das amostras não importa.

At-least-once: a mensagem é entregue uma ou mais vezes. O broker reenvia até receber ACK do consumidor. Se o consumidor processar e morrer antes de dar o ACK, o broker reenvia — o processamento aconteceu duas vezes. É a garantia padrão de RabbitMQ, SQS Standard, Kafka com autocommit desabilitado. O processamento deve ser idempotente.

Exactly-once: a mensagem é processada exatamente uma vez, mesmo com falhas. É a garantia mais cara — requer coordenação entre produtor, broker e consumidor. Kafka suporta exactly-once dentro do ecossistema Kafka (Kafka Streams, produtor idempotente + consumer com transações). Para qualquer saída que não seja Kafka (banco de dados, API externa, email), você volta ao problema do dual write — e está de volta ao at-least-once com idempotência.

Por que at-least-once é inevitável

O problema fundamental é o two generals problem: você não pode ter certeza absoluta que uma mensagem chegou sem uma confirmação, e a confirmação também pode se perder. Em termos práticos:

Cenário 1: mensagem perdida antes da entrega
  Broker → [rede falha] → Consumidor
  Resultado: consumidor nunca recebeu
  Solução: retry → at-least-once

Cenário 2: processamento ok, ACK perdido
  Consumidor processa mensagem
  Consumidor → ACK → [rede falha] → Broker
  Broker: não recebi ACK, reenvia
  Consumidor processa de novo
  Resultado: processamento duplicado

Cenário 3: consumidor morre após processar
  Consumidor processa mensagem
  Consumidor morre (crash, OOM, kill)
  Broker: visibility timeout expirou, reenvia
  Outro consumidor processa de novo
  Resultado: processamento duplicado

Os cenários 2 e 3 são inevitáveis em at-least-once. Não há configuração de broker que os elimine sem abrir mão do at-least-once. A única resposta correta é: aceite que duplicatas vão acontecer e projete os consumidores para que duplicatas sejam inofensivas.

Idempotência — definição formal e prática

Uma operação é idempotente se aplicá-la múltiplas vezes produz o mesmo resultado que aplicá-la uma única vez. Formalmente: f(f(x)) = f(x). Na prática de sistemas distribuídos: processar a mesma mensagem duas vezes deve ter o mesmo efeito que processar uma vez.

Operações naturalmente idempotentes: SET balance = 100 (não importa quantas vezes execute, o saldo fica 100), DELETE FROM sessions WHERE id = 'abc' (segunda execução não encontra nada, sem efeito), INSERT OR IGNORE / UPSERT por chave natural.

Operações que não são idempotentes por natureza: balance = balance + 50 (incremento — segunda execução soma de novo), INSERT INTO orders sem constraint de unicidade (cria duplicata), envio de email (segundo envio dispara email de novo), cobrança de cartão (segundo processamento cobra o cliente duas vezes).

A solução para operações não-idempotentes é introduzir um idempotency key — um identificador único da operação que permite detectar duplicatas e ignorá-las.

Estratégias de idempotência

Idempotency key + registro de processados

Cada mensagem carrega um ID único. O consumidor registra os IDs processados em um store (banco, Redis). Ao receber uma mensagem, verifica se o ID já foi processado antes de executar.

-- Tabela de mensagens processadas (inbox pattern)
CREATE TABLE processed_messages (
    message_id   TEXT PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    result       JSONB        -- opcional: cachear o resultado
);

-- Índice por tempo para limpeza periódica
CREATE INDEX processed_messages_time ON processed_messages (processed_at);

A chave está em verificar E registrar na mesma transação de banco de dados que executa o efeito. Se você verificar fora da transação, duas instâncias concorrentes podem passar na verificação simultaneamente e ambas processarem.

-- Padrão correto: tudo na mesma transação
BEGIN;
  -- Tenta inserir o ID. Falha se já existe (PK violation) → duplicata
  INSERT INTO processed_messages (message_id) VALUES ($1);

  -- Efeito de negócio
  UPDATE accounts SET balance = balance + $2 WHERE id = $3;
COMMIT;

-- Se a INSERT falhar por PK violation, nenhum UPDATE acontece

Conditional update com versão

Em vez de uma tabela separada, use um campo de versão ou timestamp na própria entidade. O update só acontece se a versão do evento for maior que a versão atual — duplicatas chegam com versão menor e são naturalmente ignoradas.

-- UPDATE condicional por versão
UPDATE accounts
SET balance = $new_balance, version = $event_version
WHERE id = $account_id
  AND version < $event_version;

-- Rows affected = 0 → duplicata (versão já foi processada)
-- Rows affected = 1 → processado com sucesso

Esta abordagem é eficiente (sem tabela extra) mas requer que os eventos carreguem um número de sequência monotônico por agregado — o que nem sempre é garantido pelo broker.

Upsert com chave natural

Para operações de criação, use INSERT ON CONFLICT DO NOTHING ou INSERT ON CONFLICT DO UPDATE. A chave natural do negócio (ex: order_id) serve como idempotency key implícita.

-- PostgreSQL: INSERT idempotente
INSERT INTO orders (id, customer_id, total, status)
VALUES ($1, $2, $3, 'placed')
ON CONFLICT (id) DO NOTHING;
-- Segunda execução: não faz nada, não falha, não duplica

Idempotência em APIs HTTP

Em APIs REST, o cliente envia um Idempotency-Key header. O servidor armazena o resultado da primeira requisição e retorna o mesmo resultado para requisições subsequentes com a mesma chave — sem reprocessar.

POST /payments
Idempotency-Key: a1b2c3d4-e5f6-7890-abcd-ef1234567890
Content-Type: application/json

{ "amount": 150.00, "currency": "BRL", "account_id": "acc-123" }

-- Primeira chamada: processa e salva resultado
-- Segunda chamada com mesma Idempotency-Key: retorna resultado salvo
-- HTTP 200 em ambos os casos (ou 201 na primeira, 200 na segunda)

Stripe, PayPal e a maioria das APIs de pagamento implementam idempotency keys. O TTL da chave (quanto tempo o servidor lembra) é tipicamente 24h-7 dias. Clientes devem gerar a chave antes de fazer a requisição, não depois — assim se a requisição timeout, a retry usa a mesma chave.

Deduplicação no nível do broker

Alguns brokers oferecem deduplicação nativa:

SQS FIFO: aceita MessageDeduplicationId (hash do conteúdo ou ID explícito). Mensagens com o mesmo ID dentro de uma janela de 5 minutos são descartadas pelo broker antes de chegar ao consumidor. Reduz a carga de deduplicação na aplicação, mas não elimina — o produtor pode enviar o mesmo ID em janelas diferentes.

Kafka produtor idempotente: com enable.idempotence=true, o produtor atribui sequence numbers a cada mensagem. O broker detecta e descarta retransmissões dentro de uma sessão. Garante exactly-once de publicação — não de processamento pelo consumidor.

RabbitMQ: não tem deduplicação nativa. Cada mensagem tem um message_id que o consumidor pode usar, mas a deduplicação é inteiramente responsabilidade da aplicação.

Timeout e retry com backoff exponencial

Idempotência é necessária, mas não suficiente. O retry precisa de backoff exponencial com jitter para evitar thundering herd — situação onde todas as instâncias de um consumidor tentam retry ao mesmo tempo após uma falha, sobrecarregando o sistema que já estava sob pressão.

// Fórmula de backoff exponencial com jitter
delay = min(base * 2^attempt, max_delay)
delay_with_jitter = delay * (0.5 + random(0, 0.5))

// Exemplo com base=1s, max=60s:
// Attempt 0: 0.5-1.0s
// Attempt 1: 1.0-2.0s
// Attempt 2: 2.0-4.0s
// Attempt 3: 4.0-8.0s
// Attempt 4: 8.0-16.0s
// Attempt 5: 16.0-32.0s
// Attempt 6+: 30-60s (capped)

O jitter distribui os retries no tempo, reduzindo o pico de carga. "Full jitter" (multiplicar por random 0-1) é mais eficaz que "equal jitter" (adicionar random a um base). AWS recomenda full jitter baseado em estudos empíricos com SQS.

Dead Letter Queue na perspectiva de idempotência

Após N retries sem sucesso, a mensagem vai para a DLQ. A DLQ não é um depósito de lixo — é uma fila de mensagens que precisam de intervenção manual ou automática. Toda DLQ deve ter:

Monitoramento: alerta quando há mensagens na DLQ (zero mensagens é o estado normal). Metadados de falha: motivo do erro, número de tentativas, última exceção. Mecanismo de reprocessamento: capacidade de mover mensagens da DLQ de volta para a fila original após corrigir o bug — reprocessamento deve ser manual ou com aprovação, nunca automático sem análise.

Idempotência em chamadas síncronas

At-least-once e idempotência não são exclusivos de mensageria. Toda chamada HTTP com retry tem o mesmo problema. Um POST /charges com retry automático (por timeout ou 5xx) pode criar duas cobranças. As mesmas estratégias se aplicam: idempotency key, conditional update, upsert.

O padrão de retry em HTTP deve sempre checar se a operação é idempotente antes de retentar. GET é sempre seguro para retry. PUT e DELETE são idempotentes por definição REST. POST não é — requer idempotency key explícita ou lógica de deduplicação na aplicação.

Comparativo entre linguagens — consumidor idempotente

// C# — consumidor idempotente com MassTransit e EF Core
// MassTransit Inbox (InMemory ou persistido) + retries com backoff

public class OrderPlacedConsumer(AppDbContext db, ILogger<OrderPlacedConsumer> log)
    : IConsumer<OrderPlaced> {

    public async Task Consume(ConsumeContext<OrderPlaced> context) {
        var msg = context.Message;

        // Tentativa de inserção idempotente na tabela inbox
        // MassTransit faz isso automaticamente com .UseInbox()
        // Aqui, exemplo manual com EF Core:
        var alreadyProcessed = await db.ProcessedMessages
            .AnyAsync(m => m.MessageId == context.MessageId.ToString());

        if (alreadyProcessed) {
            log.LogInformation("Mensagem {Id} já processada, ignorando", context.MessageId);
            return; // não precisa rejeitar — ACK normalmente
        }

        // Efeito de negócio + registro do inbox na mesma transação
        await using var tx = await db.Database.BeginTransactionAsync();

        try {
            await db.ProcessedMessages.AddAsync(new ProcessedMessage {
                MessageId = context.MessageId.ToString(),
                ProcessedAt = DateTimeOffset.UtcNow,
            });

            await db.InventoryReservations.AddAsync(new InventoryReservation {
                OrderId = msg.OrderId,
                Items = msg.Items,
                ReservedAt = DateTimeOffset.UtcNow,
            });

            await db.SaveChangesAsync();
            await tx.CommitAsync();
        } catch (DbUpdateException ex) when (ex.IsUniqueConstraintViolation()) {
            // Race condition: outra instância processou ao mesmo tempo
            await tx.RollbackAsync();
            log.LogWarning("Race na deduplicação de {Id}", context.MessageId);
            // ACK mesmo assim — foi processado pela outra instância
        }
    }
}

// Configuração de retry com backoff (no Program.cs)
services.AddMassTransit(x => {
    x.AddConsumer<OrderPlacedConsumer>();
    x.UsingRabbitMq((ctx, cfg) => {
        cfg.UseMessageRetry(r => r
            .Exponential(5,              // máximo de 5 tentativas
                TimeSpan.FromSeconds(1), // delay inicial
                TimeSpan.FromSeconds(60), // delay máximo
                TimeSpan.FromSeconds(1)  // delta para variação (jitter)
            )
            .Ignore<ArgumentException>()  // não retentar erros de validação
        );
        cfg.UseInMemoryOutbox(ctx);
        cfg.ConfigureEndpoints(ctx);
    });
});

# Python — consumidor idempotente com aio-pika e SQLAlchemy
# Deduplicação via tabela processed_messages com constraint UNIQUE

import asyncio
import logging
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
import aio_pika

log = logging.getLogger(__name__)

async def handle_order_placed(
    message: aio_pika.IncomingMessage,
    session_factory,
) -> None:
    message_id = message.message_id
    if not message_id:
        # Mensagens sem ID não podem ser deduplicadas — falha explícita
        raise ValueError("Mensagem sem message_id — impossível garantir idempotência")

    async with message.process(requeue=True):  # requeue em exceção não tratada
        async with session_factory() as session:
            async with session.begin():
                try:
                    # INSERT na tabela de deduplicação
                    # Falha com IntegrityError se message_id já existe (PK violation)
                    await session.execute(
                        text("INSERT INTO processed_messages (message_id, processed_at) VALUES (:id, NOW())"),
                        {"id": message_id}
                    )

                    # Efeito de negócio na mesma transação
                    payload = json.loads(message.body)
                    await reserve_inventory(session, payload["order_id"], payload["items"])

                    # Commit implícito no fim do context manager session.begin()

                except IntegrityError:
                    # PK violation → mensagem já foi processada
                    log.info("Mensagem %s já processada, ignorando", message_id)
                    await session.rollback()
                    # Sai sem requeue — mensagem será ACKada normalmente


# Retry com backoff exponencial (implementação manual)
async def consume_with_retry(
    channel: aio_pika.Channel,
    queue_name: str,
    session_factory,
    max_retries: int = 5,
) -> None:
    queue = await channel.declare_queue(queue_name, durable=True)

    async def on_message(msg: aio_pika.IncomingMessage) -> None:
        retry_count = int(msg.headers.get("x-retry-count", 0))

        try:
            await handle_order_placed(msg, session_factory)
        except Exception as e:
            if retry_count >= max_retries:
                log.error("Mensagem %s excedeu %d retries, enviando para DLQ", msg.message_id, max_retries)
                await send_to_dlq(channel, msg, str(e))
                async with msg.process():
                    pass  # ACK para remover da fila principal
                return

            # Backoff exponencial com jitter
            delay = min(2 ** retry_count, 60) * (0.5 + asyncio.get_event_loop().time() % 0.5)
            log.warning("Retry %d em %.1fs para %s", retry_count + 1, delay, msg.message_id)
            await asyncio.sleep(delay)

            # Republica com contador de retry incrementado
            await channel.default_exchange.publish(
                aio_pika.Message(
                    body=msg.body,
                    headers={**msg.headers, "x-retry-count": retry_count + 1},
                    message_id=msg.message_id,
                    delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
                ),
                routing_key=queue_name,
            )
            async with msg.process():
                pass  # ACK original após republicar

    await queue.consume(on_message)

// Go — consumidor idempotente com amqp091 e pgx
// Deduplicação via INSERT ON CONFLICT + retry com backoff

package consumer

import (
    "context"
    "encoding/json"
    "fmt"
    "math"
    "math/rand"
    "time"

    amqp "github.com/rabbitmq/amqp091-go"
    "github.com/jackc/pgx/v5/pgxpool"
    "log/slog"
)

type OrderPlacedEvent struct {
    OrderID    string   `json:"order_id"`
    CustomerID string   `json:"customer_id"`
    Items      []string `json:"items"`
}

func ConsumeWithIdempotency(ctx context.Context, ch *amqp.Channel, pool *pgxpool.Pool) error {
    msgs, err := ch.Consume("orders.placed", "", false, false, false, false, nil)
    if err != nil {
        return fmt.Errorf("consume: %w", err)
    }

    for {
        select {
        case <-ctx.Done():
            return nil
        case msg, ok := <-msgs:
            if !ok {
                return fmt.Errorf("canal fechado")
            }
            processWithRetry(ctx, pool, msg, 5)
        }
    }
}

func processWithRetry(ctx context.Context, pool *pgxpool.Pool, msg amqp.Delivery, maxRetries int) {
    messageID := msg.MessageId
    if messageID == "" {
        slog.Error("mensagem sem MessageId — rejeitando permanentemente")
        _ = msg.Reject(false) // false = não requeue
        return
    }

    var lastErr error
    for attempt := 0; attempt <= maxRetries; attempt++ {
        if attempt > 0 {
            delay := backoffWithJitter(attempt)
            slog.Info("retry", "attempt", attempt, "delay", delay, "message_id", messageID)
            select {
            case <-ctx.Done():
                return
            case <-time.After(delay):
            }
        }

        lastErr = processIdempotent(ctx, pool, msg)
        if lastErr == nil {
            _ = msg.Ack(false)
            return
        }

        slog.Error("falha no processamento", "attempt", attempt, "err", lastErr)
    }

    slog.Error("max retries excedido, rejeitando para DLQ", "message_id", messageID)
    _ = msg.Reject(false) // vai para DLQ se configurada no broker
}

func processIdempotent(ctx context.Context, pool *pgxpool.Pool, msg amqp.Delivery) error {
    var event OrderPlacedEvent
    if err := json.Unmarshal(msg.Body, &event); err != nil {
        return fmt.Errorf("unmarshal: %w", err) // não retentável
    }

    tx, err := pool.Begin(ctx)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback(ctx)

    // Deduplicação: INSERT ON CONFLICT — atômico com o efeito de negócio
    tag, err := tx.Exec(ctx,
        `INSERT INTO processed_messages (message_id, processed_at)
         VALUES ($1, NOW())
         ON CONFLICT (message_id) DO NOTHING`,
        msg.MessageId,
    )
    if err != nil {
        return fmt.Errorf("dedup insert: %w", err)
    }

    if tag.RowsAffected() == 0 {
        // Já processado — commit vazio e sai
        slog.Info("mensagem já processada", "message_id", msg.MessageId)
        return tx.Commit(ctx)
    }

    // Efeito de negócio
    _, err = tx.Exec(ctx,
        `INSERT INTO inventory_reservations (order_id, reserved_at)
         VALUES ($1, NOW())
         ON CONFLICT (order_id) DO NOTHING`,
        event.OrderID,
    )
    if err != nil {
        return fmt.Errorf("reservar estoque: %w", err)
    }

    return tx.Commit(ctx)
}

func backoffWithJitter(attempt int) time.Duration {
    base := math.Pow(2, float64(attempt))
    capped := math.Min(base, 60) // máximo 60s
    jitter := capped * (0.5 + rand.Float64()*0.5)
    return time.Duration(jitter * float64(time.Second))
}

Exercícios práticos

1. Crie um consumidor deliberadamente não-idempotente (incremento de saldo) e demonstre a duplicação: publique a mesma mensagem 3 vezes e observe o saldo final incorreto. Em seguida, corrija com idempotency key + tabela de processados e repita o experimento.
   Critério: sem idempotência, saldo final é 3× o esperado; com idempotência, saldo é exatamente 1× mesmo com 3 entregas; o fix deve usar constraint de unicidade no banco, não checagem em memória.
2. Implemente o padrão de Idempotency-Key em uma API REST de pagamento: gere a chave no cliente antes de chamar, armazene o resultado por 24h, e retorne o resultado cacheado em chamadas subsequentes com a mesma chave. Valide que nenhuma cobrança dupla ocorre mesmo com retries agressivos.
   Critério: 100 chamadas paralelas com a mesma chave resultam em exatamente 1 cobrança; a resposta das 99 repetições é idêntica à resposta original (mesmo status code, mesmo body).
3. Configure o SQS FIFO com MessageDeduplicationId baseado no hash SHA-256 do conteúdo. Publique 10 mensagens idênticas em sequência e verifique que o consumidor recebe apenas uma. Em seguida, modifique um campo e observe que é tratada como nova mensagem. Teste o comportamento na janela de deduplicação de 5 minutos.
   Critério: 10 mensagens idênticas → 1 entrega ao consumidor; após 5 minutos, a mesma mensagem é entregue novamente; mensagem com campo diferente gera entrega adicional.
4. Implemente backoff exponencial com full jitter e meça a distribuição dos retries: com 100 clientes simultâneos falhando ao mesmo tempo, compare o pico de QPS no servidor com jitter vs sem jitter. Use um histograma para visualizar a diferença.
   Critério: histograma mostra distribuição uniforme com jitter vs spike concentrado sem jitter; pico de QPS com jitter é pelo menos 10× menor que sem jitter no cenário de 100 clientes simultâneos.
5. Simule uma DLQ real: configure o RabbitMQ com x-dead-letter-exchange, faça um consumidor falhar intencionalmente após 3 tentativas, inspecione as mensagens na DLQ incluindo os headers de rastreamento (x-death), e implemente um consumer de DLQ que alerta e registra os erros com contexto suficiente para diagnóstico.
   Critério: mensagem aparece na DLQ após exatamente 3 tentativas; o consumer de DLQ registra queue de origem, timestamp de cada tentativa, e motivo da falha; um alerta é disparado quando DLQ ultrapassa 10 mensagens.