NestJS中nest-sftp定时任务触发时出现ETIMEDOUT错误求助

问题根源 #

定时任务触发BullMQ任务时SFTP上传超时，但手动REST触发正常，核心原因在于BullMQ的Worker进程与主进程的资源隔离：

• nest-sftp默认在主进程初始化单例连接，而BullMQ Worker是独立进程，无法共享主进程的SFTP连接。当定时任务触发时，Worker进程中注入的sftpClient可能是无效/已超时的连接；
• 手动REST触发时，任务在主进程的请求上下文执行，复用了主进程保持的有效SFTP连接，因此可以正常上传。

解决方案 #

方案1：在Worker任务中动态创建SFTP连接 #

直接在任务处理逻辑中临时创建SFTP连接，避免依赖主进程的单例连接，确保每次任务执行都使用新鲜的有效连接：

import { Client } from 'ssh2-sftp-client';
import * as fse from 'fs-extra';
import { InternalServerErrorException } from '@nestjs/common';

async jlineDSU() {
    this.logger.warn("[JDSU] - New request received.");

    const today = new Date().toLocaleDateString("nl-BE", {
        year: "numeric",
        month: "2-digit",
        day: "2-digit"
      }).split("/");

    const basePath = `${process.cwd()}/public/vendors/x/dsu`;
    let sftpClient: Client | null = null;

    try {
      const inputFile =`${basePath}/in/File_X_${today[2]}_${today[1]}_${today[0]}.csv`;
      const outputFile = `${basePath}/out/stock_daily.csv`;

      await this.stockUpdater.readFileWithRetries(inputFile);
      this.logger.log("[JDSU] - File read successfull.");

      await this.stockUpdater.transformFile(inputFile, outputFile);
      this.logger.log("[JDSU] - Uploading file to SFTP server.");

      // 动态创建SFTP连接
      sftpClient = new Client();
      await sftpClient.connect({
        host: this.config.get<string>("cfg.ftp.dgoFtpHost"),
        port: this.config.get<number>("cfg.ftp.dgoFtpPort"),
        username: this.config.get<string>("cfg.ftp.dgoFtpUsername"),
        password: this.config.get<string>("cfg.ftp.dgoFtpPassword")
      });

      // 执行上传
      await sftpClient.put(fse.readFileSync(outputFile), `${this.config.get<string>("cfg.ftp.dgoFtpRemotePath")}/stock_daily.csv`);

      await fse.rm(inputFile);
      await fse.rm(outputFile);

      this.logger.log("[JDSU] - update finished succesfully.");
    } catch (err) {
      throw new InternalServerErrorException(err.message);
    } finally {
      // 确保任务结束后关闭连接
      if (sftpClient) {
        await sftpClient.end();
      }
    }
  }

方案2：配置SFTP连接心跳，防止闲置超时 #

如果坚持使用注入的sftpClient，可以通过添加心跳配置，避免连接因长时间闲置被SFTP服务器主动断开：

修改App.module.ts中的SFTP配置：

SftpModule.forRootAsync({
  inject: [ConfigService],
  imports: [ConfigModule],
  useFactory: async (config: ConfigService) => ({
    host: config.get<string>("cfg.ftp.dgoFtpHost"),
    port: config.get<number>("cfg.ftp.dgoFtpPort"),
    username: config.get<string>("cfg.ftp.dgoFtpUsername"),
    password: config.get<string>("cfg.ftp.dgoFtpPassword"),
    // 添加心跳参数，维持连接活性
    keepaliveInterval: 30000, // 每30秒发送一次心跳包
    keepaliveCountMax: 10 // 连续10次心跳失败则重新连接
 })
}, false),

方案3：添加上传重试机制 #

针对偶发的超时问题，增加重试逻辑提高任务容错性：

// 封装带重试的上传方法
private async uploadWithRetry(localPath: string, remotePath: string, maxRetries = 3) {
  let attempt = 0;
  while (attempt < maxRetries) {
    try {
      await this.sftpClient.upload(fse.readFileSync(localPath), remotePath);
      return;
    } catch (err) {
      attempt++;
      this.logger.warn(`Upload attempt ${attempt} failed: ${err.message}`);
      if (attempt === maxRetries) throw err;
      // 重试前等待1秒
      await new Promise(resolve => setTimeout(resolve, 1000));
    }
  }
}

// 在任务中调用
async jlineDSU() {
  // ... 其他逻辑
  await this.uploadWithRetry(outputFile, `${this.config.get<string>("cfg.ftp.dgoFtpRemotePath")}/stock_daily.csv`);
  // ... 其他逻辑
}

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