Backpressure em pipeline de IA: quando o consumidor não acompanha

A intuição de que uma fila resolve descompasso de velocidade está certa, mas só para descompasso temporário. Uma fila é um amortecedor: ela absorve a diferença entre produção e consumo durante um intervalo, contando com um período posterior em que o consumidor está mais rápido que o produtor e drena o acúmulo. Se a taxa média de produção é maior que a taxa média de consumo, nenhum tamanho de fila resolve, porque não existe momento de drenagem. A fila só cresce. O que muda com o tamanho é quando o problema aparece, não se ele aparece.

O detalhe cruel é que uma fila crescendo piora ativamente o sistema antes de derrubá-lo. Cada item enfileirado ocupa memória, e em pipeline de IA os itens não são pequenos: carregam o histórico da conversa, o contexto recuperado, os anexos. Mil itens de cem kilobytes são cem megabytes de heap que o coletor de lixo precisa varrer a cada ciclo, e a pressão de GC deixa o consumidor mais lento, o que faz a fila crescer mais rápido, o que aumenta a pressão de GC. O sistema entra numa espiral em que a própria tentativa de absorver a carga é o que reduz a capacidade de processá-la.

E há o custo que ninguém contabiliza: o trabalho enfileirado envelhece. Uma mensagem que espera oito minutos numa fila para depois ser processada gera uma resposta que chega quando o cliente já saiu da conversa, já ligou para o telefone ou já desistiu. Você gastou tokens, gastou tempo de máquina e produziu uma resposta que não vale nada. Processar um item vencido é pior do que tê-lo recusado na entrada, porque a recusa é barata e honesta, enquanto o processamento tardio custa dinheiro e ainda entrega uma experiência ruim.

Backpressure começa com uma medida. A mais direta é a ocupação da fila: quantos itens estão esperando dividido pelo teto configurado. É melhor que medir CPU ou memória porque a ocupação responde imediatamente ao descompasso, enquanto CPU alta pode significar simplesmente trabalho útil sendo feito e memória alta é um indicador atrasado, que já vem tarde demais.

A armadilha aparece na transformação da medida em decisão. Se a regra é "acima de sessenta por cento, recusar lote", uma fila oscilando em torno de sessenta por cento faz o sistema alternar entre aceitar e recusar a cada item, e o produtor recebe um sinal que muda de segundo em segundo, impossível de seguir. A correção é histerese: o regime sobe quando a pressão cruza o limiar, mas só desce quando ela cai abaixo do limiar menos uma margem. É o mesmo princípio do termostato, que não liga e desliga o compressor a cada décimo de grau. Com histerese, o produtor recebe um regime estável o suficiente para agir sobre ele.

Três regimes cobrem praticamente todo caso real, e o valor de nomeá-los é que cada um carrega uma política declarada, escrita antes do incidente, em vez de uma decisão improvisada durante ele.

// src/pressure-gauge.js
// Medidor de pressao com histerese: sobe no limiar cheio, mas so desce
// quando a pressao cai abaixo do limiar menos a margem. Sem isso o regime
// pisca em cima da fronteira e o produtor nao consegue reagir.

export const NORMAL = 'normal';
export const DEGRADED = 'degraded';
export const CRITICAL = 'critical';

export class PressureGauge {
  constructor({ degradedAt = 0.6, criticalAt = 0.85, hysteresis = 0.1 } = {}) {
    if (!(degradedAt < criticalAt)) {
      throw new Error('degradedAt deve ser menor que criticalAt');
    }
    this.degradedAt = degradedAt;
    this.criticalAt = criticalAt;
    this.hysteresis = hysteresis;
    this.state = NORMAL;
  }

  update(pressure) {
    if (this.state === NORMAL && pressure >= this.degradedAt) {
      this.state = pressure >= this.criticalAt ? CRITICAL : DEGRADED;
      return this.state;
    }

    if (this.state === DEGRADED) {
      if (pressure >= this.criticalAt) this.state = CRITICAL;
      else if (pressure < this.degradedAt - this.hysteresis) this.state = NORMAL;
      return this.state;
    }

    if (this.state === CRITICAL && pressure < this.criticalAt - this.hysteresis) {
      this.state = pressure >= this.degradedAt ? DEGRADED : NORMAL;
    }

    return this.state;
  }

  // Prioridade maxima aceita em cada regime (menor numero = mais urgente).
  admissionCeiling() {
    if (this.state === CRITICAL) return 1;
    if (this.state === DEGRADED) return 5;
    return Number.POSITIVE_INFINITY;
  }
}

Quando a fila enche, algo vai ser perdido. Isso não é opcional, é aritmética: se entram mil e saem cem, novecentos ficam de fora de um jeito ou de outro. A única escolha real é entre decidir o que perder ou deixar o esgotamento de recurso decidir por você. A primeira opção produz perda controlada e registrada; a segunda produz perda arbitrária e silenciosa, normalmente do trabalho mais caro, porque é o que estava em processamento quando o processo morreu.

Três políticas cobrem a maior parte dos casos, e a escolha não é técnica, é de negócio. Recusar na entrada é a política certa quando o produtor pode segurar o trabalho e tentar de novo depois, como um webhook que devolve 429 e confia na retentativa do remetente. Descartar o mais antigo é a política certa quando o dado envelhece e o valor está no mais fresco, como um fluxo de telemetria em que a medição de agora vale mais que a de dois minutos atrás. Descartar o de menor prioridade é a política certa quando classes diferentes convivem na mesma fila e uma delas tem um humano esperando do outro lado.

// src/bounded-queue.js
// Fila limitada com politica de admissao explicita.
// Ela nunca cresce alem de maxSize: quando cheia, decide o que perder
// em vez de deixar a memoria decidir por ela.

export const ADMITTED = 'admitted';
export const REJECTED = 'rejected';
export const SHED = 'shed';

export class BoundedQueue {
  constructor({ maxSize, policy = 'reject' }) {
    if (!Number.isInteger(maxSize) || maxSize < 1) {
      throw new Error('maxSize deve ser um inteiro maior que zero');
    }
    this.maxSize = maxSize;
    this.policy = policy;
    this.items = [];
    this.stats = { admitted: 0, rejected: 0, shed: 0 };
  }

  get pressure() {
    return this.items.length / this.maxSize;
  }

  enqueue(item, priority = 5) {
    const entry = { item, priority, enqueuedAt: Date.now() };

    if (this.items.length < this.maxSize) {
      this.insertByPriority(entry);
      this.stats.admitted += 1;
      return { outcome: ADMITTED, dropped: null };
    }

    if (this.policy === 'drop-oldest') {
      const dropped = this.removeOldest();
      this.insertByPriority(entry);
      this.stats.admitted += 1;
      this.stats.shed += 1;
      return { outcome: SHED, dropped: dropped.item };
    }

    if (this.policy === 'drop-lowest') {
      const worst = this.items[this.items.length - 1];
      // So descarta se o recem-chegado for realmente mais urgente que o pior
      // da fila, senao a fila vira um carrossel de trocas sem progresso.
      if (worst.priority > priority) {
        this.items.pop();
        this.insertByPriority(entry);
        this.stats.admitted += 1;
        this.stats.shed += 1;
        return { outcome: SHED, dropped: worst.item };
      }
      this.stats.rejected += 1;
      return { outcome: REJECTED, dropped: null };
    }

    this.stats.rejected += 1;
    return { outcome: REJECTED, dropped: null };
  }

  insertByPriority(entry) {
    // Insercao estavel: entre prioridades iguais vale a ordem de chegada.
    let index = this.items.length;
    while (index > 0 && this.items[index - 1].priority > entry.priority) {
      index -= 1;
    }
    this.items.splice(index, 0, entry);
  }
}

O detalhe que separa uma implementação correta de uma que se sabota está na política drop-lowest: a troca só acontece se o recém-chegado for estritamente mais urgente que o pior item já enfileirado. Sem essa comparação, uma rajada de itens de mesma prioridade faz a fila descartar e readmitir os próprios itens sem parar, gastando trabalho para não avançar nada. É o tipo de bug que só aparece sob carga, exatamente quando você menos quer descobri-lo.

Limitar a fila protege o processo, mas ainda não é backpressure. Backpressure é o sinal que sobe: o consumidor precisa comunicar ao produtor que está saturado, e o produtor precisa fazer algo com essa informação. Sem esse retorno, você trocou um processo que morre por um processo saudável que recusa tudo, o que é melhor, mas ainda muito longe do ideal, porque o produtor continua martelando na mesma velocidade e desperdiçando trabalho em tentativas que serão rejeitadas.

O que forma esse sinal depende do transporte, mas a informação é sempre a mesma: houve recusa, e quanto esperar antes de tentar de novo. Numa API HTTP, é o 429 com Retry-After. Numa fila de mensagens com confirmação manual, é parar de buscar novos lotes enquanto os anteriores não forem confirmados, o que é o prefetch limitado do AMQP ou a pausa de consumo do Kafka. Dentro do mesmo processo, é a promessa retornada pelo write de um stream, ou simplesmente aguardar antes do próximo submit. O erro comum é responder a recusa com um valor fixo de espera: cinco segundos chutados fazem todos os produtores voltarem juntos, sincronizados, num efeito manada que recria o pico. O valor certo deriva da fila real, do trabalho pendente dividido pela vazão do consumidor.

  produtor                      estagio com backpressure
     |                                    |
     |  submit(job, prioridade)           |
     |----------------------------------->|
     |                             [ mede ocupacao da fila ]
     |                             [ atualiza regime c/ histerese ]
     |                                    |
     |                          prioridade <= teto do regime?
     |                                    |
     |            <---- admitido ---------+ sim
     |                                    |
     |            <-- recusado + espera --+ nao
     |                                    |
     |  desacelera na origem              v
     |  (pausa lote, mantem interativo)  consumidor
     |                                   concorrencia limitada
     v                                        |
  produz menos <----- regime critico ---------+

// src/pipeline.js
// Estagio com backpressure explicito: concorrencia limitada no consumidor,
// fila limitada no meio e resultado de admissao devolvido ao produtor.

export class BackpressureStage {
  constructor({ handler, concurrency = 2, maxQueueSize = 10, policy = 'reject', gauge }) {
    this.handler = handler;
    this.concurrency = concurrency;
    this.queue = new BoundedQueue({ maxSize: maxQueueSize, policy });
    this.gauge = gauge || new PressureGauge();
    this.inFlight = 0;
    this.metrics = { processed: 0, failed: 0, rejected: 0, shed: 0, maxQueueSeen: 0 };
  }

  submit(job, priority = 5) {
    const state = this.gauge.update(this.queue.pressure);

    // O teto do regime fecha a porta para o lote antes do interativo sofrer.
    if (priority > this.gauge.admissionCeiling()) {
      this.metrics.rejected += 1;
      return { outcome: 'rejected', state, retryAfterMs: this.retryAfterMs() };
    }

    const admission = this.queue.enqueue(job, priority);
    if (admission.outcome === 'rejected') {
      this.metrics.rejected += 1;
      return { outcome: 'rejected', state, retryAfterMs: this.retryAfterMs() };
    }
    if (admission.outcome === 'shed') this.metrics.shed += 1;

    this.metrics.maxQueueSeen = Math.max(this.metrics.maxQueueSeen, this.queue.size);
    this.pump();
    return { outcome: admission.outcome, state, dropped: admission.dropped };
  }

  // Espera estimada a partir da fila real e da vazao configurada,
  // em vez de um valor fixo que sincroniza todos os produtores.
  retryAfterMs(perItemMs = 200) {
    return Math.ceil(((this.queue.size + 1) / this.concurrency) * perItemMs);
  }

  pump() {
    while (this.inFlight < this.concurrency && this.queue.size > 0) {
      const job = this.queue.dequeue();
      this.inFlight += 1;
      Promise.resolve()
        .then(() => this.handler(job))
        .then(() => { this.metrics.processed += 1; })
        .catch(() => { this.metrics.failed += 1; })
        .finally(() => {
          this.inFlight -= 1;
          this.gauge.update(this.queue.pressure);
          this.pump();
        });
    }
  }
}

Repare que a concorrência limitada no consumidor é parte essencial do desenho, e não um detalhe. Se o pump disparasse todos os itens da fila de uma vez, a fila esvaziaria instantaneamente, a pressão medida cairia para zero e o medidor diria que está tudo bem enquanto mil chamadas simultâneas estariam em voo contra o provedor. A fila só é um sensor honesto de pressão porque existe um limite de trabalho em voo represando os itens nela.

Um pipeline real tem vários estágios em sequência: recebe a mensagem, normaliza, recupera o contexto, chama o modelo, valida a saída, entrega. Cada um tem vazão diferente, e o gargalo é o mais lento, normalmente a chamada ao modelo. O erro clássico é colocar backpressure só nesse estágio: os anteriores continuam aceitando tudo, acumulam nas próprias filas, e o problema apenas se muda de lugar, saindo do estágio lento para os que vêm antes dele.

A regra é que o sinal precisa subir a cadeia até chegar à borda, o ponto onde o sistema fala com o mundo externo e pode legitimamente dizer não. Cada estágio observa a pressão do estágio seguinte antes de aceitar mais trabalho, e o primeiro estágio traduz a pressão acumulada numa resposta ao chamador. É o inverso do fluxo de dados: os dados descem do primeiro ao último, o sinal de pressão sobe do último ao primeiro.

1. Cada estágio expõe a própria pressão como um número entre zero e um, derivado da ocupação da fila mais o trabalho em voo.
2. Antes de aceitar um item, o estágio consulta a pressão do estágio seguinte e usa o maior valor entre a própria e a dele como pressão efetiva.
3. A borda traduz a pressão efetiva em resposta ao chamador: aceita, aceita com aviso de degradação ou recusa com espera estimada.
4. O trabalho de lote consulta a pressão antes de puxar o próximo bloco e reduz o tamanho do bloco no regime degradado, pausando no crítico.
5. Nenhum estágio intermediário tem fila ilimitada, porque um único buffer sem teto no meio da cadeia anula o sinal de todos os outros.

Há um caso em que a borda não pode dizer não: quando o produtor é um webhook externo que vai desistir e considerar a mensagem perdida se você recusar. Aí a resposta correta não é aceitar em memória, é persistir imediatamente em armazenamento durável e responder 200 sobre o registro persistido, transformando a fila em memória numa fila em disco que sobrevive à queda do processo. A pressão continua existindo e continua sendo medida, mas passa a ser medida sobre a fila durável, e o backlog vira um número visível em vez de heap invisível.

Backpressure implementado sem métrica vira um mecanismo que descarta trabalho em silêncio, o que é uma forma diferente do mesmo problema. Contar apenas o total processado não revela nada, porque um sistema que recusa noventa por cento da carga e processa dez por cento rapidamente parece saudável em latência média. As métricas precisam separar o que foi aceito do que foi recusado e mostrar o custo de cada decisão.

A última linha merece destaque porque é a que mais surpreende quem instrumenta pela primeira vez. Um pipeline com fila generosa costuma ter uma fatia relevante de itens concluídos depois que já não importavam: a resposta pronta para uma conversa encerrada, o resumo gerado para um atendimento já transferido. Esses itens consumiram cota, custaram dinheiro e ocuparam o lugar de trabalho útil. Medir isso costuma ser o argumento mais convincente para reduzir o tamanho da fila, porque transforma uma discussão de arquitetura numa linha de fatura.

A leitura conjunta é o que dá o diagnóstico. Recusa alta com ocupação baixa significa teto de admissão calibrado de forma conservadora demais. Recusa baixa com ocupação sempre no topo significa que a fila está grande demais e você está acumulando latência em vez de sinalizar. Tempo em regime crítico crescendo semana após semana não é problema de configuração: é falta de capacidade, e nenhum ajuste de limiar resolve, só mais consumidores ou um modelo mais rápido no gargalo.