Node.js Streams Promise遇错误未触发拒绝问题排查

核心原因 #

你的代码中存在Promise被多次决议的风险，导致reject被后续的resolve覆盖，最终Promise状态变为resolved，因此catch块无法捕获拒绝：

• 当SFTP流触发error事件并调用reject后，若因流未被正确销毁，后续可能意外触发end事件，进而调用resolve，Promise最终状态会被覆盖为已完成。
• 链式绑定的流事件中，SFTP流的错误未被完全阻断，可能导致CSV流触发额外事件，干扰Promise的状态。

另外，你在两个流上分别绑定了error事件，两次调用reject虽不会改变已决议的Promise状态，但会造成不必要的冗余逻辑。

修复方案 #

方案1：确保Promise仅被决议一次+手动销毁流 #

在getCSV中增加标志位，避免重复决议，并在错误发生时手动销毁所有相关流，阻止后续事件触发：

const getCSV = (fileName) => {
    const results = []
    let isSettled = false // 标记Promise是否已决议
    return new Promise((resolve, reject) => {
        if (!fileName) {
            isSettled = true
            return resolve(results)
        }

        const sftpStream = sftp.createReadStream(`${directoryToFetch}/${fileName}`)
        const csvStream = sftpStream.pipe(csv({ skipLines: 1, separator: '\t' }))

        // SFTP流错误处理
        sftpStream.on('error', (e) => {
            if (isSettled) return
            isSettled = true
            console.error(`Failed parsing CSV ${e}, ${directoryToFetch}/${fileName}`)
            reject(new Error(`Failed parsing CSV ${e}: ${directoryToFetch}/${fileName}`))
            // 销毁所有流，阻止后续事件
            sftpStream.destroy()
            csvStream.destroy()
        })

        // CSV流数据收集
        csvStream.on('data', (data) => results.push(data))

        // 流结束处理
        csvStream.on('end', () => {
            if (isSettled) return
            isSettled = true
            resolve(results)
        })

        // CSV流错误处理
        csvStream.on('error', (e) => {
            if (isSettled) return
            isSettled = true
            reject(new Error(`Failed parsing CSV: ${e}`))
            sftpStream.destroy()
            csvStream.destroy()
        })
    })
}

方案2：使用stream.pipeline简化流处理（推荐） #

Node.js的stream.pipeline会自动处理流的错误传播与资源销毁，无需手动绑定多个事件，代码更简洁可靠：

const { pipeline } = require('stream').promises

const getCSV = async (fileName) => {
    if (!fileName) return []
    const results = []

    try {
        await pipeline(
            sftp.createReadStream(`${directoryToFetch}/${fileName}`),
            csv({ skipLines: 1, separator: '\t' }),
            // 收集数据的异步迭代器
            async function* (source) {
                for await (const data of source) {
                    results.push(data)
                }
            }
        )
        return results
    } catch (e) {
        console.error(`Failed parsing CSV ${e}, ${directoryToFetch}/${fileName}`)
        throw new Error(`Failed parsing CSV ${e}: ${directoryToFetch}/${fileName}`)
    }
}

验证说明 #

修复后，当SFTP流触发权限拒绝等错误时，Promise会被正确拒绝，await操作会抛出错误并被catch块捕获，你会看到catch中的错误日志输出。

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