| 1 | use super::*; |
| 2 | |
| 3 | /// A schedule that went five owed slots behind produces one catch-up run for |
| 4 | /// the oldest owed slot and resumes on the next future grid slot — it must |
| 5 | /// not replay one stale slot per tick. |
| 6 | #[tokio::test] |
| 7 | async fn overdue_recurring_schedule_coalesces_to_one_catch_up_run() -> Result<()> { |
| 8 | let root = tempfile::tempdir()?; |
| 9 | let receipts = root.path().join("executions"); |
| 10 | let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?; |
| 11 | let manager = AutomationManager::open_for_test(root.path().join("automations"))?; |
| 12 | let mut automation = automation_record_with_settings(None, None, None, None); |
| 13 | automation.id = "overdue".into(); |
| 14 | let first_due = Utc::now() - Duration::hours(5) - Duration::minutes(3); |
| 15 | automation.next_run_at = Some(first_due); |
| 16 | manager.save_automation(&automation)?; |
| 17 | let shared = Arc::new(Mutex::new(manager)); |
| 18 | |
| 19 | scheduler_tick_shared(&shared, &tasks).await?; |
| 20 | let bound_id = { |
| 21 | let manager = shared.lock().await; |
| 22 | let runs = manager.list_runs(&automation.id, None)?; |
| 23 | assert_eq!(runs.len(), 1, "one catch-up run, not one per missed slot"); |
| 24 | assert_eq!( |
| 25 | runs[0].scheduled_for, first_due, |
| 26 | "the receipt owns the oldest owed slot" |
| 27 | ); |
| 28 | let updated = manager.get_automation(&automation.id)?; |
| 29 | let next = updated |
| 30 | .next_run_at |
| 31 | .context("an hourly schedule keeps a future slot")?; |
| 32 | let now = Utc::now(); |
| 33 | assert!( |
| 34 | next > now && next <= now + Duration::hours(1), |
| 35 | "advance must land on the *first* future grid slot, got {next}" |
| 36 | ); |
| 37 | // Unanchored HOURLY keeps its established grid — minute-truncated |
| 38 | // owed slot plus exact interval multiples — never re-phased to the |
| 39 | // recovery instant. |
| 40 | let grid = first_due |
| 41 | .with_second(0) |
| 42 | .and_then(|t| t.with_nanosecond(0)) |
| 43 | .context("minute truncation")?; |
| 44 | assert_eq!( |
| 45 | (next - grid).num_seconds() % Duration::hours(1).num_seconds(), |
| 46 | 0, |
| 47 | "resume slot left the established hourly grid" |
| 48 | ); |
| 49 | runs[0].task_id.clone().context("bound task id")? |
| 50 | }; |
| 51 | let task = crate::task_manager::wait_for_terminal_state( |
| 52 | &tasks, |
| 53 | &bound_id, |
| 54 | std::time::Duration::from_secs(10), |
| 55 | ) |
| 56 | .await?; |
| 57 | assert_eq!(task.status, TaskStatus::Completed); |
| 58 | reconcile_run_statuses_shared(&shared, &tasks).await?; |
| 59 | scheduler_tick_shared(&shared, &tasks).await?; |
| 60 | assert_eq!( |
| 61 | fixture_executions(&receipts), |
| 62 | vec![bound_id], |
| 63 | "the backlog produced exactly one execution" |
| 64 | ); |
| 65 | assert_eq!( |
| 66 | shared.lock().await.list_runs(&automation.id, None)?.len(), |
| 67 | 1 |
| 68 | ); |
| 69 | tasks.shutdown(); |
| 70 | Ok(()) |
| 71 | } |
| 72 | |
| 73 | /// A weekly slot owed three weeks over collapses to the upcoming calendar |
| 74 | /// slot — same weekday and wall time — rather than replaying each missed |
| 75 | /// Monday. |
| 76 | #[test] |
| 77 | fn overdue_weekly_slots_coalesce_to_the_next_calendar_slot() { |
| 78 | let schedule = AutomationSchedule::parse_rrule("FREQ=WEEKLY;BYDAY=MO;BYHOUR=9;BYMINUTE=30") |
| 79 | .expect("weekly schedule parses"); |
| 80 | let anchor = Utc::now() - Duration::days(40); |
| 81 | let slot = schedule |
| 82 | .next_after_with_anchor(anchor, anchor) |
| 83 | .expect("first weekly slot"); |
| 84 | assert!(slot < Utc::now() - Duration::days(30)); |
| 85 | let now = Utc::now(); |
| 86 | let next = schedule |
| 87 | .next_unskipped_slot(slot, now, anchor) |
| 88 | .expect("weekly grid advance") |
| 89 | .expect("weekly schedule always has a future slot"); |
| 90 | assert!(next > now, "coalesced slot must be in the future"); |
| 91 | assert!( |
| 92 | next <= now + Duration::days(7), |
| 93 | "coalesce must land on the *next* owed slot, not a later one" |
| 94 | ); |
| 95 | let local = next.with_timezone(&Local); |
| 96 | assert_eq!(local.weekday(), Weekday::Mon); |
| 97 | assert_eq!((local.hour(), local.minute()), (9, 30)); |
| 98 | } |
| 99 | |
| 100 | /// While an occurrence is still queued or running, the next due slot is |
| 101 | /// retained as owed work. One catch-up run is admitted after the earlier run |
| 102 | /// settles, with no canceled receipt or lost one-shot occurrence. |
| 103 | #[tokio::test] |
| 104 | async fn scheduled_occurrences_wait_while_a_prior_run_is_in_flight() -> Result<()> { |
| 105 | let root = tempfile::tempdir()?; |
| 106 | let receipts = root.path().join("executions"); |
| 107 | let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?; |
| 108 | let manager = AutomationManager::open_for_test(root.path().join("automations"))?; |
| 109 | let automation = fixture_due_automation(&manager, "overlap", 1); |
| 110 | |
| 111 | // A bound occurrence durably in flight: queued, owning a task identity, |
| 112 | // not yet admitted. |
| 113 | let mut held = queued_run_for(&automation); |
| 114 | held.scheduled_for = Utc::now() - Duration::hours(2); |
| 115 | bind_run_dispatch(&mut held, &automation, &tasks.data_dir(), true)?; |
| 116 | manager.save_run(&held)?; |
| 117 | |
| 118 | // The next slot comes due while that run is still queued. The claim must |
| 119 | // not stack a second task or consume the still-owed slot. |
| 120 | let (observed, proposed) = manager.collect_due_runs(Utc::now())?.remove(0); |
| 121 | assert!( |
| 122 | manager |
| 123 | .claim_scheduled_run(&observed, proposed, &tasks.data_dir())? |
| 124 | .is_none(), |
| 125 | "an in-flight occurrence blocks the next scheduled claim" |
| 126 | ); |
| 127 | let runs = manager.list_runs(&automation.id, None)?; |
| 128 | assert_eq!(runs.len(), 1, "only the already-admitted run has a receipt"); |
| 129 | let stored = manager.get_automation(&automation.id)?; |
| 130 | assert_eq!( |
| 131 | stored.next_run_at, automation.next_run_at, |
| 132 | "owed slot is retained" |
| 133 | ); |
| 134 | assert!( |
| 135 | fixture_executions(&receipts).is_empty(), |
| 136 | "the waiting slot produced no task" |
| 137 | ); |
| 138 | |
| 139 | // Once the earlier occurrence settles, the original owed slot admits. |
| 140 | held.status = AutomationRunStatus::Completed; |
| 141 | held.ended_at = Some(Utc::now()); |
| 142 | manager.save_run(&held)?; |
| 143 | let (observed, proposed) = manager.collect_due_runs(Utc::now())?.remove(0); |
| 144 | let admitted = manager |
| 145 | .claim_scheduled_run(&observed, proposed, &tasks.data_dir())? |
| 146 | .context("a settled history no longer gates admission")?; |
| 147 | assert_eq!(admitted.status, AutomationRunStatus::Queued); |
| 148 | assert_eq!(Some(admitted.scheduled_for), automation.next_run_at); |
| 149 | assert!(manager.get_automation(&automation.id)?.next_run_at.unwrap() > Utc::now()); |
| 150 | assert!(admitted.task_id.is_some()); |
| 151 | tasks.shutdown_and_wait().await?; |
| 152 | Ok(()) |
| 153 | } |
| 154 | |
| 155 | /// One damaged definition — or one outright unparseable file — quarantines to |
| 156 | /// a diagnostic and leaves its bytes alone; every healthy owned automation |
| 157 | /// still collects and dispatches. |
| 158 | #[tokio::test] |
| 159 | async fn damaged_definitions_quarantine_without_starving_owned_work() -> Result<()> { |
| 160 | let root = tempfile::tempdir()?; |
| 161 | let receipts = root.path().join("executions"); |
| 162 | let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?; |
| 163 | let manager = AutomationManager::open_for_test(root.path().join("automations"))?; |
| 164 | fixture_due_automation(&manager, "healthy", 1); |
| 165 | |
| 166 | let mut broken = automation_record_with_settings(None, None, None, None); |
| 167 | broken.id = "broken".into(); |
| 168 | broken.rrule = "FREQ=HOURLY;INTERVAL=nope".into(); |
| 169 | broken.next_run_at = Some(Utc::now() - Duration::minutes(1)); |
| 170 | manager.save_automation(&broken)?; |
| 171 | |
| 172 | let garbage = manager.automations_dir.join("garbage.json"); |
| 173 | fs::write(&garbage, b"{not json")?; |
| 174 | let garbage_before = fs::read(&garbage)?; |
| 175 | |
| 176 | let shared = Arc::new(Mutex::new(manager)); |
| 177 | scheduler_tick_shared(&shared, &tasks).await?; |
| 178 | |
| 179 | // The dispatched occurrence runs asynchronously; settle it before the |
| 180 | // receipt assertions. |
| 181 | let bound_id = { |
| 182 | let manager = shared.lock().await; |
| 183 | let runs = manager.list_runs("healthy", None)?; |
| 184 | assert_eq!(runs.len(), 1, "the healthy owned automation still claimed"); |
| 185 | assert!( |
| 186 | manager.list_runs("broken", None)?.is_empty(), |
| 187 | "the unevaluable schedule never claimed a run" |
| 188 | ); |
| 189 | runs[0].task_id.clone().context("bound task id")? |
| 190 | }; |
| 191 | let task = crate::task_manager::wait_for_terminal_state( |
| 192 | &tasks, |
| 193 | &bound_id, |
| 194 | std::time::Duration::from_secs(10), |
| 195 | ) |
| 196 | .await?; |
| 197 | assert_eq!(task.status, TaskStatus::Completed); |
| 198 | assert_eq!( |
| 199 | fixture_executions(&receipts).len(), |
| 200 | 1, |
| 201 | "the healthy owned automation still dispatched" |
| 202 | ); |
| 203 | assert_eq!( |
| 204 | fs::read(&garbage)?, |
| 205 | garbage_before, |
| 206 | "unreadable definitions are preserved, never rewritten or deleted" |
| 207 | ); |
| 208 | let manager = shared.lock().await; |
| 209 | let stored = manager.get_automation("broken")?; |
| 210 | assert_eq!( |
| 211 | stored.rrule, broken.rrule, |
| 212 | "damaged schedule left untouched" |
| 213 | ); |
| 214 | assert_eq!(stored.status, AutomationStatus::Active); |
| 215 | tasks.shutdown_and_wait().await?; |
| 216 | Ok(()) |
| 217 | } |
| 218 | |
| 219 | /// A corrupt receipt quarantines only its own automation: the strict history |
| 220 | /// read inside the claim refuses to risk re-admitting that slot, while the |
| 221 | /// sibling automation's due work still dispatches. |
| 222 | #[tokio::test] |
| 223 | async fn damaged_run_receipt_quarantines_only_its_own_automation() -> Result<()> { |
| 224 | let root = tempfile::tempdir()?; |
| 225 | let receipts = root.path().join("executions"); |
| 226 | let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?; |
| 227 | let manager = AutomationManager::open_for_test(root.path().join("automations"))?; |
| 228 | let healthy = fixture_due_automation(&manager, "healthy", 1); |
| 229 | let damaged = fixture_due_automation(&manager, "damaged", 2); |
| 230 | |
| 231 | let runs_dir = manager.runs_dir_for(&damaged.id)?; |
| 232 | fs::create_dir_all(&runs_dir)?; |
| 233 | let corrupt = runs_dir.join("corrupt.json"); |
| 234 | fs::write(&corrupt, b"{}")?; |
| 235 | |
| 236 | let shared = Arc::new(Mutex::new(manager)); |
| 237 | scheduler_tick_shared(&shared, &tasks).await?; |
| 238 | |
| 239 | // The dispatched occurrence runs asynchronously; settle it before the |
| 240 | // receipt assertions. |
| 241 | let bound_id = { |
| 242 | let manager = shared.lock().await; |
| 243 | let runs = manager.list_runs(&healthy.id, None)?; |
| 244 | assert_eq!(runs.len(), 1, "healthy automation claimed its run"); |
| 245 | runs[0].task_id.clone().context("bound task id")? |
| 246 | }; |
| 247 | let task = crate::task_manager::wait_for_terminal_state( |
| 248 | &tasks, |
| 249 | &bound_id, |
| 250 | std::time::Duration::from_secs(10), |
| 251 | ) |
| 252 | .await?; |
| 253 | assert_eq!(task.status, TaskStatus::Completed); |
| 254 | assert_eq!( |
| 255 | fixture_executions(&receipts).len(), |
| 256 | 1, |
| 257 | "healthy work dispatched despite the sibling's corrupt receipt" |
| 258 | ); |
| 259 | assert_eq!(fs::read(&corrupt)?, b"{}", "corrupt receipt is preserved"); |
| 260 | let manager = shared.lock().await; |
| 261 | assert!( |
| 262 | manager.list_runs(&healthy.id, None)?.len() == 1, |
| 263 | "healthy automation recorded its run" |
| 264 | ); |
| 265 | // The damaged automation is quarantined: still due, never advanced by a |
| 266 | // claim it cannot verify, and never silently re-admitted. |
| 267 | let stored = manager.get_automation(&damaged.id)?; |
| 268 | assert_eq!(stored.status, AutomationStatus::Active); |
| 269 | tasks.shutdown_and_wait().await?; |
| 270 | Ok(()) |
| 271 | } |
| 272 |