Конвейеры потоков, трансформы и ловушка .pipe()

Ловушка .pipe(): утечка одного fd на каждый сбой

Урок 02 оставил тебе правило «используй pipeline(), а не ручной pipe()». Вот почему — по механизму. a.pipe(b).pipe(c) верно проводит поток данных и backpressure — эта часть работает, — но по умолчанию делает две опасные вещи. Первое: он не пробрасывает ошибки. Если b генерит 'error', эта ошибка остаётся на b; она не распространяется ни на a, ни на c. Второе: он не разрушает остаток цепочки при сбое. Когда b падает, a остаётся открытым и текущим, а c никто не велит закрыться. Все ресурсы, что держат a и c, — файловый дескриптор, сокет, буфер в полёте — утекают.

// THE TRAP: a malformed gzip body throws in gunzip; the socket leaks.
import zlib from "node:zlib";

function handler(req, res) {
  const gunzip = zlib.createGunzip();
  const parser = createNdjsonParser();

  req.pipe(gunzip).pipe(parser);   // ❌ no error wiring, no cleanup

  parser.on("data", (record) => save(record));
  parser.on("end", () => res.end("ok"));
  // gunzip throws on bad input → 'error' on gunzip only.
  // req (the socket) is never destroyed. One leaked fd per bad request.
}

В одиночном запуске CLI это невидимо: процесс завершается и ОС забирает fd обратно. На долгоживущем сервере это смертельно. Каждый утёкший дескриптор постоянен на всё время жизни процесса, и они накапливаются по одному на сбой. Стандартный ulimit -n часто 1024, поднятый до 4096 на нагруженных машинах; как только ты утёк столько, каждый новый сокет, открытие файла и accept() отказывают с EMFILE, и сервис валится — ровно та вторничная лесенка. Хуже того, 'error' на потоке без слушателя сам по себе становится uncaughtException, так что второй режим сбоя голого pipe — жёсткое падение. У ловушки два выхода, и оба плохие.

Фикс: stream.pipeline()

stream.pipeline(...streams, callback) добавили, чтобы закрыть именно эту дыру. Он проводит тот же поток данных и backpressure, что и pipe(), и дополнительно: пробрасывает ошибку из любого этапа в один callback и разрушает каждый поток в цепочке — при сбое и при нормальном завершении. Одно место узнаёт исход; ничто не утекает. Промис-форма из node:stream/promises даёт те же гарантии за await — это та форма, что нужна в современном async-коде.

// THE FIX: pipeline wires errors AND destroys the whole chain on any outcome.
import { pipeline } from "node:stream/promises";
import zlib from "node:zlib";

async function handler(req, res) {
  try {
    await pipeline(
      req,                       // Readable: the inbound socket
      zlib.createGunzip(),       // Transform: throws on a malformed body
      createNdjsonParser(),      // Transform: objectMode records out
      saveSink(),                // Writable: persists each record
    );
    res.end("ok");
  } catch (err) {
    // gunzip's error lands HERE. req, gunzip, parser, sink are ALL destroyed.
    res.statusCode = 400;
    res.end("bad body");
  }
}

Когда createGunzip() падает на битом теле, pipeline реджектится, каждый поток — включая входящий сокет req — разрушается, и fd освобождается. Никакой лесенки, никакого uncaughtException. Callback-форма, pipeline(req, gunzip, parser, sink, (err) => …), идентична по гарантиям; используй её, когда ты не в async-функции. Для одного потока, а не цепочки, stream.finished(stream, cb) (или его промис-форма) сообщит, когда поток завершён или упал, без переписывания учёта слушателей вручную. Правило заостряется: всякий раз, когда задействован больше чем один поток, первым делом тянись за pipeline() и никогда не печатай .pipe( в серверном коде.

Авторинг Transform: _transform, _flush, object mode

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

Transform — это Duplex, у которого читаемая сторона есть функция от записываемой: gzip, шифр, построчный парсер. Ты пишешь его, реализуя _transform(chunk, encoding, callback): сделай работу, отправь ноль или больше выходов через this.push(...), затем вызови callback(), сигнализируя, что с этим чанком ты закончил и готов к следующему. Опциональный _flush(callback) бежит один раз, когда записываемая сторона заканчивается, — это твой шанс отдать трейлер или то, что осталось во внутреннем накопителе. Ставь objectMode: true, когда чанки — это записи (распарсенные строки), а не байты; тогда this.push(obj) отдаёт JS-объект вместо Buffer.

import { Transform } from "node:stream";

// Поток байт на входе → один объект на строку NDJSON на выходе. objectMode на читаемой стороне.
class NdjsonParser extends Transform {
  constructor() {
    super({ readableObjectMode: true });  // bytes in, objects out
    this.tail = "";
  }
  _transform(chunk, _enc, cb) {
    const lines = (this.tail + chunk).split("\n");
    this.tail = lines.pop();             // last fragment may be a partial line
    try {
      for (const line of lines) {
        if (line) this.push(JSON.parse(line));   // emit one record per line
      }
      cb();                              // success: ready for next chunk
    } catch (err) {
      cb(err);                           // failure: propagate, do NOT throw + cb
    }
  }
  _flush(cb) {
    if (this.tail) this.push(JSON.parse(this.tail));  // last line, no newline
    cb();
  }
}

У Transform два highWaterMark — по одному на сторону — каждый со значением по умолчанию 16384 байта (16КБ) в байтовом режиме или 16 объектов в object mode, и он участвует в backpressure на обеих: он перестаёт тянуть сверху, когда его читаемая сторона полна, и сигналит верху притормозить. Тут кусаются три режима сбоя. Сигналь об ошибках через cb(err), а не throw синхронно внутри async-продолжения — throw там ускользает из машинерии потока и становится uncaughtException, ровно как ловушка EventEmitter. Никогда не вызывай cb дважды: двойной callback — это жёсткий ERR_MULTIPLE_CALLBACK, портящий внутреннее состояние потока. И кардинальный грех — Transform, который буферизует всё в _flush (накопить все чанки, отдать в конце), молча убивает стриминг: он держит весь датасет в памяти, так что вход 4ГБ, который должен был стоить ~3×16КБ на трёхэтапном конвейере, вместо этого стоит 4ГБ и роняет процесс по OOM. Весь смысл — пушить по ходу.