use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use chrono::{DateTime, Utc};
use cron::Schedule;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use super::{spawn_immediate_activation, ActivationEnvelope, ImmediateActivation};
use crate::lfd::events::EventHub;
use crate::lfd::executor::WaveExecutor;
use crate::lfd::scheduler::Scheduler;
use crate::lfd::store::SharedStore;
pub fn spawn_cron_poller(
store: SharedStore,
executor: WaveExecutor,
scheduler: Arc<Scheduler>,
event_hub: EventHub,
cancel: CancellationToken,
) -> JoinHandle<()> {
tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(30));
loop {
tokio::select! {
_ = cancel.cancelled() => {
tracing::info!("cron_poller shutting down");
break;
}
_ = interval.tick() => {
check_cron_waves(&store, &executor, &scheduler, &event_hub).await;
}
}
}
})
}
async fn check_cron_waves(
store: &SharedStore,
executor: &WaveExecutor,
scheduler: &Arc<Scheduler>,
event_hub: &EventHub,
) {
let crons = match store.list_all_active_crons().await {
Ok(crons) => crons,
Err(err) => {
tracing::error!(error = %err, "failed to list active wave crons");
return;
}
};
for cron in crons {
let last_triggered = cron
.last_triggered_at
.and_then(|value| DateTime::<Utc>::from_timestamp(value, 0));
if !should_activate_cron(&cron.schedule, last_triggered) {
continue;
}
let wave = match store.get_wave(&cron.wave_id).await {
Ok(Some(wave)) => wave,
Ok(None) => continue,
Err(err) => {
tracing::warn!(wave_id = %cron.wave_id, cron_id = %cron.id, error = %err, "failed to load wave for cron");
continue;
}
};
let reason = format!("cron [{}] schedule {} due", cron.flow, cron.schedule);
let envelope = ActivationEnvelope::new(wave.id(), None, reason, "", "", "main");
match spawn_immediate_activation(
store,
executor,
scheduler,
event_hub,
ImmediateActivation {
wave: &wave,
flow_override: Some(cron.flow.clone()),
roadmap_item: None,
force_parallel: true,
envelope,
},
)
.await
{
Ok(Some(_)) | Ok(None) => {
let now = Utc::now().timestamp();
if let Err(err) = store
.update_wave_cron_last_triggered(&cron.id, Some(now))
.await
{
tracing::warn!(wave_id = %wave.id(), cron_id = %cron.id, error = %err, "failed to update cron last_triggered_at");
}
}
Err(err) => {
tracing::warn!(wave_id = %wave.id(), cron_id = %cron.id, error = %err, "cron activation failed");
}
}
}
}
fn should_activate_cron(cron_expr: &str, last_triggered: Option<DateTime<Utc>>) -> bool {
let schedule = match Schedule::from_str(cron_expr) {
Ok(schedule) => schedule,
Err(_) => return false,
};
let now = Utc::now();
let grace_period = chrono::Duration::hours(24);
let check_from = last_triggered.unwrap_or(now - grace_period);
schedule
.after(&check_from)
.next()
.is_some_and(|scheduled| scheduled <= now)
}
#[cfg(test)]
mod tests {
use super::{check_cron_waves, should_activate_cron};
use chrono::{Datelike, Duration, Timelike, Utc};
use tempfile::tempdir;
use time::OffsetDateTime;
use crate::lfd::http::routes::test_helpers::test_http_state;
use crate::lfd::id::LfdId;
use crate::lfd::types::{Wave, WaveCron, WaveMode, WaveStatus};
#[test]
fn never_triggered_within_grace_period() {
let now = Utc::now();
let expr = format!("0 {} {} * * * *", now.minute(), now.hour());
assert!(should_activate_cron(&expr, None));
}
#[test]
fn never_triggered_outside_grace_period() {
let now = Utc::now();
let two_days_ago = now - Duration::days(2);
let expr = format!(
"0 {} {} {} {} * *",
two_days_ago.minute(),
two_days_ago.hour(),
two_days_ago.day(),
two_days_ago.month()
);
assert!(!should_activate_cron(&expr, None));
}
#[test]
fn just_triggered() {
let now = Utc::now();
let last_triggered = now - Duration::minutes(1);
let expr = format!("0 {} * * * * *", last_triggered.minute());
assert!(!should_activate_cron(&expr, Some(last_triggered)));
}
#[test]
fn past_due() {
let now = Utc::now();
let expr = format!("0 {} * * * * *", now.minute());
let last_triggered = now - Duration::hours(2);
assert!(should_activate_cron(&expr, Some(last_triggered)));
}
#[test]
fn invalid_expression() {
assert!(!should_activate_cron("not-a-cron", None));
}
#[tokio::test]
async fn cron_check_records_activation_log_and_last_triggered() {
let state = test_http_state().await;
let repo_tmp = tempdir().expect("tempdir");
crate::lfd::http::routes::test_helpers::init_git_repo(repo_tmp.path());
let wave = Wave {
id: LfdId::new(),
name: "cron-wave".to_string(),
repo: repo_tmp.path().to_string_lossy().to_string(),
mode: WaveMode::Manual,
primary_flow: "build".to_string(),
crons: Vec::new(),
direction: Vec::new(),
area: Vec::new(),
status: WaveStatus::Idle,
iteration: 0,
cycle_start_iteration: 0,
created_at: Some(OffsetDateTime::now_utc()),
workers: 0,
};
state.store.create_wave(&wave).await.expect("create wave");
let now = Utc::now();
let cron = WaveCron {
id: LfdId::new(),
wave_id: wave.id().clone(),
flow: "build".to_string(),
schedule: format!("0 {} {} * * * *", now.minute(), now.hour()),
last_triggered_at: None,
created_at: Some(OffsetDateTime::now_utc()),
};
state
.store
.create_wave_cron(&cron)
.await
.expect("create cron");
let _slot = state
.scheduler
.clone()
.acquire_guard("busy-run")
.await
.expect("occupy scheduler slot");
check_cron_waves(
&state.store,
&state.executor,
&state.scheduler,
&state.event_hub,
)
.await;
let logs = state
.store
.list_activation_log(wave.id(), 10)
.await
.expect("list activation log");
assert!(logs.iter().any(|log| {
log.reason.contains("cron [build] schedule") && log.reason.contains(&cron.schedule)
}));
let updated = state
.store
.list_wave_crons(wave.id())
.await
.expect("list wave crons");
assert_eq!(updated.len(), 1);
assert!(updated[0].last_triggered_at.is_some());
}
}