Chave de particionamento errada: a fila que trava porque um cliente sozinho ocupa tudo

Em qualquer sistema de log particionado, seja Kafka, Kinesis, Pub/Sub com chave de ordenação ou uma fila caseira em cima do Postgres, a chave cumpre duas funções ao mesmo tempo e quase todo desenho só considera uma delas. A primeira é a que aparece na documentação: mensagens com a mesma chave vão para a mesma partição e por isso são entregues na ordem em que foram produzidas. A segunda é a que aparece no incidente: mensagens com chaves diferentes que caem na mesma partição por acaso do hash passam a compartilhar uma fila serial e uma sorte comum. A chave não define apenas o que fica ordenado, define quem fica preso atrás de quem.

É por isso que a pergunta certa na hora de escolher a chave não é qual campo distribui melhor, e sim qual é a menor unidade que realmente precisa de ordem. Se o requisito é que os eventos de uma conversa sejam processados na sequência, a unidade é a conversa e não o cliente. Se a chave for o identificador do cliente, todas as conversas daquele cliente ficam serializadas por uma exigência que o negócio nunca fez, e a fila herda um gargalo artificial. A regra prática que fecha essa parte é direta: ordenar mais do que o necessário nunca é neutro, é sempre capacidade jogada fora e latência transferida para quem não pediu.

CHAVE = tenant_id  (ordem alem do necessario)

P0  [tenant-A x 70000 ...........................] <- 18 min de backlog
P1  [tenant-C][tenant-F][tenant-J]                  <- vazia em segundos
P2  [tenant-B][tenant-D]                            <- vazia em segundos
P3  [tenant-E][tenant-G][tenant-H]                  <- vazia em segundos
     ^ tenant-K caiu no hash de P0 e espera 18 min
       por um backlog que nao e dele

CHAVE = tenant_id + conversation_id  (ordem no que o negocio exige)

P0  [A/c1][K/c9][A/c4][C/c2]
P1  [A/c2][B/c7][A/c5][K/c3]
P2  [A/c3][D/c1][F/c8][A/c6]
P3  [A/c7][E/c2][A/c8][J/c4]
     ^ o volume do tenant-A continua grande, mas agora
       se espalha, e nenhum outro tenant fica atras dele

Vale separar dois fenômenos que costumam ser tratados como um só porque produzem o mesmo sintoma no painel. Desequilíbrio de partição é quando o hash distribui mal e uma partição recebe mais mensagens que as outras, e ele se corrige com mais entropia na chave. Bloqueio de cabeça de fila é quando uma única mensagem lenta ou envenenada trava tudo que está atrás dela na mesma partição, e ele não se corrige com entropia nenhuma, porque o problema não é o volume e sim a serialização. Sistemas reais quase sempre têm os dois, e a correção de um não ajuda em nada no outro.

A métrica que quase todo time tem é o lag total do grupo de consumidores, e ela é péssima para esse problema específico. O lag total é uma soma, e a soma esconde exatamente a informação que interessa: uma partição com noventa mil mensagens de atraso e trinta e uma partições zeradas produzem o mesmo número que trinta e duas partições com quase três mil cada. No primeiro caso existe um incidente em curso para um subconjunto de clientes, no segundo o sistema está apenas ocupado. O painel mostra o mesmo valor nos dois.

O que revela o problema é olhar a distribuição em vez do agregado, e três medidas bastam. O lag máximo por partição diz se existe alguma partição em situação ruim, e é ele que deve disparar alerta, não a soma. A razão entre o lag máximo e o lag mediano diz se o sistema está desequilibrado ou apenas carregado, e um valor persistente acima de cinco é sinal de chave mal escolhida e não de falta de consumidor. A participação do maior produtor por partição diz de quem é o volume, e é essa medida que transforma um alerta genérico em uma ação concreta, porque ela nomeia o cliente que precisa de tratamento diferente.

// metrics/partition-skew.js
// Distribuicao do lag por particao, nao a soma.
//
// A soma esconde o caso que importa: uma particao com 90k de lag e
// 31 zeradas somam o mesmo que 32 particoes com 2.8k cada. A primeira
// e um incidente para um subconjunto de clientes, a segunda e trafego.

const percentile = (sorted, p) => {
  if (sorted.length === 0) return 0;
  const index = Math.min(sorted.length - 1, Math.floor((sorted.length - 1) * p));
  return sorted[index];
};

// lagByPartition: Map<partitionId, number>
export const describeSkew = (lagByPartition) => {
  const entries = [...lagByPartition.entries()];
  const values = entries.map(([, lag]) => lag).sort((a, b) => a - b);

  const total = values.reduce((sum, lag) => sum + lag, 0);
  const median = percentile(values, 0.5);
  const max = values[values.length - 1] ?? 0;

  const [hottestPartition] = entries.reduce(
    (worst, entry) => (entry[1] > worst[1] ? entry : worst),
    [null, -1],
  );

  return {
    total,
    median,
    max,
    hottestPartition,
    // Razao, nao diferenca: 90000/2800 e desequilibrio, 90000/89000 e carga.
    // Mediana zero com maximo positivo e o caso mais grave, e virar Infinity
    // aqui seria perder o alerta, entao ancoramos em 1.
    skewRatio: max / Math.max(median, 1),
    // Particoes efetivamente paradas enquanto existe backlog em outra:
    // capacidade contratada, paga e ociosa.
    idlePartitions: values.filter((lag) => lag === 0).length,
  };
};

// Alerta por particao, nao por soma. O limiar de razao pega o desequilibrio
// que ainda nao virou lag absoluto grande, que e a janela util para agir.
export const skewAlerts = (skew, { maxLagThreshold = 5000, maxSkewRatio = 5 } = {}) => {
  const alerts = [];

  if (skew.max > maxLagThreshold) {
    alerts.push({
      level: 'page',
      reason: 'partition_lag_high',
      partition: skew.hottestPartition,
      value: skew.max,
    });
  }

  // Desequilibrio com lag ainda baixo: nao acorda ninguem, mas vira tarefa
  // de revisao de chave antes do proximo pico.
  if (skew.skewRatio > maxSkewRatio && skew.idlePartitions > 0) {
    alerts.push({
      level: 'ticket',
      reason: 'partition_key_skew',
      partition: skew.hottestPartition,
      value: Number(skew.skewRatio.toFixed(1)),
    });
  }

  return alerts;
};

Há um detalhe operacional que muda o valor desse alerta: ele precisa rodar sobre uma janela curta, de um ou dois minutos, e não sobre a média da hora. Desequilíbrio é um fenômeno de rajada, e ele aparece justamente quando um cliente dispara uma importação, uma campanha ou uma reprocessagem. Na média de sessenta minutos, um pico de quatro minutos com razão vinte vira uma razão dois e some do painel, e o time descobre o problema pelo canal de suporte em vez de pelo alerta.

A correção estrutural quase sempre é a mesma: trocar a chave por uma composição que carregue a menor unidade de ordem real. Em atendimento, isso normalmente é a conversa e não o cliente, porque duas conversas do mesmo cliente não têm relação causal entre si. Em cobrança, é a assinatura e não a empresa. Em estoque, é o item e não o depósito. A composição precisa manter o identificador de tenant como prefixo quando o roteamento posterior depende dele, mas o que entra no cálculo do hash passa a ser o par completo.

A troca de chave tem um custo que precisa ser planejado e é onde a maioria das migrações erra: durante a transição, eventos da mesma entidade existem nas duas partições, a antiga e a nova, e a ordem entre eles deixa de ser garantida exatamente no intervalo em que as duas convivem. A forma segura de fazer é drenar antes de cortar. O produtor passa a escrever num tópico novo com a chave nova, os consumidores continuam lendo o tópico antigo até o lag chegar a zero e só então assumem o novo. Cortar a chave no mesmo tópico, sem drenar, é criar uma janela de reordenação silenciosa que só vai aparecer depois, como um estado inconsistente que ninguém consegue reproduzir.

// queue/partition-key.js
// Chave composta: a menor unidade que o negocio exige em ordem.
//
// Manter tenant_id no prefixo serve para roteamento e depuracao, mas o
// hash usa a chave inteira. Trocar o separador ou a ordem dos campos
// remapeia tudo: a chave e um contrato, versionado como qualquer outro.

const SEPARATOR = '\u001f'; // unit separator: nao aparece em id de negocio

export const partitionKey = ({ tenantId, entityId }) => {
  if (!tenantId || !entityId) {
    // Cair no null aqui significa round-robin silencioso e perda de ordem.
    // Melhor falhar na producao do evento do que descobrir no consumo.
    throw new Error('partition_key_requires_tenant_and_entity');
  }
  return `${tenantId}${SEPARATOR}${entityId}`;
};

// Sufixo por faixa para a entidade que sozinha excede uma particao.
// Custo explicito: eventos da mesma entidade deixam de ter ordem total
// entre si. So use quando a ordem dentro da faixa for suficiente, por
// exemplo entregas de broadcast, que sao independentes entre destinatarios.
export const shardedPartitionKey = ({ tenantId, entityId, shardCount = 1, discriminator }) => {
  const base = partitionKey({ tenantId, entityId });
  if (shardCount <= 1) return base;

  // O discriminador precisa ser estavel por sub-unidade (o destinatario),
  // nao aleatorio: aleatorio quebraria a ordem tambem dentro da sub-unidade.
  const shard = stableHash(String(discriminator ?? '')) % shardCount;
  return `${base}${SEPARATOR}${shard}`;
};

// FNV-1a: estavel entre processos e versoes de runtime. Nao use
// String.prototype.hashCode improvisado nem Math.random: a chave precisa
// produzir a mesma particao daqui a seis meses, em outro deploy.
const stableHash = (value) => {
  let hash = 0x811c9dc5;
  for (let i = 0; i < value.length; i += 1) {
    hash ^= value.charCodeAt(i);
    hash = Math.imul(hash, 0x01000193) >>> 0;
  }
  return hash;
};

A reação mais comum diante do desequilíbrio é subir o número de partições, e ela falha por um motivo aritmético simples: uma chave sempre mapeia para uma partição só. Se a chave é o identificador do cliente e um cliente concentra setenta mil eventos, esses setenta mil continuam em uma única partição, tenha o tópico oito ou oitocentas. O que muda é apenas a probabilidade de outro cliente cair junto, o que reduz o número de vítimas colaterais sem reduzir em nada o tempo de espera do cliente grande nem o da vítima que ainda assim cair no mesmo hash.

Existe ainda um efeito colateral que costuma passar despercebido no momento da mudança. Aumentar o número de partições em um tópico já em uso remapeia todas as chaves, porque o destino é calculado como hash da chave módulo o número de partições. Eventos de uma mesma entidade que estavam na partição três passam a ir para a onze, enquanto os antigos ainda esperam na três, e por um intervalo a ordem entre eles simplesmente não existe. Ou seja, a medida que não resolve o desequilíbrio introduz, de graça, a mesma janela de reordenação que uma troca de chave mal feita introduziria.

• Mais partições ajudam quando o gargalo é vazão agregada e as chaves já estão bem distribuídas, porque aí a limitação é paralelismo. Não ajudam quando o gargalo é uma chave específica.
• O número de partições limita o paralelismo do grupo de consumidores: com trinta e duas partições, o trigésimo terceiro consumidor fica ocioso. Subir consumidor sem subir partição não aumenta a vazão.
• Reduzir o número de partições não é suportado na maioria dos sistemas, então o número escolhido é praticamente permanente. Vale começar acima do necessário, mas sem exagero, porque cada partição custa memória de broker e arquivo aberto.
• A ordem só é garantida dentro da partição, nunca entre partições. Qualquer lógica que dependa de comparar eventos de partições diferentes precisa de marca de tempo lógica própria, não da ordem de consumo.
• Se a distribuição já está boa e o lag máximo continua alto em todas as partições, o problema não é a chave: é capacidade de consumo ou uma dependência lenta dentro do handler.

Depois de compor a chave corretamente, sobra um caso que a composição não cobre: a entidade única cujo volume excede a capacidade de uma partição. É o disparo de broadcast para duzentos mil destinatários, a importação inicial de um catálogo, a reprocessagem de um mês inteiro de histórico. Nesses casos, mesmo com a menor unidade de ordem correta, o trabalho de uma entidade não cabe no tempo aceitável de uma partição só, e a única saída é dividir a entidade ou tirá-la da fila principal.

A divisão por sufixo funciona quando as sub-unidades são independentes entre si, que é o caso do broadcast: a ordem entre a entrega para o destinatário A e para o destinatário B nunca importou. O sufixo precisa ser derivado de forma estável do destinatário, e não sorteado, porque um sufixo aleatório quebraria também a ordem dos eventos daquele destinatário específico. Quando as sub-unidades não são independentes, a divisão não é possível e a resposta correta é a separação de tópico: trabalho de volume alto e latência tolerante sai da fila que atende o tempo real e vai para um tópico próprio, com seus próprios consumidores e seu próprio orçamento.

A última linha da tabela merece nota, porque é a que mais causa incidente longo e a que menos aparece em desenho de arquitetura. Uma mensagem que falha de forma determinística, por um campo inválido ou um registro que não existe mais, é retentada para sempre, e enquanto ela está no topo da partição nada atrás dela avança. O sintoma é característico e fácil de reconhecer depois que se viu uma vez: lag subindo em uma partição só, consumo de CPU no chão, e o mesmo identificador de mensagem repetido no log a cada poucos segundos. A correção é enviar a mensagem para uma fila de mensagens mortas depois de um número finito de tentativas, aceitando conscientemente que a ordem daquela entidade se perdeu, porque a alternativa é perder o avanço de todas as outras entidades da mesma partição.

Nenhuma dessas correções deve ser confiada apenas ao raciocínio, porque a propriedade que se quer garantir é justamente uma que só aparece sob desequilíbrio. O teste útil não mede vazão: ele mede latência do cliente pequeno enquanto o cliente grande satura a fila. Produza dezenas de milhares de eventos de um tenant único, injete no meio disso um punhado de eventos de outro tenant e meça quanto tempo o segundo levou do produtor até o consumidor. Com a chave antiga, essa latência é da ordem do backlog inteiro. Com a chave composta, ela precisa continuar na ordem de segundos.

// test/noisy-neighbor.test.js
// O teste nao mede vazao: mede se o cliente pequeno ainda e atendido
// enquanto o grande satura a fila. E a unica propriedade que a chave
// composta existe para garantir.

import { describe, it, expect } from 'vitest';
import { partitionKey } from '../queue/partition-key.js';

const PARTITIONS = 32;
const fnv1a = (value) => {
  let hash = 0x811c9dc5;
  for (let i = 0; i < value.length; i += 1) {
    hash ^= value.charCodeAt(i);
    hash = Math.imul(hash, 0x01000193) >>> 0;
  }
  return hash;
};
const partitionOf = (key) => fnv1a(key) % PARTITIONS;

// Simulacao sem broker: cada particao e uma lista serial, e a posicao de
// uma mensagem na sua particao e o numero de mensagens que ela espera.
const simulate = (events, keyOf) => {
  const depth = new Array(PARTITIONS).fill(0);
  return events.map((event) => {
    const partition = partitionOf(keyOf(event));
    const waitsBehind = depth[partition];
    depth[partition] += 1;
    return { ...event, partition, waitsBehind };
  });
};

describe('isolamento entre tenants na fila', () => {
  // Cliente grande: 70k eventos espalhados em 5k conversas.
  // Cliente pequeno: 10 eventos, injetados no meio do lote.
  const events = [
    ...Array.from({ length: 70_000 }, (_, i) => ({
      tenantId: 'tenant-a',
      entityId: `conv-${i % 5_000}`,
    })),
  ];
  events.splice(
    35_000,
    0,
    ...Array.from({ length: 10 }, (_, i) => ({ tenantId: 'tenant-k', entityId: `conv-${i}` })),
  );

  const smallTenantWait = (processed) =>
    Math.max(
      ...processed.filter((event) => event.tenantId === 'tenant-k').map((e) => e.waitsBehind),
    );

  it('chave por tenant prende o cliente pequeno atras do backlog', () => {
    const processed = simulate(events, (event) => event.tenantId);
    // tenant-k tem particao propria, mas tenant-a ocupa uma inteira:
    // qualquer tenant que caia no mesmo hash espera o backlog completo.
    const hottest = Math.max(
      ...Array.from({ length: PARTITIONS }, (_, p) =>
        processed.filter((event) => event.partition === p).length,
      ),
    );
    expect(hottest).toBeGreaterThan(60_000);
  });

  it('chave composta mantem o cliente pequeno em espera de segundos', () => {
    const processed = simulate(events, partitionKey);
    // 70k espalhados em 32 particoes: ~2.2k por particao. O cliente
    // pequeno espera uma fracao disso, nao o backlog inteiro.
    expect(smallTenantWait(processed)).toBeLessThan(3_000);
  });

  it('nenhuma particao fica ociosa enquanto outra acumula', () => {
    const processed = simulate(events, partitionKey);
    const depths = Array.from({ length: PARTITIONS }, (_, p) =>
      processed.filter((event) => event.partition === p).length,
    );
    const max = Math.max(...depths);
    const median = depths.sort((a, b) => a - b)[Math.floor(PARTITIONS / 2)];
    expect(max / median).toBeLessThan(1.5);
  });
});

Esse teste tem uma virtude que compensa a simplificação: ele não precisa de broker, roda em milissegundos e por isso cabe no pipeline como qualquer teste unitário. A simulação captura exatamente a propriedade que interessa, que é quantas mensagens estão à frente da mensagem do cliente pequeno na sua partição, e ignora tudo o mais. Quando alguém futuramente trocar a chave de volta para o identificador do cliente por conveniência, a terceira asserção quebra e o motivo fica registrado no arquivo, que é o único lugar onde esse tipo de decisão sobrevive à rotatividade do time.