fix: handle locking tasks in retry context (#20)

* fix: handle locking tasks in retry context

* lint: clippy

Author: geofmureithi

…a888ce3c14e404ace182b985b8a16999f71.json …ddf0bd6d1af82bccf97b2b39827daf0e4ca.json.sqlx/query-3b0fccfab61f95863ef5a2e5c4ca4a888ce3c14e404ace182b985b8a16999f71.json renamed to .sqlx/query-5fd3913bd5d12b5571141da1d5fc6ddf0bd6d1af82bccf97b2b39827daf0e4ca.json .sqlx/query-3b0fccfab61f95863ef5a2e5c4ca4a888ce3c14e404ace182b985b8a16999f71.json renamed to .sqlx/query-5fd3913bd5d12b5571141da1d5fc6ddf0bd6d1af82bccf97b2b39827daf0e4ca.json

@@ -1,6 +1,6 @@
 use std::time::Duration;
 
-use apalis::prelude::*;
+use apalis::{layers::retry::RetryPolicy, prelude::*};
 use apalis_postgres::*;
 use futures::stream::{self, StreamExt};
 
@@ -24,12 +24,18 @@ async fn main() {
     .take(10);
     backend.push_all(&mut items).await.unwrap();
 
-    async fn send_reminder(_item: usize, _wrk: WorkerContext) -> Result<(), BoxDynError> {
+    async fn send_reminder(item: usize, wrk: WorkerContext) -> Result<(), BoxDynError> {
+        if item % 3 == 0 {
+            println!("Reminding about item: {} but failing", item);
+            return Err("Failed to send reminder".into());
+        }
+        println!("Reminding about item: {}", item);
         Ok(())
     }
 
     let worker = WorkerBuilder::new("worker-1")
         .backend(backend)
+        .retry(RetryPolicy::retries(1))
         .build(send_reminder);
     worker.run().await.unwrap();
 }

examples/basic.rs

@@ -1,10 +1,17 @@
-UPDATE apalis.jobs
-SET 
+UPDATE
+    apalis.jobs
+SET
     status = 'Running',
     lock_at = now(),
     lock_by = $2
-WHERE 
-    status = 'Queued'
+WHERE
+    (
+        status = 'Pending'
+        OR status = 'Queued'
+        OR (
+            status = 'Failed'
+            AND attempts < max_attempts
+        )
+    )
     AND run_at < now()
-    AND id = ANY($1)
-RETURNING *;
+    AND id = ANY($1) RETURNING *;

queries/task/lock_by_id.sql

@@ -1,4 +1,5 @@
 use apalis_core::{
+    error::AbortError,
     error::BoxDynError,
     layers::{Layer, Service},
     task::{Parts, status::Status},
@@ -148,7 +149,9 @@ where
         };
         let fut = self.inner.call(req);
         async move {
-            lock_task(&pool, &task_id, &worker_id).await.unwrap();
+            lock_task(&pool, &task_id, &worker_id)
+                .await
+                .map_err(AbortError::new)?;
             fut.await.map_err(|e| e.into())
         }
         .boxed()