Node ReadStream 轉接 Web ReadableStream 的 memory leak 問題

記錄一下最近碰到的問題。

公司的桌面端 App 專案，有個需求是讀取硬碟上 file 做一些加工處理之後再存回硬碟上。

由於 Node.js 本身存在記憶體限制，所以為了應對大檔案讀寫，勢必得用 Stream 串流來處理。但實際遇上好幾十 GB 量級的檔案時，App 最終還是遭遇了記憶體 alloc 挖不出來的問題，花了許多功夫排查原因……

找了幾天之後發現，似乎是 Node.js 的 Stream 本身就存在一些 GC 不良的問題，且 stream.pipe() 串接時，也不會自動控制水流。必須要使用 node:stream module 的 pipeline 去串接才有自動控制機制。

High memory consumption when piping between streams · Issue #50762 · nodejs/node

由於 Node Stream 不會在後面忙碌時自動暫停水流，導致 ReadStream 一直無腦把 chunk 往下送，但後面又來不及處理，最後記憶體爆炸。

就算是將 Node Stream 轉接到有自動控制功能的 Web Stream 上也會有一樣的問題，畢竟 Web Stream 就是不同的介面，他沒辦法直接管理 Node Stream。

import fs from "node:fs"

const readStream = fs.createReadStream('/path/to/file')

const readable = new ReadableStream({
  async start(controller) {
    try {
      for await (const chunk of readStream) {
        controller.enqueue(chunk) // 串流一接上, 就會一直往下流
      }
    } finally {
      controller.close()
    }
  }
})

await readable
  .pipeThrough(someStream)
  .pipeThrough(heavyStream) // 處理較慢時, 前面還一直送 chunk 水管就會爆炸
  .pipeTo(writable)

後來才找到了正確的串接方式，Stream 有一種叫做 Backpressure 概念，就是在講，當上游的資料產生速度快過下游的消費速度時的處理方式。

而具體做法很單純，就是 ReadStream API 是有個暫停跟繼續的，不能直接用 for-await 去接水流，而是要用傳統串流 on 監聽的方式，去手動做暫停跟繼續。

需要把 ReadableStream 的部分改一下。

// Backpressure support

const readStream = fs.createReadStream('/path/to/file')

const readable = new ReadableStream({
  async start(controller) {
    readStream.on('data', (chunk) => {
      controller.enqueue(chunk)

      // 當下游請求 size 小於零表示忙碌中, 暫停 read
      if (controller.desiredSize <= 0) {
        readStream.pause()
      }
    })

    readStream.on('error', () => controller.error())
    readStream.on('end', () => controller.close())
  },
  /** 下游閒置時會呼叫 pull 向上游請求資源 */
  pull() {
    readStream.resume() // 恢復 read
  },
  cancel() {
    readStream.destroy()
  }
})

這樣在下游忙碌時，就可以自動暫停 Node ReadStream，並在閒置時自動繼續串流，就不會有 memory leak 的問題了。

Reference:

Node.js - Backpressuring in Streams

Streams