refactor: limit item add

app/features/tasks/router.py

@@ -439,12 +439,11 @@ async def task_metadata(request: Request, repo: TasksRepository, config: Config,
                 opts["proxy"] = proxy
 
             try:
-                from httpx_curl_cffi import AsyncCurlTransport, CurlOpt
+                from httpx_curl_cffi import AsyncCurlTransport
 
                 opts["transport"] = AsyncCurlTransport(
                     impersonate="chrome",
                     default_headers=True,
-                    curl_options={CurlOpt.FRESH_CONNECT: True},
                 )
                 opts.pop("headers", None)
             except Exception:

app/library/config.py

@@ -207,6 +207,9 @@ class Config(metaclass=Singleton):
     live_premiere_buffer: int = 5
     """The buffer time in minutes to add to video duration to wait before starting premiere download."""
 
+    add_items_concurrency: int = 4
+    """The number of concurrent add items to be processed at same time."""
+
     playlist_items_concurrency: int = 4
     """The number of concurrent playlist items to be processed at same time."""
 
@@ -274,6 +277,7 @@ class Config(metaclass=Singleton):
         "auto_clear_history_days",
         "default_pagination",
         "extract_info_concurrency",
+        "add_items_concurrency",
         "flaresolverr_max_timeout",
         "flaresolverr_client_timeout",
         "flaresolverr_cache_ttl",

app/library/downloads/item_adder.py

@@ -76,7 +76,11 @@ async def add_item(
 
 
 async def add(
-    queue: "DownloadQueue", item: "Item", already: set | None = None, entry: dict | None = None
+    queue: "DownloadQueue",
+    item: "Item",
+    already: set | None = None,
+    entry: dict | None = None,
+    playlist: bool = False,
 ) -> dict[str, str]:
     """
     Add an item to the download queue.
@@ -86,11 +90,23 @@ async def add(
         item: Item to be added to the queue
         already: Set of already downloaded items
         entry: Entry associated with the item (if already extracted)
+        playlist: Whether the item is part of a playlist
 
     Returns:
         dict[str, str]: Status dict with "status" and optional "msg" keys
 
     """
+    if playlist:
+        # If not done like this, it would lead to deadlocks when adding playlist items.
+        return await _add_impl(queue=queue, item=item, already=already, entry=entry)
+
+    async with queue.add_sem:
+        return await _add_impl(queue=queue, item=item, already=already, entry=entry)
+
+
+async def _add_impl(
+    queue: "DownloadQueue", item: "Item", already: set | None = None, entry: dict | None = None
+) -> dict[str, str]:
     _preset: Preset | None = Presets.get_instance().get(item.preset)
 
     if item.has_cli():

app/library/downloads/playlist_processor.py

@@ -98,9 +98,9 @@ async def process_playlist(
             ):
                 dct = merge_dict(merge_dict({"_type": "video"}, etr), entry)
                 dct.pop("entries", None)
-                return await queue.add(item=newItem, entry=dct, already=already)
+                return await queue.add(item=newItem, entry=dct, already=already, playlist=True)
 
-            return await queue.add(item=newItem, already=already)
+            return await queue.add(item=newItem, already=already, playlist=True)
         finally:
             if acquired:
                 queue.processors.release()

app/library/downloads/queue_manager.py

@@ -43,6 +43,8 @@ class DownloadQueue(metaclass=Singleton):
         "DataStore for the download queue."
         self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency)
         "Semaphore to limit the number of concurrent processors."
+        self.add_sem = asyncio.Semaphore(self.config.add_items_concurrency)
+        "Semaphore to limit the number of concurrent add item operations."
         self.pool = PoolManager(queue=self, config=self.config)
         "Pool manager for coordinating download execution."
 
@@ -221,7 +223,9 @@ class DownloadQueue(metaclass=Singleton):
     async def on_shutdown(self, _: web.Application):
         await self.pool.shutdown()
 
-    async def add(self, item: Item, already: set | None = None, entry: dict | None = None) -> dict[str, str]:
+    async def add(
+        self, item: Item, already: set | None = None, entry: dict | None = None, playlist: bool = False
+    ) -> dict[str, str]:
         """
         Add an item to the download queue.
 
@@ -229,12 +233,13 @@ class DownloadQueue(metaclass=Singleton):
             item: Item to be added to the queue
             already: Set of already downloaded items
             entry: Entry associated with the item (if already extracted)
+            playlist: Whether the item is part of a playlist
 
         Returns:
             dict[str, str]: Status dict with "status" and optional "msg" keys
 
         """
-        return await add_impl(queue=self, item=item, already=already, entry=entry)
+        return await add_impl(queue=self, item=item, already=already, entry=entry, playlist=playlist)
 
     async def cancel(self, ids: list[str]) -> dict[str, str]:
         """