greentic_runner_host/runner/
adapt_timer.rs

1use std::str::FromStr;
2use std::sync::Arc;
3use std::time::Duration;
4
5use anyhow::{Context, Result};
6use chrono::{DateTime, Utc};
7use cron::Schedule;
8use serde_json::json;
9use tokio::task::JoinHandle;
10use tokio::time::sleep;
11
12use crate::engine::runtime::IngressEnvelope;
13use crate::runtime::TenantRuntime;
14
15pub fn spawn_timers(runtime: Arc<TenantRuntime>) -> Result<Vec<JoinHandle<()>>> {
16    let mut handles = Vec::new();
17
18    for timer in runtime.config().timers.clone() {
19        let cron_expr = timer.cron.clone();
20        let normalized = normalize_cron(&cron_expr);
21        let schedule = Schedule::from_str(&normalized)
22            .with_context(|| format!("invalid cron expression for {}", timer.schedule_id()))?;
23        let flow_id = timer.flow_id.clone();
24        let schedule_id = timer.schedule_id().to_string();
25        let tenant = runtime.config().tenant.clone();
26        let runtime_clone = Arc::clone(&runtime);
27
28        let handle = tokio::spawn(async move {
29            tracing::info!(
30                flow_id = %flow_id,
31                schedule_id = %schedule_id,
32                cron = %cron_expr,
33                normalized_cron = %normalized,
34                "registered timer schedule"
35            );
36            for next in schedule.upcoming(Utc) {
37                if let Some(wait) = duration_until(next) {
38                    sleep(wait).await;
39                } else {
40                    continue;
41                }
42                let pack_id = match runtime_clone.engine().flow_by_id(&flow_id) {
43                    Some(flow) => flow.pack_id.clone(),
44                    None => {
45                        tracing::error!(
46                            flow_id = %flow_id,
47                            schedule_id = %schedule_id,
48                            "timer flow is ambiguous; pack_id is required"
49                        );
50                        continue;
51                    }
52                };
53                let payload = json!({
54                    "now": next.to_rfc3339(),
55                    "schedule_id": schedule_id.clone(),
56                });
57                tracing::info!(
58                    flow_id = %flow_id,
59                    schedule_id = %schedule_id,
60                    scheduled_for = %next,
61                    "triggering timer flow"
62                );
63                let envelope = IngressEnvelope {
64                    tenant: tenant.clone(),
65                    env: None,
66                    pack_id: Some(pack_id),
67                    flow_id: flow_id.clone(),
68                    flow_type: Some("timer".into()),
69                    action: Some("timer".into()),
70                    session_hint: Some(schedule_id.clone()),
71                    provider: Some("timer".into()),
72                    channel: Some(schedule_id.clone()),
73                    conversation: Some(schedule_id.clone()),
74                    user: None,
75                    activity_id: Some(format!("{}@{}", schedule_id, next)),
76                    timestamp: Some(next.to_rfc3339()),
77                    payload,
78                    metadata: None,
79                    reply_scope: None,
80                }
81                .canonicalize();
82                match runtime_clone.state_machine().handle(envelope).await {
83                    Ok(output) => {
84                        tracing::info!(
85                            flow_id = %flow_id,
86                            schedule_id = %schedule_id,
87                            now = %next,
88                            response = %output,
89                            "timer flow completed"
90                        );
91                    }
92                    Err(err) => {
93                        let chain = err.chain().map(|e| e.to_string()).collect::<Vec<_>>();
94                        tracing::error!(
95                            flow_id = %flow_id,
96                            schedule_id = %schedule_id,
97                            error.cause_chain = ?chain,
98                            "timer flow execution failed"
99                        );
100                    }
101                }
102            }
103            tracing::info!(
104                flow_id = %flow_id,
105                schedule_id = %schedule_id,
106                "timer schedule completed"
107            );
108        });
109
110        handles.push(handle);
111    }
112
113    Ok(handles)
114}
115
116fn duration_until(next: DateTime<Utc>) -> Option<Duration> {
117    let now = Utc::now();
118    let duration = next - now;
119    if duration.num_milliseconds() <= 0 {
120        return Some(Duration::from_secs(0));
121    }
122    duration.to_std().ok()
123}
124
125fn normalize_cron(expr: &str) -> String {
126    if expr.split_whitespace().count() == 5 {
127        format!("0 {expr}")
128    } else {
129        expr.to_string()
130    }
131}
132
133#[cfg(test)]
134mod tests {
135    use super::*;
136
137    #[test]
138    fn normalize_cron_adds_seconds_for_five_fields() {
139        assert_eq!(normalize_cron("*/5 * * * *"), "0 */5 * * * *");
140        assert_eq!(normalize_cron("0 */2 * * * *"), "0 */2 * * * *");
141    }
142
143    #[test]
144    fn duration_until_returns_zero_for_past_times() {
145        let past = Utc::now() - chrono::Duration::seconds(10);
146        assert_eq!(duration_until(past).unwrap(), Duration::from_secs(0));
147    }
148}
greentic_runner_host/runner/adapt_timer.rs

greentic_runner_host/runner/
adapt_timer.rs
