Стримы в Node.js: pipe, pipeline или async iterator

Задача звучит просто: прочитать файл на 2 ГБ, преобразовать построчно и записать результат в другой файл или отдать по HTTP. Если сделать это через fs.readFileSync — процесс упрётся в память и упадёт с ERR_FS_FILE_TOO_LARGE или по OOM от системы. Стримы решают проблему: данные текут кусками, память под контролем. Но у самих стримов есть четыре способа связать источник с потребителем, и они по-разному ведут себя при ошибках и обратном давлении. Ниже — что выбрать под конкретный случай на Node.js 18 и новее.

Почему нельзя просто написать readable.pipe(writable)

Классический pipe() известен всем и работает: он сам управляет backpressure¹ — притормаживает источник, когда потребитель не успевает. Проблема в обработке ошибок. Если в середине цепочки один из стримов сломается, pipe() не разрушит остальные автоматически. Вы получаете висящие дескрипторы файлов и утечку памяти.

const fs = require('node:fs');

// Node.js 18+. Опасный вариант: ошибку записи никто не поймает,
// исходный стрим останется открытым.
fs.createReadStream('big.log')
  .pipe(fs.createWriteStream('out.log'));

Чтобы это не текло, к каждому стриму придётся вручную вешать on('error') и вручную вызывать destroy() у соседей. На цепочке из трёх-четырёх трансформаций код превращается в кашу.

Главная ошибка в стримах — не забытый backpressure, его pipe разруливает сам. Главная ошибка — необработанная ошибка в середине цепочки, после которой процесс тихо течёт дескрипторами.

Четыре подхода

1. Голый pipe()

Годится только для двух стримов, где вы готовы сами обработать error на обоих концах. Плюс — минимум обвязки и предсказуемое поведение с backpressure. Минус — ручная уборка при ошибке и никакого сигнала «всё дописалось» без отдельной подписки на finish.

2. stream.pipeline()

Появился в Node.js 10, с версии 15 есть промис-версия в node:stream/promises. Связывает любое число стримов, при ошибке в любом звене разрушает все остальные и вызывает колбэк или реджектит промис. Это рабочий выбор по умолчанию для «прочитать — преобразовать — записать».

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function gzipFile(src, dest) {
  // При ошибке любого из трёх стримов pipeline разрушит все и бросит исключение
  await pipeline(
    fs.createReadStream(src),
    zlib.createGzip(),
    fs.createWriteStream(dest),
  );
}

gzipFile('big.log', 'big.log.gz')
  .then(() => console.log('готово'))
  .catch((err) => console.error('сбой конвейера:', err.message));

3. Async iterator (for await ... of)

Начиная с Node.js 10 Readable-стрим — асинхронный итератор. Вы читаете куски обычным for await, и backpressure сохраняется: пока тело цикла не завершилось, следующий кусок не запрашивается. Удобно, когда логика преобразования сложная и её неудобно упаковывать в Transform-стрим.

const fs = require('node:fs');
const readline = require('node:readline');

async function countErrors(path) {
  const rl = readline.createInterface({
    input: fs.createReadStream(path),
    crlfDelay: Infinity,
  });

  let count = 0;
  // Каждая строка обрабатывается по очереди, память не растёт
  for await (const line of rl) {
    if (line.includes('ERROR')) count += 1;
  }
  return count;
}

Цена: если вы читаете источник итератором, а результат пишете в Writable вручную, вы снова сами отвечаете за уборку при ошибке. Итератор при выходе по исключению разрушит только сам Readable, про Writable он ничего не знает.

4. Transform + pipeline с генератором

С Node.js 13.10 pipeline принимает async-генератор как одно из звеньев. Это компромисс: логика преобразования пишется как обычный генератор с for await, а уборку и backpressure берёт на себя pipeline.

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');

async function upperCaseLines(src, dest) {
  await pipeline(
    fs.createReadStream(src, { encoding: 'utf8' }),
    // Генератор-звено: преобразует поток и возвращает строки
    async function* (source) {
      for await (const chunk of source) {
        yield chunk.toUpperCase();
      }
    },
    fs.createWriteStream(dest),
  );
}

Сравнение

Когда что выбирать

• Отдаёте файл по HTTP без изменений — pipeline(readStream, res). Он корректно разрушит чтение, если клиент оборвёт соединение. Голый pipe тут оставит открытый дескриптор при разрыве.
• Считаете статистику по строкам, результат — число или объект — async iterator через readline. Писать Transform ради счётчика избыточно.
• Конвейер из нескольких преобразований — pipeline. Промис-версия из node:stream/promises плюс async/await читаются лучше колбэков.
• Логика трансформации не влезает в одно выражение — генератор-звено внутри pipeline вместо ручного Transform-класса.

Что ломается на практике

• Забытый await у pipeline-промиса. Если не дождаться промиса, необработанный reject уйдёт в unhandledRejection, а вызывающий код продолжит выполнение, будто всё записалось.
• Смешивание событий и итератора. Нельзя одновременно вешать data-обработчик и читать стрим через for await — данные разъедутся между двумя потребителями. Выберите одно.
• objectMode и Buffer. Если Transform возвращает объекты, а следующий стрим ждёт Buffer или строку, получите ERR_INVALID_ARG_TYPE. Проверяйте, что режимы соседних звеньев совпадают.
• Разрыв HTTP-соединения. Клиент закрыл вкладку — res эмитит close. С pipeline исходный файловый стрим разрушится автоматически; при ручной обвязке дескриптор остаётся висеть до сборки мусора или до предела EMFILE.
• AbortSignal мимо pipeline. Чтобы отменить долгий конвейер по таймауту, передавайте сигнал в опциях: pipeline(..., { signal }). Собственный setTimeout с ручным destroy легко оставит один из стримов недоуничтоженным.

Что делать дальше

Переведите существующие цепочки с голого pipe на pipeline из node:stream/promises — это убирает целый класс утечек дескрипторов. Под нагрузкой смотрите на два показателя: число открытых файловых дескрипторов процесса (lsof -p PID | wc -l на Linux) и RSS-память. Если дескрипторы растут при работе со стримами — где-то осталась ручная обвязка без уборки при ошибке. Инвалидация и потоковая обработка через Web Streams API (node:stream/web) — отдельная тема, здесь мы её не разбираем; актуальный статус совместимости сверяйте в документации своей версии Node.js.