Reject malformed legacy scheduler policies before migration

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();

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,