gRPC Streaming: fluxos eficientes no Node.js

gRPC Streaming no Node.js permite trocar sequências de mensagens sobre uma chamada HTTP/2. O protocolo oferece server streaming, client streaming e streaming bidirecional, além do modelo unary tradicional.

Streaming reduz a necessidade de polling e permite processar dados incrementalmente. Porém, conexões longas exigem backpressure, deadlines, cancelamento, controle de memória e uma estratégia clara de retomada depois de falhas.

Tipos de streaming

• server streaming: uma requisição, várias respostas;
• client streaming: várias requisições, uma resposta;
• bidirectional streaming: mensagens nos dois sentidos;
• unary: uma requisição e uma resposta.

Contrato Protobuf

service ReportService {
  rpc ExportReport(ExportRequest) returns (stream ReportChunk);
  rpc ImportRows(stream ImportRow) returns (ImportResult);
  rpc Sync(stream SyncMessage) returns (stream SyncMessage);
}

message ReportChunk {
  bytes data = 1;
  int64 sequence = 2;
}

Use mensagens pequenas e numeradas. Para arquivos muito grandes, object storage pode ser mais simples.

Server streaming

async function exportReport(call) {
  try {
    for await (const chunk of reportService.generate(call.request)) {
      if (call.cancelled) break;

      const canContinue = call.write({
        data: chunk,
        sequence: String(chunk.sequence)
      });

      if (!canContinue) {
        await new Promise(resolve => call.once('drain', resolve));
      }
    }

    call.end();
  } catch (error) {
    call.destroy(toGrpcError(error));
  }
}

O retorno de write() informa backpressure. Ignorá-lo pode acumular chunks em memória quando o cliente é lento.

Cancelamento pelo cliente

O servidor deve observar cancelamento e interromper banco, arquivos e chamadas externas:

const controller = new AbortController();

call.on('cancelled', () => {
  controller.abort(new Error('cliente cancelou'));
});

await generateReport({ signal: controller.signal });

Não continue produzindo dados que ninguém consumirá.

Client streaming

function importRows(call, callback) {
  let count = 0;
  let failed = false;

  call.on('data', async row => {
    if (failed) return;

    try {
      validateRow(row);
      await importer.add(row);
      count += 1;
    } catch (error) {
      failed = true;
      call.destroy(toGrpcError(error));
    }
  });

  call.on('end', async () => {
    if (failed) return;
    await importer.finish();
    callback(null, { imported: String(count) });
  });
}

Callbacks assíncronos em eventos podem executar em paralelo sem intenção. Use uma fila sequencial ou pause o stream enquanto processa.

Pausando o consumo

call.pause();
try {
  await processRow(row);
} finally {
  call.resume();
}

Outra opção é transformar eventos em um iterador assíncrono com buffer limitado.

Streaming bidirecional

function sync(call) {
  call.on('data', async message => {
    const result = await syncService.apply(message);
    call.write(result);
  });

  call.on('end', () => call.end());
}

Mensagens podem cruzar nos dois sentidos de forma independente. Não suponha que resposta N corresponde automaticamente à requisição N. Inclua requestId ou sequência.

Ordenação

HTTP/2 preserva ordem dos bytes no stream, mas processamento concorrente pode reordenar respostas. Se a ordem importa, use uma fila por chamada ou números de sequência.

Resume e checkpoints

Uma conexão pode cair depois de centenas de mensagens. Para retomar, o protocolo precisa de cursor:

message ExportRequest {
  string report_id = 1;
  int64 after_sequence = 2;
}

O servidor deve produzir chunks determinísticos ou manter checkpoints duráveis.

Idempotência

No client streaming, uma retomada pode reenviar linhas. Inclua uma chave estável por item e aplique constraint única. No streaming bidirecional, deduplique por messageId.

Deadlines

Streams longos também precisam de deadline ou política de inatividade. Um deadline global de horas pode ser adequado para exportação, enquanto um timeout de idle detecta conexões sem progresso.

const call = client.exportReport(request, {
  deadline: Date.now() + 15 * 60_000
});

Idle timeout

Reinicie um timer a cada mensagem:

let idleTimer;

function touch() {
  clearTimeout(idleTimer);
  idleTimer = setTimeout(() => call.cancel(), 30_000);
}

call.on('data', chunk => {
  touch();
  consume(chunk);
});

Tamanho de mensagem

Mesmo em streaming, cada mensagem possui limite. Prefira chunks entre dezenas e centenas de kilobytes, medindo throughput e overhead. Não transforme um arquivo inteiro em um único bytes.

Compressão

Compressão reduz rede, mas consome CPU e pode atrasar o primeiro chunk. Teste com dados reais. Conteúdo já comprimido, como JPEG ou ZIP, raramente se beneficia.

Controle de concorrência

Limite streams ativos por cliente, tenant e instância. Uma conexão longa consome memória, sockets e recursos externos.

if (activeStreamsForTenant >= maxStreams) {
  throw grpcError(grpc.status.RESOURCE_EXHAUSTED, 'limite atingido');
}

Autenticação e autorização

Autentique no início, mas considere expiração de credenciais em streams muito longos. Para operações sensíveis, limite duração ou implemente renovação explícita.

Autorize cada canal lógico e recurso solicitado. Não permita que uma mensagem subsequente altere o tenant estabelecido na abertura.

Erros parciais

Depois de enviar parte dos dados, um erro encerra a chamada. O cliente precisa saber quais itens foram confirmados. Use sequência e confirmação de checkpoint quando a recuperação for importante.

Trailers e metadata

Metadata inicial pode descrever versão e compressão. Trailers podem informar contagens finais e checksum, desde que o stream conclua normalmente.

Observabilidade

Monitore:

• streams ativos;
• duração;
• mensagens e bytes;
• tempo entre mensagens;
• backpressure;
• cancelamentos;
• deadlines;
• retomadas;
• erros por método.

Crie um span para a chamada e eventos ou métricas para progresso, evitando um span por chunk em fluxos enormes.

Load balancing

Um stream permanece na mesma conexão e instância. Deploys e rebalanceamentos devem permitir drenagem. Novos streams vão para instâncias prontas; os existentes precisam de janela para terminar ou retomar.

Shutdown gracioso

Marque a instância como não pronta, recuse novos streams, aguarde os ativos até o deadline e encerre os restantes com status apropriado. O cliente deve retomar a partir do último checkpoint.

Testes

• server streaming com vários chunks;
• cliente lento e evento drain;
• cancelamento;
• deadline;
• client streaming com payload inválido;
• bidirecional fora de ordem;
• queda e retomada;
• mensagem duplicada;
• limite de streams;
• shutdown durante exportação.

Erros comuns

• ignorar backpressure;
• buffer ilimitado;
• sem cursor de retomada;
• processamento assíncrono desordenado;
• stream sem deadline;
• mensagens gigantes;
• não cancelar dependências;
• autorizar apenas a abertura;
• não limitar conexões por tenant.

Conclusão

gRPC Streaming no Node.js permite transportar dados incrementalmente e manter comunicação bidirecional eficiente. O principal desafio não é escrever no stream, mas controlar fluxo, falhas e retomada.

Preserve backpressure, use deadlines, propague cancelamento e inclua IDs ou sequências. Com checkpoints e idempotência, conexões longas deixam de ser frágeis diante de deploys e interrupções.

Consulte a documentação oficial dos tipos de streaming gRPC e a referência de streams do grpc-node.