Стримы в Node.js: чек-лист перед продакшеном

Файловая выгрузка на 12 МБ проходит идеально: тесты зелёные, память ровная, отдаётся за 200 мс. В проде на файле в 1,4 ГБ тот же код съедает 2 ГБ RSS за десять секунд и получает OOM-kill. Причина почти всегда одна из трёх: ошибку не поймали и она уронила процесс, не соблюли backpressure и забили буфер, или не закрыли поток и утекли дескрипторы. Ниже — чек-лист по этим трём точкам плюс детали, которые всплывают только под нагрузкой. Материал для Node.js 18+ и 20 LTS; где поведение менялось между версиями — отмечаю отдельно.

Что должно быть закрыто до релиза

Стримы ломаются не в тех местах, где их тестируют. Пройдитесь по списку — каждый пункт ниже разбирается детально.

• Ошибки перехвачены на каждом потоке в цепочке, а не только на последнем.
• Соединение потоков идёт через pipeline(), а не через ручной .pipe().
• Backpressure реально работает: вы проверяете возврат write() или используете async-итераторы.
• Ресурсы (файловые дескрипторы, соединения) закрываются при ошибке и при отмене запроса.
• Есть ограничение на размер входящих данных, чтобы клиент не выел вам память.

Ошибки: одна необработанная — и процесс мёртв

Событие error на потоке — это EventEmitter. Если на него никто не подписан, Node.js бросает исключение, которое всплывает как uncaughtException и по умолчанию убивает процесс. В цепочке из трёх потоков ошибка может возникнуть на любом, и подписка на последний вас не спасёт.

Классическая ошибка — ручной pipe без обработчиков на источнике:

// Опасно: ошибка чтения файла обрушит процесс
const fs = require('node:fs');

const readable = fs.createReadStream('/data/report.csv');
readable.pipe(response); // если файла нет — error на readable без слушателя

.pipe() не пробрасывает ошибки и не закрывает потоки-источники, если сломался потребитель. Именно поэтому в проде его использовать не стоит.

pipeline вместо pipe

Функция stream.pipeline() соединяет потоки, пробрасывает ошибки в один колбэк и корректно уничтожает все потоки цепочки при сбое любого из них. С Node.js 15+ есть промис-версия в node:stream/promises.

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

async function sendGzippedFile(filePath, response) {
  try {
    // pipeline сам закроет readable и gzip, если response оборвётся
    await pipeline(
      fs.createReadStream(filePath),
      zlib.createGzip(),
      response
    );
  } catch (err) {
    // сюда попадёт ошибка ЛЮБОГО звена цепочки
    if (!response.headersSent) {
      response.statusCode = 500;
    }
    response.end();
    // логируем и не даём процессу упасть
    console.error('Ошибка отдачи файла:', err.code || err.message);
  }
}

Правило простое: если в цепочке больше одного потока, используйте pipeline. Ручной pipe оставьте для одноразовых скриптов, где падение процесса не проблема.

Backpressure: почему память растёт быстрее, чем отдаётся

Backpressure — механизм, при котором быстрый источник притормаживает, пока медленный потребитель разгребает данные.^* Если его проигнорировать, данные копятся во внутреннем буфере потока, и на большом файле буфер вырастает до размеров всего файла.

Ручной write() без проверки возврата — прямой путь к переполнению:

// Опасно: пишем, не глядя на возврат write()
for (const chunk of hugeArrayOfChunks) {
  response.write(chunk); // вернёт false, когда буфер заполнен — но мы игнорируем
}
response.end();

write() возвращает false, когда внутренний буфер превысил highWaterMark (по умолчанию 16 КБ для объектных потоков — 16 элементов). Игнорируя false, вы продолжаете пихать данные в память.

Правильно — дождаться события drain, либо, что проще и надёжнее, использовать async-итераторы, где backpressure соблюдается автоматически:

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

// Генератор отдаёт данные порциями; pipeline сам притормозит его,
// пока response не готов принять следующую порцию
async function* generateRows(dbCursor) {
  for await (const row of dbCursor) {
    yield JSON.stringify(row) + '\n';
  }
}

await pipeline(
  Readable.from(generateRows(cursor)),
  response
);

Здесь backpressure работает без единой ручной проверки: pipeline не запросит следующее значение из генератора, пока response не примет предыдущее.

Утечка ресурсов: клиент ушёл, а дескриптор остался

Сценарий, который не видно в тестах: клиент открыл выгрузку большого файла и закрыл вкладку на середине. Соединение оборвано, но если вы соединяли потоки через .pipe(), файловый дескриптор createReadStream останется открытым. Тысяча таких запросов — и вы упираетесь в лимит открытых файлов (EMFILE).

pipeline эту проблему решает: при обрыве response он вызывает destroy() на всех потоках выше. Если вы по какой-то причине пишете вручную, вешайте очистку на события закрытия:

const readable = fs.createReadStream(filePath);

// Клиент оборвал соединение — уничтожаем источник
response.on('close', () => {
  if (!readable.destroyed) {
    readable.destroy();
  }
});

С Node.js 16+ событие close на HTTP-ответе срабатывает и при нормальном завершении, и при обрыве, поэтому проверка readable.destroyed обязательна, чтобы не дёргать destroy() дважды.

Сравнение способов соединения потоков

Что ещё всплывает под нагрузкой

highWaterMark по умолчанию не всегда подходит

Для файловых потоков highWaterMark по умолчанию — 64 КБ. При отдаче множества мелких файлов это нормально, но при потоковой обработке гигабайтных файлов иногда выгоднее увеличить размер чанка, чтобы снизить число системных вызовов. Мерять надо на своём объёме данных: увеличили — сняли профиль по CPU и памяти, сравнили. Без замеров подбор highWaterMark — гадание.

Лимит на входящие данные

Если вы принимаете загрузку и стримите её в файл или БД, обязательно ограничьте объём. Без лимита клиент отправит вам поток на десятки гигабайт и займёт весь диск или память. Отслеживайте накопленный размер и уничтожайте поток при превышении:

const MAX_BYTES = 50 * 1024 * 1024; // 50 МБ
let received = 0;

request.on('data', (chunk) => {
  received += chunk.length;
  if (received > MAX_BYTES) {
    request.destroy(); // обрываем приём
    response.statusCode = 413; // Payload Too Large
    response.end();
  }
});

Смешивание stream и async/await

Внутри обработчика data нельзя нормально ждать промис — событие не притормозит источник, пока вы делаете await. Для этого и нужны async-итераторы (for await...of) или pipeline с генератором. Если видите readable.on('data', async () => ...) с await внутри — это баг, backpressure там сломан.

Что мерить и делать дальше

Перед релизом прогоните выгрузку на файле, близком по размеру к реальному максимуму в проде, а не на тестовом. Снимите три показателя:

1. RSS-память процесса во время потоковой операции — должна оставаться ровной, а не расти линейно с объёмом данных. Рост означает сломанный backpressure.
2. Число открытых файловых дескрипторов (lsof -p PID | wc -l на Linux) до и после серии оборванных запросов — не должно накапливаться.
3. Поведение при uncaughtException: намеренно скормите несуществующий путь и убедитесь, что процесс не падает, а отдаёт 500.

Дальше — настройте graceful shutdown, чтобы при остановке процесса активные стримы дописались или корректно оборвались, и добавьте таймаут на медленных клиентов (server.requestTimeout), иначе один зависший потребитель будет держать поток открытым бесконечно. Инвалидация кеша отданных файлов и работа с S3-стримингом — отдельные темы, здесь мы разбирали только надёжность самой потоковой цепочки.

^* Backpressure — обратное давление: сигнал от медленного потребителя быстрому источнику приостановить подачу данных, чтобы не переполнить буфер в памяти.