Труба и её обрыв: свой pipeline с ошибками и уборкой

Способ закончить поток уже есть — end. Но end — окончание вежливое: «кусков больше не будет, допиши очередь и закрой». Оно годится, пока устройство пишет. Если сокет сообщил, что клиента больше нет, дописывать некуда. Куски в очереди никуда не уйдут, finish не придёт, а очередь и соединение так и останутся занятыми. Для этого случая у потоков Node есть второй способ — destroy: выбросить очередь, отпустить ресурс сейчас и сообщить об этом событием close. Если причина — ошибка, её передают аргументом, и destroy(err) испускает сначала error, потом close.

Что значит «отпустить ресурс», знает только сам ресурс: файл закрывает дескриптор, сокет — соединение. Поэтому поток, как и в случае с write и read, получает функцию destroy в конструкторе и зовёт её из своего метода destroy. Ниже Writable из прошлой статьи с четырьмя правками. Первая — флаг destroyed и сам метод. Вторая — done смотрит на свой аргумент: done(err) значит, что устройство кусок не записало, и поток уничтожает себя с этой ошибкой. В задаче про переполненный диск такую ошибку уже ловили, теперь к ней добавилась уборка. Третья — уничтоженный поток молчит: write ничего не берёт, next ничего не отдаёт, а запись, которая шла в момент уничтожения, счётчик не трогает. Четвёртая — после finish поток уничтожает себя сам: всё записано, и ресурс больше не нужен. Сокет в примере теряет клиента на третьем куске. Прогноз: какие события придут, сколько кусков останется в очереди и что ответит write после обрыва?

class Writable extends Emitter {
  buffer = [];
  length = 0;
  writing = false;
  ending = false;
  finished = false;
  needDrain = false;
  destroyed = false;

  constructor({ write, destroy, highWaterMark = 64 * 1024 }) {
    super();
    this.writeChunk = write;
    this.release = destroy;
    this.highWaterMark = highWaterMark;
  }

  write(chunk) {
    // ресурс уже отпущен: кусок некуда ни записать, ни положить
    if (this.destroyed) return false;
    if (this.ending) {
      this.emit("error", new Error("write после end"));
      return false;
    }
    this.length += chunk.length;
    this.buffer.push(chunk);
    this.next();
    // ответ после попытки записи: кусок, записанный сразу, место уже освободил
    const ok = this.length < this.highWaterMark;
    if (!ok) this.needDrain = true;
    return ok;
  }

  end() {
    this.ending = true;
    this.next();
  }

  next() {
    if (this.writing || this.destroyed) return;
    if (this.buffer.length === 0) {
      if (this.needDrain && !this.ending) {
        this.needDrain = false;
        this.emit("drain");
      }
      if (this.ending && !this.finished) {
        this.finished = true;
        this.emit("finish");
        // всё записано — ресурс больше не нужен
        this.destroy();
      }
      return;
    }
    this.writing = true;
    const chunk = this.buffer.shift();
    this.writeChunk(chunk, (err) => {
      this.writing = false;
      // запись шла, пока поток уничтожали: очередь и счётчик уже обнулены
      if (this.destroyed) return;
      // кусок держит память, пока устройство не позвало done
      this.length -= chunk.length;
      if (err) {
        this.destroy(err);
        return;
      }
      this.next();
    });
  }

  destroy(err) {
    if (this.destroyed) return this;
    this.destroyed = true;
    this.buffer = [];
    this.length = 0;
    this.release?.(err);
    if (err) this.emit("error", err);
    this.emit("close");
    return this;
  }
}

let open = true;
const socket = new Writable({
  highWaterMark: 16,
  write(chunk, done) {
    setTimeout(() => {
      // клиент закрыл вкладку: третий кусок отправить уже некуда
      if (chunk === "кусок-3!") done(new Error("write EPIPE"));
      else done();
    }, 5);
  },
  // у настоящего сокета здесь закрывается соединение
  destroy() {
    open = false;
  },
});
socket.on("error", (err) => console.log("error:", err.message));
socket.on("finish", () => console.log("finish"));
socket.on("close", () => console.log("close | в очереди:", socket.length, "| соединение открыто:", open));

for (const chunk of ["кусок-1!", "кусок-2!", "кусок-3!", "кусок-4!", "кусок-5!"]) socket.write(chunk);
socket.end();
setTimeout(() => console.log("write после обрыва:", socket.write("кусок-6!")), 50);

Пришли error и close, а finish — нет, хотя end позвали: четвёртый и пятый куски не записались, и объявить «всё записано» было бы ложью. Очередь пуста не потому, что куски ушли, — её выбросил destroy. Соединение закрыла функция destroy из конструктора, а write после обрыва ответил false и кусок не взял.

Порядок событий не случаен: error раньше close. close — последнее событие потока, после него поток ничего не испускает. И приходит оно при любом исходе: и после finish, и после обрыва. Поэтому код, который убирает что-то своё вслед за потоком, слушает close, а причину узнаёт из error, пришедшего раньше.

class Readable extends Emitter {
  buffer = [];
  length = 0;
  flowing = false;
  reading = false;
  ended = false;
  endEmitted = false;
  destroyed = false;

  constructor({ read, destroy, highWaterMark = 64 * 1024 } = {}) {
    super();
    this.readChunk = read;
    this.release = destroy;
    this.highWaterMark = highWaterMark;
  }

  push(chunk) {
    this.reading = false;
    // кусок прочитан до destroy, а пришёл после: класть его некуда
    if (this.destroyed) return false;
    if (this.ended) {
      this.emit("error", new Error("push после конца потока"));
      return false;
    }
    if (chunk === null) {
      this.ended = true;
    } else {
      this.buffer.push(chunk);
      this.length += chunk.length;
    }
    this.flow();
    return !this.ended && this.length < this.highWaterMark;
  }

  on(name, fn) {
    super.on(name, fn);
    if (name === "data") this.resume();
    return this;
  }

  pause() {
    this.flowing = false;
  }

  resume() {
    queueMicrotask(() => {
      this.flowing = true;
      this.flow();
    });
  }

  flow() {
    // флаг на каждом куске: pause() из слушателя data действует сразу
    while (this.flowing && this.buffer.length > 0) {
      const chunk = this.buffer.shift();
      this.length -= chunk.length;
      this.emit("data", chunk);
    }
    // слушатель data мог уничтожить поток: ни конца, ни чтения
    if (this.destroyed) return;
    if (this.ended) {
      if (this.flowing && this.buffer.length === 0 && !this.endEmitted) {
        this.endEmitted = true;
        this.emit("end");
        // всё отдано — ресурс больше не нужен
        this.destroy();
      }
      return;
    }
    if (this.readChunk && !this.reading && this.length < this.highWaterMark) {
      this.reading = true;
      // после текущего кода: push из read снова зовёт flow, и без отсрочки
      // чтение большого файла шло бы рекурсией глубиной в число кусков
      queueMicrotask(() => {
        if (!this.destroyed) this.readChunk();
      });
    }
  }

  destroy(err) {
    if (this.destroyed) return this;
    this.destroyed = true;
    this.buffer = [];
    this.length = 0;
    this.release?.(err);
    if (err) this.emit("error", err);
    this.emit("close");
    return this;
  }
}

function openFile(name, chunks) {
  let made = 0;
  return new Readable({
    highWaterMark: 16,
    read() {
      if (made === chunks) {
        this.push(null);
        return;
      }
      made += 1;
      this.push(`кусок-${made}!`);
    },
    // у настоящего файла здесь закрывается дескриптор
    destroy() {
      console.log(`${name}: файл закрыт, прочитано кусков: ${made}`);
    },
  });
}

const whole = openFile("целиком", 3);
whole.on("data", () => {});
whole.on("end", () => console.log("целиком: end"));

const cut = openFile("оборван", 100);
cut.on("data", (chunk) => {
  if (chunk === "кусок-2!") cut.destroy();
});
cut.on("close", () => console.log("оборван: close, push отвечает:", cut.push("ещё кусок")));

Emitter, Readable и Writable с destroy теперь лежат в подложке.

Теперь соединение. Функция pipe ниже — три строки из прошлой статьи и одна добавка, которая есть и у pipe в Node: закрылся приёмник — соединение снимается. Слушатели отписываются — закрытому приёмнику куски больше не нужны, — а источник встаёт на паузу, если других приёмников у него не осталось. Файл в примере большой: read отдаёт куски, пока их просят. Сокет, как в прошлом разделе, теряет клиента на третьем куске. Слушатели error стоят на обоих потоках: без слушателя error роняет процесс. Прогноз: получит ли ошибку файл, уничтожат ли его и сколько кусков он прочитает?

// Emitter, Readable и Writable с destroy — в подложке
function pipe(source, dest) {
  const onData = (chunk) => {
    if (!dest.write(chunk)) source.pause();
  };
  const onDrain = () => source.resume();
  const onEnd = () => dest.end();
  // приёмник закрылся — соединение снимается; пауза, только если приёмников не осталось
  const unpipe = () => {
    source.off("data", onData);
    source.off("end", onEnd);
    dest.off("drain", onDrain);
    if (source.listenerCount("data") === 0) source.pause();
  };
  source.on("data", onData);
  source.on("end", onEnd);
  dest.on("drain", onDrain);
  dest.on("close", unpipe);
}

let made = 0;
let fileOpen = true;
const file = new Readable({
  highWaterMark: 16,
  // файл большой: куски кончатся нескоро
  read() {
    made += 1;
    this.push(`кусок-${made}!`);
  },
  destroy() {
    fileOpen = false;
  },
});
const socket = new Writable({
  highWaterMark: 16,
  write(chunk, done) {
    setTimeout(() => {
      if (chunk === "кусок-3!") done(new Error("write EPIPE"));
      else done();
    }, 5);
  },
});
file.on("error", (err) => console.log("error файла:", err.message));
socket.on("error", (err) => console.log("error сокета:", err.message));

pipe(file, socket);

setTimeout(() => {
  console.log("файл уничтожен:", file.destroyed, "| открыт:", fileOpen);
  console.log("прочитано кусков:", made, "| в буфере:", `${file.length}/${file.highWaterMark}`);
}, 100);

Ошибку получил только сокет. Файл о ней не узнал: его слушатель error молчал, destroyed — false, функция закрытия не звана. Разберём шесть кусков. Два записались, третий сорвался, четвёртый ждал в очереди сокета и пропал вместе с ней. Пятый и шестой лежат в буфере файла: pipe снял соединение и поставил файл на паузу, а поток на паузе дочитывает буфер до порога. Так файл и останется — открытым, с полным буфером, в ожидании resume, которого никто не позовёт.

Почему pipe не уничтожает источник? Потому что источник ему не принадлежит. У одного источника бывает несколько приёмников — как два клиента в задаче прошлой статьи, — и обрыв одного клиента не должен обрывать остальных. Настоящий pipe так и поступает: упавший приёмник отсоединяется, остальные получают файл до конца. Решать судьбу источника должен код, который его открыл, а pipe делает единственное, в чём уверен, — снимает соединение. Код, который этого решения не принял, оставляет по открытому файлу на каждое оборванное скачивание.

Обратный случай — сбоит диск. Файл уничтожает себя с ошибкой чтения на третьем куске, сокет здоров. Emitter, потоки с destroy и pipe — в подложке. Прогноз: что будет с сокетом и узнает ли клиент, что файла он не получит?

// Emitter, потоки с destroy и pipe — в подложке
let made = 0;
const file = new Readable({
  highWaterMark: 16,
  read() {
    made += 1;
    // сбой диска на третьем куске
    if (made === 3) {
      this.destroy(new Error("EIO: i/o error, read"));
      return;
    }
    this.push(`кусок-${made}!`);
  },
});
let sent = 0;
let finished = false;
let socketOpen = true;
const socket = new Writable({
  highWaterMark: 16,
  write(chunk, done) {
    setTimeout(() => {
      sent += 1;
      done();
    }, 5);
  },
  destroy() {
    socketOpen = false;
  },
});
file.on("error", (err) => console.log("error файла:", err.message));
socket.on("finish", () => {
  finished = true;
});

pipe(file, socket);

setTimeout(() => {
  console.log("отправлено кусков:", sent, "| finish:", finished, "| соединение открыто:", socketOpen);
}, 100);

Файл получил свою ошибку и уничтожил себя, а сокет о ней не узнал: pipe слушает у источника data и end, но не error и не close. Два куска ушли, finish не пришёл, соединение открыто. На сервере это ответ, оборванный на полуслове и повисший: клиент ждёт продолжения, пока не сработает таймаут, и соединение занято всё это время.

Первое, что хочется дописать, — file.on("error", () => socket.end()): читать больше нечего, значит, закончить ответ. Это хуже зависания. end значит «всё отправлено»: сокет допишет очередь и закроет соединение как положено, и клиент получит обрезок файла как целый. Когда статус и часть тела уже ушли, сообщить клиенту об обрыве можно только обрывом — destroy.

import { createServer } from "node:http";
import { Readable, pipeline } from "node:stream";

// диск отдаёт по куску в 1000 байт и сбоит на четвёртом
function openFile() {
  let made = 0;
  return new Readable({
    read() {
      made += 1;
      if (made === 4) {
        this.destroy(new Error("EIO: i/o error, read"));
        return;
      }
      setTimeout(() => this.push("x".repeat(1000)), 5);
    },
  });
}

const server = createServer((req, res) => {
  const file = openFile();
  if (req.url === "/end") {
    // сбой диска — просто закончить ответ
    file.on("error", () => res.end());
    file.pipe(res);
  } else {
    pipeline(file, res, () => {});
  }
});

server.listen(0, "127.0.0.1", async () => {
  const base = `http://127.0.0.1:${server.address().port}`;
  for (const path of ["/end", "/pipeline"]) {
    const response = await fetch(base + path);
    try {
      const body = await response.text();
      console.log(`${path}: статус ${response.status}, тело получено, байт: ${body.length}`);
    } catch (err) {
      console.log(`${path}: статус ${response.status}, тело не получено: ${err.name}: ${err.message}`);
    }
  }
  server.closeAllConnections();
  server.close();
});

/end: статус 200, тело получено, байт: 3000
/pipeline: статус 200, тело не получено: TypeError: terminated

Из двух обрывов выходит правило уборки: ошибка любого потока уничтожает оба. Файл закрывается, соединение обрывается, и тот, кто соединял, узнаёт, чем кончилось. Это и есть pipeline: pipe, слушатели error на обоих потоках и колбэк, который получает ошибку — или ничего, если всё записано. Первая версия прямолинейна: на ошибку — уничтожить оба потока и позвать колбэк, на finish — позвать колбэк без ошибки. destroy получает ту же ошибку: у каждого потока могут быть свои слушатели error, и им полезно знать причину. Сокет снова теряет клиента на третьем куске. Прогноз: сколько раз позовут колбэк и что останется открытым?

// Emitter, потоки с destroy и pipe — в подложке
function pipeline(source, dest, callback) {
  const fail = (err) => {
    source.destroy(err);
    dest.destroy(err);
    callback(err);
  };
  pipe(source, dest);
  source.on("error", fail);
  dest.on("error", fail);
  dest.on("finish", () => callback());
}

let made = 0;
let fileOpen = true;
const file = new Readable({
  highWaterMark: 16,
  read() {
    made += 1;
    this.push(`кусок-${made}!`);
  },
  destroy() {
    fileOpen = false;
  },
});
const socket = new Writable({
  highWaterMark: 16,
  write(chunk, done) {
    setTimeout(() => {
      if (chunk === "кусок-3!") done(new Error("write EPIPE"));
      else done();
    }, 5);
  },
});

pipeline(file, socket, (err) => {
  console.log("колбэк:", err ? err.message : "без ошибки", "| файл открыт:", fileOpen);
});

Файл закрыт, соединение тоже — уборка работает. Но колбэк пришёл дважды, и первым его позвал не сокет, а файл. Сокет испустил error, fail уничтожил файл с той же ошибкой, и файл испустил свой error — на нём стоит тот же fail. Этот второй вызов уничтожать уже нечего: повторный destroy ничего не делает. Он зовёт колбэк, а потом до своей строки с колбэком доходит и первый вызов.

Двойной колбэк — не косметика. В колбэке сервер пишет журнал, отвечает клиенту, удаляет недописанный файл, считает ошибки. Одна ошибка, записанная дважды, два ответа на один запрос, повторное удаление уже удалённого файла — каждое ломается по-своему. Лечение — флаг: колбэк зовётся один раз, по первой вести, а всё, что приходит следом, отбрасывается.

Флаг чинит двойной колбэк, но у обрыва есть ещё одна дорога, которой pipeline не слышит, — закрытие без ошибки. Клиент закрыл вкладку, и соединение закрылось чисто: ответ сервера на Node в этом случае получает close и никакого error. В миниатюре так закрывает сокет destroy() без аргумента. Ниже pipeline с флагом: колбэк теперь зовётся из одного места, функции report. Прогноз: позовут ли колбэк и что станет с файлом?

// Emitter, потоки с destroy и pipe — в подложке
function pipeline(source, dest, callback) {
  let reported = false;
  const report = (err) => {
    // destroy второго потока испустит error ещё раз: в счёт идёт только первая весть
    if (reported) return;
    reported = true;
    if (err) {
      source.destroy(err);
      dest.destroy(err);
    }
    callback(err);
  };
  pipe(source, dest);
  source.on("error", report);
  dest.on("error", report);
  dest.on("finish", () => report());
}

let made = 0;
let fileOpen = true;
const file = new Readable({
  highWaterMark: 16,
  read() {
    made += 1;
    this.push(`кусок-${made}!`);
  },
  destroy() {
    fileOpen = false;
  },
});
const socket = new Writable({
  highWaterMark: 16,
  write(chunk, done) {
    setTimeout(done, 5);
  },
});

let called = false;
pipeline(file, socket, (err) => {
  called = true;
  console.log("колбэк:", err ? err.message : "без ошибки");
});
// клиент закрыл вкладку: сокет закрыт без всякой ошибки
setTimeout(() => socket.destroy(), 12);
setTimeout(() => {
  console.log("колбэк позван:", called, "| файл открыт:", fileOpen, "| сокет уничтожен:", socket.destroyed);
}, 100);

Колбэк молчит, файл открыт. Сокет уничтожен, pipe снял соединение, но не было ни error, ни finish, а pipeline слушал только их. Это тот же открытый файл, что в прошлом разделе, только хуже: тот, кто соединял, уверен, что ждёт конца отдачи. Закрытие раньше finish — тоже обрыв, просто без названной причины: данные не дошли, а объяснить почему некому. В Node у этой ошибки есть имя — ERR_STREAM_PREMATURE_CLOSE, «Premature close», преждевременное закрытие.

Значит, pipeline слушает ещё и close обоих потоков. close приёмника после finish — успех, без finish — обрыв. close источника без end — обрыв. Успех объявляется на close приёмника, а не на finish, и из этого есть польза: колбэк приходит, когда приёмник уже отпустил свой ресурс, и в нём, например, можно удалять файл, не опасаясь, что его ещё пишут. А какая весть придёт первой, решает порядок событий в destroy: error раньше close, поэтому настоящая причина обрыва опережает «Premature close». Ниже итоговый pipeline и три отдачи: целиком, со сбоем сокета и клиентом, который ушёл после первого куска. Прогноз: что напечатает каждая?

// Emitter, потоки с destroy и pipe — в подложке
function pipeline(source, dest, callback) {
  let reported = false;
  const report = (err) => {
    // destroy второго потока испустит error ещё раз: в счёт идёт только первая весть
    if (reported) return;
    reported = true;
    if (err) {
      source.destroy(err);
      dest.destroy(err);
    }
    callback(err);
  };
  pipe(source, dest);
  source.on("error", report);
  dest.on("error", report);
  // close без end или finish — обрыв, даже если ошибки не было
  source.on("close", () => {
    if (!source.endEmitted) report(new Error("Premature close"));
  });
  dest.on("close", () => {
    report(dest.finished ? undefined : new Error("Premature close"));
  });
}

function serveFile(name, trouble) {
  let made = 0;
  let fileOpen = true;
  const file = new Readable({
    highWaterMark: 16,
    read() {
      if (made === 5) {
        this.push(null);
        return;
      }
      made += 1;
      this.push(`кусок-${made}!`);
    },
    destroy() {
      fileOpen = false;
    },
  });
  let sent = 0;
  const socket = new Writable({
    highWaterMark: 16,
    write(chunk, done) {
      setTimeout(() => {
        // клиент ушёл после первого куска: сокет закрыт без ошибки
        if (trouble === "close" && sent === 1) {
          socket.destroy();
          return;
        }
        if (trouble === "error" && sent === 3) {
          done(new Error("write EPIPE"));
          return;
        }
        sent += 1;
        done();
      }, 5);
    },
  });

  pipeline(file, socket, (err) => {
    console.log(`${name}: ${err ? err.message : "без ошибки"} | отправлено: ${sent} | файл открыт: ${fileOpen}`);
  });
}

serveFile("клиент закрыл вкладку", "close");
serveFile("сокет упал", "error");
serveFile("целиком");

Все три отдачи кончились колбэком и закрытым файлом. Обрыв без ошибки получил свою, «Premature close». У сбоя сокета осталась настоящая причина: close сокета пришёл вслед за error, когда флаг уже стоял. Итоговый pipeline — два десятка строк, и почти все они про обрыв: само соединение в нём — одна строка pipe. Emitter, потоки с destroy, pipe и pipeline лежат в подложке задач.

Кто соединил потоки, тот за них и отвечает: обрыв одного — уничтожить все и сообщить об исходе ровно один раз.

В Node всё это уже написано: pipeline из node:stream и его версия на промисах из node:stream/promises. Прогон ниже повторяет обрывы на настоящих потоках. Файл и сокет — Readable и Writable с теми же порогами и кусками, что в миниатюре. Функция destroy в опциях так же отмечает, что ресурс отпущен, только у Node она получает ещё колбэк, который зовут, когда ресурс закрыт. Случаев четыре: голый pipe и сбой сокета на третьем куске — как в разделе про pipe; то же через pipeline; pipeline и клиент, ушедший после первого куска; await pipeline(…) и сбой диска. Прогноз: совпадёт ли pipe с миниатюрой и что придёт в колбэк и в catch?

import { Readable, Writable, pipeline } from "node:stream";
import * as promises from "node:stream/promises";

// файл и сокет как в миниатюре: куски по 8 символов, пороги по 16
function openFile(state, { failAt = 0 } = {}) {
  let made = 0;
  return new Readable({
    encoding: "utf8",
    highWaterMark: 16,
    read() {
      made += 1;
      state.made = made;
      if (made === failAt) {
        this.destroy(new Error("EIO: i/o error, read"));
        return;
      }
      this.push(`кусок-${made}!`);
    },
    destroy(err, callback) {
      state.fileOpen = false;
      callback(err);
    },
  });
}

function openSocket(state, trouble) {
  let sent = 0;
  const socket = new Writable({
    decodeStrings: false,
    highWaterMark: 16,
    write(chunk, encoding, done) {
      setTimeout(() => {
        if (trouble === "error" && sent === 2) {
          done(new Error("write EPIPE"));
          return;
        }
        // клиент ушёл после первого куска: сокет закрыт без ошибки
        if (trouble === "close" && sent === 1) {
          socket.destroy();
          return;
        }
        sent += 1;
        done();
      }, 5);
    },
    destroy(err, callback) {
      state.socketOpen = false;
      callback(err);
    },
  });
  return socket;
}

const later = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
// каждый случай ждёт своего исхода, а не часов: на занятой машине таймеры потоков
// опаздывали, и строки соседних случаев менялись местами
const closed = (stream) => new Promise((resolve) => stream.on("close", resolve));

{
  const state = { fileOpen: true, socketOpen: true };
  const file = openFile(state);
  const socket = openSocket(state, "error");
  file.on("error", (err) => console.log("pipe: error файла:", err.message));
  socket.on("error", (err) => console.log("pipe: error сокета:", err.message));
  file.pipe(socket);
  await closed(socket);
  // файл дочитывает буфер до порога уже после обрыва
  await later(20);
  console.log("pipe: файл уничтожен:", file.destroyed, "| открыт:", state.fileOpen);
  console.log("pipe: прочитано кусков:", state.made, "| в буфере:", `${file.readableLength}/${file.readableHighWaterMark}`);
}

{
  const state = { fileOpen: true, socketOpen: true };
  const file = openFile(state);
  const socket = openSocket(state, "error");
  await new Promise((resolve) => {
    pipeline(file, socket, (err) => {
      console.log("pipeline: колбэк:", err.message, "| файл уничтожен:", file.destroyed, "| открыт:", state.fileOpen);
      resolve();
    });
  });
}

{
  const state = { fileOpen: true, socketOpen: true };
  const file = openFile(state);
  const socket = openSocket(state, "close");
  await new Promise((resolve) => {
    pipeline(file, socket, (err) => {
      console.log("клиент ушёл: колбэк:", err.code, err.message, "| файл открыт:", state.fileOpen);
      resolve();
    });
  });
}

{
  const state = { fileOpen: true, socketOpen: true };
  const file = openFile(state, { failAt: 3 });
  const socket = openSocket(state);
  try {
    await promises.pipeline(file, socket);
    console.log("сбой диска: отправлено целиком");
  } catch (err) {
    console.log("сбой диска: catch:", err.message, "| соединение открыто:", state.socketOpen, "| finish:", socket.writableFinished);
  }
}

pipe: error сокета: write EPIPE
pipe: файл уничтожен: false | открыт: true
pipe: прочитано кусков: 6 | в буфере: 16/16
pipeline: колбэк: write EPIPE | файл уничтожен: true | открыт: false
клиент ушёл: колбэк: ERR_STREAM_PREMATURE_CLOSE Premature close | файл открыт: false
сбой диска: catch: EIO: i/o error, read | соединение открыто: false | finish: false

pipe совпал с миниатюрой строка в строку: ошибка только у сокета, файл не уничтожен и открыт, прочитано шесть кусков, буфер полон. pipeline уничтожил файл и отдал ошибку сокета в колбэк. Клиент, ушедший без ошибки, превратился в ERR_STREAM_PREMATURE_CLOSE. А версия на промисах отклонилась с ошибкой диска и оборвала соединение, так и не объявив finish: в асинхронном обработчике обрыв любого из потоков ловит один try.

Чем ещё оригинал отличается от миниатюры — в таблице.

Правило для кода на Node короткое: потоки соединяет pipeline, а в асинхронном коде — await pipeline(…) из node:stream/promises внутри try. Голый pipe остаётся для случаев, когда источник живёт дольше соединения и его судьбу решают сами, — тогда слушатели error и close ставят руками. Одного pipeline не убирает: данных. Недописанный файл остаётся на диске, и удалять его — дело колбэка: об этом первая задача, а в статье про загрузку файлов обрезок появится и без обрыва, на лимите размера. А чем соединяемые потоки бывают на самом деле — файл, тело запроса, ответ сервера — и как то же чтение пишут через for await вместо слушателя data, разберёт статья про настоящие потоки.