Reuse failed queue jobs for duplicate retries

queue_jobs.py

@@ -1136,6 +1136,51 @@ class DownloadQueue:
         with self.lock:
             return self._find(job_id)
 
+    def _job_identity_fields(self, job):
+        return (
+            str(job.get("show_id") or "").strip(),
+            str(job.get("title") or "").strip(),
+            str(job.get("anime_name") or "").strip(),
+            str(job.get("media_type") or "").strip(),
+            str(job.get("season") or "").strip(),
+            str(job.get("episode_offset") or "").strip(),
+            str(job.get("query") or "").strip(),
+            job.get("result_index"),
+            str(job.get("mode") or "").strip().lower(),
+            str(job.get("quality") or "").strip().lower(),
+            str(job.get("episodes") or "").strip(),
+            str(job.get("download_dir") or "").strip(),
+        )
+
+    def _find_reusable_failed_job_locked(self, job):
+        expected = self._job_identity_fields(job)
+        with self._connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT *
+                FROM jobs
+                WHERE status = 'failed'
+                ORDER BY created_at DESC
+                """
+            ).fetchall()
+        for row in rows:
+            candidate = self._job_from_row(row)
+            if self._job_identity_fields(candidate) == expected:
+                return candidate
+        return None
+
+    def _reset_retryable_job_locked(self, job):
+        job["status"] = "pending"
+        job["exit_code"] = None
+        job["pid"] = None
+        job["started_at"] = None
+        job["finished_at"] = None
+        job["updated_at"] = now_iso()
+        job["cancel_requested"] = False
+        job["log"] = []
+        self._save_job_locked(job)
+        return job
+
     def add(self, payload):
         job = build_job(payload, self.config_getter())
         if self.job_prepare_fn is not None:
@@ -1143,7 +1188,11 @@ class DownloadQueue:
             if prepared is not None:
                 job = prepared
         with self.lock:
-            self._save_job_locked(job)
+            reusable = self._find_reusable_failed_job_locked(job)
+            if reusable is not None:
+                job = self._reset_retryable_job_locked(reusable)
+            else:
+                self._save_job_locked(job)
         self.wakeup.set()
         return job
 
@@ -1221,15 +1270,7 @@ class DownloadQueue:
                 raise ValueError("Running jobs cannot be retried")
             if job["status"] not in {"failed", "canceled"}:
                 raise ValueError("Only failed or canceled jobs can be retried")
-            job["status"] = "pending"
-            job["exit_code"] = None
-            job["pid"] = None
-            job["started_at"] = None
-            job["finished_at"] = None
-            job["updated_at"] = now_iso()
-            job["cancel_requested"] = False
-            job["log"] = []
-            self._save_job_locked(job)
+            job = self._reset_retryable_job_locked(job)
         self.wakeup.set()
         return job