Node.js 14升级至18后Stream行为异常，单元测试失败求助

问题背景 #

从Node.js 14升级到18后，项目大部分功能正常运行，但一个CSV表头验证的单元测试失败。Node 14中执行console.log(data)会输出['yo', 'playerId']，但Node 18中输出空数组[]，尝试添加readable.resume()调用无效。

单元测试代码 #

import { ValidationError } from "../../errors"
import { validateCSVHeader } from "../validateCSVHeader"

describe("validateCSVHeader", () => {
  it("Should not throw error if validHeaders included in buffer", async () => {
    const csvBuffer = Buffer.from(`yo,playerId
    1234567,abcd
    7654321,efgh
    `)

    await expect(
      validateCSVHeader(csvBuffer, { validHeaders: ["accountId", "playerId"] })
    ).resolves.not.toThrowError()
  })

  it("Should pass if no specific headers is required", async () => {
    const csvBuffer = Buffer.from(`yo,playerId
    1234567,abcd
    7654321,efgh
    `)

    await expect(validateCSVHeader(csvBuffer)).rejects.toThrowError()
  })

  it("Should throw error if validHeaders not included in buffer", async () => {
    const csvBuffer = Buffer.from(`yo,playerId
    1234567,abcd
    7654321,efgh

    `)

    await expect(
      validateCSVHeader(csvBuffer, { validHeaders: ["accountId"] })
    ).rejects.toThrow(new ValidationError("Invalid CSV Header"))
  })

  it("Should throw error if csv format", async () => {
    const csvBuffer = Buffer.from("")

    await expect(
      validateCSVHeader(csvBuffer, { validHeaders: ["accountId", "playerId"] })
    ).rejects.toThrow(new ValidationError("Invalid CSV Header"))
  })
})

报错信息 #

第一个测试用例未通过：预期不抛出错误，但实际抛出了ValidationError，核心原因是代码中data数组为空，导致表头验证逻辑判定失败。

源码实现 #

import parse from "csv-parse"
import stream, { Readable, Writable } from "stream"
import util from "util"

import { ValidationError } from "../errors"

const pipeline = util.promisify(stream.pipeline)

type ValidationsType = {
  validHeaders?: string[]
}

export const validateCSVHeader = async (
  fileBuffer: Buffer,
  validations?: ValidationsType
) => {
  const { validHeaders = [] } = validations || {}

  const csvParser = parse({
    delimiter: ",",
    bom: true,
    columns: false,
    skip_empty_lines: true,
    to_line: 1,
    record_delimiter: ["\r", "\n", "\r\n"],
  })

  const readable = Readable.from(fileBuffer)

  const data: any[] = []
  const writable = new Writable({
    objectMode: true,
    write(chunk, _encoding, callback) {
      data.push(chunk)
      callback()
    },
  })

  await pipeline(readable, csvParser, writable)

  console.log(data)

  const [firstLine] = data
  if (firstLine && validHeaders.some((header) => firstLine.includes(header))) {
    return
  }

  throw new ValidationError("Invalid CSV Header", {
    fileHeaders: firstLine,
    validHeaders: validHeaders,
  })
}

问题原因与解决方案 #

原因分析 #

Node.js 18中stream.pipeline的流结束时机处理有调整，结合csv-parse的to_line:1配置，解析器在完成第一行解析后提前结束流，但自定义Writable流还未完成数据写入，就触发了pipeline的完成回调，导致data数组为空。

解决方案 #

直接使用csv-parse的异步API替代手动流处理，避免Node.js版本间的流行为差异，同时简化代码逻辑：

修改后的validateCSVHeader代码：

import parse from "csv-parse"
import { ValidationError } from "../errors"

type ValidationsType = {
  validHeaders?: string[]
}

export const validateCSVHeader = async (
  fileBuffer: Buffer,
  validations?: ValidationsType
) => {
  const { validHeaders = [] } = validations || {}

  const records = await parse(fileBuffer, {
    delimiter: ",",
    bom: true,
    columns: false,
    skip_empty_lines: true,
    to_line: 1,
    record_delimiter: ["\r", "\n", "\r\n"],
  })

  console.log(records)
  const [firstLine] = records
  
  if (firstLine && validHeaders.some((header) => firstLine.includes(header))) {
    return
  }

  throw new ValidationError("Invalid CSV Header", {
    fileHeaders: firstLine,
    validHeaders: validHeaders,
  })
}

改动说明 #

1. 利用csv-parse原生支持的异步Promise API，直接传入Buffer获取解析结果，无需手动处理流和pipeline
2. 去掉自定义Writable流，避免流处理的版本兼容性问题
3. 代码逻辑更简洁，同时兼容Node.js 14和18版本，能正确获取表头数据，让单元测试通过

内容的提问来源于stack exchange，提问作者HKIT9