Três duplicatas, três causas, um sintoma

Três bugs numa semana

Na última semana de agosto olhei números duplicados em três sistemas diferentes. Dois deles moram no pipeline que leva os logs de exibição das telas até o ClickHouse passando pelo Kafka. O terceiro está numa API de e-commerce que envia pedidos pagos para um sistema comercial. Times diferentes, linguagens diferentes, código de anos diferentes. Só percebi que era o mesmo problema quando escrevi os três lado a lado.

1. Dois consumers que nunca estiveram no grupo

Nosso job de importação é simples. Um cron sobe um worker em Python, o worker lê um lote do Kafka e insere no ClickHouse. Com subscribe(), o consumer entra no grupo, o coordenador divide as partições e cada um de dois workers fica com uma metade. Com assign(), você escolhe as partições na mão e o coordenador sai de cena. O worker continua commitando offsets com o group id, mas nunca é membro do grupo.

from confluent_kafka import Consumer, TopicPartition

c = Consumer({
    "bootstrap.servers": BROKERS,
    "group.id": "audit-import",
})

# o que o worker fazia: pega as 6 partições e começa do offset commitado
c.assign([TopicPartition("playlog", p) for p in range(6)])

# o que dois workers precisam: deixar o coordenador dividir as partições
c.subscribe(["playlog"])

Com um worker isso não faz mal nenhum, e é justamente o que torna o bug perigoso. Quando um segundo worker sobe, os dois pegam as seis partições a partir do mesmo offset commitado e toda mensagem entra duas vezes. A única dedup do worker era um set() do Python dentro do lote, que não enxerga o outro processo, então o fator no insert deu exatamente 2,0. O crontab até tentava disparar os dois em momentos diferentes, com passos */2+1 e */2+2. Essa sintaxe não é válida no cron que usamos, e ela não dá erro: a parte do deslocamento é ignorada e os dois workers dispararam no mesmo segundo, como o syslog mostrou.

Se você olhar, o próprio grupo conta: os offsets commitados andam a cada 2 minutos e o grupo tem 0 membros.

$ kafka-consumer-groups.sh --bootstrap-server $BROKERS --describe --group audit-import --state

GROUP          COORDINATOR (ID)   ASSIGNMENT-STRATEGY   STATE   #MEMBERS
audit-import   broker-2 (2)                             Empty   0

Empty, sem membros e sem estratégia de atribuição, enquanto o lag continua pequeno: é assim que o assign() aparece do lado do broker quando você descreve o grupo.

Havia uma segunda coisa na mesma função. O worker chamava consumer.close() num finally antes do insert, e com auto-commit isso grava o offset primeiro. Se o insert falha, o lote some. Commitar depois da escrita é a ordem que eu usaria em qualquer consumer, mesmo que um insert com falha faça o mesmo lote voltar e a dedup tenha que dar conta dele:

# consumer criado com "enable.auto.commit": False
msgs = c.consume(num_messages=30_000, timeout=10)
rows = [parse(m) for m in msgs if not m.error()]
clickhouse.insert("playlog", rows)      # se isto falhar, nada é commitado
c.commit(asynchronous=False)            # offset só depois do dado gravado

2. Um sumarizado que lia antes do merge

Na tabela crua, um ReplacingMergeTree, duplicatas com a mesma chave de ordenação colapsam quando as partes se fundem, então a tabela crua se cura sozinha, em algum momento. Em cima dela um sumarizador roda a cada 10 minutos, conta linhas por aparelho, campanha e dia, e insere num SummingMergeTree. Esse engine só soma. A contagem dobrada que entra ali fica, porque o SummingMergeTree não tem como saber que duas das suas entradas eram o mesmo evento.

Quão eventual é o eventual? Medi nesta tabela em agosto, depois de um período em que toda linha era gravada 2 vezes:

Nosso sumarizador rodava bem na linha dos 10 minutos. Ele usava count() sobre a tabela crua sem FINAL, então contava linhas físicas. Para medir isso na sua tabela, compare linhas com chaves distintas numa janela recente de 10 minutos:

SELECT
    count()                          AS rows,
    uniqExact(device_id, event_id)   AS keys,
    round(rows / keys, 4)            AS dup_factor
FROM playlog
WHERE inserted_at >= now() - INTERVAL 10 MINUTE;

Eu não quis acreditar no meu próprio número, então ele passou por três análises independentes, uma delas com a tarefa explícita de refutá-lo. Duas checagens fecharam a questão. O grupo de controle: outro ambiente roda o mesmo código com um worker só, e o fator dele era 1,0000. A fonte externa: a soma da tabela sumarizada batia com as linhas que o próprio ClickHouse diz ter inserido, pelo system.part_log, numa razão de 0,99994 para um mês inteiro. Ou seja, o sumarizado contava linhas gravadas, e a partir do dia em que o segundo worker subiu as linhas gravadas eram o dobro.

SELECT sum(rows)
FROM system.part_log
WHERE event_type = 'NewPart'
  AND database = 'audit' AND table = 'playlog'
  AND event_date BETWEEN '2026-01-01' AND '2026-01-31';

Para o sumarizado, qualquer uma destas serve: ler com FINAL (custa CPU e dá o número certo), sumarizar só janelas mais antigas que o tempo que os merges levam, ou agrupar pela chave de dedup antes de contar. A primeira é a mais simples de revisar e é por onde eu começaria, porque o custo aparece no query log e dá para medir:

INSERT INTO playlog_summary
SELECT device_id, campaign_id, toDate(ts) AS day, count() AS total
FROM playlog FINAL
WHERE inserted_at >= {from:DateTime} AND inserted_at < {to:DateTime}
GROUP BY device_id, campaign_id, day;

Tirar o segundo worker estancou o problema no mesmo dia, e ressumarizamos os meses anteriores a partir do histórico cru. Um detalhe que quase me escapou: sem o segundo worker o fator voltou para perto de 1,10, e não para 1,00. Havia uma fonte de duplicatas mais antiga e menor, que é outra história. Se eu tivesse declarado vitória no primeiro número, o problema mais antigo continuaria no sumarizado até hoje sem ninguém procurando por ele.

3. Dois webhooks para um pagamento

O gateway de pagamento manda order.paid e charge.paid quase no mesmo instante. Os dois handlers faziam a coisa razoável e enfileiravam "integrar o pedido X". O sistema comercial recebia o mesmo pedido duas vezes. A correção no BullMQ 5 é uma opção no add():

await this.queue.add(
  "integrate-order",
  { orderId },
  { deduplication: { id: `order-${orderId}` } },
);

A minha primeira versão tinha ttl: 60_000. Na revisão do mesmo dia tirei o TTL. Sem ele, o BullMQ guarda a chave enquanto existir um job esperando ou rodando para aquele pedido e a libera quando o job termina; assim uma reintegração manual mais tarde continua possível, e não existe número mágico para a distância entre dois webhooks. Isso só protege o enfileiramento. O lado que recebe ainda deve tratar o id do pedido como chave de idempotência, porque um retry depois de um timeout pode chegar de qualquer lugar. (Foi a mesma semana do sync travado de que falei em query_timeout não cancela nada.)

O mesmo formato

Processamento duplicado quase nunca é o mesmo bug duas vezes. Quase sempre é a mesma peça faltando: uma chave que diz "isso já aconteceu", e cada camada conferindo essa chave. Em todos os casos alguma camada confiava que outra fazia a dedup. A tabela crua tinha dedup, em algum momento. O worker tinha um set(), por processo, e a fila não tinha nada até aquela semana de agosto.

Nenhum desses foi descuido. Cada peça era razoável sozinha, e bugs como o do assign() só acordam quando você escala, o que normalmente acontece quando há backlog e ninguém está olhando consumer group. A pergunta que faço agora para cada camada é o que acontece se ela rodar duas vezes, e o teste mais barato é rodar duas vezes de propósito em homologação e contar linhas contra chaves distintas.

Arquivado pelo problema que resolve

• deduplicação
• idempotência
• kafka consumer
• clickhouse
• webhooks
• bullmq
• confiabilidade