La arquitectura de ejecución de Node.js está diseñada para procesar operaciones de entrada/salida de red con un consumo mínimo de hardware. Bajo un modelo de buffer monolítico convencional (como la lectura con fs.readFile), la totalidad de los datos se carga en la memoria del proceso. Si un servidor procesa concurrentemente 100 peticiones de subida de archivos de 500 MB cada una, el proceso demandará 50 GB de RAM, lo que provocará una caída fatal por JavaScript heap out of memory o la terminación por el OOM Killer del kernel del sistema operativo.
Los flujos de datos (Streams) implementan el principio fundamental de procesamiento continuo acotado: dividir un flujo binario en fragmentos discretos (chunks) y procesarlos secuencialmente a medida que ingresan. El consumo de memoria del proceso queda limitado al tamaño de los chunks activos en tránsito (apenas unas decenas de kilobytes), sin importar si el archivo subyacente pesa 100 MB o 100 GB, o cuánto tiempo permanezca abierta la conexión de red.
El núcleo de todos los streams de Node.js reside en el módulo node:stream, estructurado sobre el patrón EventEmitter y una lista enlazada de buffers (BufferList). La plataforma clasifica los streams en cuatro tipos fundamentales:
Readable: Fuente de datos desde la que se pueden leer bytes (descriptor de lectura de archivos con fs.createReadStream, cuerpo de petición entrante http.IncomingMessage, socket de red).Writable: Receptor o destino en el que se pueden escribir bytes (descriptor de escritura con fs.createWriteStream, respuesta HTTP saliente http.ServerResponse, salida estándar process.stdout).Duplex: Stream bidireccional que combina canales independientes de lectura y escritura (por ejemplo, un socket de red TCP net.Socket).Transform: Subclase especializada de Duplex donde los datos salientes se calculan algorítmicamente a partir de los datos entrantes (por ejemplo, compresión con zlib.createGzip(), cifrado con crypto.createCipheriv() o parsers de formatos).Cada stream mantiene estructuras internas de estado: _readableState para streams de lectura y _writableState para streams de escritura. El parámetro esencial de configuración es highWaterMark:
fs.createReadStream) es también de 64 KB.objectMode: true), el umbral por defecto es de 16 objetos de JavaScript.highWaterMark no es un límite estricto de memoria, sino un umbral de señalización. Si el receptor es más lento que el emisor, los chunks continuarán acumulándose en memoria superando el highWaterMark, a menos que se aplique un control explícito de contrapresión (Backpressure).
El fenómeno de Backpressure surge en cualquier sistema asíncrono o distribuido cuando la tasa de generación de datos del productor supera la capacidad de procesamiento o el ancho de banda del canal del consumidor:
Tasa de Generación del Productor >> Capacidad de Procesamiento del Consumidor
(ejemplo: Disco local NVMe a 3 GB/s) (ejemplo: Cliente sobre conexión 3G a 50 KB/s)Sin un mecanismo de regulación, la memoria del servidor se desbordará con chunks en espera de ser enviados al socket. Node.js gestiona la contrapresión mediante un protocolo en dos fases:
writable.write(chunk) devuelve un valor booleano. Si el tamaño del buffer interno del receptor supera highWaterMark, write() devuelve false. Esto indica imperativamente al productor que debe pausar la lectura de inmediato (readable.pause()).'drain'. Al capturar 'drain', el productor reanuda la lectura (readable.resume()).El método clásico readable.pipe(writable) se utilizaba habitualmente para encadenar flujos, pero en la actualidad se considera un antipatrón en entornos empresariales por los siguientes motivos:
pipe() no propaga los eventos de error a lo largo de la cadena. Si un stream intermedio o el receptor fallan, el resto de los streams permanecen abiertos. Los descriptores de archivos quedan bloqueados, provocando el error EMFILE: too many open files.pipe() no se integra con AbortController y no puede ser abortado de forma limpia ante desconexiones de clientes.'error', 'close' y 'finish' en cada eslabón del pipeline.El estándar moderno en producción es la función stream.promises.pipeline, que garantiza transaccionalidad: si se produce un error en cualquier eslabón o se emite un AbortSignal, todos los streams de la cadena se destruyen de inmediato (destroy(err)), cerrando descriptores de archivos y liberando memoria para el Garbage Collector.
El control de flujo con contrapresión puede representarse como una esclusa hidráulica con válvula de despresurización automática:
[ Readable Stream ] (Origen en disco: 500 MB/s)
|
| 1. chunk (64 KB)
v
[ Writable Stream ] (Cola interna: capacidad highWaterMark = 64 KB)
+-------------------------------------------------------+
| Estado cola: [ XXXXXXXX XXXXXXXX XXXXXXXX XXXXXXXX ] | -> ¡Buffer lleno!
+-------------------------------------------------------+
|
+---> writable.write(chunk) === false
|
v
[ readable.pause() ] <--- El productor PAUSA la lectura desde el disco
|
(El stack de red del SO transmite paquetes TCP al cliente...)
|
+-------------------------------------------------------+
| Buffer liberado: [ ] |
+-------------------------------------------------------+
|
v
[ emit('drain') ]
|
+---> [ readable.resume() ] <--- ¡El productor REANUDA la emisión!zlib.createBrotliCompress()), adaptando el throughput al ancho de banda del cliente.El siguiente stream Transform de grado de producción procesa flujos binarios de CSV, gestiona divisiones de líneas en los bordes de los chunks y emite objetos estructurados:
import { Transform, TransformCallback } from 'node:stream';
export interface CsvTransformOptions {
delimiter?: string;
hasHeader?: boolean;
}
export class CsvToJsonTransform extends Transform {
private delimiter: string;
private hasHeader: boolean;
private headers: string[] = [];
private leftover: string = '';
private isFirstChunk: boolean = true;
constructor(options: CsvTransformOptions = {}) {
super({
readableObjectMode: true, // Emite objetos JavaScript hacia downstream
writableObjectMode: false, // Acepta buffers binarios crudos desde upstream
highWaterMark: 64 * 1024, // Umbral de 64 KB de buffer
});
this.delimiter = options.delimiter ?? ',';
this.hasHeader = options.hasHeader ?? true;
}
/**
* Se ejecuta por cada chunk binario entrante.
*/
public _transform(chunk: Buffer, encoding: BufferEncoding, callback: TransformCallback): void {
try {
// Concatenamos el resto de texto del chunk anterior
const content = this.leftover + chunk.toString('utf-8');
const lines = content.split(/\r?\n/);
// El último elemento puede ser una línea incompleta en el borde del chunk
this.leftover = lines.pop() ?? '';
for (const line of lines) {
const trimmed = line.trim();
if (!trimmed) continue;
const columns = trimmed.split(this.delimiter).map((col) => col.trim());
if (this.isFirstChunk && this.hasHeader) {
this.headers = columns;
this.isFirstChunk = false;
continue;
}
const record: Record<string, string> = {};
for (let i = 0; i < columns.length; i++) {
const key = this.headers[i] ?? `col_${i}`;
record[key] = columns[i];
}
// Enviamos el objeto procesado al siguiente eslabón
this.push(record);
}
callback(); // Preparado para el siguiente chunk
} catch (err) {
callback(err instanceof Error ? err : new Error(String(err)));
}
}
/**
* Se ejecuta una sola vez cuando la fuente agota los datos.
*/
public _flush(callback: TransformCallback): void {
try {
if (this.leftover.trim()) {
const columns = this.leftover.trim().split(this.delimiter).map((col) => col.trim());
const record: Record<string, string> = {};
for (let i = 0; i < columns.length; i++) {
const key = this.headers[i] ?? `col_${i}`;
record[key] = columns[i];
}
this.push(record);
}
callback();
} catch (err) {
callback(err instanceof Error ? err : new Error(String(err)));
}
}
}En versiones modernas de Node.js, el consumo de streams Readable se realiza de forma limpia mediante for await (const chunk of stream). El runtime sincroniza automáticamente la contrapresión con la ejecución de cada vuelta del bucle:
import { createReadStream, createWriteStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';
import { Transform } from 'node:stream';
async function processBigDataset(sourcePath: string, targetPath: string, signal: AbortSignal): Promise<void> {
const readStream = createReadStream(sourcePath, { highWaterMark: 64 * 1024 });
const writeStream = createWriteStream(targetPath);
// Transform personalizado para enmascarar información sensible
const anonymizer = new Transform({
transform(chunk: Buffer, encoding, callback) {
const sanitized = chunk.toString('utf-8').replace(/\b\d{4}-\d{4}-\d{4}-\d{4}\b/g, 'XXXX-XXXX-XXXX-XXXX');
callback(null, Buffer.from(sanitized));
}
});
try {
await pipeline(
readStream,
anonymizer,
writeStream,
{ signal } // Cancelación segura mediante AbortSignal externo
);
console.log('Pipeline completado sin fugas de descriptores');
} catch (error) {
if ((error as { name?: string }).name === 'AbortError') {
console.warn('Procesamiento de archivo cancelado por señal externa');
} else {
console.error('Fallo crítico en la cadena de streams:', error);
throw error;
}
}
}El stream de lectura carga bloques binarios de 64 KB desde el disco a su cola interna de memoria.
El stream Transform parsea delimitadores, sanea campos y genera registros NDJSON serializados.
writable.write() devuelve false cuando la cola de destino supera highWaterMark. El origen entra en pause().
El stack de red del sistema operativo transmite los paquetes TCP al cliente, vaciando la cola.
El stream Writable emite el evento drain. La fuente llama a resume() y continúa con el siguiente bloque.
Un procesador de logs en streaming de nivel empresarial con control de contrapresión, saneamiento de datos sensibles, compresión Gzip y apagado limpio:
import { Readable, Writable, Transform, TransformCallback } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createGzip } from 'node:zlib';
export interface LogEntry {
timestamp: string;
level: 'info' | 'warn' | 'error' | 'fatal';
service: string;
message: string;
context?: Record<string, unknown>;
}
export interface ProcessorMetrics {
totalProcessed: number;
filteredOut: number;
bytesWritten: number;
backpressurePauses: number;
}
export class HighThroughputLogProcessor {
private metrics: ProcessorMetrics = {
totalProcessed: 0,
filteredOut: 0,
bytesWritten: 0,
backpressurePauses: 0,
};
/**
* Crea un stream Transform para validación y saneamiento de PII.
*/
private createSanitizerTransform(): Transform {
const self = this;
return new Transform({
objectMode: true,
highWaterMark: 256, // 256 registros en cola
transform(entry: LogEntry, encoding: BufferEncoding, callback: TransformCallback) {
self.metrics.totalProcessed++;
// Filtrado de logs ruidosos de comprobación de salud
if (entry.level === 'info' && entry.message.includes('HEALTHCHECK')) {
self.metrics.filteredOut++;
return callback(); // Omitimos el registro sin llamar a push
}
// Saneamiento de campos confidenciales (PII)
if (entry.context) {
if ('password' in entry.context) entry.context.password = '[REDACTED]';
if ('token' in entry.context) entry.context.token = '[REDACTED]';
if ('authorization' in entry.context) entry.context.authorization = '[REDACTED]';
}
// Serialización a Newline Delimited JSON
const serialized = JSON.stringify(entry) + '\n';
callback(null, Buffer.from(serialized, 'utf-8'));
},
});
}
/**
* Envuelve el stream receptor para monitorizar eventos de contrapresión.
*/
private createMetricsMonitor(targetWritable: Writable): Writable {
const self = this;
const monitored = new Writable({
highWaterMark: 64 * 1024,
write(chunk: Buffer, encoding: BufferEncoding, callback) {
self.metrics.bytesWritten += chunk.byteLength;
const canContinue = targetWritable.write(chunk, encoding, callback);
if (!canContinue) {
self.metrics.backpressurePauses++;
targetWritable.once('drain', () => {
// Notificación de vaciado
});
}
},
final(callback) {
targetWritable.end(callback);
},
});
return monitored;
}
/**
* Ejecuta un pipeline completo y resiliente con compresión GZIP.
*/
public async executePipeline(
sourceStream: Readable,
destinationStream: Writable,
signal?: AbortSignal
): Promise<ProcessorMetrics> {
const sanitizer = this.createSanitizerTransform();
const gzip = createGzip({ level: 6 });
const monitoredTarget = this.createMetricsMonitor(destinationStream);
try {
await pipeline(
sourceStream,
sanitizer,
gzip,
monitoredTarget,
{ signal }
);
return { ...this.metrics };
} catch (error) {
if ((error as { name?: string }).name === 'AbortError') {
throw new Error('Log processing pipeline aborted by signal');
}
throw error;
}
}
public getMetrics(): ProcessorMetrics {
return { ...this.metrics };
}
}Llamar a stream.write(chunk) dentro de un bucle sin comprobar si devuelve false provoca un consumo descontrolado de memoria. Si el receptor está saturado, los datos continúan acumulándose en la RAM del proceso. Con tráfico elevado, el servidor colapsará por Out Of Memory.
Pausa siempre la escritura cuando stream.write() devuelva false y aguarda al evento 'drain' (await once(stream, 'drain')) antes de transmitir nuevos datos.
Asignar un listener con stream.on('data', handler) sitúa al stream en "modo fluido" (flowing mode), deshabilitando la contrapresión. Intentar consumir simultáneamente ese stream con for await (const chunk of stream) provoca pérdida de chunks, estados de carrera y datos corruptos.
Nunca mezcles estilos de lectura. Adopta exclusivamente for await ... of, stream.promises.pipeline() o el consumo manual mediante stream.on('readable').
Olvidar llamar a callback() en alguna rama lógica de un stream Transform congela la cadena de forma indefinida. Node.js detiene el envío de nuevos chunks y no emitirá los eventos 'finish' o 'end'. El pipeline quedará bloqueado de forma permanente.
Asegúrate de invocar callback(err) o callback(null, output) en todos los caminos lógicos, envolviendo el cuerpo en un bloque try ... catch.
Encadenar flujos mediante readStream.pipe(transform).pipe(writeStream) en un servidor web ocasiona fugas de descriptores. Si el cliente interrumpe abruptamente la conexión TCP, el stream de salida falla, pero pipe() mantiene abierto readStream. El descriptor de archivo del sistema operativo queda bloqueado hasta que el servicio arroje el error EMFILE.
Usa siempre stream.promises.pipeline(), que asegura la destrucción atómica de todos los eslabones ante fallos.
El origen invoca writable.write(chunk) transmitiendo un buffer binario o un objeto.
El runtime comprueba el tamaño de la cola: si excede el umbral, el método write() devuelve false.
El stream Readable detiene la lectura desde el sistema operativo mediante pause().
El socket de red o el controlador de disco transmite los datos acumulados al hardware.
El stream Writable emite el evento drain, lo que lleva a la fuente a invocar resume() y continuar.