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    use std::sync::Arc;
137
138    #[test]
139    fn normalize_cron_adds_seconds_for_five_fields() {
140        assert_eq!(normalize_cron("*/5 * * * *"), "0 */5 * * * *");
141        assert_eq!(normalize_cron("0 */2 * * * *"), "0 */2 * * * *");
142    }
143
144    #[test]
145    fn duration_until_returns_zero_for_past_times() {
146        let past = Utc::now() - chrono::Duration::seconds(10);
147        assert_eq!(duration_until(past).unwrap(), Duration::from_secs(0));
148    }
149
150    #[test]
151    fn duration_until_returns_positive_duration_for_future_times() {
152        let future = Utc::now() + chrono::Duration::milliseconds(150);
153        let wait = duration_until(future).unwrap();
154        assert!(wait > Duration::from_secs(0));
155        assert!(wait <= Duration::from_secs(1));
156    }
157
158    #[test]
159    fn normalize_cron_preserves_non_five_field_expressions() {
160        assert_eq!(normalize_cron("@daily"), "@daily");
161        assert_eq!(normalize_cron("0 0 */2 * * *"), "0 0 */2 * * *");
162    }
163
164    #[tokio::test]
165    async fn spawn_timers_is_empty_when_config_has_no_timers() {
166        let (_workspace, runtime) = crate::test_support::build_test_runtime()
167            .await
168            .expect("runtime");
169        let timers = spawn_timers(Arc::clone(&runtime)).expect("spawn timers");
170        assert!(timers.is_empty());
171    }
172}
greentic_runner_host/runner/adapt_timer.rs

greentic_runner_host/runner/
adapt_timer.rs
