Fastify+S3文件下载：Node.js流多环节管控与ClamAV扫描问题

我正在搭建一个Fastify服务器，用于实现文件与S3的上传/下载功能。下载文件时，我通过Node.js流获取文件并返回，以此避免将整个文件加载到内存中。我希望实现以下流程：

1. 从存储桶获取文件web stream
2. 使用ClamAV对流式数据进行扫描，仅当文件未被感染时才将文件流返回给客户端

我尝试使用pipeline实现，但需要等待扫描完成后再处理，且将S3流传给ClamAV后，原流会被消费，无法再发送给客户端。请问各位有什么建议或需要注意的要点吗？

当前代码如下：

const stream = body.transformToWebStream();

const antivirusPassthrough = app.avClient.passthrough();
const downloadPassthrough = new PassThrough();


antivirusPassthrough.once("error", (err) => {
  throw err
});
antivirusPassthrough.once("scan-complete", (result) => {
  const { isInfected } = result;
  if (isInfected) {
    throw new Errpr("File is infected");
  }
});


pipeline(
  stream,
  antivirusPassthrough,
  downloadPassthrough,
  (err) => {
    if (err) {
      app.log.error(err);
    }
  }
);
return reply.send(downloadPassthrough);

核心问题拆解 #

你的代码存在两个关键问题：

• 扫描未完成就将流返回给客户端，可能导致感染文件被提前发送
• 原流被ClamAV消费后无法复用，且事件回调中的throw无法被Fastify捕获，错误处理无效

具体改进方案 #

1. 先完成扫描，再返回文件流（推荐） #

ClamAV必须读完整个流才能给出扫描结果，因此可以先完成扫描，确认文件安全后再重新从S3获取流返回给客户端。这种方式逻辑简单，不会出现流复用问题，唯一代价是多一次S3请求。

示例代码：

// 第一步：从S3获取流进行扫描
const { Body: scanBody } = await s3Client.getObject({ Bucket: '你的存储桶名', Key: '文件Key' });
const scanStream = scanBody.transformToWebStream();

const antivirusPassthrough = app.avClient.passthrough();

// 等待扫描结果
await new Promise((resolve, reject) => {
  antivirusPassthrough.once('scan-complete', (result) => {
    result.isInfected ? reject(new Error('文件已感染')) : resolve();
  });
  antivirusPassthrough.once('error', reject);
  
  pipeline(scanStream, antivirusPassthrough, (err) => err && reject(err));
});

// 第二步：扫描通过，重新从S3获取流返回给客户端
const { Body: safeBody } = await s3Client.getObject({ Bucket: '你的存储桶名', Key: '文件Key' });
return reply.send(safeBody.transformToWebStream());

2. 分流缓存避免重复S3请求（大文件场景可选） #

如果不想重复请求S3，可以通过分流将原流同时导入缓存和扫描器，扫描通过后再从缓存读取流返回。注意大文件场景下优先用磁盘缓存，避免内存溢出。

内存缓存（小文件适用）

const { Body } = await s3Client.getObject({ Bucket: '你的存储桶名', Key: '文件Key' });
const sourceStream = Body.transformToWebStream();

// 创建缓存流和扫描分流
const cacheStream = new PassThrough();
const scanBranchStream = new PassThrough();

// 将原流分流到缓存和扫描分支
sourceStream.pipe(cacheStream);
sourceStream.pipe(scanBranchStream);

// 执行扫描
await new Promise((resolve, reject) => {
  const antivirusPassthrough = app.avClient.passthrough();
  
  antivirusPassthrough.once('scan-complete', (result) => {
    if (result.isInfected) {
      reject(new Error('文件已感染'));
      cacheStream.destroy(); // 销毁缓存流，停止数据传输
    } else {
      resolve();
    }
  });
  antivirusPassthrough.once('error', (err) => {
    reject(err);
    cacheStream.destroy();
  });
  
  pipeline(scanBranchStream, antivirusPassthrough, (err) => err && reject(err));
});

// 扫描通过，返回缓存流
return reply.send(cacheStream);

磁盘缓存（大文件适用）

使用fs.createWriteStream临时存储文件，扫描通过后用fs.createReadStream读取返回，请求结束后记得清理临时文件：

const tempFilePath = `/tmp/${Date.now()}_temp_file`;
const writeStream = fs.createWriteStream(tempFilePath);

// 先把S3流写入临时文件
await new Promise((resolve, reject) => {
  sourceStream.pipe(writeStream);
  writeStream.on('finish', resolve);
  writeStream.on('error', reject);
});

// 读取临时文件进行扫描
const scanStream = fs.createReadStream(tempFilePath);
// ...执行扫描逻辑（同前文）

// 扫描通过，返回临时文件流
const responseStream = fs.createReadStream(tempFilePath);
// 请求结束后删除临时文件
reply.on('finish', () => fs.unlink(tempFilePath, () => {}));
return reply.send(responseStream);

3. 错误处理优化 #

• 事件回调中的throw无法被Fastify捕获，需直接通过reply返回错误响应
• 确保所有流的错误都被捕获，避免内存泄漏

优化示例：

antivirusPassthrough.once('scan-complete', (result) => {
  if (result.isInfected) {
    reply.code(403).send('文件已感染，无法下载');
    return;
  }
  // 继续返回文件流逻辑
});

antivirusPassthrough.once('error', (err) => {
  app.log.error('病毒扫描失败:', err);
  reply.code(500).send('病毒扫描服务异常');
});

关键注意事项 #

• 大文件场景下，内存缓存可能导致内存溢出，优先选择重复S3请求或磁盘缓存
• 确认ClamAV客户端的passthrough模式支持流式扫描，避免出现扫描不完整的情况
• 返回流时需正确设置响应头（如Content-Type、Content-Length），保证客户端正常解析文件
• 添加超时逻辑，防止扫描或S3请求长时间阻塞服务

内容的提问来源于stack exchange，提问作者F. Maida