Rocket Fairing、Tokio调度器与定时任务的生命周期问题

背景 #

基于Rocket框架构建API引擎，需在API运行时执行异步定时任务。选择通过Rocket的Fairing机制，在liftoff事件中启动基于tokio-cron-scheduler的调度器。完成数据库日志、结构体与trait等组件搭建后，编译时触发Fairing生命周期相关错误。

相关代码 #

命令模块结构 #

/// 命令执行结果枚举
pub enum CommandResult {
    SUCCESS,
    ERROR(String),
    SKIPPED(String),
}

/// 定义可作为异步定时任务的结构体的trait
#[async_trait]
pub trait Command: Send + Sync {
    /// 返回当前命令名称
    fn get_command_name(&self) -> String;

    /// 返回当前命令的参数载荷
    fn get_command_args(&self) -> Option<HashMap<String, String>>;

    /// 返回cron日志中间件
    fn get_cron_middleware(&self) -> CronLogMiddleware<CronLogRepository>;

    /// 命令执行的核心逻辑，需在实现中重写
    async fn do_run(&self) -> Result<CommandResult>;

    /// 启动命令流程：验证命令锁，在数据库中记录开启的定时任务日志
    async fn begin(&self) -> Result<CronLog> {
        // ...
    }

    /// 结束命令流程：释放命令锁，在数据库中更新定时任务日志的执行结果
    async fn end(&self, cron_log: &CronLog, result: CommandResult) -> Result<()> {
        // ...
    }

    /// 命令的内部执行逻辑，整合begin、end与do_run，供调度器调用
    async fn run(&self) -> Result<()> {
        // ...
    }

    /// 生成命令名称+参数的唯一键，用于锁机制
    fn generate_unicity_key(&self) -> String {
        // ...
    }

    /// 将命令参数转换为字符串载荷
    #[allow(clippy::or_fun_call)]
    fn get_command_args_as_string(&self) -> String {
        // ...
    }
}

/// 封装命令与其定时调度规则的结构体
pub struct CommandHandle<T: Command + ?Sized + Send + Sync> {
    pub command: Box<T>,
    pub schedule: String,
}

测试命令结构体 #

/// 用于测试的命令结构体
pub struct TestCommand {
    pub name: String,
    pub args: Option<HashMap<String, String>>,
    pub cron_log_middleware: CronLogMiddleware<CronLogRepository>,
}

#[async_trait]
impl Command for TestCommand {
    // 实现各类getter方法

    async fn do_run(&self) -> Result<CommandResult> {
        debug!("执行完成！");

        Ok(CommandResult::SUCCESS)
    }
}

Rocket构建器代码 #

let mut sched = CronScheduler::default();

    sched.add_cron(CommandHandle {
        command: Box::new(TestCommand {
            name: "app:test".to_string(),
            args: None,
            cron_log_middleware: cron_log_middleware.clone(),
        }),
        schedule: "*/1 * * * *".to_string(),
    });

    // 将sched添加到Rocket的管理容器中

Fairing实现代码 #

/// 用于启动异步定时任务的Rocket Fairing
#[derive(Default)]
pub struct CronScheduler {
    crons: Vec<CommandHandle<dyn Command>>,
}

impl CronScheduler {
    /// 添加定时任务（包含命令与调度规则）
    pub fn add_cron(&mut self, cron: CommandHandle<dyn Command>) {
        self.crons.push(cron);
    }
}

#[rocket::async_trait]
impl Fairing for CronScheduler {
    //...
                        v -- 错误发生在此处
    async fn on_liftoff(&self, _rocket: &Rocket<Orbit>) {
        let sched = SchedulerBuilder::build().await;

        for handle in self.crons.iter() {
            let job = Job::new_cron_job_async(handle.schedule.as_str(), |_uid, _lock| {
                Box::pin(async move {
                    handle.command.run().await;
                })
            })
            .unwrap();

            sched.add(job).await.unwrap();
        }

        sched.start().await.unwrap();
    }
}

编译错误信息 #

error[E0759]: `self` 具有生命周期 `'life0`，但需要满足 `'static` 生命周期要求
  --> src/core/fairings/cron_scheduler.rs:34:26
   |
34 |       async fn on_liftoff(&self, rocket: &Rocket<Orbit>) {
   |                            ^^^^ 此数据的生命周期为 `'life0`...
...
39 | /                 Box::pin(async move {
40 | |                     handle.command.run().await;
41 | |                 })
   | |__________________- ...被使用且需要存活至 `'static` 生命周期
   |
note: `'static` 生命周期要求由返回类型引入
  --> src/core/fairings/cron_scheduler.rs:34:5
   |
34 |       async fn on_liftoff(&self, rocket: &Rocket<Orbit>) {
   |       ^^^^^ 此返回类型引入了生命周期要求
...
39 | /                 Box::pin(async move {
40 | |                     handle.command.run().await;
41 | |                 })
   | |__________________- 因为此返回表达式

尝试将Box::pin内的代码替换为简单打印时无错误，推测问题与Command相关，但不确定是否是async-trait与Rocket异步trait的冲突。

修复方案 #

将CommandHandle中的Box<T>改为Arc<T>，手动实现Clone trait，并调整Fairing中的任务添加逻辑：

修改后的命令模块代码 #

// command 

pub struct CommandHandle<T: Command + ?Sized + Send + Sync> {
    pub command: Arc<T>,
    pub schedule: String,
}

impl<T> Clone for CommandHandle<T>
where
    T: Command + ?Sized + Send + Sync,
{
    fn clone(&self) -> Self {
        Self {
            command: self.command.clone(),
            schedule: self.schedule.clone(),
        }
    }
}

修改后的Fairing代码 #

// fairing 

    async fn on_liftoff(&self, _rocket: &Rocket<Orbit>) {
        let sched = SchedulerBuilder::build().await;

        for handle in self.crons.iter() {
            let schedule = handle.schedule.clone();

            let handle = handle.clone();

            let job = Job::new_cron_job_async(schedule.as_str(), move |_uid, _lock| {
                let handle = handle.clone();

                Box::pin(async move {
                    handle.command.run().await.unwrap();
                })
            })
            .unwrap();

            sched.add(job).await.unwrap();
        }

        sched.start().await.unwrap();
    }

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