fix; proper fix for dispatching many websocket events

app/features/tasks/definitions/handlers/generic.py

@@ -34,7 +34,7 @@ if TYPE_CHECKING:
     import httpx
     from parsel.selector import SelectorList
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("handlers.generic")
 CACHE: Cache = Cache()
 
 

app/features/tasks/definitions/handlers/rss.py

@@ -14,7 +14,7 @@ from ._base_handler import BaseHandler
 if TYPE_CHECKING:
     from xml.etree.ElementTree import Element
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("handlers.rss")
 CACHE: Cache = Cache()
 
 

app/features/tasks/definitions/handlers/tver.py

@@ -6,7 +6,7 @@ from app.library.Utils import get_archive_id
 
 from ._base_handler import BaseHandler
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("handlers.tver")
 
 
 class TverHandler(BaseHandler):

app/features/tasks/definitions/handlers/twitch.py

@@ -11,7 +11,7 @@ from ._base_handler import BaseHandler
 if TYPE_CHECKING:
     from xml.etree.ElementTree import Element
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("handlers.twitch")
 
 
 class TwitchHandler(BaseHandler):

app/features/tasks/definitions/handlers/youtube.py

@@ -11,7 +11,7 @@ from ._base_handler import BaseHandler
 if TYPE_CHECKING:
     from xml.etree.ElementTree import Element
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("handlers.youtube")
 
 
 class YoutubeHandler(BaseHandler):

app/features/tasks/definitions/service.py

@@ -22,7 +22,7 @@ if TYPE_CHECKING:
     from app.library.config import Config
     from app.library.Scheduler import Scheduler
 
-LOG: logging.Logger = logging.getLogger("tasks.definitions.service")
+LOG: logging.Logger = logging.getLogger("definitions.service")
 
 
 class TaskHandle:

app/features/tasks/service.py

@@ -14,7 +14,7 @@ from app.library.Singleton import Singleton
 if TYPE_CHECKING:
     from aiohttp import web
 
-LOG: logging.Logger = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger("tasks.service")
 
 
 class Tasks(metaclass=Singleton):

app/library/HttpSocket.py

@@ -1,3 +1,4 @@
+import asyncio
 import functools
 import json
 import logging
@@ -186,10 +187,20 @@ class HttpSocket:
 
             await handle_connect(sid)
 
+            async def handle_message_safe(sid: str, message: str) -> None:
+                """Wrapper to handle message with error logging"""
+                try:
+                    await handle_message(sid, message)
+                except Exception as e:
+                    LOG.exception(e)
+                    LOG.error(f"Error handling WebSocket message from client '{sid}': {e}")
+
             try:
+                i: int = 0
                 async for msg in ws:
                     if msg.type == web.WSMsgType.TEXT:
-                        await handle_message(sid, msg.data)
+                        i = i + 1
+                        asyncio.create_task(handle_message_safe(sid, msg.data), name=f"ws_msg_{sid}_{i}")
                     elif msg.type == web.WSMsgType.ERROR:
                         LOG.error(f"WebSocket connection closed with exception {ws.exception()}")
             finally:

app/library/config.py

@@ -207,9 +207,6 @@ 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."""
 
-    playlist_items_concurrency: int = 4
-    """The number of concurrent playlist items to be processed at same time."""
-
     auto_clear_history_days: int = 0
     """Number of days after which completed download history is automatically cleared. 0 to disable."""
 
@@ -268,7 +265,6 @@ class Config(metaclass=Singleton):
         "max_workers_per_extractor",
         "extract_info_timeout",
         "debugpy_port",
-        "playlist_items_concurrency",
         "download_path_depth",
         "download_info_expires",
         "auto_clear_history_days",

app/library/downloads/playlist_processor.py

@@ -1,13 +1,10 @@
 """Playlist processing."""
 
-import asyncio
 import logging
 from typing import TYPE_CHECKING, Any
 
 from app.library.Utils import merge_dict, ytdlp_reject
 
-from .utils import handle_task_exception
-
 if TYPE_CHECKING:
     from app.library.ItemDTO import Item
 
@@ -59,87 +56,54 @@ async def process_playlist(
         "n_entries": len(entries),
     }
 
-    async def playlist_processor(i: int, etr: dict):
+    async def process_item(i: int, etr: dict) -> dict[str, str]:
-        acquired = False
+        """Process a single playlist item."""
-        try:
+        item_name: str = (
-            item_name: str = (
+            f"'{entry.get('title')}: {i}/{playlist_keys['n_entries']}' - '{etr.get('id')}: {etr.get('title')}'"
-                f"'{entry.get('title')}: {i}/{playlist_keys['n_entries']}' - '{etr.get('id')}: {etr.get('title')}'"
+        )
-            )
+        LOG.info(f"Processing '{item_name}'.")
-            LOG.debug(f"Waiting to acquire lock for {item_name}")
-            await queue.processors.acquire()
-            acquired = True
-            LOG.debug(f"Acquired lock for {item_name}")
 
-            LOG.info(f"Processing '{item_name}'.")
+        _status, _msg = ytdlp_reject(entry=etr, yt_params=yt_params)
+        if not _status:
+            return {"status": "error", "msg": _msg}
 
-            _status, _msg = ytdlp_reject(entry=etr, yt_params=yt_params)
+        extras: dict[str, Any] = {
-            if not _status:
+            **playlist_keys,
-                return {"status": "error", "msg": _msg}
+            "playlist_index": i,
+            "playlist_index_number": i,
+            "playlist_autonumber": i,
+        }
 
-            extras: dict[str, Any] = {
+        for property in ("id", "title", "uploader", "uploader_id"):
-                **playlist_keys,
+            if property in entry:
-                "playlist_index": i,
+                extras[f"playlist_{property}"] = entry.get(property)
-                "playlist_index_number": i,
-                "playlist_autonumber": i,
-            }
 
-            for property in ("id", "title", "uploader", "uploader_id"):
+        extractor_key = entry.get("ie_key") or entry.get("extractor_key") or entry.get("extractor") or ""
-                if property in entry:
+        if "thumbnail" not in etr and "youtube" in str(extractor_key).lower():
-                    extras[f"playlist_{property}"] = entry.get(property)
+            extras["thumbnail"] = "https://img.youtube.com/vi/{id}/maxresdefault.jpg".format(**etr)
 
-            extractor_key = entry.get("ie_key") or entry.get("extractor_key") or entry.get("extractor") or ""
+        newItem: Item = item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras)
-            if "thumbnail" not in etr and "youtube" in str(extractor_key).lower():
-                extras["thumbnail"] = "https://img.youtube.com/vi/{id}/maxresdefault.jpg".format(**etr)
 
-            newItem: Item = item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras)
+        if ("video" == etr.get("_type") and etr.get("url")) or (
+            "formats" in etr and isinstance(etr["formats"], list) and len(etr["formats"]) > 0
+        ):
+            dct = merge_dict(merge_dict({"_type": "video"}, etr), entry)
+            dct.pop("entries", None)
+            return await queue.add(item=newItem, entry=dct, already=already)
 
-            if ("video" == etr.get("_type") and etr.get("url")) or (
+        return await queue.add(item=newItem, already=already)
-                "formats" in etr and isinstance(etr["formats"], list) and len(etr["formats"]) > 0
-            ):
-                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, already=already)
-        finally:
-            if acquired:
-                queue.processors.release()
 
     max_downloads: int = -1
     ytdlp_opts: dict[str, Any] = item.get_ytdlp_opts().get_all()
     if ytdlp_opts.get("max_downloads") and isinstance(ytdlp_opts.get("max_downloads"), int):
         max_downloads: int = ytdlp_opts.get("max_downloads")
 
-    batch_size: int = max(50, int(queue.config.playlist_items_concurrency) * 10)
+    results: list[dict[str, str]] = []
-
-    async def run_batch(batch: list[tuple[int, dict]]) -> list[dict]:
-        tasks: list[asyncio.Task] = []
-        for i, etr in batch:
-            task = asyncio.create_task(
-                playlist_processor(i, etr),
-                name=f"playlist_processor_{etr.get('id')}_{i}",
-            )
-            task.add_done_callback(lambda t: handle_task_exception(t, LOG))
-            tasks.append(task)
-
-        batch_results: list[dict] = await asyncio.gather(*tasks)
-        await asyncio.sleep(0)
-        return batch_results
-
-    results: list[dict] = []
-    batch: list[tuple[int, dict]] = []
-
     for i, etr in enumerate(entries, start=1):
         if max_downloads > 0 and i > max_downloads:
             break
 
-        batch.append((i, etr))
+        results.append(await process_item(i, etr))
-        if len(batch) >= batch_size:
-            results.extend(await run_batch(batch))
-            batch.clear()
-
-    if batch:
-        results.extend(await run_batch(batch))
 
     log_msg: str = f"Playlist '{playlist_name}' processing completed with '{len(results)}' entries."
     if max_downloads > 0 and len(entries) > max_downloads:

app/library/downloads/queue_manager.py

@@ -1,4 +1,3 @@
-import asyncio
 import functools
 import glob
 import logging
@@ -42,8 +41,6 @@ class DownloadQueue(metaclass=Singleton):
         "DataStore for the completed downloads."
         self.queue = DataStore(type=StoreType.QUEUE, connection=SqliteStore.get_instance())
         "DataStore for the download queue."
-        self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency)
-        "Semaphore to limit the number of concurrent processors."
         self.pool = PoolManager(queue=self, config=self.config)
         "Pool manager for coordinating download execution."