Trabajando en mi laboratorio en richi.codes, he visto mi parte de problemas complicados a lo largo de los años. Algunos bugs piden a gritos atención, otros son silenciosos, casi educados, hasta que han puesto de rodillas a todo tu sistema. Recientemente, luché con uno de estos asesinos silenciosos en una aplicación de servidor Node.js: una fuga de memoria causada por streams que no respetaban la contrapresión.
Es una de las quejas más repetidas en los hilos de Node.js: una exportación por streams que va bien en el portátil y, con tráfico real, se come la memoria poco a poco hasta que el proceso muere con FATAL ERROR: ... JavaScript heap out of memory. La reproduje en el laboratorio para ver exactamente dónde se va la memoria. Solo aparece con respuestas grandes, lectores lentos y varias peticiones a la vez, por eso se escapa en las pruebas en local.
El Asesino Silencioso: Buffering Sin Control
El núcleo del problema residía en una malinterpretación común de cómo los streams de escritura (writable streams) de Node.js manejan los datos cuando el consumidor es más lento que el productor. Estaba usando fs.createReadStream() para leer archivos grandes, transformar los datos y luego 'pipearlos' (redirigirlos) a un stream de respuesta HTTP (que es un writable stream). En la superficie, esto parecía un ejemplo de libro de texto de streaming eficiente. La promesa de los streams es que procesan datos en fragmentos (chunks), manteniendo el consumo de memoria bajo, independientemente del tamaño total de los datos.
Sin embargo, hay un detalle crucial: la contrapresión. Cuando escribes datos a un stream de escritura usando writable.write(chunk), el método devuelve un valor booleano. Si devuelve false, es una señal de aviso: el búfer interno del stream ha excedido su highWaterMark, y deberías dejar de escribir hasta que el stream emita un evento 'drain'. Si ignoras este valor de retorno false y continúas empujando datos, Node.js seguirá felizmente almacenando esos fragmentos en un búfer en memoria. No perderá datos; simplemente los acumulará. Esta acumulación es precisamente lo que lleva al agotamiento de la memoria.
En mi escenario, el stream de respuesta HTTP era el cuello de botella. Los clientes con redes lentas, o aquellos que simplemente cerraban las pestañas de sus navegadores prematuramente, no podían consumir datos tan rápido como mi servidor podía producirlos desde la base de datos y transformarlos. Debido a que mi código no pausaba el stream de lectura (readable stream) 'upstream' cuando el stream de escritura (writable stream) señalaba false, la memoria del servidor se inflaba constantemente.
Reproducción Mínima: Observa Cómo Escala Tu Memoria
Permíteme mostrarte una versión simplificada del problema. Este servidor Node.js generará un stream continuo de datos e intentará enviarlos a un cliente. Introduciremos un retraso en el lado del cliente para simular un consumidor lento.
Primero, creemos problematic-server.js:
// problematic-server.js
const http = require('http');
const { Readable } = require('stream');
// A custom Readable stream that generates data very quickly
class FastProducer extends Readable {
constructor(options) {
super(options);
this.index = 0;
this.maxChunks = 10000; // Try sending 10,000 chunks
this.chunkSize = 1024 * 1024; // 1MB per chunk
}
_read(size) {
if (this.index < this.maxChunks) {
const chunk = Buffer.alloc(this.chunkSize, `Data chunk ${this.index++} - `);
this.push(chunk);
} else {
this.push(null); // No more data
}
}
}
const server = http.createServer((req, res) => {
if (req.url === '/download-leak') {
console.log(`[${new Date().toLocaleTimeString()}] Client connected for /download-leak`);
res.writeHead(200, {
'Content-Type': 'application/octet-stream',
'Transfer-Encoding': 'chunked',
'X-Content-Type-Options': 'nosniff'
});
const producer = new FastProducer();
let chunks = 0;
producer.on('data', (chunk) => {
// DANGER: We are ignoring the return value of res.write()!
const canWriteMore = res.write(chunk);
if (!canWriteMore) {
// This block is reached, but we aren't doing anything about it.
// The producer will keep pushing data into the buffer.
console.log(`[${new Date().toLocaleTimeString()}] WARNING: Writable buffer is full!`);
// In a real leak scenario, you might not even log this or notice.
}
// Log memory usage periodically
const memory = process.memoryUsage();
if (++chunks % 100 === 0) { // Log every 100 chunks
console.log(`[${new Date().toLocaleTimeString()}] Heap used: ${Math.round(memory.heapUsed / 1024 / 1024)} MB`);
}
});
producer.on('end', () => {
res.end();
console.log(`[${new Date().toLocaleTimeString()}] Download finished for /download-leak`);
});
producer.on('error', (err) => {
console.error(`[${new Date().toLocaleTimeString()}] Producer error:`, err);
res.end('Stream error');
});
res.on('finish', () => {
console.log(`[${new Date().toLocaleTimeString()}] Response finished for /download-leak`);
});
res.on('close', () => {
// This is important for cleanup if the client disconnects early
// but in this problematic code, producer.destroy() is not called
// which can still lead to resources being held.
if (!res.writableEnded) {
console.log(`[${new Date().toLocaleTimeString()}] Client disconnected unexpectedly for /download-leak`);
producer.destroy(); // Manually destroy the producer stream
}
});
} else {
res.writeHead(404);
res.end('Not Found');
}
});
const PORT = 3000;
server.listen(PORT, () => {
console.log(`Server running on http://localhost:${PORT}`);
console.log('Access http://localhost:3000/download-leak to trigger the issue.');
console.log('Monitor memory usage with `node --expose-gc problematic-server.js` and a tool like `top` or `htop`.');
});
Para ver esto en acción, ejecuta el servidor:
node problematic-server.js
Ahora, simula un cliente lento. Puedes hacerlo con curl y un limitador de ancho de banda (throttler), o simplemente descargando un archivo grande e interrumpiéndolo, pero un simple curl a un endpoint de red lento o uno que no lea inmediatamente mostrará el problema.
Para una observación básica, puedes simplemente ejecutar curl -o /dev/null http://localhost:3000/download-leak y observar la memoria del servidor. Mejor aún, simula un lector lento añadiendo un retraso a curl si es posible, o usando una herramienta más sofisticada como pv o throttle.
Alternativamente, puedes observar los logs. Aparecerá el mensaje "WARNING: Writable buffer is full!", y si monitorizas la memoria del proceso Node.js (por ejemplo, usando top o htop en Linux/macOS, o el Administrador de Tareas en Windows), verás cómo sube significativamente.
El problema es que producer.on('data', ...) llama continuamente a res.write(chunk). Incluso cuando res.write() devuelve false, indicando que el búfer de red subyacente está lleno, el producer continúa generando y push() datos. Estos datos se quedan entonces en el búfer interno de res, consumiendo memoria.
La Solución: Respetando la Contrapresión
La solución implica usar el mecanismo de contrapresión incorporado de Node.js. El stream de escritura te dice cuándo está sobrecargado, y el stream de lectura necesita pausarse.
Hay dos formas principales de manejar esto:
pause()yresume()manuales con el evento'drain': Cuandowritable.write()devuelvefalse, llamas areadable.pause(). Cuando el stream de escritura emite un evento'drain'(lo que significa que ha vaciado su búfer interno y está listo para más datos), llamas areadable.resume().- Usando
stream.pipeline(): Este es el enfoque moderno y recomendado.stream.pipeline()está diseñado para manejar la contrapresión y la propagación de errores automáticamente, haciendo que tu código de streaming sea mucho más robusto.
Refactoricemos el servidor para usar stream.pipeline() para un manejo correcto de la contrapresión.
// fixed-server.js
const http = require('http');
const { Readable, pipeline } = require('stream');
const util = require('util');
const streamPipeline = util.promisify(pipeline);
// A custom Readable stream that generates data very quickly
class FastProducer extends Readable {
constructor(options) {
super(options);
this.index = 0;
this.maxChunks = 10000; // Try sending 10,000 chunks
this.chunkSize = 1024 * 1024; // 1MB per chunk
console.log('FastProducer initialized.');
}
_read(size) {
if (this.index < this.maxChunks) {
const chunk = Buffer.alloc(this.chunkSize, `Data chunk ${this.index++} - `);
const canPush = this.push(chunk);
if (!canPush) {
// If push returns false, the internal readable buffer is full.
// This means the downstream consumer isn't pulling fast enough,
// which is exactly what backpressure handles automatically with pipeline.
console.log(`[${new Date().toLocaleTimeString()}] Producer internal buffer full, pausing.`);
}
} else {
console.log(`[${new Date().toLocaleTimeString()}] FastProducer: All chunks pushed.`);
this.push(null); // No more data
}
}
}
const server = http.createServer(async (req, res) => {
if (req.url === '/download-fixed') {
console.log(`[${new Date().toLocaleTimeString()}] Client connected for /download-fixed`);
res.writeHead(200, {
'Content-Type': 'application/octet-stream',
'Transfer-Encoding': 'chunked',
'X-Content-Type-Options': 'nosniff'
});
const producer = new FastProducer();
// Use stream.pipeline for automatic backpressure handling and error propagation
try {
await streamPipeline(
producer,
res
);
console.log(`[${new Date().toLocaleTimeString()}] Stream pipeline finished successfully for /download-fixed`);
} catch (error) {
if (error.code === 'ERR_STREAM_PREMATURE_CLOSE') {
console.warn(`[${new Date().toLocaleTimeString()}] Client disconnected prematurely for /download-fixed.`);
} else {
console.error(`[${new Date().toLocaleTimeString()}] Stream pipeline failed for /download-fixed:`, error);
if (!res.headersSent) {
res.writeHead(500);
}
res.end('Internal Server Error');
}
} finally {
// Ensure the response is always ended, even on error or early close
if (!res.writableEnded) {
res.end();
}
}
// Periodically log memory usage to observe the difference
const memoryInterval = setInterval(() => {
const memory = process.memoryUsage();
console.log(`[${new Date().toLocaleTimeString()}] FIXED Heap used: ${Math.round(memory.heapUsed / 1024 / 1024)} MB`);
}, 5000); // Log every 5 seconds
res.on('finish', () => {
console.log(`[${new Date().toLocaleTimeString()}] Response finished for /download-fixed`);
clearInterval(memoryInterval);
});
res.on('close', () => {
// pipeline handles this destruction for us.
// If client disconnects early, pipeline will emit 'ERR_STREAM_PREMATURE_CLOSE'
// and destroy upstream streams.
if (!res.writableEnded) {
console.log(`[${new Date().toLocaleTimeString()}] Response closed prematurely (client disconnected) for /download-fixed. Pipeline should handle cleanup.`);
}
});
} else {
res.writeHead(404);
res.end('Not Found');
}
});
const PORT = 3001;
server.listen(PORT, () => {
console.log(`Fixed Server running on http://localhost:${PORT}`);
console.log('Access http://localhost:3001/download-fixed to observe proper backpressure.');
console.log('Monitor memory usage with `node fixed-server.js` and a tool like `top` or `htop`.');
});
Ejecuta este servidor corregido:
node fixed-server.js
Ahora, accede a http://localhost:3001/download-fixed con tu simulación de cliente lento. Observarás que el uso de memoria permanece estable, o al menos dentro de los límites esperados, en lugar de subir continuamente. La función stream.pipeline() escucha automáticamente el retorno false de res.write() y pausa el stream producer hasta que res emite 'drain', para luego reanudarlo. También maneja correctamente los errores y las desconexiones prematuras del cliente, asegurando que todos los streams en el pipeline se destruyan adecuadamente para evitar fugas de recursos.
Por Qué Importa
Esto no es solo un ejercicio académico. En aplicaciones Node.js del mundo real, especialmente aquellas que tratan con subidas de archivos, descargas, exportaciones de bases de datos o transformaciones de datos complejas, los streams son fundamentales. Ignorar la contrapresión convierte lo que debería ser una operación eficiente y de baja memoria en una bomba de memoria de tiempo.
La opción highWaterMark en los streams (con un valor predeterminado de 16KB para streams de bytes y 16 objetos para streams en modo objeto en versiones antiguas de Node.js, 64KB en Node 22+) define el umbral en el que se emiten las señales de contrapresión. No es un límite estricto; es una advertencia consultiva. Depende de tu código (o de stream.pipeline()) respetar esa advertencia.
La lección aquí es clara: si bien los streams de Node.js son potentes, su naturaleza cooperativa significa que necesitas gestionar activamente el control de flujo, o delegar esa responsabilidad a herramientas como stream.pipeline(). No hacerlo puede llevar a una degradación del rendimiento, inestabilidad y fugas de memoria que son difíciles de diagnosticar hasta que alcanzan una masa crítica en producción.
Qué Revisaría A Continuación
- Streams de Transformación (Transform Streams): Si tu pipeline incluye streams
Transform, asegúrate de que también respeten las señales de contrapresión de su lado de escritura 'downstream' y pausen correctamente su lado de lectura 'upstream'. - Manejo de Errores: Un manejo adecuado de errores en los streams es crucial.
stream.pipeline()gestiona la propagación y destrucción de errores automáticamente, pero si estás encadenando streams manualmente con.pipe(), necesitarás oyentes de errores explícitos en cada stream. - Streams en Modo Objeto (Object Mode Streams): Ten en cuenta el
highWaterMarken los streams en modo objeto, donde se refiere al número de objetos en lugar de bytes. Los objetos grandes aún pueden consumir una memoria significativa incluso si el recuento de objetos es bajo. - Emisores de Eventos de Larga Vida (Long-lived Event Emitters): Las fugas de memoria también pueden ocurrir cuando se añaden oyentes de eventos a objetos de larga vida (como un servidor o una conexión a la base de datos) pero nunca se eliminan, lo que provoca la acumulación de referencias. Este es un patrón de fuga de memoria común en Node.js, separado pero relacionado.