Node.js streams và backpressure: khi nơi ghi chậm hơn nơi đọc Trung cấp

Vấn đề

Một API nhận file upload và ghi từng chunk sang object storage. Client gửi nhanh hơn storage ghi. Nếu code cứ write() mà không đọc giá trị trả về, chunk dồn vào memory của process. Một file nhỏ vẫn pass test; 200 upload 500 MB chạy cùng lúc thì RSS tăng, GC chạy dày, latency tăng, cuối cùng OOM. Thêm RAM chỉ trì hoãn lỗi.

Backpressure là tín hiệu từ nơi tiêu thụ báo “đừng đẩy nhanh hơn tôi xử lý”. Với Writable, tín hiệu đó là write() trả false và sự kiện drain. Bài này dùng một Writable cố ý chậm để nhìn thấy tín hiệu, rồi chỉ ra cách ghi đúng.

Bảng tín hiệu

Một bảng duy nhất cho mọi tín hiệu của Writable:

Tín hiệu	Nghĩa là	Không có nghĩa là
write() trả true	Buffer sau khi nhận chunk vẫn nhỏ hơn highWaterMark	Chunk đã được ghi bền
write() trả false	Chunk đã được nhận; đừng ghi chunk tiếp theo cho tới drain	Chunk bị từ chối, cần ghi lại
writableNeedDrain	true khi buffer đã đầy và stream sẽ phát drain	Một metric memory
drain	Buffer đã được tiêu thụ hết, ghi tiếp được	Toàn pipeline đã xong
finish	Sau end(), mọi dữ liệu đã được flush xuống hệ thống bên dưới	Remote storage đã commit
close	Stream và tài nguyên bên dưới (ví dụ file descriptor) đã đóng; không còn sự kiện nào nữa	Ghi thành công (cũng phát khi bị destroy)
error	Có lỗi; stream thường bị huỷ ngay sau đó

highWaterMark là ngưỡng đổi tín hiệu, không phải giới hạn cứng của memory: nếu cứ write() bất chấp false, Node vẫn buffer tiếp.

highWaterMark mặc định

Loại stream	Mặc định	Ghi chú
Byte stream (Writable, Readable, Transform)	64 KiB từ Node 22	Trước Node 22 là 16 KiB
Object mode	16 object	Đếm theo số object, không theo byte
fs.createReadStream / createWriteStream	64 KiB	Đã đo trên Node 24.17

Đã đo trên Node 24.17.0: new Writable().writableHighWaterMark là 65536, object mode là 16, stream.getDefaultHighWaterMark(false) là 65536 và (true) là 16. Node 22.0.0 liệt kê “stream: bump default highWaterMark” trong các thay đổi semver-major.

Nếu service chạy trên nhiều phiên bản Node, đừng dựa vào mặc định: đặt highWaterMark tường minh hoặc đọc stream.getDefaultHighWaterMark().

Buffer fill theo thời gian

Demo ghi các chunk 8 byte vào Writable có highWaterMark: 16 và sink mất 15 ms mỗi chunk. Trace thật (Node 24.17):

Thời điểm	Sự kiện	writableLength	writableNeedDrain
+0 ms	write #1 trả true	8 / 16	false
+0 ms	write #2 trả false	16 / 16	true
+32 ms	drain, producer tiếp tục	0	false
+32 ms	write #3 trả true, write #4 trả false	16 / 16	true
+64 ms	drain	0	false

Mỗi chu kỳ producer dừng khoảng 32 ms, đúng thời gian sink xử lý hai chunk. Đó là backpressure: tốc độ producer bị kéo về tốc độ consumer. Demo đầy đủ 10 chunk in { chunks: 10, paused: 5 }.

Producer dừng khi Writable buffer đạt ngưỡng, tiếp tục sau drain

Phóng to sơ đồ

Producer → Writable bufferchunk

◀ Trước▶ PhátTiếp ▶↺ Từ đầuBước 1/4

Producer gọi write. Chunk được nhận vào buffer của Writable và chuyển dần tới sink.

Cách ghi đúng

Cách 1: dùng pipeline (mặc định nên chọn). stream/promises.pipeline nối backpressure, truyền lỗi và dọn tài nguyên cho cả chuỗi:

import { pipeline } from 'node:stream/promises';

const ac = new AbortController();
res.on('close', () => { if (!res.writableFinished) ac.abort(); }); // client ngắt kết nối

await pipeline(req, gzip, storageWritable, { signal: ac.signal });

Khi signal bị abort, pipeline gọi destroy với AbortError lên toàn bộ chuỗi (theo tài liệu Node). Nếu một stream bị huỷ giữa chừng, pipeline reject với ERR_STREAM_PREMATURE_CLOSE (đã kiểm chứng ở Node 24.17).

Cách 2: tự ghi bằng vòng lặp. Cần khi nguồn không phải stream. Điểm hay bị bỏ sót: await once(sink, 'drain') tự reject nếu sink phát error trong lúc chờ, nhưng nếu sink bị destroy() không kèm lỗi thì cả drain lẫn error đều không bao giờ phát và vòng lặp treo mãi (đã đo: sau 300 ms vẫn treo). Phải chờ thêm close:

import { once } from 'node:events';

async function writeAll(sink: Writable, chunks: Iterable<Buffer>) {
  for (const chunk of chunks) {
    if (sink.write(chunk)) continue;
    const ac = new AbortController();
    try {
      await Promise.race([
        once(sink, 'drain', { signal: ac.signal }),
        once(sink, 'close', { signal: ac.signal }).then(() => {
          throw new Error('sink đóng sớm');
        }),
      ]);
    } finally {
      ac.abort(); // gỡ listener còn lại
    }
  }
  sink.end();
  await once(sink, 'finish');
}

Với sink bị huỷ giữa chừng, hàm này ném sink đóng sớm; với sink bình thường nó chạy xong. Cả hai đã kiểm chứng ở Node 24.17.

Ví dụ thực tế

Upload 500 MB làm API OOM

Handler gom mọi chunk vào Buffer.concat rồi mới gửi storage. Với 40 upload 500 MB đồng thời, process cần khoảng 20 GB chỉ để giữ body (40 nhân 500 MB). Đội chuyển sang pipeline(req, gzip, storageWritable, { signal }), giới hạn 10 upload đồng thời và kích thước body tối đa, đặt timeout cho storage (timeout budget).

Sau thay đổi, memory mỗi upload còn xấp xỉ tổng các buffer của chuỗi (vài lần 64 KiB cộng buffer của thư viện nén và SDK), nhân với 10. Con số cuối vẫn phải đo bằng heap profile, vì buffer nội bộ của gzip và SDK không bị highWaterMark của bạn kiểm soát.

Hai điểm liên quan đến phía đọc: for await (const chunk of readable) rồi writable.write(chunk) vẫn phải xử lý false hoặc dùng pipeline; vòng lặp đọc không tự nhường cho Writable bất kỳ. Với res.write() trong HTTP, client đọc chậm tạo false giống hệt, và route sinh dữ liệu vô hạn không chờ drain sẽ làm memory tăng theo một client duy nhất.

Checklist production

1. Dùng stream/promises.pipeline cho mọi chuỗi stream; chỉ tự write() khi nguồn không phải stream.
2. Truyền AbortSignal vào pipeline và huỷ khi client ngắt kết nối hoặc hết deadline.
3. Tự ghi thì xử lý false bằng drain, đồng thời nghe error và close để không treo vĩnh viễn.
4. Đặt highWaterMark tường minh nếu hành vi không được phụ thuộc phiên bản Node (mặc định đổi ở Node 22).
5. Giới hạn số upload/stream đồng thời và kích thước body: backpressure từng stream không thay cho budget tổng.
6. Theo dõi writableLength (hoặc writableNeedDrain) của các stream quan trọng và RSS của process.
7. Chỉ coi dữ liệu là bền sau khi storage xác nhận, không phải sau finish của một adapter tự viết.

Sai lầm thường gặp

Sai lầm thường gặp

Coi write() trả false là ghi thất bại và ghi lại chunk đó. Chunk đã nằm trong buffer; ghi lại sẽ tạo dữ liệu trùng. false chỉ nghĩa là chờ drain trước chunk kế tiếp.

Sai lầm thường gặp

Bỏ qua false rồi tin stream tự giới hạn memory. Node vẫn buffer. Với producer nhanh hoặc dữ liệu vô hạn, process vẫn OOM.

Sai lầm thường gặp

Chờ finish rồi báo “đã lưu xong”. finish chỉ nói dữ liệu đã flush xuống tầng bên dưới của stream. Với Writable tự viết gọi callback sớm, remote storage có thể chưa commit.

Tóm tắt

Tóm tắt

• write() === false nghĩa là chờ drain trước chunk tiếp theo; chunk vừa ghi không bị bỏ.
• highWaterMark mặc định 64 KiB (byte) và 16 (object mode) từ Node 22; là ngưỡng tín hiệu, không phải memory cap.
• finish là flush xong sau end(), close là tài nguyên đã đóng (cả khi lỗi hoặc destroy).
• pipeline kèm AbortSignal là mặc định đúng; tự ghi thì phải nghe cả error và close.
• Backpressure từng stream không thay cho giới hạn concurrency và kích thước tổng.

Tự kiểm tra: (1) Sau write() trả false, có ghi lại chunk đó không? (2) Khi nào await once(sink, 'drain') treo mãi? (3) finish và close khác nhau thế nào khi stream bị destroy()?

Chạy demo:

npx --prefix examples ts-node examples/src/node-backpressure/backpressure-demo.ts

Tham khảo

• Stream API — Node.js
• Backpressuring in Streams — Node.js
• Node.js 22.0.0 release notes
• events.once — Node.js