如何用NodeJS(NestJS)实现直传Google Cloud Storage(GCS)的断点续传

业务背景 #

基于NestJS开发面向Google Cloud Storage（GCS）的断点续传服务，业务场景为前端上传文件后，后端不做临时存储直接将文件发送至GCS。

初始实现（存在问题） #

我最初使用的代码如下：

try {
  const filePath = path.join(directory, nameWithExtension);
  const file = this.bucket.file(filePath);
  const passthroughStream = new stream.PassThrough();
  passthroughStream.write(image.buffer);
  passthroughStream.end();

  const streamFileUpload = async () => {
    passthroughStream
      .pipe(file.createWriteStream({ resumable: true, gzip: true, public: true }))
      .on('finish', () => console.log(`resumable upload succeed`));
    return filePath;
  };

  const res = await streamFileUpload().catch((error) => {
    throw new Error(`${logPrefix} Error uploading ${filePath} ${error.message}`);
  });
  return `${process.env.GOOGLE_STORAGE_ENDPOINT}/${this.bucket.name}/${res}`;
} catch (error) {
  throw new Error(`${logPrefix} Error uploading ${error.message}`);
}

我在createWriteStream参数中添加了resumable: true配置，但功能未达到预期效果，查阅官方文档后仍对实现逻辑不够清晰，希望获得相关建议。

修复后可运行版本 #

2021年10月6日更新：我将代码调整为如下版本后功能正常运行：

async resumableUpload(directory: string, image: MultipartFile, nameWithExtension: string): Promise<string> {
  const logPrefix = 'GoogleStorageService.resumableUpload:';
  const filePath = path.join(directory, nameWithExtension);
  const { buffer } = image;
  const blob = this.bucket.file(filePath);
  const promiseUpload = new Promise((resolve, reject) => {
    const blobStream = blob.createWriteStream({
      resumable: true,
      gzip: true,
      public: true,
    });
    blobStream
      .on('error', () => {
        reject(`${logPrefix} Unable to upload image, something went wrong`);
      })
      .on('finish', async () => {
        const publicUrl = new URL(process.env.GOOGLE_STORAGE_ENDPOINT || '');
        publicUrl.pathname = path.join(this.bucket.name, filePath);
        resolve(publicUrl.toString());
      })
      .end(buffer);
  });
  const response = promiseUpload
    .then((res: string) => res)
    .catch((err: Error) => {
      throw new Error(`${logPrefix} Error uploading ${err.message}`);
    });
  return response;
}

目前代码已可正常运行，请问是否有更优的实现方案？

初始代码问题根因 #

• 异步逻辑时序错误：streamFileUpload函数没有等待流上传完成就直接返回了filePath，你写的await实际等待的是这个同步返回的字符串，而非上传流的完成事件，因此上传报错无法捕获，resumable配置也无法按预期生效，还会出现接口返回成功但GCS侧无文件的偶发问题。
• 冗余流开销：额外声明的PassThrough流完全没有必要，你已经拿到了完整的文件Buffer，直接调用GCS写入流的end(buffer)方法写入即可，多套一层流只会增加内存开销和异常风险。

现有版本优化建议 #

你当前的可运行版本逻辑已经通顺，可从可靠性、性能、可维护性三个方向做优化：

1. 错误捕获优化 #

现有error事件回调没有携带原始错误信息，排查问题时无法区分是网络错误、权限错误还是参数错误，建议修改为：

.on('error', (err) => {
  reject(new Error(`${logPrefix} Unable to upload image: ${err.message}, stack: ${err.stack}`));
})

2. 大文件性能优化 #

你当前是等NestJS把整个文件解析为完整Buffer后再处理，对于100MB以上的大文件会占用很高的内存。建议将NestJS的文件解析配置改为流式接收，直接把前端的上传流pipe到GCS的写入流中，内存占用可降到KB级，更适合大文件断点续传场景。

3. 断点续传复用优化 #

当前的断点续传由GCS客户端自动处理，但如果上传过程中服务重启，已经上传的分片会被浪费。你可以给resumable参数传入自定义的sessionId缓存逻辑，将GCS返回的续传session存储在Redis等持久化组件中，同一个文件下次上传时直接复用session，即可实现跨请求的断点续传，无需重传已上传分片。

4. 返回值生成优化 #

不要手动拼接公开访问URL，建议直接调用GCS SDK提供的getPublicUrl()方法获取地址，避免手动拼接时出现转义错误、路径拼接错误的问题。如果需要使用自定义域名，只需替换返回地址的前缀即可。

5. 超时配置补充 #

建议给上传流添加超时配置，避免出现流卡住长时间占用服务器连接资源的问题。

优化后参考实现 #

async resumableUpload(directory: string, image: MultipartFile, nameWithExtension: string): Promise<string> {
  const logPrefix = 'GoogleStorageService.resumableUpload:';
  const filePath = path.join(directory, nameWithExtension);
  const blob = this.bucket.file(filePath);

  return new Promise((resolve, reject) => {
    const blobStream = blob.createWriteStream({
      resumable: true,
      gzip: true,
      public: true,
      timeout: 300000, // 配置5分钟超时
      // 如需跨请求续传，可传入自定义session逻辑
      // resumable: {
      //   sessionId: await getResumableSessionFromRedis(filePath)
      // }
    });

    blobStream
      .on('error', (err) => {
        reject(new Error(`${logPrefix} Upload failed: ${err.message}`));
      })
      .on('finish', () => {
        // 直接调用SDK方法获取公开地址，避免拼接错误
        const publicUrl = blob.publicUrl();
        // 如需自定义域名可替换前缀
        // const customUrl = publicUrl.replace('storage.googleapis.com', process.env.GOOGLE_STORAGE_ENDPOINT)
        resolve(publicUrl);
      })
      // 若使用流式接收，此处直接写 .pipe(前端上传流) 即可，无需传入buffer
      .end(image.buffer);
  }).catch((err: Error) => {
    throw new Error(`${logPrefix} Error uploading ${nameWithExtension}: ${err.message}`);
  });
}

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