Merge pull request #213 from tobymao/toby/aborted

fix: allow aborting jobs to be swept

saq/job.py

@@ -242,7 +242,7 @@ def _duration(self, a: int, b: int) -> int | None:
     def stuck(self) -> bool:
         """Checks if an active job is passed its timeout or heartbeat."""
         current = now()
-        return (self.status == Status.ACTIVE) and bool(
+        return (self.status == Status.ACTIVE or self.status == Status.ABORTING) and bool(
             (self.timeout and seconds(current - self.started) > self.timeout)
             or (self.heartbeat and seconds(current - self.touched) > self.heartbeat)
         )

saq/worker.py

@@ -274,7 +274,7 @@ def get_duration(job: Job) -> float:
             return
 
         for job in await self.queue.jobs(job.key for job in jobs):
-            if not job or job.status != Status.ABORTING:
+            if not job or job.status not in (Status.ABORTING, Status.ABORTED):
                 continue
 
             task_data: JobTaskContext = self.job_task_contexts.get(job, {})

tests/test_job.py

@@ -71,7 +71,10 @@ async def test_retry(self) -> None:
         self.assertEqual(self.job.error, "error")
 
     async def test_stuck(self) -> None:
-        self.assertFalse(Job("", status=Status.ACTIVE, timeout=0).stuck)
+        self.assertFalse(Job("", status=Status.ACTIVE, timeout=0, started=0).stuck)
+        self.assertFalse(Job("", status=Status.COMPLETE, timeout=0, started=0).stuck)
+        self.assertTrue(Job("", status=Status.ACTIVE, timeout=1, started=0).stuck)
+        self.assertTrue(Job("", status=Status.ABORTING, timeout=1, started=0).stuck)
 
     async def test_update(self) -> None:
         self.assertEqual(self.job.attempts, 0)