使用@aws-sdk/client-s3时，Pipeline中Transform Stream报错未被捕获怎么办？

问题背景 #

在Node.js 18环境下，使用S3客户端的send方法处理流式上传时，Pipeline内的Transform Stream抛出的错误无法被外层try/catch块捕获。已将问题隔离到示例项目，运行步骤如下：

cd into the folder
nvm use
npm ci
docker compose up -d
npm start

项目中通过async generators创建Readable和Transform流，且发现若将uploadS3改为async generator writable，错误可正常被try/catch捕获。

问题原因 #

这是因为S3客户端的send方法处理流时，内部可能将流的错误处理逻辑与Promise链解耦，导致Transform Stream的错误无法冒泡到外层try/catch。而async generator writable遵循Node.js流与异步迭代器的兼容规范，错误能顺着Promise链正常传递，因此可以被捕获。

可行解决方案 #

1. 使用stream/promises的pipeline方法统一处理 #

Node.js的stream/promises模块提供了Promise化的pipeline方法，能更好地捕获流链中的所有错误。将S3上传逻辑包装到该pipeline中，即可确保错误被try/catch捕获：

const { pipeline } = require('stream/promises');
const { S3Client, PutObjectCommand } = require('@aws-sdk/client-s3');

async function uploadWithPipeline() {
  const s3Client = new S3Client({ /* 你的S3配置 */ });
  const readable = /* 你的async generator生成的Readable流 */;
  const transform = /* 你的Transform流 */;
  
  try {
    await pipeline(
      readable,
      transform,
      async function* (source) {
        const uploadCommand = new PutObjectCommand({
          Bucket: '目标桶名称',
          Key: '文件键名',
          Body: source
        });
        await s3Client.send(uploadCommand);
        yield* source;
      }
    );
    console.log('上传完成');
  } catch (err) {
    console.error('捕获到上传错误:', err);
    // 此处添加错误处理逻辑，如重试、告警等
  }
}

2. 手动监听Transform Stream的error事件 #

如果必须沿用原有send调用方式，可给Transform Stream手动绑定error事件，避免错误丢失：

const { S3Client, PutObjectCommand } = require('@aws-sdk/client-s3');

async function uploadWithErrorListener() {
  const s3Client = new S3Client({ /* 你的S3配置 */ });
  const readable = /* 你的Readable流 */;
  const transform = /* 你的Transform流 */;
  
  return new Promise((resolve, reject) => {
    // 监听Transform流的错误，直接reject Promise
    transform.on('error', (err) => {
      console.error('Transform流出错:', err);
      reject(err);
    });
    
    const uploadCommand = new PutObjectCommand({
      Bucket: '目标桶名称',
      Key: '文件键名',
      Body: readable.pipe(transform)
    });
    
    s3Client.send(uploadCommand)
      .then(resolve)
      .catch(reject);
  });
}

// 调用时用try/catch捕获
async function runUpload() {
  try {
    await uploadWithErrorListener();
    console.log('上传完成');
  } catch (err) {
    console.error('捕获到错误:', err);
  }
}

3. 改用async generator writable（已验证有效） #

正如你发现的，将uploadS3改为async generator writable可解决问题，因为异步迭代器的错误会自然通过Promise链传递，符合try/catch的捕获逻辑：

const { pipeline } = require('stream/promises');
const { S3Client, PutObjectCommand } = require('@aws-sdk/client-s3');

async function* s3UploadWritable(source) {
  const s3Client = new S3Client({ /* 你的S3配置 */ });
  const chunks = [];
  // 遍历获取所有流数据
  for await (const chunk of source) {
    chunks.push(chunk);
  }
  // 合并chunk后上传
  const uploadCommand = new PutObjectCommand({
    Bucket: '目标桶名称',
    Key: '文件键名',
    Body: Buffer.concat(chunks)
  });
  await s3Client.send(uploadCommand);
}

async function upload() {
  const readable = /* 你的Readable流 */;
  const transform = /* 你的Transform流 */;
  
  try {
    await pipeline(readable, transform, s3UploadWritable);
    console.log('上传完成');
  } catch (err) {
    console.error('捕获到上传错误:', err);
  }
}

内容的提问来源于stack exchange，提问作者Hálex Viotto Gomes