Node.js tar包流write方法未调用回调致程序挂起问题排查

代码目标 #

这段代码的目标是将上传的文件提取到Express应用中并保存其中一个文件，接着解析该yaml格式的文件，解析后的内容将用于调用方法把原始tar包上传至GCP存储桶。

问题现象 #

当日志打印出will write pubspec后，程序就不再继续执行，甚至调试时无法暂停。这里的Pubspec指的是要写入的yaml文件pubspec.yaml，使用的是tar包。

代码示例 #

app.post("/api/packages/upload", authenticate, versioning, async (req: express.Request, res: express.Response) => {
    const files = req.files;
    if (!files) {
        return res.json({
            "error": {
                "code": "400",
                "message": "No files were uploaded."
            }
        })
    }
    Object.keys(files!).forEach(async (key) => {
        const uploadedFile: UploadedFile = files![key] as UploadedFile;
        await fs.mkdir("./tmp/pubspecs", {
            recursive: true
        });
        const tarStream = tar.extract({
            cwd: "./tmp/pubspecs"
            }, ["pubspec.yaml"])

        const data = await fs.readFile(uploadedFile.tempFilePath);
        await new Promise((resolve, reject) => {
            console.log("Will write pubspec");
            tarStream.write(data, (err) => {
                if (err) {
                    console.log(err);
                    reject(err);
                }
                console.log("wrote pubspec");
                resolve(() => {});
            });
        });
        console.log("Will read yaml");
        const pubspec = YAML.parse((await fs.readFile("./tmp/pubspecs/pubspec.yaml")).toString());
        console.log("read yaml");
        const name = pubspec.name;
        const version = pubspec.version;
        console.log("name: " + name);
        await uploadPackage(name, PackageType.DART, version, uploadedFile);
        res.status(204);
        res.location(`${process.env.URL_BASE}/dart/${name}/${version}/package.tar.gz`)
        await fs.rm(".tmp/pubspecs/pubspec.yaml");
        res.send();
    });
});

问题原因与修复方案 #

核心问题 #

1. Stream未结束导致Promise挂起：tarStream.write()仅完成数据写入缓冲区，并未触发提取操作的结束逻辑，Stream一直处于打开状态，Promise无法resolve，程序卡住。
2. forEach异步处理冲突：forEach中的async函数不会阻塞循环，多文件上传时会出现竞态条件，还可能重复发送响应。
3. 文件路径不一致：删除临时文件时使用.tmp/pubspecs/pubspec.yaml，但之前创建的目录是./tmp/pubspecs，路径不匹配导致删除失败。

修复后的代码 #

app.post("/api/packages/upload", authenticate, versioning, async (req: express.Request, res: express.Response) => {
    const files = req.files;
    if (!files) {
        return res.json({
            "error": {
                "code": "400",
                "message": "No files were uploaded."
            }
        });
    }

    // 用for...of替代forEach，确保异步操作顺序执行
    for (const key of Object.keys(files!)) {
        const uploadedFile: UploadedFile = files![key] as UploadedFile;
        const tmpDir = "./tmp/pubspecs";
        await fs.mkdir(tmpDir, { recursive: true });
        
        const tarStream = tar.extract({
            cwd: tmpDir
        }, ["pubspec.yaml"]);

        const data = await fs.readFile(uploadedFile.tempFilePath);
        
        await new Promise((resolve, reject) => {
            console.log("Will write pubspec");
            // 监听finish事件确认提取完成
            tarStream.on("finish", () => {
                console.log("Extraction finished");
                resolve(null);
            });
            tarStream.on("error", (err) => {
                console.error("Tar extraction error:", err);
                reject(err);
            });
            
            // 写入数据后结束Stream，触发finish事件
            tarStream.write(data, (err) => {
                if (err) {
                    console.error("Write error:", err);
                    reject(err);
                }
                console.log("Wrote pubspec data");
                tarStream.end();
            });
        });

        console.log("Will read yaml");
        const pubspecPath = `${tmpDir}/pubspec.yaml`;
        const pubspecContent = await fs.readFile(pubspecPath, "utf8");
        const pubspec = YAML.parse(pubspecContent);
        console.log("Read yaml");
        
        const name = pubspec.name;
        const version = pubspec.version;
        console.log("name: " + name);
        
        await uploadPackage(name, PackageType.DART, version, uploadedFile);
        
        res.status(204);
        res.location(`${process.env.URL_BASE}/dart/${name}/${version}/package.tar.gz`);
        await fs.rm(pubspecPath);
        res.send();
    }
});

关键修复点 #

• 监听Stream的finish事件：通过finish事件确认tar包提取操作完全完成，确保Promise能正常resolve。
• 主动结束Stream：调用tarStream.end()关闭流，触发finish事件，解决程序卡住的问题。
• 替换forEach为for...of：保证异步操作按顺序执行，避免多文件处理时的竞态条件和重复响应。
• 统一文件路径：修正删除临时文件的路径，确保操作有效。

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