cron-task-scheduler 0.1.7

A high-performance async cron scheduler for Rust, supporting both async and blocking tasks with fine-grained scheduling policies.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

use crate::actor::worker::WorkerMessage;
use crate::cron_parser::CronParser;
use crate::models::{ExecutionPolicy, ReactiveTask, SchedulingPolicy, TaskContext};
use chrono::Utc;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::mpsc;
use tracing::{debug, error, info};

pub struct ScheduledTaskItem {
    pub task: Arc<dyn ReactiveTask>,
    pub cron_parser: CronParser,
    pub execution_policy: ExecutionPolicy,
    pub scheduling_policy: SchedulingPolicy,
    pub weight: i8,
}

pub struct SchedulerActor {
    tasks: Vec<ScheduledTaskItem>,
    worker_tx: mpsc::Sender<WorkerMessage>,
}

impl SchedulerActor {
    pub fn new(worker_tx: mpsc::Sender<WorkerMessage>) -> Self {
        Self {
            tasks: Vec::new(),
            worker_tx,
        }
    }

    pub fn add_task(
        &mut self,
        task: Arc<dyn ReactiveTask>,
        cron_expr: &str,
        execution_policy: ExecutionPolicy,
        scheduling_policy: SchedulingPolicy,
        weight: i8,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        let cron_parser = CronParser::new(cron_expr)?;
        self.tasks.push(ScheduledTaskItem {
            task,
            cron_parser,
            execution_policy,
            scheduling_policy,
            weight,
        });
        Ok(())
    }

    pub async fn start_all(self) {
        info!("Starting all scheduled tasks (recursively)");
        for item in self.tasks {
            let worker_tx = self.worker_tx.clone();
            tokio::spawn(async move {
                schedule_loop(item, worker_tx).await;
            });
        }
    }
}

async fn schedule_loop(item: ScheduledTaskItem, worker_tx: mpsc::Sender<WorkerMessage>) {
    let mut next_run = item.cron_parser.next_after(Utc::now());

    while let Some(scheduled_time) = next_run {
        let now = Utc::now();
        let duration_to_wait = scheduled_time.signed_duration_since(now).to_std();

        match duration_to_wait {
            Ok(duration) => {
                debug!("Task {} sleeping for {:?}", item.task.id(), duration);
                tokio::time::sleep(duration).await;
            }
            Err(_) => {
                // If the duration is negative, we are already past the scheduled time
                debug!(
                    "Task {} scheduled time {:?} is in the past, running immediately",
                    item.task.id(),
                    scheduled_time
                );
            }
        }

        let context = TaskContext {
            scheduled_time,
            actual_time: Utc::now(),
            weight: item.weight,
            metadata: {
                let mut m = HashMap::new();
                m.insert("task_id".to_string(), item.task.id().to_string());
                m
            },
        };

        if let Err(e) = worker_tx
            .send(WorkerMessage::Execute {
                task: item.task.clone(),
                context,
                execution_policy: item.execution_policy,
                scheduling_policy: item.scheduling_policy,
            })
            .await
        {
            error!(
                "Failed to send execution message for task {}: {}",
                item.task.id(),
                e
            );
            break;
        }

        next_run = item.cron_parser.next_after(scheduled_time);
    }
}