Idempotent consumers: evitando processamento duplicado em filas

Por que duplicatas são inevitáveis

Em qualquer sistema com at-least-once delivery, duplicatas não são um bug: são uma característica do modelo de garantia. Elas acontecem em cenários comuns:

Consumer processa a mensagem mas morre antes de enviar o ack
Timeout de ack: o broker reenvia antes do consumer confirmar
Consumer restarted durante o processamento
Reprocessamento intencional de mensagens da DLQ

A questão não é eliminar duplicatas no broker, mas garantir que processar a mesma mensagem duas vezes não tenha efeito diferente de processar uma vez.

O padrão: verificar antes de executar

@Injectable()
export class PaymentConsumer {
  @RabbitSubscribe({ queue: "payments.processing" })
  async processPayment(@Payload() message: PaymentMessage) {
    // 1. Verifica se já processamos essa mensagem
    const processed = await this.idempotencyService.isProcessed(message.messageId);
    if (processed) {
      this.logger.log(`Duplicata detectada: ${message.messageId}`);
      return; // retorna sem fazer nada, ack implícito
    }
 
    // 2. Processa com marcação atômica
    await this.db.transaction(async (trx) => {
      await this.paymentService.charge(message, trx);
      await this.idempotencyService.markProcessed(message.messageId, trx);
    });
  }
}

O ponto crítico: markProcessed acontece na mesma transação que o charge. Se qualquer um falhar, ambos são revertidos. Se o consumer morrer após a transação mas antes do ack, o broker reenvia: e a verificação de idempotência detecta a duplicata.

Implementação com banco de dados

CREATE TABLE processed_messages (
  message_id    VARCHAR(255) PRIMARY KEY,
  processed_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  expires_at    TIMESTAMPTZ NOT NULL
);
 
CREATE INDEX ON processed_messages (expires_at);

@Injectable()
export class IdempotencyService {
  constructor(private readonly db: Knex) {}
 
  async isProcessed(messageId: string): Promise<boolean> {
    const row = await this.db("processed_messages")
      .where({ messageId })
      .where("expiresAt", ">", new Date())
      .first();
    return !!row;
  }
 
  async markProcessed(messageId: string, trx: Knex.Transaction): Promise<void> {
    await trx("processed_messages").insert({
      messageId,
      expiresAt: new Date(Date.now() + 7 * 24 * 3600 * 1000), // 7 dias
    });
  }
}

O expires_at evita que a tabela cresça indefinidamente. Use um job periódico para limpar registros expirados:

@Cron("0 3 * * *") // 3h da manhã
async cleanExpiredMessages() {
  await this.db("processed_messages")
    .where("expiresAt", "<", new Date())
    .delete();
}

Implementação com Redis

Para alta vazão, Redis é mais rápido:

@Injectable()
export class RedisIdempotencyService {
  constructor(private readonly redis: Redis) {}
 
  async isProcessed(messageId: string): Promise<boolean> {
    return (await this.redis.exists(`processed:${messageId}`)) === 1;
  }
 
  async markProcessed(messageId: string): Promise<void> {
    await this.redis.setex(`processed:${messageId}`, 7 * 24 * 3600, "1");
  }
}

Limitação: Redis não participa de transações de banco de dados. Para operações que precisam de atomicidade entre o efeito e a marcação, use o banco de dados como store de idempotência.

Idempotência por design: evitando o problema

Às vezes, a melhor estratégia é desenhar as operações para serem naturalmente idempotentes:

// Não idempotente: cria um novo pagamento toda vez
async createPayment(orderId: string, amount: number) {
  return this.db("payments").insert({ orderId, amount });
}
 
// Idempotente: INSERT OR IGNORE: não duplica se já existir
async createPayment(orderId: string, amount: number) {
  return this.db("payments")
    .insert({ orderId, amount })
    .onConflict("orderId") // unique constraint em orderId
    .ignore();             // se já existir, não faz nada e não retorna erro
}

Com ON CONFLICT DO NOTHING, processar o mesmo evento duas vezes resulta em apenas um pagamento no banco: sem verificação explícita de duplicata.

Para updates, upsert com condições:

async updateOrderStatus(orderId: string, status: string, version: number) {
  const updated = await this.db("orders")
    .where({ id: orderId })
    .where("version", "<", version) // só atualiza se for uma versão mais nova
    .update({ status, version });
 
  // Se updated === 0, significa que já temos essa versão ou mais nova
  // Não é erro: é idempotência
}

Deduplicação em múltiplos consumers

Se você tem múltiplos consumers do mesmo tipo rodando em paralelo, dois podem receber a mesma mensagem ao mesmo tempo. O banco de dados resolve com unique constraint:

async markProcessed(messageId: string, trx: Knex.Transaction): Promise<boolean> {
  try {
    await trx("processed_messages").insert({ messageId, expiresAt: /* ... */ });
    return true; // este consumer processou
  } catch (error) {
    if (isUniqueConstraintError(error)) {
      return false; // outro consumer já processou primeiro
    }
    throw error;
  }
}
 
// No consumer:
const shouldProcess = await this.idempotencyService.markProcessed(messageId, trx);
if (!shouldProcess) return; // outro consumer ganhou a corrida

A unique constraint garante que só um consumer executa o efeito, mesmo com múltiplas tentativas simultâneas.

Conclusão

Idempotent consumers são a contrapartida necessária do at-least-once delivery. Sem eles, duplicatas causam cobranças duplas, emails duplicados e inconsistências de dados. Com eles, o sistema se torna robusto a falhas de rede, restarts e reprocessamento.

A implementação ideal combina: verificação antes de executar + marcação atômica com o efeito + unique constraint como última linha de defesa.