Flujos de Node.js más allá de lo básico: memoria, contrapresión y fallos reales
Aprenda cómo las transmisiones de Node.js interactúan con Web Streams, los ahorros de memoria medidos mediante pruebas reales, y los errores en producción que solo aparecen bajo carga.
Fuentes: Node.js Streams in Production: The Parts Most Tutorials Skip
Muchas introducciones a las transmisiones de Node.js explican bien los conceptos básicos: lectura, escritura, modo dúplex, transformación, pipe(), contrapresión, una demostración con gzip y, a veces, un servidor de solicitudes por rango de video como cierre. Si ya has estudiado una o dos de ellas, entenderás qué es una transmisión y por qué cargar un archivo de 2 GB completo en la memoria durante una demostración en vivo es un problema.
Lo que esas introducciones rara vez abordan es lo que viene a continuación. ¿Cuánta memoria se ahorra realmente al pasar a los flujos, con cifras concretas en lugar de una vaga expresión de “menos”? ¿Cómo interactúan los flujos nativos de Node con la API Web Streams utilizada por fetch y los navegadores? ¿Cómo se pueden transmitir los resultados de un modelo que corre en su propio hardware en lugar de pagar por token a través de una API alojada? ¿Y cuáles son los dos o tres errores que terminan arruinando un pipeline que parecía funcionar bien en pruebas cuando se enfrenta al tráfico real?
Este artículo trata exactamente de eso. Supone que ya sabe qué es un flujo legible. Si no es así, el breve resumen a continuación le ayudará a ponerse al día rápidamente. Todo lo que sigue se basa en esos fundamentos.
Resumen de 60 segundos
Readable -> produces data (fs.createReadStream, HTTP request body)
Writable -> consumes data (fs.createWriteStream, HTTP response)
Duplex -> both, independently (a TCP socket)
Transform -> consumes, changes, produces (zlib.createGzip)
Readable stream -> Transform stream -> Writable stream
Los datos fluyen a través del sistema en pequeñas partes en lugar de como un único bloque grande. pipe() conecta las corrientes entre sí y gestiona la situación en la que el destino no puede seguir el ritmo del origen, lo cual se conoce como contrapresión. pipeline() logra la misma conexión, pero también transmite los errores de cualquier corriente de la cadena y garantiza su limpieza cuando ocurre un fallo, por lo que es la función que se desea utilizar en cualquier servicio de larga duración en lugar de en un script temporal.
import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
await pipeline(
createReadStream('./input.txt'),
createGzip(),
createWriteStream('./input.txt.gz')
);
Si ese ejemplo ya le resulta completamente comprensible, puede saltárselo. Lo que sigue es el contenido necesario conocer antes de introducir código basado en corrientes por primera vez en un entorno de producción.
El número que nadie menciona en estos artículos
La mayoría de los artículos sobre transmisiones afirman que el enfoque basado en búferes “utiliza mucha memoria”, mientras que el enfoque de transmisión continua “usa mucho menos”, sin nunca incluir una cifra que respalde ninguna de estas afirmaciones. Para solucionarlo, se generó un archivo de texto plano de 501 MB y se ejecutaron ambos enfoques en la misma máquina, uno tras otro, dos veces cada uno.
// buffered.js: read the whole file, then scan it
import { readFileSync } from 'node:fs';
const before = process.memoryUsage().rss;
const data = readFileSync('./big.txt', 'utf8');
const afterRead = process.memoryUsage().rss;
let count = 0;
for (let i = 0; i < data.length; i++) if (data[i] === 'q') count++;
console.log({
rssBeforeMB: (before / 1024 / 1024).toFixed(1),
rssAfterReadMB: (afterRead / 1024 / 1024).toFixed(1),
count
});
// streamed.js: read in 64KB chunks, scan as it arrives
import { createReadStream } from 'node:fs';
let peakRss = 0;
const timer = setInterval(() => {
peakRss = Math.max(peakRss, process.memoryUsage().rss);
}, 10);
let count = 0;
const stream = createReadStream('./big.txt', { encoding: 'utf8', highWaterMark: 64 * 1024 });
stream.on('data', (chunk) => {
for (let i = 0; i < chunk.length; i++) if (chunk[i] === 'q') count++;
});
stream.on('end', () => {
clearInterval(timer);
console.log({ peakRssMB: (peakRss / 1024 / 1024).toFixed(1), count });
});
Esto es lo que realmente resultó al ejecutarlos en un archivo ASCII de 500 MB con Node.js 22, dos ejecuciones por enfoque:
Approach Peak RSS Wall time Notes
--------------------------------------------------------------------
readFileSync (buffered) ~539 MB 1.24 - 1.76s Full file materialized as a string
createReadStream, 64KB chunks ~52 MB 1.56 - 1.67s Roughly 10x less peak memory
createReadStream, 1MB chunks ~50 MB 1.25 - 1.34s Bigger chunks, same ballpark memory
Algunas consideraciones importantes que vale la pena mencionar claramente, ya que omitirlas haría que los resultados fueran engañosos: esto refleja una sola máquina, un solo archivo y una sola carga de trabajo, que consiste en un escaneo de caracteres trivial. El tiempo real transcurrido fue aproximadamente igual entre los dos enfoques, siendo que la versión por streaming incluso resultó ligeramente más rápida en algunos casos, lo cual fue algo inesperado. La suposición inicial era que la sobrecarga de procesar muchos fragmentos pequeños ralentizaría las cosas más de lo que realmente ocurrió. Por otro lado, los valores de memoria coincidieron estrechamente con las expectativas: la ejecución con búfer alcanzó un pico cercano al tamaño total del archivo, mientras que la ejecución por streaming alcanzó un valor bajo que se mantuvo prácticamente constante independientemente del tamaño del archivo. Ese contraste resume realmente todo el caso a favor del uso de flujos de datos, condensado en una sola tabla; además, es fácil verificarlo de forma independiente ejecutando los dos scripts anteriores sobre un archivo.
el tuyo propio.La implicación práctica: para una única solicitud atendida por una máquina lo suficientemente potente, leer todo el archivo en memoria suele ser suficiente. El verdadero problema nunca ha sido una sola solicitud de forma aislada; se trata de la siguiente pregunta que siempre surge: ¿qué ocurre cuando hay 100 solicitudes ejecutándose simultáneamente? El uso de memoria con el enfoque de transmisión en flujo apenas cambia a medida que aumenta la concurrencia, ya que cada solicitud individual solo almacena un búfer del tamaño de un fragmento y no todo el contenido a la vez.
Node streams y Web Streams ahora pertenecen a la misma familia
Durante años, las transmisiones de Node.js y la API Streams de WHATWG (ReadableStream, WritableStream, TransformStream —las clases que subyacen a fetch() tanto en el navegador como en Node) parecían dos ecosistemas separados con poca superposición. Eso ya no es cierto. El módulo node:stream/web expone directamente las clases de WHATWG, y los propios tipos Readable, Writable y Duplex de Node incluyen los métodos .toWeb() y .fromWeb(), marcados como Estabilidad 2 (“Estable”) en las versiones actuales de Node.js.
Esto se vuelve relevante en el momento en que se llama a fetch() desde dentro de un servicio Node, ya que el response.body que se obtiene es un ReadableStream web, no un Readable de Node.
import { createWriteStream } from 'node:fs';
import { Writable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
const response = await fetch('https://example.com/large-report.csv');
if (!response.ok || !response.body) {
throw new Error(`Download failed with status ${response.status}`);
}
// response.body is a Web ReadableStream, so convert the Node side instead
const fileStream = Writable.toWeb(createWriteStream('./report.csv'));
await response.body.pipeTo(fileStream);
console.log('Download saved without buffering the response in memory');
La conversión funciona de manera igualmente fluida en sentido inverso: puedes transformar un objeto Node Readable en un flujo que cualquier API basada en Web Streams aceptará:
import { createReadStream } from 'node:fs';
import { Readable } from 'node:stream';
const webStream = Readable.toWeb(createReadStream('./large-file.bin'));
// Now this works with fetch's request body, a browser upload API,
// or any library built against the WHATWG interface instead of Node's.
await fetch('https://uploads.example.com/put', {
method: 'PUT',
body: webStream,
duplex: 'half' // required by fetch when the body is a stream
});
Un detalle que confunde a la gente la primera vez: fetch lanzará un error si se le proporciona un cuerpo de solicitud en formato streaming sin también pasar duplex: 'half'. Esto existe porque los cuerpos de solicitud en streaming no son compatibles uniformemente en todos lados, por lo que la especificación exige que se active explícitamente. Es un detalle menor que solo causa problemas quince minutos una vez y luego nunca más.
La verdadera ventaja es que ahora tu backend de Node, tus funciones edge y tu código del navegador pueden compartir en gran medida la misma lógica de streaming, en lugar de tener que mantener una versión adaptada a fetch y otra separada basada en pipe(), construida con los módulos fs y http de Node.
stream.compose(): un pipeline que devuelve un flujo
Técnicamente, stream.compose() existe desde Node.js 16.9.0, pero permaneció durante mucho tiempo en estado experimental antes de volverse estable en una versión reciente de Node.js. Ahora que su API ya está definida, merece ser analizada más detenidamente.
La idea es que, en lugar de crear una subclase completa de Transform con los métodos _transform y _flush, se escribe una función generadora asíncrona, y compose() combina toda esa cadena en un único flujo Duplex que se puede canalizar como cualquier otro flujo.
import { compose } from 'node:stream';
import { createReadStream, createWriteStream } from 'node:fs';
async function* toUpperCase(source) {
for await (const chunk of source) {
yield chunk.toString().toUpperCase();
}
}
async function* onlyNonEmptyLines(source) {
let leftover = '';
for await (const chunk of source) {
const lines = (leftover + chunk).split('\n');
leftover = lines.pop() ?? '';
for (const line of lines) {
if (line.trim()) yield line + '\n';
}
}
if (leftover.trim()) yield leftover;
}
const pipeline = compose(
createReadStream('./raw.txt', { encoding: 'utf8' }),
onlyNonEmptyLines,
toUpperCase
);
pipeline.pipe(createWriteStream('./clean.txt'));
Al combinar esto con la creación de dos clases Transform separadas —incluyendo una lógica de búfer manual para la parte que divide las líneas— se obtiene el mismo comportamiento subyacente con una estructura significativamente más sencilla, además de un código que se lee de arriba hacia abajo en el mismo orden en que fluyen los datos realmente. compose() también es flexible con sus extremos: si se le proporciona un AsyncIterable simple como primer argumento, se convierte en un Duplex legible; si se le da una AsyncFunction como último argumento, se convierte en un destino escribible. Eso significa que todo un pipeline puede construirse únicamente con generadores asíncronos si ese es el estilo que prefiera.
Nada de esto es una razón para eliminar las clases Transform que ya funcionan. Pero en el caso de código nuevo, especialmente aquel que se reduce a “analizar esto, filtrar aquello, darle otra forma a esto”, recurrir por defecto a un generador asíncrono junto con compose() suele evitar la necesidad de escribir varias subclases pequeñas de Transform que al final terminan teniendo la misma estructura.
Flujo de tokens desde un modelo local
Casi todas las secciones sobre “flujos del mundo real” en los tutoriales se limitan a los mismos tres ejemplos: descargas de archivos, compresión gzip y solicitudes de rangos de video. Estos son casos de uso legítimos, pero existe un patrón más reciente que se adapta igualmente bien al modelo de streaming y que rara vez se aborda: obtener salida token por token de un gran modelo de lenguaje.
Dado que un ejemplo realmente apto para copiar y pegar no debería depender de la clave API de nadie, el que se muestra a continuación se comunica con Ollama ejecutado localmente en lugar de con un proveedor alojado:
# one-time setup
ollama pull llama3.2
ollama serve # usually already running as a background service
Por defecto, el endpoint /api/generate de Ollama transmite su respuesta como JSON delimitado por saltos de línea (application/x-ndjson): un objeto JSON por línea, cada uno con un fragmento response y una bandera done, siendo la última línea la que indica "done": true. Si se elimina el formato, se trata simplemente de un flujo legible que realiza un análisis básico de texto; el mismo razonamiento basado en bloques utilizado en otras partes de este artículo se aplica sin modificaciones.
// server.js: proxies Ollama's token stream to the browser over SSE
import { createServer } from 'node:http';
const server = createServer(async (request, response) => {
if (request.url !== '/ask') {
response.statusCode = 404;
response.end('Not found');
return;
}
response.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive'
});
const ollamaResponse = await fetch('http://localhost:11434/api/generate', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
model: 'llama3.2',
prompt: 'Explain backpressure in one short paragraph.',
stream: true
})
});
if (!ollamaResponse.body) {
response.end();
return;
}
// Web ReadableStream from fetch -> async iterate it directly (Node 18+)
let buffered = '';
for await (const chunk of ollamaResponse.body) {
buffered += Buffer.from(chunk).toString('utf8');
const lines = buffered.split('\n');
buffered = lines.pop() ?? '';
for (const line of lines) {
if (!line.trim()) continue;
const event = JSON.parse(line);
response.write(`data: ${JSON.stringify({ text: event.response })}\n\n`);
if (event.done) {
response.write('event: done\ndata: {}\n\n');
response.end();
return;
}
}
}
// client disconnected mid-stream? stop pulling tokens we can't deliver
request.on('close', () => {
if (!response.writableEnded) response.end();
});
});
server.listen(3000, () => console.log('Ask at http://localhost:3000/ask'));
Si desea inspeccionar una línea NDJSON en bruto de Ollama sin el encuadre SSE alrededor, se verá así:
{"model":"llama3.2","created_at":"2026-07-30T10:15:24.121Z","response":"Back","done":false}
{"model":"llama3.2","created_at":"2026-07-30T10:15:24.132Z","response":"pressure","done":false}
{"model":"llama3.2","created_at":"2026-07-30T10:15:24.166Z","response":" happens...","done":true,"done_reason":"stop"}
Dos detalles pueden confundir a los usuarios la primera vez que lo intentan. Primero, en las versiones actuales de Node.js, response.body de fetch ya es iterable asíncronamente por sí mismo, por lo que no hay necesidad de llamar a Readable.fromWeb() solo para este caso; aunque ese helper sigue estando disponible si alguna biblioteca posterior espera un objeto Readable clásico de Node. Segundo, las líneas de NDJSON pueden abarcar múltiples límites de bloque, lo cual representa exactamente el mismo problema que aparece al parsear CSV en formato de streaming: no se garantiza que un único bloque de fetch coincida con una única línea JSON. Por eso el ejemplo anterior almacena en búfer líneas parciales en lugar de llamar directamente a JSON.parse(chunk) con lo que llegue.
Si prefiere utilizar una API alojada —OpenAI, Anthropic o similares— la estructura general apenas cambia: sigue consumiendo una respuesta HTTP fragmentada y reenviándola al cliente en formato SSE. La versión local de Ollama es simplemente aquella que puede ejecutarse de inmediato sin necesidad de cuenta ni clave, y constituye una opción razonable para desarrollar la parte del cliente de una funcionalidad como esta durante la fase de desarrollo, antes de dirigir el mismo código a un modelo alojado una vez que se requiera calidad de salida de nivel profesional.
Transmisión en flujo continuo de un archivo directamente al almacenamiento de objetos
Enviar un archivo “a S3” casi nunca implica que el archivo sea lo suficientemente pequeño como para caber cómodamente en la memoria de antemano. El paquete @aws-sdk/lib-storage existe precisamente porque el comando simple PutObjectCommand requiere conocer la longitud del contenido antes de que se inicie la solicitud. En cambio, su ayuda Upload acepta un flujo de datos, decide por sí mismo si es apropiado utilizar una sola solicitud o una carga multipart, y muestra eventos de progreso durante el proceso.
import { S3Client } from '@aws-sdk/client-s3';
import { Upload } from '@aws-sdk/lib-storage';
import { createReadStream } from 'node:fs';
const s3 = new S3Client({ region: 'us-east-1' });
async function uploadFile(localPath, bucket, key) {
const upload = new Upload({
client: s3,
params: {
Bucket: bucket,
Key: key,
Body: createReadStream(localPath)
}
});
upload.on('httpUploadProgress', (progress) => {
console.log(`${key}: ${progress.loaded} / ${progress.total ?? '?'} bytes`);
});
await upload.done();
console.log(`${key} uploaded`);
}
Para experimentar sin necesidad de una cuenta real de AWS, MinIO ofrece un almacén de objetos compatible con S3 que se puede configurar localmente con una sola orden. El mismo código @aws-sdk/lib-storage mencionado anteriormente funciona con él sin cambios, solo siendo necesario dirigir al cliente al endpoint de MinIO en lugar del de AWS:
docker run -p 9000:9000 -p 9001:9001 \
-e "MINIO_ROOT_USER=admin" \
-e "MINIO_ROOT_PASSWORD=admin12345" \
minio/minio server /data --console-address ":9001"
const s3 = new S3Client({
region: 'us-east-1',
endpoint: 'http://localhost:9000',
forcePathStyle: true, // required for MinIO's path-style URLs
credentials: { accessKeyId: 'admin', secretAccessKey: 'admin12345' }
});
Este es el enfoque que se debe adoptar por defecto para archivos de más de unas pocas decenas de megabytes: evitar cargar todo el archivo en un Buffer antes de pasarlo al SDK. Utilice createReadStream() y deje que Upload se encargue del resto; el uso de memoria permanece constante independientemente de si el archivo es de 10 MB o 10 GB.
Análisis de una carga sin almacenar en búfer el cuerpo de la solicitud
El otro lado de las cargas de archivos es el servidor que debe recibirlos, y aquí es donde con mayor frecuencia ocurre un almacenamiento accidental en búfer: generalmente porque un middleware de análisis del cuerpo no fue diseñado teniendo en cuenta el streaming y leyó silenciosamente toda la solicitud multipart en memoria antes de que su procesador tuviera la oportunidad de acceder a ella.
busboy procesa el análisis de formularios multipart de forma incremental: tan pronto como comienzan a llegar los bytes de un campo de archivo determinado, te devuelve un flujo legible para ese campo en lugar de esperar a que la solicitud llegue completa:
import { createServer } from 'node:http';
import { createWriteStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';
import busboy from 'busboy';
const server = createServer((request, response) => {
if (request.method !== 'POST' || request.url !== '/upload') {
response.statusCode = 404;
response.end();
return;
}
const bb = busboy({ headers: request.headers, limits: { fileSize: 200 * 1024 * 1024 } });
const uploads = [];
bb.on('file', (fieldName, fileStream, info) => {
const destinationPath = `./uploads/${Date.now()}-${info.filename}`;
uploads.push(
pipeline(fileStream, createWriteStream(destinationPath)).catch((error) => {
fileStream.resume(); // drain on failure so the request can still finish
throw error;
})
);
});
bb.on('close', async () => {
try {
await Promise.all(uploads);
response.writeHead(200, { 'Content-Type': 'application/json' });
response.end(JSON.stringify({ status: 'ok' }));
} catch (error) {
response.statusCode = 500;
response.end(JSON.stringify({ status: 'error', message: error.message }));
}
});
request.pipe(bb);
});
server.listen(3000);
Nada en este gestor almacena nunca el cuerpo completo de la solicitud de una sola vez. Los bytes pasan desde el socket, a través del analizador de busboy, y directamente al disco. Esa distinción es lo que separa a un servicio que funciona bien con varias subidas grandes que ocurren en paralelo de uno que colapsa la primera vez que dos usuarios suben un video simultáneamente.
Prueba de un flujo de transformación personalizado
Los flujos Transform generados manualmente son candidatos ideales para un deterioro silencioso: nadie se molesta en escribir pruebas para ellos, todos asumen que “simplemente funcionan”, y luego un caso límite de _flush se lanza defectuoso sin que nadie se dé cuenta hasta la producción. Sin embargo, probar uno no es tan doloroso como parece. No necesitas crear un archivo real ni un servidor HTTP real. Readable.from() combinado con stream/promises te permite hacerlo.
// uppercase-transform.js
import { Transform } from 'node:stream';
export function createUppercaseTransform() {
return new Transform({
transform(chunk, encoding, callback) {
callback(null, chunk.toString().toUpperCase());
}
});
}
// uppercase-transform.test.js
import { describe, it, expect } from 'vitest'; // works the same with Jest
import { Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createUppercaseTransform } from './uppercase-transform.js';
async function collect(readable) {
const parts = [];
for await (const chunk of readable) parts.push(chunk.toString());
return parts.join('');
}
describe('uppercase transform', () => {
it('uppercases each chunk', async () => {
const source = Readable.from(['hello ', 'world']);
const transform = createUppercaseTransform();
source.pipe(transform);
expect(await collect(transform)).toBe('HELLO WORLD');
});
it('propagates errors from upstream instead of hanging', async () => {
const source = new Readable({
read() {
this.destroy(new Error('upstream failed'));
}
});
const transform = createUppercaseTransform();
await expect(pipeline(source, transform, async function drain(it) {
for await (const _ of it) { /* discard */ }
})).rejects.toThrow('upstream failed');
});
});
La segunda prueba en ese archivo es más importante de lo que parece a primera vista. Confirma que un error generado en la etapa anterior realmente se propaga hacia afuera, en lugar de dejar al flujo de transformación atascado a mitad de su proceso con una promesa que nunca se resuelve. Ese bloqueo silencioso es la forma más común en que una cadena de procesos que parece funcionar se convierte en una solicitud atascada una vez está en producción.
Un componente de métricas ligero
No necesitas una plataforma completa de observabilidad solo para verificar si un pipeline se está comportando correctamente. A menudo, basta con un pequeño flujo PassThrough que cuente los bytes y los fragmentos a medida que pasan, y este se integra en cualquier pipeline existente sin afectar nada más:
import { PassThrough } from 'node:stream';
export function createMetricsTap(label) {
let bytes = 0;
let chunks = 0;
const startedAt = process.hrtime.bigint();
const tap = new PassThrough();
tap.on('data', (chunk) => {
bytes += chunk.length;
chunks += 1;
});
tap.on('end', () => {
const seconds = Number(process.hrtime.bigint() - startedAt) / 1e9;
console.log(`[${label}] ${chunks} chunks, ${(bytes / 1024 / 1024).toFixed(2)} MB, ` +
`${(bytes / 1024 / 1024 / seconds).toFixed(2)} MB/s`);
});
return tap;
}
await pipeline(
createReadStream('./large-file.bin'),
createMetricsTap('upload'),
createGzip(),
createWriteStream('./large-file.bin.gz')
);
Si deseas algo más que la salida de console.log, reemplaza la lógica dentro de los manejadores 'data' y 'end' por un Counter o Histogram de prom-client. El componente en sí mantiene la misma estructura: es simplemente otro flujo ubicado dentro de un pipeline, que informa sobre lo que pasa a través de él.
Tres errores que solo se manifiestan bajo carga
Unhandled 'error' events bring down the whole process. A stream that emits an error with no 'error' listener attached doesn't just fail its own request. In Node.js, an 'error' event on an EventEmitter that nobody handles gets thrown, and if nothing catches that throw, the process exits. It only takes one malformed upload to take an entire service offline because a manually wired pipe() chain had error listeners on two out of three streams. pipeline() exists specifically to make this class of bug structurally difficult, so prefer it, or make sure every stream you build by hand gets its own .on('error', ...), with no exceptions.
Reutilizar una instancia de flujo provoca fugas de listeners. Supongamos que envolvemos la creación del flujo dentro de una función que se llama en cada solicitud, pero algo fuera de esa función mantiene una referencia y adjunta sus propios listeners “por si acaso”. Con el tiempo verás aparecer MaxListenersExceededWarning en tus registros, que es la forma en que Node indica una fuga de memoria en curso. Cada solicitud necesita sus propios flujos recién creados, y nada fuera del ciclo de vida de esa solicitud debería mantener una referencia a ellos una vez que haya finalizado.
Nadie destruye el flujo cuando el cliente se retira. Este es el error que realmente cuesta dinero en la producción. Si un usuario cierra la pestaña del navegador durante la descarga y no se está escuchando response.on('close', ...) para llamar a .destroy() en el flujo de origen, el servidor sigue extrayendo datos del disco; o, en un caso peor, deja abierto un cursor de base de datos o mantiene viva una conexión HTTP ascendente, todo esto por un cliente que ya se ha ido. Con suficiente carga concurrente, así es como un servidor agota silenciosamente sus descriptores de archivo o su piscina de conexiones sin que aparezca una causa evidente en los registros. La solución siempre son esas mismas tres líneas: escuchar que el destino se cierre, destruir el origen como respuesta y tratar ese patrón como una costumbre por defecto en cada ruta basada en flujos, en lugar de algo que se recuerda solo después de un incidente.
¿Cuándo usar cada API?
pipe()
Simple, two-or-three-stream chains. Handles backpressure.
Error handling is your responsibility per-stream.
pipeline()
Anything running in production. Handles backpressure,
forwards errors from every stream, cleans up on failure.
Has a promise-based version and supports AbortSignal.
stream.compose()
New pipelines built from async generators instead of
Transform subclasses. Returns one Duplex you can pipe
like any other stream. Stable as of recent Node.js releases.
node:stream/web (ReadableStream/WritableStream/TransformStream)
Anywhere you're mixing Node streams with fetch(), browser
code, or a library built against the WHATWG standard.
Convert with .toWeb() / .fromWeb() at the boundary.
Ninguno de estos enfoques hace que los demás sean innecesarios. Cada uno resuelve una parte diferente del mismo flujo general de datos a través de su sistema. Hoy en día es común que un único pipeline comience con un ReadableStream web proveniente de fetch, convierta esos datos, los envíe a través de una transformación creada con compose(), y finalmente pase el resultado a pipeline() para que se encargue de la escritura. Antes, mezclar APIs de esta manera parecía inusual; hoy en día es un patrón bastante habitual.
Reflexiones finales
Nada de lo que se menciona aquí elimina la necesidad de comprender primero qué es la contrapresión, highWaterMark y las cuatro clases fundamentales de flujos. Esos conocimientos básicos siguen siendo un requisito indispensable antes de que cualquiera de las técnicas anteriores sea útil. Lo que sí cambia una vez que se cuenta con ese conocimiento previo es el punto en el que comienza el verdadero trabajo de ingeniería: confirmar las mejoras en el uso de memoria mediante cifras concretas en lugar de afirmaciones vagas, recordar que un flujo basado en fetch y un flujo de Node solo necesitan una conversión mediante .toWeb() para poder interactuar, y diseñar un plan para manejar la desconexión del cliente durante una solicitud como parte integral de la implementación de la ruta, y no como un detalle que se soluciona posteriormente.
Si ejecuta los ejemplos de prueba mencionados anteriormente en hardware diferente y los resultados difieren notablemente de lo que se muestra aquí, eso constituye un dato útil que vale la pena compartir en lugar de ignorarlo, ya que las cifras relativas al rendimiento y al uso de memoria son mucho más fiables una vez se han verificado en más de una máquina.
Recursos
- Documentación de Node.js sobre la API Stream clásica: https://nodejs.org/api/stream.html
- Documentación de Node.js sobre la API Web Streams: https://nodejs.org/api/webstreams.html
- Referencia de Ollama para su endpoint de transmisión en tiempo real: https://docs.ollama.com/api/streaming
- Lista del paquete
@aws-sdk/lib-storageen npm: https://www.npmjs.com/package/@aws-sdk/lib-storage - Repositorio del proyecto
busboy: https://github.com/mscdex/busboy