Node.js archiver包归档S3文件时卡在Archive.finalize()方法

问题场景 #

使用Node.js的archiver包归档AWS S3中的对象键，执行到await archive.finalize();时进程卡住。测试情况：

• 文件夹内文件总大小小于1MB时可能正常运行
• 包含1.62MB图片的文件夹会卡住，小文件也会出现相同问题
• 单独转换单个文件流时正常工作

问题代码 #

归档函数 #

async function createsZipOrGzip(
  client: S3ExtendedClient,
  keys: string[],
  folderCompressionMode: FolderCompressionMode,
  filePath: string
) {
  try {
    client.logger.info(`createsZipOrGzip function started`);
    client.logger.debug(
      `input got in createsZipOrGzip method is keys:${JSON.stringify(
        keys
      )} filePath: ${filePath} folderCompressionMode:${folderCompressionMode}`
    );
    const compresionMode =
      folderCompressionMode === FolderCompressionMode.ZIP ? "zip" : "tar";
    const archive = archiver(compresionMode, {
      zlib: { level: 1 }, // Set compression level
    });

    // Iterate through each object key
    client.logger.info("getting object from the keys");
    for (const key of keys) {
      // Get the object from S3
      const command = new GetObjectCommand({
        Bucket: client.bucket,
        Key: key,
      });

      client.logger.info("getting object the keys", key);

      const response = await client.s3client.send(command);
      // Append the file to the zip archive
       archive.append(response.Body, { name: key });

      client.logger.info("appended  the key", key);
    }

    await archive.finalize();

    client.logger.debug(`worked from here?`);

    if (filePath.endsWith("/")) filePath = filePath.slice(0, -1);

    if (folderCompressionMode === FolderCompressionMode.GZIP) {
      const gzip = zlib.createGzip();
      // compressing the tar to gzip as gzip can't alone compress multiple files
      archive.pipe(gzip);
      return {
        archive: gzip,
        key: filePath !== "" ? `${filePath}.tgz` : "root.tgz",
      };
    }

    client.logger.info("compressed  successfully function ended");

    return {
      archive,
      key:
        filePath !== ""
          ? `${filePath}.${compresionMode}`
          : `root.${compresionMode}`,
    };
  } catch (error) {
    throw new Error(error.message);
  }
}

调用代码 #

const streamZipOrTar = await createsZipOrGzip(
  client,
  keys,
  folderMode,
  filePath
);

问题原因 #

• Archiver流未被消费：Archiver基于流实现，若未将其输出流pipe到可写流（如文件、HTTP响应），内部缓冲区会被填满导致阻塞，finalize()无法完成。
• GZIP处理顺序错误：代码在finalize()之后才尝试将archive pipe到gzip流，颠倒了流处理顺序——必须先建立pipe关系，再添加内容、调用finalize，否则archive输出无去处，必然卡住。

修复方案 #

核心修复点 #

• 提前创建目标流（直接返回archive或gzip包装后的流），立即将archive pipe到该流。
• 严格遵循流操作顺序：创建archive → pipe到目标流 → 添加文件 → finalize。

修复后的代码 #

async function createsZipOrGzip(
  client: S3ExtendedClient,
  keys: string[],
  folderCompressionMode: FolderCompressionMode,
  filePath: string
) {
  try {
    client.logger.info(`createsZipOrGzip function started`);
    client.logger.debug(
      `input got in createsZipOrGzip method is keys:${JSON.stringify(
        keys
      )} filePath: ${filePath} folderCompressionMode:${folderCompressionMode}`
    );
    
    const compressionMode = folderCompressionMode === FolderCompressionMode.ZIP ? "zip" : "tar";
    const archive = archiver(compressionMode, {
      zlib: { level: 1 },
    });

    // 提前创建目标流并建立pipe关系
    let outputStream = archive;
    if (folderCompressionMode === FolderCompressionMode.GZIP) {
      const gzip = zlib.createGzip();
      archive.pipe(gzip);
      outputStream = gzip;
    }

    // 处理文件路径
    if (filePath.endsWith("/")) {
      filePath = filePath.slice(0, -1);
    }

    // 遍历添加S3文件
    client.logger.info("getting objects from keys");
    for (const key of keys) {
      client.logger.info("fetching object", key);
      const command = new GetObjectCommand({
        Bucket: client.bucket,
        Key: key,
      });
      const response = await client.s3client.send(command);
      // 监听S3流错误，避免静默阻塞
      response.Body.on('error', (err) => {
        client.logger.error(`Error reading S3 object ${key}:`, err);
        archive.abort();
        throw err;
      });
      archive.append(response.Body, { name: key });
      client.logger.info("appended object", key);
    }

    // 监听Archiver错误
    archive.on('error', (err) => {
      client.logger.error("Archiver error:", err);
      throw err;
    });

    // 调用finalize，此时流已有输出目标
    await archive.finalize();
    client.logger.debug(`Archive finalized successfully`);

    // 生成最终键名
    const finalKey = filePath !== "" 
      ? `${filePath}.${folderCompressionMode === FolderCompressionMode.GZIP ? 'tgz' : compressionMode}` 
      : `root.${folderCompressionMode === FolderCompressionMode.GZIP ? 'tgz' : compressionMode}`;

    client.logger.info("Compression completed successfully");
    return {
      archive: outputStream,
      key: finalKey,
    };
  } catch (error) {
    client.logger.error("Compression failed:", error);
    throw new Error(error.message);
  }
}

额外注意事项 #

• 添加流错误监听：给S3的Body流和archive流绑定错误事件，避免单个文件读取失败导致静默阻塞。
• 内存管理：归档大量文件时，需关注流式处理的内存限制，防止内存溢出。

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