Backpressure en streams de Node.js: por qué un pipe mal configurado tira abajo un microservicio de ingesta

Cada vez que investigamos un microservicio de ingesta que empieza a consumir memoria sin límite hasta que el proceso muere con un OutOfMemory o el orquestador lo reinicia por exceder su límite de RAM, el patrón se repite: alguien escribió datos hacia un stream de salida sin comprobar si ese stream estaba listo para recibirlos. El código funciona en desarrollo, donde los volúmenes son pequeños y la fuente nunca produce más rápido de lo que el destino puede consumir. Falla en producción, donde un archivo grande, una conexión lenta hacia el destino o un pico de tráfico rompen esa suposición implícita. El mecanismo que existe justo para evitar esto se llama backpressure, y Node lo expone desde siempre en su API de streams. El problema no es que falte la herramienta, es que pipe() deja demasiado margen para ignorarla.

Qué es el backpressure en un stream

Un stream de Node conecta una fuente de datos con un consumidor, y ambos lados rara vez procesan a la misma velocidad. Un Readable que lee de un socket TCP o de un archivo en disco puede producir chunks mucho más rápido de lo que un Writable que escribe hacia otro servicio por HTTP, hacia un disco lento o hacia una base de datos puede aceptarlos. Sin ningún mecanismo de control, el lado lector seguiría empujando datos hacia el escritor, y esos datos se acumularían en un buffer interno mientras esperan a ser procesados. Ese buffer no tiene un límite natural: crece con cada chunk que llega antes de que el anterior termine de escribirse, y ese crecimiento sin freno es exactamente lo que agota la memoria del proceso.

Backpressure es la señal que un stream Writable envía hacia atrás cuando su buffer interno alcanza el highWaterMark configurado —16 KB por defecto para streams en modo binario—. El método write() de un Writable devuelve false cuando esto ocurre, y esa devolución es una instrucción explícita: deja de escribir hasta que el stream emita el evento drain, que indica que el buffer bajó lo suficiente como para aceptar más datos.

const { createWriteStream } = require('fs');

const destino = createWriteStream('/var/data/salida.log');

function escribirLote(datos, indice) {
  if (indice >= datos.length) return;

  const puedeSeguir = destino.write(datos[indice]);

  if (puedeSeguir) {
    escribirLote(datos, indice + 1);
  } else {
    destino.once('drain', () => escribirLote(datos, indice + 1));
  }
}

Este patrón manual funciona, pero nadie lo escribe así en un microservicio real con varias transformaciones encadenadas. Ahí es donde entra pipe(), y ahí es donde empieza el problema que da título a este artículo.

Por qué pipe() no es suficiente por sí solo

pipe() respeta el backpressure del lado del flujo de datos: si el Writable de destino devuelve false, pipe() pausa automáticamente el Readable de origen hasta el próximo drain. Hasta ahí, hace exactamente lo que promete. El problema aparece en el manejo de errores, que es donde la mayoría de los microservicios de ingesta realmente fallan.

const { createReadStream, createWriteStream } = require('fs');
const { createGunzip } = require('zlib');

const origen = createReadStream('/tmp/entrada.csv.gz');
const descompresor = createGunzip();
const destino = createWriteStream('/var/data/salida.csv');

origen.pipe(descompresor).pipe(destino);

Si destino emite un error —por ejemplo, porque el disco se llenó o la conexión de red que respalda ese stream se cayó— pipe() no propaga ese error hacia origen ni hacia descompresor. Cada stream de la cadena queda con su propio listener de error, y si no se instaló uno en cada uno de ellos, Node lanza una excepción no capturada por el stream que sí tiene el error sin control, mientras los streams anteriores de la cadena quedan sin cerrar, sin liberar sus descriptores de archivo ni sus buffers. En un microservicio que procesa miles de archivos por hora, ese descriptor de archivo colgado y ese buffer sin liberar se acumulan hasta que el proceso se queda sin memoria o sin file descriptors disponibles, y el síntoma que se reporta es "memory leak intermitente", no "faltó un manejador de error en un stream intermedio de una cadena de pipes".

El otro problema de pipe() es que no cierra automáticamente los streams de la cadena cuando uno termina antes que los demás. Si origen termina de emitir datos pero destino todavía tiene el buffer lleno procesando el último lote, la limpieza de recursos queda en manos de quien escribió el código, y es habitual que nadie la escriba porque en las pruebas locales, con archivos pequeños, nunca llega a manifestarse.

pipeline(): el mismo backpressure, con el ciclo de vida resuelto

stream.pipeline(), disponible desde Node 10 y con soporte para promesas desde Node 15 vía stream/promises, resuelve exactamente estos dos puntos sin cambiar el mecanismo de backpressure subyacente, que sigue siendo el mismo write() que devuelve false y el mismo evento drain.

const { pipeline } = require('stream/promises');
const { createReadStream, createWriteStream } = require('fs');
const { createGunzip } = require('zlib');

async function procesarArchivo(rutaEntrada, rutaSalida) {
  try {
    await pipeline(
      createReadStream(rutaEntrada),
      createGunzip(),
      createWriteStream(rutaSalida)
    );
    console.log('procesamiento completo');
  } catch (error) {
    console.error('fallo en la cadena de streams', error);
  }
}

pipeline() propaga el error de cualquier stream de la cadena hacia un único punto de manejo, y además destruye todos los streams involucrados —llamando a su destroy() interno— en cuanto cualquiera de ellos falla o termina, sin importar en qué posición de la cadena ocurrió. Esto es lo que elimina los descriptores de archivo colgados y los listeners huérfanos que pipe() deja atrás cuando algo sale mal a mitad de la cadena.

El patrón típico que satura memoria en un microservicio de ingesta

El caso que vemos con más frecuencia no es un pipe() simple como el del ejemplo anterior, sino un microservicio que recibe un stream de entrada por HTTP, lo transforma con un Transform propio, y lo reenvía hacia otro servicio también por HTTP:

app.post('/ingesta', (req, res) => {
  const transformador = new MiTransform();

  req.pipe(transformador).pipe(destinoRemoto);

  destinoRemoto.on('finish', () => res.status(200).end());
});

Si destinoRemoto es un stream hacia una conexión HTTP saliente que se vuelve lenta —por ejemplo, porque el servicio río abajo empieza a degradarse bajo carga— el backpressure se propaga correctamente hacia req gracias a pipe(), y hasta ahí no hay leak. El problema aparece cuando ese Transform personalizado tiene un buffer propio dentro de su método _transform que no respeta el callback de forma correcta: si el Transform llama a push() con más datos de los que ha recibido, o si acumula chunks en un array interno antes de emitirlos "para procesarlos en lote" sin verificar el valor de retorno de push(), ese array crece sin que ningún backpressure lo detenga, porque el backpressure de streams solo controla el flujo entre streams nativos, no las estructuras de datos internas que un Transform mal escrito decide mantener por su cuenta.

class MiTransformMalo extends Transform {
  _transform(chunk, encoding, callback) {
    this.buffer = this.buffer || [];
    this.buffer.push(procesar(chunk));

    if (this.buffer.length >= 1000) {
      for (const item of this.buffer) this.push(item);
      this.buffer = [];
    }
    callback();
  }
}

Este Transform acumula hasta mil elementos antes de emitir nada, y mientras acumula, sigue aceptando chunks de req porque callback() se llama en cada invocación sin importar cuánto haya crecido this.buffer. Si destinoRemoto es lento y el flujo de entrada es constante, este buffer interno crece indefinidamente porque nada en la API de streams sabe que existe: es responsabilidad exclusiva del código del Transform respetar backpressure hacia adentro, no solo dejarlo pasar hacia atrás.

La corrección no es compleja, pero exige entender que push() dentro de un Transform también devuelve false cuando el lado de lectura está saturado, y que acumular datos "para procesar en lote" dentro de un stream anula la garantía que el resto de la cadena da por hecha:

class MiTransformCorregido extends Transform {
  _transform(chunk, encoding, callback) {
    this.push(procesar(chunk));
    callback();
  }
}

Emitir cada chunk procesado de inmediato, sin acumulación intermedia, deja que el backpressure fluya de extremo a extremo de la cadena sin puntos ciegos.

Qué revisar antes de dar por buena una cadena de streams

Reemplazar pipe() por pipeline() resuelve la limpieza de recursos y la propagación de errores, pero no sustituye la revisión de cada Transform personalizado en la cadena: cualquier acumulación interna de datos que no pase por push() en cuanto está lista es un punto donde el backpressure deja de propagarse, sin importar qué tan bien esté escrito el resto del pipeline. La lista de verificación que aplicamos antes de aprobar un microservicio de ingesta nuevo es corta: usar pipeline() en lugar de pipe() encadenado a mano, revisar que ningún Transform acumule chunks sin verificar backpressure, y confirmar que el highWaterMark de cada stream esté dimensionado para el tamaño de chunk real que produce la fuente, no para el valor por defecto que asumió quien escribió el código sin medir el tráfico real del servicio.