diff --git a/CITATION.cff b/CITATION.cff index b711c89..a499a8b 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -27,5 +27,5 @@ keywords: - self-hosted - portabase license: Apache-2.0 -version: 1.16.1 +version: 1.16.3 date-released: '2026-02-24' diff --git a/Cargo.lock b/Cargo.lock index b723425..9f36e77 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3503,7 +3503,7 @@ dependencies = [ [[package]] name = "portabase-agent" -version = "1.16.1" +version = "1.16.3" dependencies = [ "aes", "aes-gcm", diff --git a/Cargo.toml b/Cargo.toml index 84cc2a0..1034b0e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "portabase-agent" -version = "1.16.1" +version = "1.16.3" edition = "2024" [dependencies] diff --git a/src/tests/utils/normalize_cron_tests.rs b/src/tests/utils/normalize_cron_tests.rs index 2144e05..29a8c47 100644 --- a/src/tests/utils/normalize_cron_tests.rs +++ b/src/tests/utils/normalize_cron_tests.rs @@ -31,12 +31,60 @@ fn normalized_expression_is_valid_for_cron_schedule() { #[test] fn next_run_timestamp_returns_future_timestamp() { let expr = normalize_cron("*/1 * * * *"); - let ts = next_run_timestamp(&expr); + let ts = next_run_timestamp(&expr).unwrap(); let now = chrono::Local::now().timestamp(); assert!(ts > now); } +#[test] +fn normalize_converts_unix_sunday_zero_to_crate_dow() { + let input = "00 06 * * 0"; + let normalized = normalize_cron(input); + + assert_eq!(normalized, "0 00 06 * * 1"); + + let schedule = Schedule::from_str(&normalized); + assert!(schedule.is_ok()); +} + +#[test] +fn normalize_converts_unix_saturday_to_crate_dow() { + assert_eq!(normalize_cron("00 06 * * 6"), "0 00 06 * * 7"); +} + +#[test] +fn normalized_sunday_actually_fires_on_sunday() { + use chrono::{Datelike, Timelike, Weekday}; + + let normalized = normalize_cron("00 03 * * 0"); + assert_eq!(normalized, "0 00 03 * * 1"); + + let schedule = Schedule::from_str(&normalized).unwrap(); + let next = schedule.upcoming(chrono::Utc).next().unwrap(); + + assert_eq!(next.weekday(), Weekday::Sun); + assert_eq!(next.hour(), 3); + assert_eq!(next.minute(), 0); +} + +#[test] +fn normalize_converts_unix_dow_range() { + assert_eq!(normalize_cron("00 06 * * 0-4"), "0 00 06 * * 1-5"); +} + +#[test] +fn next_run_timestamp_returns_none_for_invalid_cron() { + assert!(next_run_timestamp("not a cron").is_none()); +} + +#[test] +fn normalize_leaves_out_of_range_dow_untouched() { + let normalized = normalize_cron("* * * * 100"); + assert_eq!(normalized, "0 * * * * 100"); + assert!(Schedule::from_str(&normalized).is_err()); +} + #[test] fn normalization_does_not_break_schedule_parsing() { let input = "0 */10 * * * *"; diff --git a/src/utils/task_manager/cron.rs b/src/utils/task_manager/cron.rs index 381ccd4..002662f 100644 --- a/src/utils/task_manager/cron.rs +++ b/src/utils/task_manager/cron.rs @@ -10,9 +10,9 @@ use std::str::FromStr; use tracing::debug; use tracing::info; -pub fn next_run_timestamp(expr: &str) -> i64 { - let schedule = Schedule::from_str(expr).unwrap(); - schedule.upcoming(Local).next().unwrap().timestamp() +pub fn next_run_timestamp(expr: &str) -> Option { + let schedule = Schedule::from_str(expr).ok()?; + Some(schedule.upcoming(Local).next()?.timestamp()) } pub async fn check_and_update_cron( @@ -38,8 +38,9 @@ pub async fn check_and_update_cron( } Some(cron) => { - let cron = normalize_cron(&cron); - debug!("Task cron (normalized): {:?}", cron); + let raw_cron = cron; + let cron = normalize_cron(&raw_cron); + debug!("Task cron normalized: unix \"{}\" -> crate \"{}\"", raw_cron, cron); if exists { let raw: String = conn.hget(&redis_key, "data").await.unwrap(); @@ -50,24 +51,21 @@ pub async fn check_and_update_cron( let metadata_changed = stored.metadata != metadata; if cron_changed || args_changed || metadata_changed { - upsert_task(conn, &task_name, task, &cron, args.clone(), metadata) - .await - .unwrap_or_else(|e| { - tracing::error!("Failed to update task {}: {:?}", task_name, e); - }); - - info!( - "Task {} updated (cron: {}, args: {}, metadata: {})", - task_name, cron_changed, args_changed, metadata_changed - ); + match upsert_task(conn, &task_name, task, &cron, args.clone(), metadata).await { + Ok(()) => info!( + "Task {} updated (cron: {}, args: {}, metadata: {})", + task_name, cron_changed, args_changed, metadata_changed + ), + Err(e) => { + tracing::error!("Failed to update task {}: {:?}", task_name, e) + } + } } } else { - upsert_task(conn, &task_name, task, &cron, args, metadata) - .await - .unwrap_or_else(|e| { - tracing::error!("Failed to create task {}: {:?}", task_name, e); - }); - info!("Task {} created", task_name); + match upsert_task(conn, &task_name, task, &cron, args, metadata).await { + Ok(()) => info!("Task {} created", task_name), + Err(e) => tracing::error!("Failed to create task {}: {:?}", task_name, e), + } } } } diff --git a/src/utils/task_manager/scheduler.rs b/src/utils/task_manager/scheduler.rs index 1018c79..2972f60 100644 --- a/src/utils/task_manager/scheduler.rs +++ b/src/utils/task_manager/scheduler.rs @@ -48,8 +48,17 @@ pub async fn scheduler_loop(mut conn: MultiplexedConnection) { task_clone.task, e ); } - let next_ts = next_run_timestamp(&task_clone.cron); - let _: () = conn_clone.zadd(SCHEDULE_KEY, &key, next_ts).await.unwrap(); + match next_run_timestamp(&task_clone.cron) { + Some(next_ts) => { + let _: () = conn_clone.zadd(SCHEDULE_KEY, &key, next_ts).await.unwrap(); + } + None => { + error!( + "Invalid cron expression for task={}: {}", + task_clone.task, task_clone.cron + ); + } + } }); } tokio::time::sleep(std::time::Duration::from_secs(1)).await; diff --git a/src/utils/task_manager/tasks.rs b/src/utils/task_manager/tasks.rs index 28bb115..8e5ed20 100644 --- a/src/utils/task_manager/tasks.rs +++ b/src/utils/task_manager/tasks.rs @@ -16,7 +16,16 @@ pub async fn upsert_task( metadata: Option, ) -> redis::RedisResult<()> { let key = format!("redbeat:{}", name); - let next_ts = next_run_timestamp(cron); + let next_ts = match next_run_timestamp(cron) { + Some(ts) => ts, + None => { + return Err(redis::RedisError::from(( + redis::ErrorKind::Client, + "invalid cron expression", + cron.to_string(), + ))); + } + }; let entry = PeriodicTask { task: task.to_string(), diff --git a/src/utils/text.rs b/src/utils/text.rs index c72b743..a8814ea 100644 --- a/src/utils/text.rs +++ b/src/utils/text.rs @@ -1,7 +1,55 @@ pub fn normalize_cron(expr: &str) -> String { - if expr.split_whitespace().count() == 5 { - format!("0 {}", expr) + let fields: Vec<&str> = expr.split_whitespace().collect(); + + let (sec, mut rest): (String, Vec) = match fields.len() { + 5 => ("0".to_string(), fields.iter().map(|s| s.to_string()).collect()), + 6 => ( + fields[0].to_string(), + fields[1..].iter().map(|s| s.to_string()).collect(), + ), + _ => return expr.to_string(), + }; + + if let Some(last) = rest.last_mut() { + *last = convert_dow(last); + } + + format!("{} {}", sec, rest.join(" ")) +} + +fn convert_dow(field: &str) -> String { + field + .split(',') + .map(convert_dow_part) + .collect::>() + .join(",") +} + +fn convert_dow_part(part: &str) -> String { + let (base, step) = match part.split_once('/') { + Some((b, s)) => (b, Some(s)), + None => (part, None), + }; + + let converted = if let Some((start, end)) = base.split_once('-') { + match (start.parse::(), end.parse::()) { + (Ok(a), Ok(b)) if a <= 7 && b <= 7 => { + format!("{}-{}", (a % 7) + 1, (b % 7) + 1) + } + _ => base.to_string(), + } + } else if let Ok(n) = base.parse::() { + if n <= 7 { + ((n % 7) + 1).to_string() + } else { + base.to_string() + } } else { - expr.to_string() + base.to_string() + }; + + match step { + Some(s) => format!("{}/{}", converted, s), + None => converted, } }