Added more protection to workers pool being stuck in preparing state.

app/library/DownloadQueue.py

@@ -24,7 +24,7 @@ from .ItemDTO import Item, ItemDTO
 from .Presets import Presets
 from .Scheduler import Scheduler
 from .Singleton import Singleton
-from .Utils import arg_converter, calc_download_path, dt_delta, extract_info, is_downloaded, load_cookies
+from .Utils import ag, arg_converter, calc_download_path, dt_delta, extract_info, is_downloaded, load_cookies
 from .YTDLPOpts import YTDLPOpts
 
 LOG = logging.getLogger("DownloadQueue")
@@ -50,7 +50,8 @@ class DownloadQueue(metaclass=Singleton):
     pool: AsyncPool | None = None
     """Pool of workers to download the files."""
 
-    _active_downloads: dict[str, Download] = {}
+    _active: dict[str, Download] = {}
+    """Dictionary of active downloads."""
 
     _instance = None
     """Instance of the DownloadQueue."""
@@ -94,8 +95,8 @@ class DownloadQueue(metaclass=Singleton):
 
         Scheduler.get_instance().add(
             timer="* * * * *",
-            func=self.monitor_queue_for_stale_items,
+            func=self.monitor_stale,
-            id=f"{__class__.__name__}.{__class__.monitor_queue_for_stale_items.__name__}",
+            id=f"{__class__.__name__}.{__class__.monitor_stale.__name__}",
         )
 
         Scheduler.get_instance().add(
@@ -176,10 +177,10 @@ class DownloadQueue(metaclass=Singleton):
 
     async def on_shutdown(self, _: web.Application):
         LOG.debug("Canceling all active downloads.")
-        if self._active_downloads:
+        if self._active:
             self.pause()
             try:
-                await self.cancel(list(self._active_downloads.keys()))
+                await self.cancel(list(self._active.keys()))
             except Exception as e:
                 LOG.error(f"Failed to cancel downloads. {e!s}")
 
@@ -603,14 +604,14 @@ class DownloadQueue(metaclass=Singleton):
         items = {"queue": {}, "done": {}}
 
         for k, v in self.queue.saved_items():
-            items["queue"][k] = v
+            items["queue"][k] = self._active[k].info if k in self._active else v
 
         for k, v in self.done.saved_items():
             items["done"][k] = v
 
         for k, v in self.queue.items():
             if k not in items["queue"]:
-                items["queue"][k] = v.info
+                items["queue"][k] = self._active[k].info if k in self._active else v
 
         for k, v in self.done.items():
             if k not in items["done"]:
@@ -690,7 +691,7 @@ class DownloadQueue(metaclass=Singleton):
         )
 
         try:
-            self._active_downloads[entry.info._id] = entry
+            self._active[entry.info._id] = entry
             await entry.start()
 
             if "finished" != entry.info.status:
@@ -703,8 +704,8 @@ class DownloadQueue(metaclass=Singleton):
 
                 entry.info.status = "error"
         finally:
-            if entry.info._id in self._active_downloads:
+            if entry.info._id in self._active:
-                self._active_downloads.pop(entry.info._id, None)
+                self._active.pop(entry.info._id, None)
 
             await entry.close()
 
@@ -756,26 +757,68 @@ class DownloadQueue(metaclass=Singleton):
         keys = ("playlist", "external_downloader", "live_in")
         return any(key == k or key.startswith(k) for k in keys)
 
-    async def monitor_queue_for_stale_items(self):
+    async def monitor_stale(self):
         """
-        Monitor the queue for stale items and cancel them if needed.
+        Monitor the queue and pool for stale downloads and cancel them if needed.
         """
         if self.is_paused() or self.queue.empty():
             return
 
-        LOG.debug("Checking for stale items in the download queue.")
+        if not self.queue.empty():
-        for id, item in list(self.queue.items()):
+            LOG.debug("Checking for stale items in the download queue.")
-            item_ref = f"{id=} {item.info.id=} {item.info.title=}"
+            for _id, item in list(self.queue.items()):
-            if not item.is_stale():
+                item_ref = f"{_id=} {item.info.id=} {item.info.title=}"
-                LOG.debug(f"Item '{item_ref}' is not stale.")
+                if not item.is_stale():
-                continue
+                    LOG.debug(f"Item '{item_ref}' is not stale.")
+                    continue
 
-            LOG.warning(f"Cancelling staled item '{item_ref}' from download queue.")
+                LOG.warning(f"Cancelling staled item '{item_ref}' from download queue.")
-            try:
+                try:
-                await self.cancel([id])
+                    await self.cancel([_id])
-            except Exception as e:
+                except Exception as e:
-                LOG.error(f"Failed to cancel staled item '{item_ref}'. {e!s}")
+                    LOG.error(f"Failed to cancel staled item '{item_ref}'. {e!s}")
-                LOG.exception(e)
+                    LOG.exception(e)
+
+        if self.pool:
+            time_now = datetime.now(tz=UTC)
+            workers = self.pool.get_workers_status()
+            if len(workers) > 0:
+                LOG.debug(f"Checking for stale workers. {len(workers)} workers found.")
+
+            for worker_id in workers:
+                worker = ag(workers, worker_id, {})
+                started = ag(worker, "started", None)
+                if not started:
+                    LOG.debug(f"Worker '{worker_id}' not working yet.")
+                    continue
+
+                started = datetime.fromisoformat(started)
+
+                if time_now - started < timedelta(minutes=5):
+                    LOG.debug(f"Worker '{worker_id}' is not consider stale yet.")
+                    continue
+
+                status = ag(worker, "data.status", None)
+                if "preparing" != status:
+                    LOG.debug(f"Worker '{worker_id}' not stuck. Status '{status}'.")
+                    continue
+
+                _id = ag(worker, "data._id", None)
+                if not _id:
+                    LOG.debug(f"Worker '{worker_id}' has no id.")
+                    continue
+
+                id = ag(worker, "data.id", None)
+                title = ag(worker, "data.title", None)
+
+                item_ref = f"{_id=} {id=} {title=}"
+
+                LOG.warning(f"Cancelling staled item '{item_ref}' from worker pool.")
+                try:
+                    await self.cancel([_id])
+                except Exception as e:
+                    LOG.error(f"Failed to cancel staled item '{item_ref}' from worker pool. {e!s}")
+                    LOG.exception(e)
 
     async def monitor_queue_live(self):
         """

app/library/HttpAPI.py

@@ -1029,11 +1029,10 @@ class HttpAPI(Common):
 
         """
         data: dict = {"queue": [], "history": []}
+        q = self.queue.get()
 
-        for _, v in self.queue.queue.saved_items():
+        data["queue"].extend([q.get("queue", {}).get(k) for k in q.get("queue", {})])
-            data["queue"].append(v)
+        data["history"].extend([q.get("done", {}).get(k) for k in q.get("done", {})])
-        for _, v in self.queue.done.saved_items():
-            data["history"].append(v)
 
         return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode)