`.pipe()`와 `pipeline()` — 오류와 역압력 다루기

".pipe()는 짧지만 오류 정리를 모두 책임지지 않아. 프로덕션 코드에서는 pipeline()으로 연결 전체의 성공과 실패를 한곳에서 관리해."

짧고 익숙한 .pipe()

Node 스트림을 연결하는 고전적인 방법은 다음처럼 보였어:

import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';

createReadStream('input.txt')
  .pipe(createGzip())
  .pipe(createWriteStream('output.txt.gz'));

.pipe()는 읽기 스트림을 쓰기 스트림에 연결하고, 소비자가 느리면 읽기를 늦추며, 입력이 끝났을 때 출력도 닫아 줘. 반환값이 다음 목적지라서 여러 단계를 짧게 이어 붙일 수도 있어. 정상 경로만 보면 간결하고 잘 동작해.

문제는 오류가 난 뒤의 정리야

const src = createReadStream('input.txt');
const gz = createGzip();
const dst = createWriteStream('output.txt.gz');

src.on('error', cleanup);
gz.on('error', cleanup);
dst.on('error', cleanup);

src.pipe(gz).pipe(dst);

function cleanup(err) {
  src.destroy();
  gz.destroy();
  dst.destroy();
  console.error(err);
}

pipeline()으로 연결 전체를 기다리기

node:stream/promises의 pipeline()은 모든 단계를 하나의 작업으로 묶어. 어느 단계에서든 오류가 나면 참여한 스트림을 정리하고 프로미스를 실패시키며, 전부 끝나면 프로미스를 성공시켜.

import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
import { pipeline } from 'node:stream/promises';

await pipeline(
  createReadStream('input.txt'),
  createGzip(),
  createWriteStream('output.txt.gz')
);

호출하는 쪽에서는 await 하나로 완료와 실패를 모두 다룰 수 있어. 새 코드에서 여러 스트림을 직선으로 연결한다면 .pipe()보다 pipeline()을 기본으로 삼는 이유야.

빠른 생산자를 늦추는 역압력

역압력은 데이터를 받는 쪽이 처리 속도를 따라가지 못할 때 보내는 쪽을 잠시 멈추는 장치야. 이 제어가 없으면 생산자가 소비자의 내부 버퍼를 계속 채우고, 결국 메모리 사용량이 한계까지 늘어날 수 있어.

Writable의 write()가 false를 돌려주면 버퍼가 찼다는 뜻이므로 drain 이벤트를 기다려야 해:

async function copy(src, dst) {
  for await (const chunk of src) {
    if (!dst.write(chunk)) {
      await new Promise(r => dst.once('drain', r));
    }
  }
  dst.end();
}

pipeline()은 이 흐름 제어를 연결 전체에 적용해. 큰 업로드나 파일 복사에서 메모리가 계속 늘어난다면 수동 쓰기 코드가 write()의 반환값을 무시하는지 먼저 확인해.

직접 제어해야 하는 경우

여러 입력을 하나의 출력에 임의 순서로 합치거나, 출력 시점을 세밀하게 조절하는 사용자 정의 스트림을 만들 때는 수동 쓰기가 필요할 수 있어. Transform 한 단계를 없애는 최적화가 실제 병목으로 측정된 경우도 예외가 될 수 있지. 그 밖의 직선 처리 흐름에서는 pipeline()이 오류 정리와 역압력을 함께 맡도록 두는 편이 안전해.

Pippa의 고백