Introduce lease tokens for task fencing

migrations/20250212120000_4.sql

@@ -0,0 +1,11 @@
+-- Force anything running this migration to use the right search path.
+set local search_path to underway;
+
+alter table underway.task
+add column if not exists lease_token uuid;
+
+alter table underway.task
+add column if not exists lease_expires_at timestamp with time zone;
+
+alter table underway.task_attempt
+add column if not exists lease_token uuid;

src/job.rs

@@ -652,9 +652,7 @@ use uuid::Uuid;
 use crate::{
     queue::{Error as QueueError, InProgressTask, Queue},
     scheduler::{Error as SchedulerError, Scheduler, ZonedSchedule},
-    task::{
-        Error as TaskError, Result as TaskResult, RetryPolicy, State as TaskState, Task, TaskId,
-    },
+    task::{Error as TaskError, Result as TaskResult, RetryPolicy, State as TaskState, Task},
     worker::{Error as WorkerError, Worker},
 };
 
@@ -917,7 +915,9 @@ impl<T: Task> EnqueuedJob<T> {
               retry_policy as "retry_policy: RetryPolicy",
               timeout,
               heartbeat,
-              concurrency_key
+              concurrency_key,
+              0::int as "attempt_number!",
+              lease_token as "lease_token?"
             from underway.task
             where input->>'job_id' = $1
               and state = $2