refactor: limit concurrent extract info calls to prevent loop blocking

FAQ.md

@@ -57,6 +57,7 @@ or the `environment:` section in `compose.yaml` file.
 | YTP_TASK_HANDLER_RANDOM_DELAY   | The maximum random delay in seconds before starting a task handler. | `60`                  |
 | YTP_IGNORE_ARCHIVED_ITEMS       | Don't report archived items in the download history.                | `false`               |
 | YTP_CHECK_FOR_UPDATES           | Whether to check for application updates.                           | `true`                |
+| YTP_EXTRACT_INFO_CONCURRENCY    | The number of concurrent extract info operations.                   | `4`                   |
 
 > [!NOTE]
 > To raise the maximum workers for specific extractor, you need to add a ENV variable that follows the pattern `YTP_MAX_WORKERS_FOR_<EXTRACTOR_NAME>`.

app/library/Events.py

@@ -9,7 +9,7 @@ from typing import Any
 from .BackgroundWorker import BackgroundWorker
 from .Singleton import Singleton
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("library.events")
 
 
 class Events:

app/library/config.py

@@ -117,6 +117,9 @@ class Config(metaclass=Singleton):
     extract_info_timeout: int = 70
     """The timeout to use for extracting video information."""
 
+    extract_info_concurrency: int = 4
+    """The number of concurrent extract_info calls allowed."""
+
     db_file: str = "{config_path}{os_sep}ytptube.db"
     """The path to the database file."""
 

app/library/downloads/item_adder.py

@@ -45,7 +45,7 @@ if TYPE_CHECKING:
 
     from .queue_manager import DownloadQueue
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("downloads.add")
 
 
 async def add_item(
@@ -201,23 +201,26 @@ async def add(
                 LOG.error(msg)
                 return {"status": "error", "msg": msg}
 
-        LOG.info(f"Extracting '{item.url}'{' with cookies' if yt_conf.get('cookiefile') else ''}.")
+        if entry:
+            LOG.info(f"[P] Extracting '{item.url}'{' with cookies' if yt_conf.get('cookiefile') else ''}.")
 
         if not entry:
-            entry: dict | None = await asyncio.wait_for(
-                fut=asyncio.get_running_loop().run_in_executor(
-                    None,
-                    functools.partial(
-                        extract_info,
-                        config=yt_conf,
-                        url=item.url,
-                        debug=bool(queue.config.ytdlp_debug),
-                        no_archive=False,
-                        follow_redirect=True,
+            async with queue.extractors:
+                LOG.info(f"Extracting '{item.url}'{' with cookies' if yt_conf.get('cookiefile') else ''}.")
+                entry: dict | None = await asyncio.wait_for(
+                    fut=asyncio.get_running_loop().run_in_executor(
+                        None,
+                        functools.partial(
+                            extract_info,
+                            config=yt_conf,
+                            url=item.url,
+                            debug=bool(queue.config.ytdlp_debug),
+                            no_archive=False,
+                            follow_redirect=True,
+                        ),
                     ),
-                ),
-                timeout=queue.config.extract_info_timeout,
-            )
+                    timeout=queue.config.extract_info_timeout,
+                )
 
         if not entry:
             LOG.error(f"Unable to extract info for '{item.url}'. Logs: {logs}")

app/library/downloads/monitors.py

@@ -12,7 +12,7 @@ from app.library.Utils import dt_delta, str_to_dt
 if TYPE_CHECKING:
     from .queue_manager import DownloadQueue
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("downloads.monitors")
 
 
 async def check_for_stale(queue: "DownloadQueue") -> None:

app/library/downloads/playlist_processor.py

@@ -13,7 +13,7 @@ if TYPE_CHECKING:
 
     from .queue_manager import DownloadQueue
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("downloads.playlist")
 
 
 async def process_playlist(

app/library/downloads/pool_manager.py

@@ -15,7 +15,7 @@ if TYPE_CHECKING:
 
     from .queue_manager import DownloadQueue
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("downloads.pool")
 
 
 class PoolManager:

app/library/downloads/queue_manager.py

@@ -25,7 +25,7 @@ from .pool_manager import PoolManager
 if TYPE_CHECKING:
     from app.library.DataStore import StoreType
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("downloads.queue")
 
 
 class DownloadQueue(metaclass=Singleton):
@@ -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.extractors = asyncio.Semaphore(self.config.extract_info_concurrency)
+        "Semaphore to limit the number of concurrent extract_info calls."
         self.pool = PoolManager(queue=self, config=self.config)
         "Pool manager for coordinating download execution."
 

app/library/downloads/video_processor.py

@@ -17,7 +17,7 @@ if TYPE_CHECKING:
 
     from .queue_manager import DownloadQueue
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("downloads.video")
 
 
 async def add_video(queue: "DownloadQueue", entry: dict, item: "Item", logs: list[str] | None = None) -> dict[str, str]:

app/tests/test_item_adder_extract_concurrency.py

@@ -0,0 +1,66 @@
+import time
+import asyncio
+import threading
+import pytest
+
+from app.library.config import Config
+from app.library.downloads.queue_manager import DownloadQueue
+from app.library.ItemDTO import Item
+
+
+@pytest.mark.asyncio
+async def test_extract_concurrency_limited(monkeypatch):
+    """Ensure that concurrent extract_info calls are limited by the configured semaphore."""
+    # Configure a low concurrency to make the timing assertions stable
+    cfg = Config.get_instance()
+    cfg.extract_info_concurrency = 2
+
+    # Reset singleton so new DownloadQueue picks up updated config
+    DownloadQueue._reset_singleton()
+    queue = DownloadQueue.get_instance()
+
+    sleep_time = 0.18
+
+    # Thread-safe counters to observe concurrency usage in the executor threads
+    lock = threading.Lock()
+    current = {"val": 0}
+    max_seen = {"max": 0}
+
+    def fake_extract_info(config, url, **kwargs):
+        # Track concurrent starts
+        with lock:
+            current["val"] += 1
+            if current["val"] > max_seen["max"]:
+                max_seen["max"] = current["val"]
+
+        # Blocking call to simulate expensive IO/work executed in executor
+        time.sleep(sleep_time)
+
+        with lock:
+            current["val"] -= 1
+
+        return {"_type": "video", "id": url.split("/")[-1], "title": "t", "url": url}
+
+    monkeypatch.setattr("app.library.Utils.extract_info", fake_extract_info)
+
+    items = [Item(url=f"http://example.com/{i}") for i in range(6)]
+
+    start = time.perf_counter()
+    tasks = [asyncio.create_task(queue.add(item=item)) for item in items]
+
+    await asyncio.gather(*tasks)
+    elapsed = time.perf_counter() - start
+
+    # Assert we never exceeded the configured concurrency
+    assert max_seen["max"] <= cfg.extract_info_concurrency, (
+        f"Max concurrent extractions {max_seen['max']} exceeded limit {cfg.extract_info_concurrency}"
+    )
+
+    # Sanity timing check (relaxed to avoid flakiness)
+    rounds = -(-len(items) // cfg.extract_info_concurrency)  # ceil division
+    assert elapsed >= rounds * sleep_time * 0.8, (
+        f"Elapsed {elapsed:.2f}s too low; expected at least {rounds * sleep_time * 0.8:.2f}s"
+    )
+
+    # Cleanup
+    DownloadQueue._reset_singleton()