Backpressure e flow control

Em produção, o cenário se repete com regularidade depressiva. Um serviço novo é lançado e funciona perfeitamente em testes. Sob tráfego real, começa "ok" — métricas verdes, latência baixa. Algumas horas depois, queries para o banco começam a engargalhar. A latência sobe gradualmente. As queues internas crescem. A memória do processo escala. Em algum momento, garbage collection vira o gargalo, latência tail estoura, alarmes disparam, e eventualmente o processo é OOM-killed pelo kernel ou pelo Kubernetes. Restart automático, e o ciclo recomeça. Esse padrão é tão comum que tem nome em literatura de SRE: cascading failure. A causa raiz, quase sempre, é a mesma: ausência de backpressure.

Backpressure é o mecanismo pelo qual o consumidor de um stream sinaliza ao produtor que está sobrecarregado, e o produtor reage diminuindo a taxa em vez de continuar empurrando trabalho. Sem backpressure, sistemas concorrentes têm uma propriedade ruim chamada unbounded queue growth: quando o consumidor é mais lento que o produtor, a queue intermediária cresce, e a memória cresce com ela, e a latência observada por novos itens explode (Lei de Little: tempo médio de espera = tamanho médio da fila ÷ taxa de saída). Eventualmente algo quebra.

Backpressure é o tópico que separa sistemas que aguentam tráfego real dos que parecem aguentar até falharem. É uma ferramenta universalmente aplicável: aparece em TCP (window size), em HTTP/2 (flow control credits), em Kafka (consumer lag), em Reactive Streams (request(n)), em frameworks de async modernos (Channel.CreateBounded, asyncio.Queue maxsize, Akka Streams). Cada manifestação é variação do mesmo princípio: nenhum sistema deveria aceitar trabalho que não pode processar.

Este conceito apresenta backpressure formalmente, cataloga estratégias quando buffers enchem, mostra como Reactive Streams formalizou a especificação, e explora padrões em Go, Python e C#. Conceito 13 cobrirá cancelamento, que é tema relacionado; juntos, formam a infraestrutura para sistemas concorrentes que sobrevivem a carga adversa.

O problema — unbounded queue growth

Considere o caso simples: produtor gera mensagens a 1.000 msg/s; consumidor processa a 800 msg/s. Em um sistema sem backpressure, essa diferença de 200 msg/s acumula em queue. Após 1 minuto, a queue tem 12.000 itens; após 10 minutos, 120.000; após uma hora, 720.000. Se cada item ocupa 1 KB, isso é 720 MB de memória só em queue. Mais importante: a latência média de processamento é tamanho_da_queue / taxa_de_saída — itens recém-chegados esperam 15 minutos para serem processados se a queue tem 720K itens e o consumidor processa 800/s.

O resultado prático: throughput estabiliza em 800 msg/s (limite do consumidor), latência cresce indefinidamente, memória cresce indefinidamente. O sistema parece "estar funcionando" — métricas de throughput estão estáveis, taxa de erro é zero — mas está claramente fora de saúde. É um modo de falha que deceptive porque o sintoma é gradual: ninguém percebe nada errado por minutos ou horas, e quando percebe, o sistema está perto de cair.

A Lei de Little, formalizada por John Little em 1961, é a equação fundamental que organiza todo esse fenômeno: em sistema em equilíbrio, L = λ × W (número médio de itens no sistema = taxa de chegada × tempo médio no sistema). Se λ excede a taxa de processamento, L cresce sem limite, e W (latência) cresce com ela. Backpressure existe para forçar λ a cair quando o sistema está sobrecarregado — manter L bounded.

Estratégias quando o buffer enche

Backpressure é o nome genérico para qualquer mecanismo que previne unbounded growth. Há várias estratégias, com trade-offs diferentes; cada framework moderno oferece um menu para você escolher.

Block — bloqueia o produtor

Quando o buffer enche, o produtor que tenta colocar bloqueia até haver espaço. É a forma mais simples e a mais segura para sistemas onde nenhum item pode ser perdido. É a estratégia default de asyncio.Queue com maxsize, channels bufferizados em Go, BoundedChannelFullMode.Wait em C#. Trade-off: latência tail do produtor explode quando o consumidor está lento. Se o produtor é um web server, isso vira "request demora" — pode ser melhor ou pior que outras opções dependendo do contexto.

Drop oldest — derruba o item mais antigo

Quando o buffer enche, o item mais antigo é descartado para abrir espaço. Apropriado para streams onde dados antigos têm menos valor (telemetria, sensor data, eventos de UI). TCP usa uma variante disso em situações de congestion. Trade-off: perda silenciosa de dados; pode mascarar problemas reais.

Drop newest — rejeita o item atual

O produtor recebe erro/null em vez do item entrar. Trade-off: o produtor sabe que perdeu (pode logar, alertar, reagir), mas a lógica de tratamento sobe para o produtor. Apropriado quando o produtor consegue lidar com falha (retry com backoff, descartar com log).

Drop with cost / sample

Em vez de dropar tudo ou nada, descarta probabilisticamente para reduzir taxa. Se 50% do buffer está cheio, drop 0%; se 90%, drop 50%. É comum em telemetria — você quer amostragem, não todos os eventos. Sistemas como Datadog Agent fazem isso para backpressure interno.

Throttle — rate limit no produtor

Em vez de buffer reactive, o produtor tem rate limit explícito. Se a taxa máxima permitida é 1.000 msg/s, ele dorme entre mensagens para respeitar. Token bucket e leaky bucket são as implementações canônicas. Útil quando você controla o produtor e quer comportamento previsível.

Spill — escreve para disco

Quando memória enche, transborda para disco. Apropriado quando perda de dados é inaceitável e bloquear o produtor é caro. Common em mensageria persistente (Kafka, RabbitMQ — que são essencialmente "spill to disk" como design principal). Trade-off: latência cresce significativamente; complexidade de gerência de arquivos.

Load shedding — rejeita conexões antes da fila

Estratégia mais ampla: detectar saturação e rejeitar trabalho no ingress, antes mesmo de entrar na queue interna. HTTP 503 com retry-after; recusa de novas conexões TCP; circuit breaker aberto. Discutido em detalhe em Site Reliability Engineering (Google, 2016) — o famoso "Handling Overload" chapter. A ideia: melhor 1% de requests com 503 imediato do que 100% com latência 30s.

TCP — backpressure built-in no protocolo

O TCP/IP, projetado nos anos 1970-80, já tem backpressure como mecanismo central — bem antes de a palavra ser comum em programação de aplicação. Cada conexão TCP tem uma window: o receptor anuncia quanto buffer ele tem disponível; o emissor não pode ter mais bytes "em vôo" do que essa window. Quando o receptor está sobrecarregado, ele anuncia window menor (ou zero); o emissor pausa.

É backpressure end-to-end implementado no nível do protocolo. Aplicações que usam TCP herdam o mecanismo gratuitamente — se seu socket está fazendo write e o socket buffer enche, write bloqueia (ou retorna EAGAIN em modo non-blocking). Você precisa apenas respeitar isso e não tentar contornar com buffering próprio sem limite.

HTTP/2 (RFC 7540, 2015) levou essa idéia para nível de stream: cada stream HTTP/2 tem flow control credits independentes. Você pode pausar uploads em um stream sem afetar outros na mesma conexão. gRPC herda disso. Em geral, qualquer protocolo moderno em escala (TCP, HTTP/2, gRPC, AMQP, MQTT) tem flow control de algum tipo.

Reactive Streams — especificação cross-JVM

Em 2013, engenheiros de Lightbend (Akka), Pivotal (Reactor), Netflix (RxJava) e Red Hat colaboraram em uma especificação: Reactive Streams. Objetivo: definir interfaces comuns para streaming async com backpressure que toda biblioteca pudesse implementar e interoperar. A spec foi publicada em 2015, virou parte de Java 9 (interfaces em java.util.concurrent.Flow), e influenciou virtualmente todo framework de streaming moderno.

O modelo central: pull-based. Em vez do publisher empurrar itens, o subscriber pede N itens via request(n); o publisher entrega no máximo N. Se o subscriber está lento, ele simplesmente pede menos — backpressure emerge naturalmente da semântica.

// Java — Flow API (java.util.concurrent.Flow)
interface Subscriber<T> {
    void onSubscribe(Subscription s);
    void onNext(T item);
    void onError(Throwable t);
    void onComplete();
}

interface Subscription {
    void request(long n);    // pede até n itens
    void cancel();
}

interface Publisher<T> {
    void subscribe(Subscriber<? super T> s);
}

A elegância está na composição: cada estágio em um pipeline Reactive Streams é simultaneamente Subscriber (recebe do anterior) e Publisher (entrega ao próximo). A pressão do final do pipeline propaga para trás: se o sink está lento, ele faz menos request(); o estágio antes dele recebe menos request(); e assim por diante até a fonte. A fonte autoritária — banco, Kafka, file — desacelera de fato.

Implementações: Akka Streams (Scala/Java), Project Reactor (Java/Spring WebFlux), RxJava 2+, RxJS 5+, Mutiny (Quarkus). Em .NET, System.Threading.Channels e TPL Dataflow têm modelos similares mas com APIs diferentes. Em Go, channels nativos são pull-based naturalmente (receive bloqueia o sender).

Backpressure em Go — channels como ferramenta

Go expõe backpressure de forma minimalista mas poderosa via channels bufferizados. Bounded channel é a primitiva direta: send bloqueia quando o buffer enche, receive bloqueia quando vazio. Toda comunicação concorrente em Go idiomática usa isso por padrão.

// Go — pipeline com backpressure natural
func produtor(ctx context.Context, out chan<- Job) {
    defer close(out)
    for job := range gerar() {
        select {
        case <-ctx.Done():
            return
        case out <- job:    // bloqueia se 'out' está cheio
        }
    }
}

func consumidor(ctx context.Context, in <-chan Job) {
    for {
        select {
        case <-ctx.Done():
            return
        case job, ok := <-in:
            if !ok { return }
            processar(job)   // demora; while isso, 'out' enche em produtor
        }
    }
}

// Em main:
ch := make(chan Job, 10)   // buffer = 10
go produtor(ctx, ch)
consumidor(ctx, ch)

Capacidade do channel é a única decisão de tuning. 0 (unbuffered) é máximo backpressure — produtor e consumidor sincronizam em cada item. 10 ou 100 dá um pouco de tolerância para variações de taxa. Muito alto remove backpressure efetivo (volta a crescer indefinidamente em prática). Heurística: capacidade = variação esperada de taxa em janela curta. Se você espera bursts de 50 itens, use capacidade ~50.

Backpressure em asyncio — Queue maxsize

asyncio.Queue(maxsize=N) é a primitiva direta em Python. queue.put() bloqueia (na semântica cooperativa do event loop) quando o buffer enche; queue.get() bloqueia quando vazio. Ambos retornam None ou levantam asyncio.QueueEmpty/ asyncio.QueueFull em modos non-blocking.

import asyncio

async def produtor(queue: asyncio.Queue):
    async for item in fonte_externa():
        await queue.put(item)   # bloqueia em modo cooperativo se cheia
    await queue.put(None)        # sentinela de fim

async def consumidor(queue: asyncio.Queue):
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            return
        try:
            await processar(item)
        finally:
            queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=100)   # bounded — backpressure ativo
    async with asyncio.TaskGroup() as tg:
        tg.create_task(produtor(queue))
        tg.create_task(consumidor(queue))

maxsize=0 em asyncio significa unbounded — evite em código de produção. maxsize apropriado depende do problema; bom default é "número de items que cabem em ~1 segundo de processamento do consumidor" — dá tempo para ele drenar variações momentâneas sem permitir crescimento patológico.

Backpressure em .NET — Channel options

System.Threading.Channels oferece o vocabulário mais explícito sobre estratégias. Ao criar um bounded channel, você escolhe entre cinco modos quando o buffer enche:

var canal = Channel.CreateBounded<Job>(new BoundedChannelOptions(100) {
    FullMode = BoundedChannelFullMode.Wait,           // produtor espera
    // alternativas:
    // BoundedChannelFullMode.DropOldest,             // remove o mais antigo
    // BoundedChannelFullMode.DropNewest,             // rejeita o atual já no buffer
    // BoundedChannelFullMode.DropWrite,              // rejeita o que está sendo escrito
});

Note DropNewest vs DropWrite: DropNewest derruba o item mais novo já no buffer, depois insere o atual — bom quando o item atual é o "mais relevante". DropWrite rejeita o item atual e mantém o buffer — bom quando os itens antigos têm prioridade. A diferença sutil expõe que estratégia certa depende de semântica de negócio.

Kafka — consumer lag como backpressure observável

Apache Kafka, como broker de mensagens, é interessante porque não tem backpressure direto entre produtor e consumidor. O broker absorve mensagens em disco sem feedback ao produtor. Em vez disso, backpressure aparece como consumer lag: a diferença entre o offset mais recente do tópico e o offset que o consumidor já processou.

Em sistema saudável, lag é estável (poucos segundos a alguns minutos). Lag crescendo é sinal de que consumidor está mais lento que produtor. Estratégias de resposta:

• Adicionar consumidores: Kafka consumer groups permitem paralelizar consumo entre instâncias. Útil se o consumo é independente entre mensagens.
• Otimizar o consumidor: profilar para identificar o gargalo (banco lento, network, parsing).
• Throttle no produtor: se o produtor é controlável, fazê-lo gerar menos.
• Aumentar retenção: comprar tempo enquanto você resolve o gargalo, ao custo de armazenamento.

Lag é uma métrica obrigatória em monitoring de qualquer sistema que usa Kafka seriamente. Alertar em "lag > X minutos" pega problemas antes de virarem outage.

Rate limiting — backpressure proativa

Rate limiting é a forma proativa de backpressure: em vez de esperar o sistema saturar para reagir, você impõe taxa máxima desde o começo. Útil para integrações externas (API limita 1000 req/s e você precisa respeitar), para proteger seu próprio backend (no máximo X req/s por usuário), e para fairness (cada tenant tem cota).

Algoritmos clássicos:

Token bucket

Bucket começa cheio com N tokens; tokens são adicionados a taxa R por segundo até o teto N. Cada request consome um token. Se há tokens, request passa; senão, é rejeitado ou bloqueia. Permite bursts até N (consumir tokens acumulados), depois afina para taxa R sustentada. É o algoritmo mais usado em APIs públicas.

Leaky bucket

Bucket "vaza" a taxa R constante. Requests entram no bucket; se ele transborda, são rejeitadas. Diferente de token bucket, não permite burst — taxa de saída é sempre constante. Útil quando você precisa de saída estável (downstream tem capacidade fixa).

Sliding window

Conta requests em janela deslizante (últimos N segundos). Mais preciso que fixed window (que tem problema de "boundary burst"), mas mais caro de calcular. Variação sliding window log guarda timestamps individuais; sliding window counter aproxima com counters.

Distributed rate limiting

Em sistemas distribuídos, rate limiting precisa coordenar entre instâncias. Soluções: Redis com token bucket atômico (INCR + EXPIRE), serviço central (com latência sua), ou aproximação local (cada instância tem cota proporcional, com tolerância para erro). Stripe, Cloudflare, Lyft publicaram artigos detalhados sobre seus designs.

O mesmo problema nas três linguagens

Para concretizar backpressure, considere o cenário comum: consumir mensagens de um stream rápido, processar com operação cara (chamada de banco), e proteger o sistema de saturar. Nas três linguagens:

using System.Threading.Channels;

// Telemetria: prefere mais recente. Drop oldest.
var canal = Channel.CreateBounded<Evento>(new BoundedChannelOptions(1000) {
    FullMode = BoundedChannelFullMode.DropOldest,
    SingleReader = true,
    SingleWriter = false,
});

async Task ProdutorAsync(IAsyncEnumerable<Evento> fonte, CancellationToken ct) {
    await foreach (var ev in fonte.WithCancellation(ct)) {
        await canal.Writer.WriteAsync(ev, ct);
    }
    canal.Writer.Complete();
}

async Task ConsumidorAsync(CancellationToken ct) {
    await foreach (var ev in canal.Reader.ReadAllAsync(ct)) {
        await SalvarEmBanco(ev, ct);   // 50ms cada
    }
}

import asyncio
from collections.abc import AsyncIterable

async def pipeline_com_backpressure(
    fonte: AsyncIterable[Evento]
):
    queue: asyncio.Queue[Evento | None] = asyncio.Queue(maxsize=1000)

    async def produtor():
        async for ev in fonte:
            # bloqueia em modo cooperativo se queue estiver cheia
            # → backpressure propaga para 'fonte' (que é async)
            await queue.put(ev)
        await queue.put(None)   # sentinela

    async def consumidor():
        while True:
            ev = await queue.get()
            if ev is None:
                return
            try:
                await salvar_em_banco(ev)   # ~50ms cada
            finally:
                queue.task_done()

    async with asyncio.TaskGroup() as tg:
        tg.create_task(produtor())
        tg.create_task(consumidor())

package main

import (
    "context"
)

// Para dados que NÃO podem ser perdidos: simples send (bloqueia)
func produtorBloqueante(ctx context.Context, out chan<- Evento, fonte <-chan Evento) {
    defer close(out)
    for {
        select {
        case <-ctx.Done():
            return
        case ev, ok := <-fonte:
            if !ok { return }
            select {
            case <-ctx.Done():
                return
            case out <- ev:   // bloqueia se 'out' cheio
            }
        }
    }
}

// Para telemetria: drop newest se não consegue enviar agora
func produtorDropNewest(ctx context.Context, out chan<- Evento, fonte <-chan Evento) {
    defer close(out)
    for {
        select {
        case <-ctx.Done():
            return
        case ev, ok := <-fonte:
            if !ok { return }
            select {
            case out <- ev:
            default:
                // canal cheio — drop silencioso (logar/metrificar)
            }
        }
    }
}

// uso
out := make(chan Evento, 1000)
go produtorDropNewest(ctx, out, fonte)
consumir(ctx, out)

Anti-padrões comuns

"Vou só usar uma queue grande"

Aumentar buffer adia o problema, não resolve. Se a taxa de produção excede a de consumo de forma sustentada, qualquer buffer finito enche. A solução não é buffer maior; é backpressure honesto. Buffer existe para absorver variações momentâneas de taxa, não diferença sustentada.

"Vou só dropar requests"

Drop silencioso é frequentemente bug em forma de feature. Sem métrica clara de quantos requests foram dropados, você vai descobrir o problema em produção dias depois. Se escolher drop, sempre meça e alerte.

"Auto-scale resolve"

Auto-scaling tem latência inerente (provisioning de instância leva minutos em Kubernetes/EC2). Em burst de tráfego, auto-scale chega tarde. Pior: scale-up sem limite custa muito dinheiro e pode mascarar problemas reais (consumidor ineficiente que deveria ser otimizado em vez de paralelizado). Auto-scaling é complemento, não substituto, para backpressure.

Buffers profundos sem alarmes

Você tem queue interna com tamanho 100K como "salvaguarda". Em produção, ela enche silenciosamente porque o consumidor desacelerou. Sem alarme em "tamanho da queue", você descobre quando OOM kill acontece. Toda queue bounded em produção precisa de métrica + alarme em sua profundidade.

Como praticar

1. Reproduza unbounded queue growth. Em qualquer linguagem, escreva produtor a 1.000 msg/s e consumidor a 800 msg/s, conectados por queue unbounded. Logue o tamanho da queue a cada segundo. Observe a taxa de crescimento (~200 items/s). Calcule quando vai estourar memória dado RAM disponível. Depois substitua por bounded queue de 100; observe o produtor desacelerando para 800/s automaticamente.
2. Construa rate limiter token bucket. Em Go ou Python, implemente um rate limiter que aceita N requests por segundo com burst de B. Teste: dispare 1.000 requests em rajada, depois 5.000 espalhadas em 10 segundos. Verifique que a taxa observada respeita o limite e bursts são absorvidos até B. Compare sua implementação com bibliotecas standard (golang.org/x/time/rate ou aiolimiter).
3. Profile lag em pipeline real. Pegue um sistema concorrente seu (ou exemplo do conceito 11). Adicione métricas que loguem o tamanho de cada queue intermediária a cada segundo. Force carga acima da capacidade do consumidor final. Observe qual queue cresce primeiro — esse é seu gargalo. Documente a curva e o que faria para resolver.