diff --git a/crates/tinyflows-drivers/src/schedule/mod_tests.rs b/crates/tinyflows-drivers/src/schedule/mod_tests.rs index d46cae1b..a9b7734a 100644 --- a/crates/tinyflows-drivers/src/schedule/mod_tests.rs +++ b/crates/tinyflows-drivers/src/schedule/mod_tests.rs @@ -6,6 +6,65 @@ use chrono::Utc; use serde_json::json; use tinyflows_storage::Precondition; +async fn refuses_malformed_legacy_policy(policy: serde_json::Value) { + use tinystoragedrivers_types::CollectionSpec; + let store = store(); + let job = store + .add_shell_job(Some("system".into()), daily(), "system:seeded") + .await + .unwrap(); + let before = store.docs.get(JOBS, &job.id).await.unwrap().unwrap(); + store + .docs + .ensure_collection(&CollectionSpec::new("cron_job_policies")) + .await + .unwrap(); + store + .docs + .put( + "cron_job_policies", + &job.id, + policy.clone(), + Precondition::Absent, + ) + .await + .unwrap(); + + assert!( + store.get_job(&job.id).await.is_err(), + "malformed legacy policy must prevent scheduling: {policy}" + ); + assert_eq!( + store.docs.get(JOBS, &job.id).await.unwrap().unwrap().doc, + before.doc, + "refusing a policy must preserve the seeded job" + ); + assert_eq!( + store + .docs + .get("cron_job_policies", &job.id) + .await + .unwrap() + .unwrap() + .doc, + policy, + "an invalid policy must remain available for repair" + ); +} + +#[tokio::test] +async fn malformed_legacy_retry_budgets_fail_closed_without_mutating_seeded_jobs() { + for retries in [json!("2"), json!(-1), json!(u64::from(u32::MAX) + 1)] { + refuses_malformed_legacy_policy(json!({"retries": retries, "single_flight": true})).await; + } +} + +#[tokio::test] +async fn malformed_legacy_single_flight_fails_closed_without_deleting_the_policy() { + refuses_malformed_legacy_policy(json!({"retries": 0, "single_flight": "true"})).await; + refuses_malformed_legacy_policy(json!({"retries": 0})).await; +} + #[test] fn limits_default_like_the_sqlite_store_and_clamp_to_one() { let store = store(); diff --git a/crates/tinyflows-drivers/src/schedule/policy.rs b/crates/tinyflows-drivers/src/schedule/policy.rs index 97aeb132..62697f3c 100644 --- a/crates/tinyflows-drivers/src/schedule/policy.rs +++ b/crates/tinyflows-drivers/src/schedule/policy.rs @@ -24,16 +24,20 @@ impl CronDocuments { return Ok(None); } let mut doc = doc.clone(); - let retries = policy - .doc - .get("retries") - .and_then(Value::as_u64) - .map(|n| u32::try_from(n).unwrap_or(u32::MAX)); + let retries = match policy.doc.get("retries") { + None | Some(Value::Null) => None, + Some(value) => Some( + value + .as_u64() + .and_then(|n| u32::try_from(n).ok()) + .ok_or_else(|| anyhow!("Invalid legacy cron policy retries"))?, + ), + }; let single_flight = policy .doc .get("single_flight") .and_then(Value::as_bool) - .unwrap_or(false); + .ok_or_else(|| anyhow!("Invalid legacy cron policy single_flight"))?; doc["policy"] = serde_json::to_value(tinyflows_schedule::JobPolicy { retries, single_flight,