| 1 | use super::*; |
| 2 | |
| 3 | #[test] |
| 4 | fn foreign_and_unbound_definitions_are_not_repaired_or_paused_by_a_tick() -> Result<()> { |
| 5 | let root = tempfile::tempdir()?; |
| 6 | let owner = AutomationManager::open_for_test(root.path().to_path_buf())?; |
| 7 | let mut foreign = automation_record_with_settings(None, None, None, None); |
| 8 | foreign.id = "foreign".into(); |
| 9 | foreign.next_run_at = None; |
| 10 | owner.save_automation(&foreign)?; |
| 11 | let mut legacy = foreign.clone(); |
| 12 | legacy.id = "legacy".into(); |
| 13 | legacy.execution_scope = None; |
| 14 | legacy.rrule = "unparseable legacy schedule".into(); |
| 15 | owner.save_automation(&legacy)?; |
| 16 | let before = [ |
| 17 | fs::read(owner.automation_path(&foreign.id)?)?, |
| 18 | fs::read(owner.automation_path(&legacy.id)?)?, |
| 19 | ]; |
| 20 | let mut other = AutomationManager::open(root.path().to_path_buf())?; |
| 21 | other.execution_scope = Some(crate::task_manager::test_execution_scope("other")); |
| 22 | assert!(other.collect_due_runs(Utc::now())?.is_empty()); |
| 23 | assert_eq!(fs::read(owner.automation_path(&foreign.id)?)?, before[0]); |
| 24 | assert_eq!(fs::read(owner.automation_path(&legacy.id)?)?, before[1]); |
| 25 | Ok(()) |
| 26 | } |
| 27 | |
| 28 | #[tokio::test] |
| 29 | async fn explicit_run_now_adopts_expired_legacy_once_but_does_not_rebind_old_admissions() |
| 30 | -> Result<()> { |
| 31 | let root = tempfile::tempdir()?; |
| 32 | let receipts = root.path().join("executions"); |
| 33 | let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?; |
| 34 | let manager = AutomationManager::open_for_test(root.path().join("automations"))?; |
| 35 | let mut legacy = automation_record_with_settings(None, None, None, None); |
| 36 | legacy.execution_scope = None; |
| 37 | legacy.rrule = format!( |
| 38 | "FREQ=ONCE;AT={}", |
| 39 | (Utc::now() - Duration::hours(1)).to_rfc3339() |
| 40 | ); |
| 41 | legacy.next_run_at = None; |
| 42 | manager.save_automation(&legacy)?; |
| 43 | let mut old = queued_run_for(&legacy); |
| 44 | let mut formerly_bound = legacy.clone(); |
| 45 | formerly_bound.execution_scope = Some(crate::task_manager::test_execution_scope("test")); |
| 46 | bind_run_dispatch(&mut old, &formerly_bound, &tasks.data_dir(), false)?; |
| 47 | old.dispatch.as_mut().unwrap().execution_scope = None; |
| 48 | manager.save_run(&old)?; |
| 49 | let old_path = manager.run_path(&old)?; |
| 50 | let before = fs::read(&old_path)?; |
| 51 | let shared = Arc::new(Mutex::new(manager)); |
| 52 | let explicit = run_now_shared(&shared, &legacy.id, &tasks).await?; |
| 53 | let id = explicit.task_id.context("explicit task")?; |
| 54 | let task = crate::task_manager::wait_for_terminal_state( |
| 55 | &tasks, |
| 56 | &id, |
| 57 | std::time::Duration::from_secs(10), |
| 58 | ) |
| 59 | .await?; |
| 60 | assert_eq!(task.status, TaskStatus::Completed); |
| 61 | assert!( |
| 62 | task.owner_session_id.is_none(), |
| 63 | "None visibility is valid for a new scoped automation" |
| 64 | ); |
| 65 | assert_eq!( |
| 66 | task.execution_scope.as_deref(), |
| 67 | Some(tasks.execution_scope()) |
| 68 | ); |
| 69 | assert_eq!( |
| 70 | tasks.get_task_for_active_runtime(&task.id).await?.id, |
| 71 | task.id |
| 72 | ); |
| 73 | scheduler_tick_shared(&shared, &tasks).await?; |
| 74 | reconcile_run_statuses_shared(&shared, &tasks).await?; |
| 75 | assert_eq!(fixture_executions(&receipts), vec![id]); |
| 76 | assert_eq!( |
| 77 | fs::read(old_path)?, |
| 78 | before, |
| 79 | "definition adoption cannot adopt an old occurrence" |
| 80 | ); |
| 81 | let adopted = shared.lock().await.get_automation(&legacy.id)?; |
| 82 | assert_eq!( |
| 83 | adopted.execution_scope.as_deref(), |
| 84 | Some(tasks.execution_scope()) |
| 85 | ); |
| 86 | assert_eq!(adopted.status, AutomationStatus::Paused); |
| 87 | tasks.shutdown_and_wait().await?; |
| 88 | Ok(()) |
| 89 | } |
| 90 | |
| 91 | #[tokio::test] |
| 92 | async fn foreign_scoped_automation_and_unbound_trigger_never_dispatch_through_current_service() |
| 93 | -> Result<()> { |
| 94 | let root = tempfile::tempdir()?; |
| 95 | let receipts = root.path().join("executions"); |
| 96 | let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?; |
| 97 | let manager = AutomationManager::open_for_test(root.path().join("automations"))?; |
| 98 | let mut foreign = fixture_due_automation(&manager, "foreign", 1); |
| 99 | foreign.execution_scope = Some(crate::task_manager::test_execution_scope("foreign")); |
| 100 | manager.save_automation(&foreign)?; |
| 101 | let mut trigger = manager.create_trigger(CreateDelayedTriggerRequest { |
| 102 | fire_at: Utc::now() + Duration::hours(1), |
| 103 | message: "legacy continuation".into(), |
| 104 | workspace: Some(root.path().to_path_buf()), |
| 105 | owner_session_id: None, |
| 106 | parent_trigger_id: None, |
| 107 | })?; |
| 108 | trigger.execution_scope = None; |
| 109 | trigger.fire_at = Utc::now() - Duration::seconds(1); |
| 110 | manager.save_trigger(&trigger)?; |
| 111 | let path = manager.trigger_path(&trigger.trigger_id)?; |
| 112 | let before = fs::read(&path)?; |
| 113 | let shared = Arc::new(Mutex::new(manager)); |
| 114 | scheduler_tick_shared(&shared, &tasks).await?; |
| 115 | fire_due_triggers_shared(&shared, &tasks).await?; |
| 116 | assert!(run_now_shared(&shared, &foreign.id, &tasks).await.is_err()); |
| 117 | assert!(fixture_executions(&receipts).is_empty()); |
| 118 | assert!(tasks.list_tasks(None).await?.is_empty()); |
| 119 | assert_eq!(fs::read(path)?, before); |
| 120 | tasks.shutdown_and_wait().await?; |
| 121 | Ok(()) |
| 122 | } |
| 123 | |
| 124 | /// Two runtimes sharing a store must not reconcile each other's receipts: a |
| 125 | /// foreign-scope pending run stays byte-identical and its binding is never |
| 126 | /// probed by the local process. |
| 127 | #[tokio::test] |
| 128 | async fn reconcile_leaves_foreign_scope_run_receipts_untouched() -> Result<()> { |
| 129 | let root = tempfile::tempdir()?; |
| 130 | let receipts = root.path().join("executions"); |
| 131 | let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?; |
| 132 | let manager = AutomationManager::open_for_test(root.path().join("automations"))?; |
| 133 | let automation = fixture_due_automation(&manager, "owned", 1); |
| 134 | let mut foreign = queued_run_for(&automation); |
| 135 | bind_run_dispatch(&mut foreign, &automation, &tasks.data_dir(), false)?; |
| 136 | foreign |
| 137 | .dispatch |
| 138 | .as_mut() |
| 139 | .context("bound dispatch")? |
| 140 | .execution_scope = Some(crate::task_manager::test_execution_scope("foreign")); |
| 141 | manager.save_run(&foreign)?; |
| 142 | let path = manager.run_path(&foreign)?; |
| 143 | let before = fs::read(&path)?; |
| 144 | |
| 145 | let shared = Arc::new(Mutex::new(manager)); |
| 146 | reconcile_run_statuses_shared(&shared, &tasks).await?; |
| 147 | |
| 148 | assert_eq!( |
| 149 | fs::read(&path)?, |
| 150 | before, |
| 151 | "a foreign-scope receipt is not touched by local reconciliation" |
| 152 | ); |
| 153 | let stored = shared |
| 154 | .lock() |
| 155 | .await |
| 156 | .list_runs(&automation.id, None)? |
| 157 | .into_iter() |
| 158 | .find(|run| run.id == foreign.id) |
| 159 | .context("foreign receipt still listed")?; |
| 160 | assert_eq!(stored.status, AutomationRunStatus::Queued); |
| 161 | assert!(stored.error.is_none()); |
| 162 | tasks.shutdown_and_wait().await?; |
| 163 | Ok(()) |
| 164 | } |
| 165 | |
| 166 | /// An accepted run whose bound task vanished from the store is a finished |
| 167 | /// fact: the receipt goes terminally Failed once instead of being retried |
| 168 | /// every tick forever. |
| 169 | #[tokio::test] |
| 170 | async fn accepted_run_with_missing_bound_task_settles_failed() -> Result<()> { |
| 171 | let root = tempfile::tempdir()?; |
| 172 | let receipts = root.path().join("executions"); |
| 173 | let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?; |
| 174 | let manager = AutomationManager::open_for_test(root.path().join("automations"))?; |
| 175 | let automation = fixture_due_automation(&manager, "orphan", 1); |
| 176 | let mut run = queued_run_for(&automation); |
| 177 | bind_run_dispatch(&mut run, &automation, &tasks.data_dir(), false)?; |
| 178 | run.dispatch.as_mut().context("bound dispatch")?.accepted = true; |
| 179 | manager.save_run(&run)?; |
| 180 | |
| 181 | let shared = Arc::new(Mutex::new(manager)); |
| 182 | reconcile_run_statuses_shared(&shared, &tasks).await?; |
| 183 | |
| 184 | let stored = shared |
| 185 | .lock() |
| 186 | .await |
| 187 | .list_runs(&automation.id, None)? |
| 188 | .into_iter() |
| 189 | .find(|row| row.id == run.id) |
| 190 | .context("settled receipt still listed")?; |
| 191 | assert_eq!(stored.status, AutomationRunStatus::Failed); |
| 192 | assert!(stored.ended_at.is_some()); |
| 193 | assert!( |
| 194 | stored |
| 195 | .error |
| 196 | .as_deref() |
| 197 | .unwrap_or_default() |
| 198 | .contains("missing"), |
| 199 | "terminal receipt names the cause: {:?}", |
| 200 | stored.error |
| 201 | ); |
| 202 | let failed_at = stored.ended_at.context("ended_at set")?; |
| 203 | reconcile_run_statuses_shared(&shared, &tasks).await?; |
| 204 | let settled = shared |
| 205 | .lock() |
| 206 | .await |
| 207 | .list_runs(&automation.id, None)? |
| 208 | .into_iter() |
| 209 | .find(|row| row.id == run.id) |
| 210 | .context("settled receipt still listed")?; |
| 211 | assert_eq!( |
| 212 | settled.ended_at, |
| 213 | Some(failed_at), |
| 214 | "settled once, idempotent" |
| 215 | ); |
| 216 | tasks.shutdown_and_wait().await?; |
| 217 | Ok(()) |
| 218 | } |
| 219 |