wider support for cron schedules (#3123)

This commit is contained in:
Max Novich
2025-06-27 11:50:57 -07:00
committed by GitHub
parent 3ad908eacd
commit 1abb7da852
3 changed files with 65 additions and 62 deletions
+36 -40
View File
@@ -3,58 +3,54 @@ mod cron_parsing_tests {
use crate::scheduler::normalize_cron_expression; use crate::scheduler::normalize_cron_expression;
use tokio_cron_scheduler::Job; use tokio_cron_scheduler::Job;
// Helper: drop the last field if we have 7 so tokio_cron_scheduler (6-field) can parse
fn to_tokio_spec(spec: &str) -> String {
let parts: Vec<&str> = spec.split_whitespace().collect();
if parts.len() == 7 {
parts[..6].join(" ")
} else {
spec.to_string()
}
}
#[test] #[test]
fn test_normalize_cron_expression() { fn test_normalize_cron_expression() {
// Test 5-field to 6-field conversion // 5-field → 7-field
assert_eq!(normalize_cron_expression("0 12 * * *"), "0 0 12 * * *"); assert_eq!(normalize_cron_expression("0 12 * * *"), "0 0 12 * * * *");
assert_eq!(normalize_cron_expression("*/5 * * * *"), "0 */5 * * * *"); assert_eq!(normalize_cron_expression("*/5 * * * *"), "0 */5 * * * * *");
assert_eq!(normalize_cron_expression("0 0 * * 1"), "0 0 0 * * 1"); assert_eq!(normalize_cron_expression("0 0 * * 1"), "0 0 0 * * 1 *");
// Test 6-field expressions (should remain unchanged) // 6-field → 7-field (append *)
assert_eq!(normalize_cron_expression("0 0 12 * * *"), "0 0 12 * * *"); assert_eq!(normalize_cron_expression("0 0 12 * * *"), "0 0 12 * * * *");
assert_eq!( assert_eq!(
normalize_cron_expression("*/30 */5 * * * *"), normalize_cron_expression("*/30 */5 * * * *"),
"*/30 */5 * * * *" "*/30 */5 * * * * *"
); );
// Test invalid expressions (should remain unchanged but warn) // Weekday expressions (unchanged apart from 7-field format)
assert_eq!(normalize_cron_expression("* * *"), "* * *"); assert_eq!(normalize_cron_expression("0 * * * 1-5"), "0 0 * * * 1-5 *");
assert_eq!(normalize_cron_expression("* * * * * * *"), "* * * * * * *"); assert_eq!(
assert_eq!(normalize_cron_expression(""), ""); normalize_cron_expression("*/20 * * * 1-5"),
"0 */20 * * * 1-5 *"
);
} }
#[tokio::test] #[tokio::test]
async fn test_cron_expression_formats() { async fn test_cron_expression_formats() {
// Test different cron formats to see which ones work let samples = [
let test_expressions = vec![ "0 0 * * *", // 5-field
("0 0 * * *", "5-field: every day at midnight"), "0 0 0 * * *", // 6-field
("0 0 0 * * *", "6-field: every day at midnight"), "*/5 * * * *", // 5-field
("* * * * *", "5-field: every minute"),
("* * * * * *", "6-field: every second"),
("0 */5 * * *", "5-field: every 5 minutes"),
("0 0 */5 * * *", "6-field: every 5 minutes"),
("0 0 12 * * *", "6-field: every day at noon"),
("0 12 * * *", "5-field: every day at noon"),
]; ];
for expr in samples {
for (expr, desc) in test_expressions { let norm = normalize_cron_expression(expr);
println!("Testing cron expression: '{}' ({})", expr, desc); let tokio_spec = to_tokio_spec(&norm);
let expr_owned = expr.to_string(); assert!(
Job::new_async(&tokio_spec, |_id, _l| Box::pin(async {})).is_ok(),
// Test with normalization "failed to parse {} -> {}",
let normalized = normalize_cron_expression(expr); expr,
println!(" Normalized to: '{}'", normalized); norm
);
match Job::new_async(&normalized, move |_uuid, _l| {
let expr_clone = expr_owned.clone();
Box::pin(async move {
println!("Job executed for: {}", expr_clone);
})
}) {
Ok(_) => println!(" ✅ Successfully parsed normalized: '{}'", normalized),
Err(e) => println!(" ❌ Failed to parse normalized '{}': {}", normalized, e),
}
println!();
} }
} }
} }
+23 -16
View File
@@ -27,32 +27,39 @@ use crate::session::storage::SessionMetadata;
type RunningTasksMap = HashMap<String, tokio::task::AbortHandle>; type RunningTasksMap = HashMap<String, tokio::task::AbortHandle>;
type JobsMap = HashMap<String, (JobId, ScheduledJob)>; type JobsMap = HashMap<String, (JobId, ScheduledJob)>;
/// Converts a 5-field cron expression to a 6-field expression by prepending "0" for seconds /// Normalize a cron string so that:
/// If the expression already has 6 fields, returns it unchanged /// 1. It is always in **quartz 7-field format** expected by Temporal
/// If the expression is invalid, returns the original expression /// (seconds minutes hours dom month dow year).
pub fn normalize_cron_expression(cron_expr: &str) -> String { /// 2. Five-field → prepend seconds `0` and append year `*`.
let fields: Vec<&str> = cron_expr.split_whitespace().collect(); /// Six-field → append year `*`.
/// 3. Everything else returned unchanged (with a warning).
pub fn normalize_cron_expression(src: &str) -> String {
let mut parts: Vec<&str> = src.split_whitespace().collect();
match fields.len() { match parts.len() {
5 => { 5 => {
// 5-field cron: minute hour day month weekday // min hour dom mon dow → 0 min hour dom mon dow *
// Convert to 6-field: second minute hour day month weekday parts.insert(0, "0");
format!("0 {}", cron_expr) parts.push("*");
} }
6 => { 6 => {
// Already 6-field, return as-is // sec min hour dom mon dow → sec min hour dom mon dow *
cron_expr.to_string() parts.push("*");
}
7 => {
// already quartz do nothing
} }
_ => { _ => {
// Invalid number of fields, return original (will likely fail parsing later)
tracing::warn!( tracing::warn!(
"Invalid cron expression '{}': expected 5 or 6 fields, got {}", "Unrecognised cron expression '{}': expected 5, 6 or 7 fields (got {}). Leaving unchanged.",
cron_expr, src,
fields.len() parts.len()
); );
cron_expr.to_string() return src.to_string();
} }
} }
parts.join(" ")
} }
pub fn get_default_scheduler_storage_path() -> Result<PathBuf, io::Error> { pub fn get_default_scheduler_storage_path() -> Result<PathBuf, io::Error> {
+6 -6
View File
@@ -501,7 +501,7 @@ impl TemporalScheduler {
let request = JobRequest { let request = JobRequest {
action: "create".to_string(), action: "create".to_string(),
job_id: Some(job.id.clone()), job_id: Some(job.id.clone()),
cron: Some(normalized_cron), cron: Some(normalized_cron.clone()),
recipe_path: Some(job.source.clone()), recipe_path: Some(job.source.clone()),
execution_mode: job.execution_mode.clone(), execution_mode: job.execution_mode.clone(),
}; };
@@ -1336,15 +1336,15 @@ mod tests {
use crate::scheduler::normalize_cron_expression; use crate::scheduler::normalize_cron_expression;
// Test cases that should be normalized // Test cases that should be normalized
assert_eq!(normalize_cron_expression("0 12 * * *"), "0 0 12 * * *"); assert_eq!(normalize_cron_expression("0 12 * * *"), "0 0 12 * * * *");
assert_eq!(normalize_cron_expression("*/5 * * * *"), "0 */5 * * * *"); assert_eq!(normalize_cron_expression("*/5 * * * *"), "0 */5 * * * * *");
assert_eq!(normalize_cron_expression("0 0 * * 1"), "0 0 0 * * 1"); assert_eq!(normalize_cron_expression("0 0 * * 1"), "0 0 0 * * 1 *");
// Test cases that should remain unchanged // Test cases that should remain unchanged
assert_eq!(normalize_cron_expression("0 0 12 * * *"), "0 0 12 * * *"); assert_eq!(normalize_cron_expression("0 0 12 * * *"), "0 0 12 * * * *");
assert_eq!( assert_eq!(
normalize_cron_expression("*/30 */5 * * * *"), normalize_cron_expression("*/30 */5 * * * *"),
"*/30 */5 * * * *" "*/30 */5 * * * * *"
); );
println!("✅ Cron normalization works correctly in TemporalScheduler"); println!("✅ Cron normalization works correctly in TemporalScheduler");