Streams Node.js au-delà des bases : mémoire, contrainte de flux et pannes réelles

De nombreuses introductions aux flux Node.js couvrent bien les bases : lecture, écriture, mode duplex, transformation, la fonction pipe(), la contrepression, une démonstration avec gzip, et parfois un serveur de demande de plage vidéo en guise de conclusion. Si vous avez déjà lu l’une ou l’autre de ces introductions, vous comprenez ce qu’est un flux et pourquoi essayer de charger entièrement un fichier de 2 Go en mémoire lors d’une démonstration en direct est une source de problèmes.

Ce que ces introductions omettent rarement, c’est ce qui vient ensuite. De combien de mémoire économisez-vous réellement en passant aux flux, avec des chiffres concrets plutôt qu’une vague notion de « moins » ? Comment les flux natifs de Node interagissent-ils avec l’API Web Streams utilisée par fetch et les navigateurs ? Comment pouvez-vous diffuser les résultats d’un modèle exécuté sur votre propre matériel au lieu de payer par token via une API hébergée ? Et quels sont les deux ou trois erreurs qui finissent par détruire un pipeline qui semblait fonctionner correctement en phase de test lorsqu’il est confronté à un trafic réel ?

Cet article aborde précisément ces questions. Il part du principe que vous connaissez déjà ce qu’est un flux lisible. Si ce n’est pas le cas, le bref résumé ci-dessous vous permettra de vous mettre rapidement à niveau. Tout ce qui suit s’appuie sur ces bases fondamentales.

Résumé en 60 secondes

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

Les données circulent dans le système par petits morceaux plutôt que sous forme d’un seul bloc massif. La fonction pipe() relie des flux entre eux et gère la situation où la destination ne peut pas suivre le rythme de la source — c’est ce qu’on appelle la contrepression. La fonction pipeline() réalise la même connexion, mais elle propage également les erreurs provenant de n’importe quel flux de la chaîne et garantit un nettoyage en cas d’échec, c’est pourquoi c’est la fonction à privilégier dans tout service longue durée plutôt que dans un script ponctuel.

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 cet exemple vous semble déjà parfaitement clair, n’hésitez pas à sauter directement à la suite. Ce qui suit concerne les éléments essentiels à connaître avant d’intégrer du code basé sur des flux dans un environnement de production pour la première fois.

Le chiffre que personne ne mentionne dans ces articles

La plupart des écrits sur le streaming affirment que l’approche à mémoire tamponne « utilise beaucoup de mémoire » tandis que l’approche en streaming « en utilise bien moins », sans jamais indiquer de chiffre à l’appui de ces affirmations. Pour y remédier, un fichier texte simple de 501 MB a été généré, et les deux approches ont été testées sur la même machine, l’une après l’autre, à deux reprises chacune.

// 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 });
});

Voici ce qui ressort réellement des tests effectués sur un fichier ASCII de 500 MB avec Node.js 22, à deux reprises pour chaque approche :

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

Quelques mises en garde à mentionner clairement, car les omettre rendrait les résultats trompeurs : il s’agit ici d’une seule machine, d’un seul fichier et d’une seule charge de travail — un scan de caractères trivial. Le temps écoulé a été à peu près identique entre les deux approches, la version en flux étant même légèrement plus rapide à certains endroits, ce qui était un peu inattendu. On partait du principe que le surcoût lié au traitement de nombreux petits fragments ralentirait les choses davantage qu’en réalité. En revanche, les chiffres concernant la mémoire correspondaient bien aux attentes : l’exécution avec tamponnement a atteint un pic proche de la taille totale du fichier, tandis que l’exécution en flux a atteint un pic faible qui est resté globalement stable quel que soit la taille du fichier. Ce contraste résume vraiment l’intérêt d’utiliser les flux, condensé en un seul tableau — et il est facile de le vérifier indépendamment en exécutant les deux scripts ci-dessus sur un fichier.

votre propre.

L’implication pratique : pour une seule requête traitée par une machine suffisamment puissante, il est souvent suffisant de simplement lire tout le fichier en mémoire. Le véritable problème n’a jamais été une seule requête isolée — c’est la question qui suit toujours : que se passe-t-il lorsque 100 de ces requêtes s’exécutent simultanément ? L’utilisation de la mémoire avec l’approche par flux ne varie presque pas à mesure que la concurrence augmente, car chaque requête individuelle ne conserve qu’un buffer de taille définie et non l’intégralité du chargement en même temps.

Node streams et Web Streams appartiennent désormais à la même famille

Pendant des années, les flux de Node.js et l’API Streams WHATWG (ReadableStream, WritableStream, TransformStream — les classes sous-jacentes à fetch() tant dans le navigateur que chez Node) semblaient représenter deux écosystèmes distincts avec peu de points communs. Cela n’est plus vrai aujourd’hui. Le module node:stream/web expose directement les classes WHATWG, et les propres classes Readable, Writable et Duplex de Node disposent toutes des méthodes .toWeb() et .fromWeb(), classées comme étant de stabilité 2 (“Stable”) dans les versions actuelles de Node.js.

Cela devient pertinent dès que vous appelez fetch() depuis un service Node, car le response.body retourné est un ReadableStream Web, et non un Readable 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 conversion fonctionne tout aussi bien dans le sens inverse — vous pouvez transformer un objet Node Readable en un flux que n’importe quelle API basée sur Web-stream acceptera :

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 détail qui prête souvent à confusion la première fois : fetch lancera une erreur si vous lui transmettez un corps de requête en flux sans également spécifier duplex: 'half'. Cela existe parce que les corps de requête en flux ne sont pas uniformément pris en charge partout, donc la spécification exige que vous l’activiez explicitement. C’est une petite chose qui prend quinze minutes une fois, puis plus jamais.

Le véritable avantage est que votre backend Node, vos fonctions edge et votre code navigateur peuvent désormais partager dans une large mesure la même logique de flux, plutôt que de devoir maintenir une version adaptée à fetch et une version distincte basée sur pipe(), construite autour des modules fs et http de Node.

stream.compose() : un pipeline qui renvoie un seul flux

stream.compose() existe techniquement depuis Node.js 16.9.0, mais il est resté longtemps en état expérimental avant de devenir stable dans une récente version de Node.js. Maintenant que son API est stabilisée, il mérite une attention plus approfondie.

L’idée : au lieu de créer une sous-classe complète Transform avec des méthodes _transform et _flush, on écrit une fonction générateur asynchrone, et compose() regroupe toute cette chaîne en un seul flux Duplex que l’on peut acheminer comme n’importe quel autre flux.

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'));

En plus de la création de deux classes Transform distinctes — incluant une logique de tamponnement manuel pour la partie chargée de séparer les lignes — cela permet d’obtenir le même comportement de base avec beaucoup moins de structure supplémentaire, ainsi que du code qui s’affiche de haut en bas dans le même ordre que le flux réel des données. compose() est également flexible quant à ses extrémités : si l’on lui fournit un simple AsyncIterable en premier argument, il devient un Duplex lisible ; si l’on lui donne une AsyncFunction en dernier argument, il devient un émetteur écritible. Cela signifie qu’un pipeline complet peut être construit uniquement à partir de générateurs asynchrones si c’est le style que l’on préfère.

Rien de tout cela ne justifie l’élimination des classes Transform qui fonctionnent déjà. Mais pour du nouveau code — en particulier tout ce qui se résume à « parser ceci, filtrer cela, reformater cela » — recourir par défaut à un générateur asynchrone associé à compose() permet d’éviter d’écrire plusieurs sous-classes de Transform petites et finalement identiques.

Débit de tokens depuis un modèle local

Presque toutes les sections sur les « flux du monde réel » dans les tutoriels se basent sur les mêmes trois exemples : les téléchargements de fichiers, la compression gzip et les demandes de segments vidéo. Ce sont des cas d’usage légitimes, mais il existe un modèle plus récent qui convient tout aussi bien au streaming et qui est rarement abordé : obtenir des résultats token par token à partir d’un grand modèle de langage.

Puisqu’un exemple véritablement copiable-collable ne devrait pas dépendre de la clé API de qui que ce soit, celui ci-dessous communique avec Ollama exécuté localement plutôt qu’avec un fournisseur hébergé :

# one-time setup
ollama pull llama3.2
ollama serve   # usually already running as a background service

Par défaut, l’endpoint /api/generate d’Ollama transmet sa réponse sous forme de JSON séparé par des sauts de ligne (application/x-ndjson) : un objet JSON par ligne, chacun contenant un fragment response et un indicateur done, la dernière ligne indiquant "done": true. En supprimant ce formatage, il s’agit simplement d’un flux lisible permettant une analyse légère du texte — le même principe basé sur des blocs utilisé partout ailleurs dans cet article peut être appliqué sans modification.

// 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 vous souhaitez examiner une ligne NDJSON brute provenant d’Ollama, sans le cadre SSE qui l’entoure, elle a cet aspect :

{"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"}

Deux points peuvent prêter à confusion la première fois qu’on aborde ce sujet. Premièrement, dans les versions actuelles de Node.js, response.body de fetch est déjà itérable de manière asynchrone, il n’est donc pas nécessaire d’appeler Readable.fromWeb() uniquement pour ce cas — bien que cet outil reste disponible si une bibliothèque intermédiaire attend un objet Readable classique de Node. Deuxièmement, les lignes NDJSON peuvent s’étendre sur plusieurs limites de blocs, ce qui représente exactement le même problème que lors de la lecture en flux de fichiers CSV : on ne peut pas garantir qu’un seul bloc obtenu via fetch corresponde à une seule ligne JSON. C’est précisément pour cette raison que l’exemple ci-dessus stocke en mémoire des lignes incomplètes plutôt que d’appeler directement JSON.parse(chunk) sur le contenu reçu.

Si vous préférez utiliser une API hébergée — OpenAI, Anthropic ou similaire — la structure globale reste presque identique : vous continuez à consommer une réponse HTTP fragmentée et à la transmettre au client sous forme SSE. La version locale d’Ollama est simplement celle que vous pouvez exécuter immédiatement sans besoin de compte ou de clé, et c’est un choix raisonnable pour développer la partie client d’une telle fonctionnalité en phase de développement, avant de diriger le même code vers un modèle hébergé lorsque vous avez besoin d’une qualité de sortie adaptée à un environnement professionnel.

Streamer un fichier directement vers le stockage d’objets

Envoyer un fichier « vers S3 » implique presque jamais un fichier suffisamment petit pour tenir confortablement en mémoire à l’avance. Le paquet @aws-sdk/lib-storage existe précisément parce que la commande simple PutObjectCommand nécessite de connaître la longueur du contenu avant le démarrage de la requête. Son outil Upload, quant à lui, prend un flux en entrée, décide par lui-même s’il est approprié d’utiliser une seule requête ou un téléchargement multipart, et affiche des événements de progression au fur et à mesure.

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`);
}

Pour expérimenter sans disposer d’un compte AWS réel, MinIO vous offre un stockage d’objets compatible S3 que vous pouvez mettre en place localement avec une seule commande. Le même code @aws-sdk/lib-storage mentionné précédemment fonctionne également avec MinIO, sans aucune modification, il suffit de pointer le client vers l’endpoint de MinIO plutôt que celui d’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' }
});

C’est l’approche à privilégier pour tout fichier dépassant quelques dizaines de mégaoctets : évitez de charger l’intégralité du fichier dans un Buffer avant de le transmettre au SDK. Utilisez plutôt createReadStream() et laissez Upload gérer le reste — l’utilisation de la mémoire reste constante, que le fichier pèse 10 MB ou 10 GB.

Analyse d’un téléchargement sans mise en mémoire tampon du corps de la requête

L’autre aspect des téléchargements de fichiers concerne le serveur qui doit les recevoir, et c’est là que la mise en mémoire tampon accidentelle apparaît le plus souvent — généralement parce qu’un middleware d’analyse du corps de la requête n’a pas été conçu pour le streaming et lit silencieusement toute la requête multipart en mémoire avant que votre gestionnaire n’ait l’occasion de s’en occuper.

busboy gère l’analyse des formulaires multipart de manière progressive : dès que les octets d’un champ de fichier donné commencent à arriver, il vous renvoie un flux lisible pour ce champ au lieu d’attendre que la demande soit complète :

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);

Rien dans ce gestionnaire ne conserve jamais l’intégralité du corps de la demande en même temps. Les octets passent directement depuis le socket, via l’analyseur de busboy, jusqu’au disque. C’est cette distinction qui sépare un service capable de gérer plusieurs téléchargements volumineux en parallèle d’un autre qui tombe en panne dès que deux utilisateurs téléchargent une vidéo simultanément.

Test d’un flux de transformation personnalisé

Les flux Transform générés manuellement sont des candidats idéaux pour une dégradation silencieuse : personne ne prend la peine d’écrire des tests pour eux, tout le monde suppose qu’ils « fonctionnent simplement », puis un cas limite de _flush est livré en état défectueux sans que personne ne s’en aperçoive avant la mise en production. Cependant, tester un tel flux est moins difficile qu’il n’y paraît. Il n’est pas nécessaire de créer un fichier réel ou un serveur HTTP réel ; Readable.from() combiné à stream/promises suffit à y parvenir.

// 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');
  });
});

Le deuxième test dans ce fichier est plus important qu’il n’y paraît au premier abord. Il confirme qu’une erreur générée en amont se propage bien vers l’extérieur, au lieu de laisser le flux de transformation bloqué à mi-chemin de son traitement avec une promesse qui ne se résout jamais. Ce blocage silencieux est la manière la plus courante pour qu’un pipeline qui semble fonctionner se transforme en une requête bloquée une fois mis en service.

Un outil de mesure léger

Vous n’avez pas besoin d’une plateforme d’observabilité complète simplement pour vérifier le comportement d’un pipeline. Souvent, un petit flux PassThrough qui compte les octets et les blocs au fur et à mesure qu’ils passent suffit, et il s’intègre dans n’importe quel pipeline existant sans perturber quoi que ce soit d’autre :

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 vous souhaitez quelque chose de plus que les sorties console.log, remplacez la logique contenue dans les gestionnaires 'data' et 'end' par un Counter ou un Histogram de prom-client. La structure même de l’outil reste identique : il s’agit simplement d’un autre flux situé au milieu du pipeline, qui rapporte ce qui le traverse.

Trois erreurs qui ne se manifestent qu’en charge

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.

La réutilisation d’une instance de flux entraîne des fuites de listeners. Supposons que vous encapsuliez la création du flux à l’intérieur d’une fonction qui est appelée à chaque requête, mais qu’un élément en dehors de cette fonction conserve une référence à ce flux et y attache ses propres listeners « au cas où ». Finalement, vous verrez apparaître l’avertissement MaxListenersExceededWarning dans vos journaux, ce qui correspond à la manière dont Node indique une fuite de mémoire en cours. Chaque requête a besoin de flux créés à nouveau, et rien en dehors de la durée de vie de cette requête ne doit conserver de référence à eux une fois celle-ci terminée.

Nul ne détruit le flux lorsque le client s’éloigne. C’est cette erreur qui entraîne réellement des pertes financières en production. Si un utilisateur ferme la fenêtre de son navigateur au milieu du téléchargement et que vous ne surveillez pas response.on('close', ...) pour appeler .destroy() sur le flux source, le serveur continue de récupérer des données depuis le disque, ou, dans le pire des cas, laisse un curseur de base de données ouvert ou maintient une connexion HTTP ascendante active, tout cela pour un client qui est déjà parti. Avec suffisamment de charge simultanée, c’est précisément ainsi qu’un serveur épuise silencieusement ses descripteurs de fichiers ou son pool de connexions sans cause évidente dans les journaux. La solution reste toujours les mêmes trois lignes : surveiller la fermeture de la destination, détruire le flux source en réponse, et considérer ce schéma comme une habitude par défaut pour chaque route basée sur des flux, plutôt que quelque chose à se rappeler après un incident.

Quand utiliser quelle 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.

Aucune de ces approches ne rend les autres inutiles. Chacune résout un aspect différent du même flux global de données à travers votre système. Il est aujourd’hui courant qu’un seul pipeline commence par un ReadableStream Web provenant de fetch, qu’il convertisse ces données, les transmette à travers une transformation assemblée avec compose(), et qu’il enfin passe le résultat à pipeline() pour gérer l’écriture. Mélanger des APIs de cette manière semblait autrefois inhabituel ; aujourd’hui, c’est un schéma plutôt courant.

Conclusions

Rien ici ne remplace la nécessité de comprendre d’abord ce qu’est le backpressure, highWaterMark, ainsi que les quatre classes de flux fondamentales. Ces connaissances de base restent une exigence indispensable avant que les techniques mentionnées ci-dessus ne soient utiles. Ce qui change une fois ces bases acquises, c’est le point où commence le véritable travail d’ingénierie : confirmer les améliorations en matière de mémoire à l’aide de chiffres mesurés plutôt que d’affirmations vagues, se rappeler qu’un flux basé sur fetch et un flux Node n’ont besoin que d’une conversion .toWeb() pour interagir, et intégrer dans la conception un plan pour faire face à une déconnexion du client en cours de requête, en tant que partie normale de l’écriture de la route, et non comme un détail à corriger ultérieurement.

Si vous exécutez les exemples de benchmark mentionnés précédemment sur différents matériels et que les résultats diffèrent notablement de ceux présentés ici, il s’agit d’une donnée utile à partager plutôt que de l’ignorer, car les chiffres relatifs au débit et à l’utilisation de la mémoire sont bien plus fiables une fois vérifiés sur plusieurs machines.

Ressources

• Documentation Node.js concernant l’API Stream classique : https://nodejs.org/api/stream.html
• Documentation Node.js concernant l’API Web Streams : https://nodejs.org/api/webstreams.html
• Référence Ollama pour son endpoint de streaming : https://docs.ollama.com/api/streaming
• Liste du paquet @aws-sdk/lib-storage sur npm : https://www.npmjs.com/package/@aws-sdk/lib-storage
• Repositoire du projet busboy : https://github.com/mscdex/busboy

Le site du projet MinIO : https://min.io/