test(job): add multi-day scheduling integration test

Cargo.toml

@@ -42,7 +42,7 @@ anyhow = { workspace = true }
 
 [workspace.dependencies]
 
-es-entity = "0.10.30"
+es-entity = { git = "https://github.com/GaloyMoney/es-entity", rev = "06e3dd0" }
 
 anyhow = "1.0"
 async-trait = "0.1"

src/poller.rs

@@ -211,9 +211,10 @@ impl JobPoller {
         let job_lost_interval = self.config.job_lost_interval;
         let pool = self.repo.pool().clone();
         let clock = self.clock.clone();
+        let supported_job_types = self.registry.registered_job_types();
         OwnedTaskHandle::new(spawn_named_task!("job-poller-lost-handler", async move {
             loop {
-                clock.sleep(job_lost_interval / 2).await;
+                clock.sleep_coalesce(job_lost_interval / 2).await;
                 let now = clock.now();
                 let check_time = now - job_lost_interval;
 
@@ -230,9 +231,11 @@ impl JobPoller {
                         UPDATE job_executions
                         SET state = 'pending', execute_at = $1, attempt_index = attempt_index + 1, poller_instance_id = NULL
                         WHERE state = 'running' AND alive_at < $1::timestamptz
+                        AND job_type = ANY($2)
                         RETURNING id as id
                         "#,
                         check_time,
+                        &supported_job_types as _,
                     )
                     .fetch_all(&pool)
                     .await
@@ -292,7 +295,7 @@ impl JobPoller {
                         }
                     };
                     drop(_guard);
-                    clock.sleep(timeout).await;
+                    clock.sleep_coalesce(timeout).await;
                 }
             }
         ))