Enable gevent in uWSGI workers; fix WS delivery and DB exhaustion

- Enable gevent cooperative multitasking in all uWSGI worker configs
  (gevent-early-monkey-patch + import dispatcharr.gevent_patch)
- Rewrite WebSocket group sends to bypass asyncio in gevent workers:
  _gevent_ws_send() replicates the channels_redis 4.x wire format
  directly via synchronous Redis so send_websocket_update() and
  _send_async() no longer fail silently after epoll is patched out
- Fix PostgreSQL connection exhaustion: CONN_MAX_AGE=0 + explicit
  close_old_connections() in stream manager and cleanup watchdog loops
- Fix stream proxy race: register client before the connect-wait loop
  so the cleanup watchdog never sees zero clients on a live channel
- Channel list/logo/profile queryset optimisations: conditional DISTINCT,
  EXISTS semi-joins for filter-options, channel_count annotation to
  eliminate N+1 in LogoSerializer, prefetched memberships in
  ChannelProfileSerializer
- JsonResponse for channel ID list and summary endpoints

CHANGELOG.md

@@ -77,6 +77,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - **Bulk channel edit auto-routes auto-created channels through the override path.** When a bulk edit includes auto-synced channels, the changed fields are written to each channel's `ChannelOverride` row instead of the raw `Channel.*` columns, so the edit survives the next sync. Manual channels in the same selection write directly to `Channel.*`. A "Clear all overrides" affordance pre-clears overrides on the selection before applying new edits in the same submit (clear runs before the routing PATCH so a same-submit clear cannot wipe just-written overrides).
   - **Inline edits on the channels table route through the override row for auto-synced channels.** Editing a cell directly in the table (number, name, group, EPG, logo) now writes to `ChannelOverride.<field>` for auto-synced rows so the edit survives the next refresh; manual rows continue to write directly. The save mechanic and validation are unchanged from the user's perspective.
   - **Channel-number duplicates are explicitly allowed.** Sync's `used_numbers` seed still avoids assigning the same number to two newly-created auto channels in the same run, so accidental duplication during sync remains impossible. Manual or override-driven duplication is permitted; downstream client behavior on duplicates varies by client.
+- **gevent cooperative multitasking enabled in all uWSGI workers.** `gevent-early-monkey-patch = true` and `import = dispatcharr.gevent_patch` are now set in all four uWSGI configuration files (`uwsgi.ini`, `uwsgi.modular.ini`, `uwsgi.dev.ini`, `uwsgi.debug.ini`). The new `dispatcharr/gevent_patch.py` module ensures gevent's stdlib monkey-patching is applied before application code loads (replacing blocking socket, threading, and OS primitives with cooperative gevent equivalents) and installs psycogreen's wait-callback so psycopg2 database I/O yields to the gevent hub instead of blocking the OS thread. Without these settings, any blocking psycopg2, `requests`, or DNS call froze every greenlet on the affected worker for the duration of the call.
+- **WebSocket group sends rewritten to bypass asyncio in gevent workers.** `gevent.monkey.patch_all()` removes `select.epoll` from the stdlib `select` module, which breaks asyncio event loop creation in threadpool threads. The previous `send_websocket_update` and `_send_async` paths dispatched via `async_to_sync(channel_layer.group_send)()` from a threadpool thread, which failed with `AttributeError: module 'select' has no attribute 'epoll'`; the exception was caught at WARNING level, so every WebSocket push from REST views was silently discarded. A new `_gevent_ws_send()` in `core/utils.py` replicates the channels_redis 4.x `group_send` wire format directly using the synchronous Redis client - group membership lookup from the `asgi:group:{name}` sorted set, msgpack serialization with a 12-byte random prefix, and `ZADD` to per-channel sorted sets. Both `send_websocket_update()` and `_send_async()` detect gevent patching at call time and dispatch via `gevent.spawn(_gevent_ws_send)` instead. Celery workers, which are not gevent-patched, continue to use the `async_to_sync` path unchanged.
 
 ### Fixed
 
@@ -90,6 +92,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Plugin event dispatch aborted silently on first disabled plugin.** `trigger_event` in `apps/connect/utils.py` iterated `pm.list_plugins()` and, for disabled plugins, logged a debug message using `plugin.key` / `plugin.name` (attribute access). Because `list_plugins()` returns dicts, this raised `AttributeError` on the first disabled plugin encountered. Fixed by changing the two accesses to `plugin['key']` / `plugin['name']`. (Fixes #1231) - Thanks [@R3XCHRIS](https://github.com/R3XCHRIS)
 - **Plugin periodic tasks silently missed the first beat tick after every Celery worker restart.** Plugin modules live outside `INSTALLED_APPS` so `autodiscover_tasks()` never imports them, and worker startup skips plugin discovery via `should_skip_initialization()`. Any plugin using module-level `@shared_task` had its tasks unregistered with the worker until a lazy event import warmed the module; beat fired on schedule but the worker rejected with `Received unregistered task` and advanced `last_run_at` anyway, hiding the miss. A `worker_ready` hook in `dispatcharr/celery.py` now eagerly calls `PluginManager.discover_plugins(sync_db=False)` on every worker boot so plugin tasks are registered before beat starts firing. (Fixes #1244) - Thanks [@R3XCHRIS](https://github.com/R3XCHRIS)
 - **Plugin monkey-patches and module-level hooks were never applied in uWSGI workers.** Both uWSGI configs use `lazy-apps=true`, meaning each worker boots independently and never inherits state from the master. `should_skip_initialization()` correctly skipped one-shot startup tasks in workers, but also blocked `discover_plugins`, so plugin modules were never imported in any of the 4 request-serving workers. Plugins relying on patching request handling (monkey-patches, signal registrations) were silently inactive until a Connect event lazily triggered discovery in that specific worker. Discovery is now run in every process that serves requests, gated only for Celery processes (which use `worker_ready`) and management commands that don't serve requests.
+- **PostgreSQL connection pool exhausted under load in gevent workers.** uWSGI's gevent pool runs many greenlets concurrently on a single OS thread. With `CONN_MAX_AGE = 60`, each greenlet that touched the database retained its own open connection for the full 60 seconds (gevent thread-locals are greenlet-locals), rapidly exhausting PostgreSQL's `max_connections` limit under moderate concurrency. `DATABASE_CONN_MAX_AGE` is now `0` so connections are closed after each request. `close_old_connections()` is also called at the top of the long-running stream manager and cleanup watchdog loops so stale handles accumulated across greenlet switches are released promptly.
+- **Stream proxy race: cleanup watchdog could stop a channel still in the connecting phase.** When the first viewer triggered channel initialization, the client was registered only after the connect-wait loop completed. The cleanup watchdog runs concurrently and stops channels with zero connected clients after a grace period; if the grace period elapsed during the connect-wait, the watchdog killed the channel and left the viewer stuck. The client is now registered before the connect-wait loop begins so the watchdog always sees at least one client.
 
 ### Performance
 
@@ -101,7 +105,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - **EPG dispatch deduplication.** `bulk_create` and `bulk_update` bypass `post_save`, so the post-loop step explicitly dispatches `parse_programs_for_tvg_id.delay` once per unique `epg_data_id` actually changed in the run, not per channel. Avoids fan-out where one EPG source change otherwise queued thousands of redundant parse tasks.
   - **Logo and EPGData lookup caching during sync.** `sync_auto_channels` now caches Logo and EPGData lookups in a per-run dict so repeated provider URLs / IDs across many streams do not produce duplicate `Logo.objects.get_or_create` or `EPGData.objects.filter` calls.
   - **Override-aware bulk PATCH endpoint.** `update_channels_with_override_routing` partitions the bulk-edit selection into auto-created (override write) vs manual (direct write) once, then issues two bulk PATCHes instead of N single-row updates.
-- **`xc_get_live_streams` response-build overhead reduced.** Previously the function iterated the full channel queryset three times (two passes for collision-free integer number mapping, one pass to build the response list) and called `reverse()` and `build_absolute_uri_with_port()` once per channel to construct logo URLs, amounting to N URL-pattern lookups and N header-parse calls per request. The number-mapping passes are now combined into one full queryset pass that immediately classifies channels with integer numbers, deferring only the fractional-number subset to a second, smaller loop. Logo URL construction is precomputed once per request using a single `reverse()` call; per-channel URLs are assembled with string interpolation. `_get_default_group_id()` is also called at most once per null-group channel instead of twice. (Fixes #1220)
+- **`xc_get_live_streams` response-build overhead reduced.** Previously the function iterated the full channel queryset three times (two passes for collision-free integer number mapping, one pass to build the response list) and called `reverse()` and `build_absolute_uri_with_port()` once per channel to construct logo URLs, amounting to N URL-pattern lookups and N header-parse calls per request. The number-mapping passes are now combined into one full queryset pass that immediately classifies channels with integer numbers, deferring only the fractional-number subset to a second, smaller loop. Logo URL construction is precomputed once per request using a single `reverse()` call; per-channel URLs are assembled with string interpolation. `_get_default_group_id()` is also called at most once per null-group channel instead of twice. The `get_live_streams` API endpoint now returns a `StreamingHttpResponse` backed by a generator that serializes one channel entry at a time instead of building the full JSON array in memory before writing; time-to-first-byte is lower on large channel libraries and peak worker memory is O(1) per channel rather than O(N). (Fixes #1220)
+- **Stream filter-options fast path when no filters are active.** `GET /api/channels/streams/filter-options/` previously ran `DISTINCT` queries across the full streams table while also mutating the underlying Django request object to strip inapplicable filter params. With no active filters the fast path now queries the streams table directly with two simple `DISTINCT` aggregations, skipping the request mutation and filterset instantiation overhead.
+- **Channel list queryset drops unconditional `DISTINCT`.** `ChannelViewSet.get_queryset()` previously appended `.distinct()` unconditionally. `DISTINCT` is only needed when the query joins a one-to-many table - specifically when `channel_profile_id` or `only_stale` filters are active. The queryset now returns plain results in the common case, eliminating sort-and-deduplicate overhead on most channel list fetches.
+- **`JsonResponse` for ID list and channel summary endpoints.** The `stream_ids`, `channel_ids`, and channel `summary` endpoints now use `django.http.JsonResponse` instead of DRF's `Response`, bypassing the DRF renderer pipeline (content-type negotiation, serializer dispatch) for responses that are already plain Python structures. Removes overhead on every channel-table load and stream-table load.
+- **Logo queryset annotates `channel_count` to eliminate N+1 in `LogoSerializer`.** `LogoViewSet.get_queryset()` now annotates each row with `Count('channels')`. `LogoSerializer.get_channel_count()` and `get_is_used()` read the annotation directly instead of issuing a separate `COUNT(*)` per logo. The `used=true` and `used=false` list filters use the annotation for their conditions, removing the `DISTINCT` that was previously required.
+- **`ChannelProfileSerializer` reads prefetched memberships.** `ChannelProfileViewSet.get_queryset()` now prefetches enabled `ChannelProfileMembership` rows into `enabled_memberships`. `ChannelProfileSerializer.get_channels()` uses the prefetched set when available, eliminating one query per profile in any response that lists multiple profiles.
 
 ## [0.24.0] - 2026-05-03
 

apps/channels/api_views.py

@@ -196,8 +196,7 @@ class StreamViewSet(viewsets.ModelViewSet):
         # Return only the IDs from the queryset
         stream_ids = queryset.values_list("id", flat=True)
 
-        # Return the response with the list of IDs
-        return Response(list(stream_ids))
+        return JsonResponse(list(stream_ids), safe=False)
 
     @extend_schema(
         parameters=[
@@ -461,6 +460,40 @@ class StreamViewSet(viewsets.ModelViewSet):
         Get available filter options based on current filter state.
         Uses a hierarchical approach: M3U is the parent filter, Group filters based on M3U.
         """
+        # Fast path: no filters supplied - skip DISTINCT over the full streams
+        # table and answer from parent tables via EXISTS semi-joins instead.
+        _group_filter_params = (
+            "name", "m3u_account", "m3u_account_name",
+            "m3u_account_is_active", "tvg_id",
+        )
+        _m3u_filter_params = (
+            "name", "m3u_account_name", "m3u_account_is_active", "tvg_id",
+        )
+        _has_group_filters = any(request.GET.get(p) for p in _group_filter_params)
+        _has_m3u_filters = any(request.GET.get(p) for p in _m3u_filter_params)
+
+        if not _has_group_filters and not _has_m3u_filters:
+            base_qs = Stream.objects.exclude(m3u_account__is_active=False)
+            group_names = list(
+                base_qs.exclude(channel_group__isnull=True)
+                .order_by("channel_group__name")
+                .values_list("channel_group__name", flat=True)
+                .distinct()
+            )
+            m3u_data = list(
+                base_qs.exclude(m3u_account__isnull=True)
+                .order_by("m3u_account__name")
+                .values("m3u_account__id", "m3u_account__name")
+                .distinct()
+            )
+            return Response({
+                "groups": group_names,
+                "m3u_accounts": [
+                    {"id": m["m3u_account__id"], "name": m["m3u_account__name"]}
+                    for m in m3u_data
+                ],
+            })
+
         # For group options: we need to bypass the channel_group custom queryset filtering
         # Store original request params
         original_params = request.query_params
@@ -940,7 +973,13 @@ class ChannelViewSet(viewsets.ModelViewSet):
         if q_filters:
             qs = qs.filter(q_filters)
 
-        return qs.distinct()
+        # DISTINCT is only needed when a filter joins to a one-to-many table
+        # and can produce duplicate channel rows. channel_profile_id joins
+        # channelprofilemembership; only_stale joins streams. All other
+        # filters use FK or one-to-one joins that cannot produce duplicates.
+        if channel_profile_id or only_stale:
+            return qs.distinct()
+        return qs
 
     def get_serializer_context(self):
         context = super().get_serializer_context()
@@ -1556,8 +1595,8 @@ class ChannelViewSet(viewsets.ModelViewSet):
         # Return only the IDs from the queryset
         channel_ids = queryset.values_list("id", flat=True)
 
-        # Return the response with the list of IDs
-        return Response(list(channel_ids))
+        # JsonResponse skips DRF's renderer pipeline for a flat int list.
+        return JsonResponse(list(channel_ids), safe=False)
 
     @action(detail=False, methods=["get"], url_path="summary")
     def summary(self, request, *args, **kwargs):
@@ -1574,27 +1613,29 @@ class ChannelViewSet(viewsets.ModelViewSet):
         queryset = with_effective_values(
             self.filter_queryset(self.get_queryset())
         )
-        data = [
-            {
-                "id": row["id"],
-                "uuid": row["uuid"],
-                "name": row["effective_name"],
-                "logo_id": row["effective_logo_id"],
-                "channel_number": row["effective_channel_number"],
-                "epg_data_id": row["effective_epg_data_id"],
-                "channel_group_id": row["effective_channel_group_id"],
-            }
-            for row in queryset.values(
-                "id",
-                "uuid",
-                "effective_name",
-                "effective_logo_id",
-                "effective_channel_number",
-                "effective_epg_data_id",
-                "effective_channel_group_id",
-            )
-        ]
-        return Response(data)
+        return JsonResponse(
+            [
+                {
+                    "id": row["id"],
+                    "uuid": row["uuid"],
+                    "name": row["effective_name"],
+                    "logo_id": row["effective_logo_id"],
+                    "channel_number": row["effective_channel_number"],
+                    "epg_data_id": row["effective_epg_data_id"],
+                    "channel_group_id": row["effective_channel_group_id"],
+                }
+                for row in queryset.values(
+                    "id",
+                    "uuid",
+                    "effective_name",
+                    "effective_logo_id",
+                    "effective_channel_number",
+                    "effective_epg_data_id",
+                    "effective_channel_group_id",
+                )
+            ],
+            safe=False,
+        )
 
     @extend_schema(
         parameters=[
@@ -2567,8 +2608,13 @@ class LogoViewSet(viewsets.ModelViewSet):
 
     def get_queryset(self):
         """Optimize queryset with prefetch and add filtering"""
-        # Start with basic prefetch for channels
-        queryset = Logo.objects.prefetch_related('channels').order_by('name')
+        # Annotate channel_count and prefetch channels to avoid N+1 in LogoSerializer.
+        queryset = (
+            Logo.objects
+            .annotate(channel_count=Count('channels'))
+            .prefetch_related('channels')
+            .order_by('name')
+        )
 
         # Filter by specific IDs
         ids = self.request.query_params.getlist('ids')
@@ -2585,11 +2631,9 @@ class LogoViewSet(viewsets.ModelViewSet):
         # Filter by usage
         used_filter = self.request.query_params.get('used', None)
         if used_filter == 'true':
-            # Logo is used if it has any channels
-            queryset = queryset.filter(channels__isnull=False).distinct()
+            queryset = queryset.filter(channel_count__gt=0)
         elif used_filter == 'false':
-            # Logo is unused if it has no channels
-            queryset = queryset.filter(channels__isnull=True)
+            queryset = queryset.filter(channel_count=0)
 
         # Filter by name
         name_filter = self.request.query_params.get('name', None)
@@ -2823,14 +2867,18 @@ class ChannelProfileViewSet(viewsets.ModelViewSet):
     serializer_class = ChannelProfileSerializer
 
     def get_queryset(self):
+        from django.db.models import Prefetch
+        enabled_memberships_prefetch = Prefetch(
+            'channelprofilemembership_set',
+            queryset=ChannelProfileMembership.objects.filter(enabled=True),
+            to_attr='enabled_memberships',
+        )
         user = self.request.user
 
-        # If user_level is 10, return all ChannelProfiles
         if hasattr(user, "user_level") and user.user_level == 10:
-            return ChannelProfile.objects.all()
+            return ChannelProfile.objects.prefetch_related(enabled_memberships_prefetch)
 
-        # Otherwise, return only ChannelProfiles related to the user
-        return self.request.user.channel_profiles.all()
+        return self.request.user.channel_profiles.prefetch_related(enabled_memberships_prefetch)
 
     def get_permissions(self):
         if self.action == "duplicate":
@@ -3449,9 +3497,7 @@ class RecordingViewSet(viewsets.ModelViewSet):
         instance.custom_properties = cp
         instance.save(update_fields=["custom_properties"])
 
-        # Send the WebSocket notification before returning the response.
-        # send_websocket_update is gevent-safe (offloads async_to_sync to a
-        # real OS thread when monkey-patching is active).
+        # send_websocket_update offloads async_to_sync to a real OS thread when gevent is active.
         channel_uuid = str(instance.channel.uuid)
         recording_id = instance.id
         task_id = instance.task_id

apps/channels/serializers.py

@@ -67,22 +67,28 @@ class LogoSerializer(serializers.ModelSerializer):
 
     def get_channel_count(self, obj):
         """Get the number of channels using this logo"""
+        # `channel_count` is provided as an annotation in LogoViewSet.get_queryset().
+        # Fall back to a query only when serializing a single un-annotated Logo
+        # (e.g. nested inside ChannelSerializer.get_logo()).
+        annotated = getattr(obj, "channel_count", None)
+        if annotated is not None:
+            return annotated
         return obj.channels.count()
 
     def get_is_used(self, obj):
         """Check if this logo is used by any channels"""
-        return obj.channels.exists()
+        return self.get_channel_count(obj) > 0
 
     def get_channel_names(self, obj):
         """Get the names of channels using this logo (limited to first 5)"""
         names = []
 
-        # Get channel names
-        channels = obj.channels.all()[:5]
+        # When LogoViewSet.get_queryset() prefetches `channels`, iterating
+        # obj.channels.all() reuses the cached set; slicing happens in Python.
+        channels = list(obj.channels.all()[:5])
         for channel in channels:
             names.append(f"Channel: {channel.name}")
 
-        # Calculate total count for "more" message
         total_count = self.get_channel_count(obj)
         if total_count > 5:
             names.append(f"...and {total_count - 5} more")
@@ -275,10 +281,15 @@ class ChannelProfileSerializer(serializers.ModelSerializer):
         fields = ["id", "name", "channels"]
 
     def get_channels(self, obj):
-        memberships = ChannelProfileMembership.objects.filter(
-            channel_profile=obj, enabled=True
+        # Use prefetched attr when available, fall back to a direct query.
+        memberships = getattr(obj, 'enabled_memberships', None)
+        if memberships is not None:
+            return [m.channel_id for m in memberships]
+        return list(
+            ChannelProfileMembership.objects.filter(
+                channel_profile=obj, enabled=True
+            ).values_list('channel_id', flat=True)
         )
-        return [membership.channel.id for membership in memberships]
 
 
 class ChannelProfileMembershipSerializer(serializers.ModelSerializer):

apps/proxy/live_proxy/input/manager.py

@@ -7,7 +7,7 @@ import requests
 import subprocess
 import gevent
 import re
-from django.db import connection
+from django.db import connection, close_old_connections
 from apps.proxy.config import TSConfig as Config
 from apps.channels.models import Channel, Stream
 from core.utils import log_system_event
@@ -202,6 +202,7 @@ class StreamManager:
 
             # Main stream switching loop - we'll try different streams if needed
             while self.running and stream_switch_attempts <= max_stream_switches:
+                close_old_connections()
                 # Check for stuck switching state
                 if self.url_switching and time.time() - self.url_switch_start_time > self.url_switch_timeout:
                     logger.warning(f"URL switching state appears stuck for channel {self.channel_id} "

apps/proxy/live_proxy/server.py

@@ -17,6 +17,7 @@ import gevent
 from apps.proxy.config import TSConfig as Config
 from apps.channels.models import Channel, Stream
 from core.utils import RedisClient, log_system_event
+from django.db import close_old_connections
 from redis.exceptions import ConnectionError, TimeoutError
 from .input.manager import StreamManager
 from .input.buffer import StreamBuffer
@@ -1432,6 +1433,7 @@ class ProxyServer:
         def cleanup_task():
             while True:
                 try:
+                    close_old_connections()
                     # Send worker heartbeat first
                     if self.redis_client:
                         worker_heartbeat_key = f"live:worker:{self.worker_id}:heartbeat"

apps/proxy/live_proxy/views.py

@@ -189,6 +189,8 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
                                     f"[{client_id}] Channel {channel_id} owner {owner} is dead, will reinitialize"
                                 )
 
+        _client_pre_registered = False
+
         # Start initialization if needed
         if needs_initialization or not proxy_server.check_if_channel_exists(channel_id):
             logger.info(f"[{client_id}] Starting channel {channel_id} initialization")
@@ -430,6 +432,23 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
 
             # If we're the owner, wait for connection to establish
             if proxy_server.am_i_owner(channel_id):
+                # Register the client before waiting for the stream to connect.
+                # The cleanup watchdog stops channels in connecting state with 0
+                # clients after the grace period - registering early prevents that.
+                output_profile = _resolve_output_profile(request, user)
+                output_format = _resolve_output_format(user, force_output_format, request)
+                resolved_format = f'{output_format}:p{output_profile.id}' if output_profile else output_format
+                client_manager = proxy_server.client_managers[channel_id]
+                client_manager.add_client(
+                    client_id, client_ip, client_user_agent, user,
+                    output_format=output_format,
+                    output_profile_id=output_profile.id if output_profile else None,
+                )
+                logger.info(
+                    f"[{client_id}] Client registered with channel {channel_id} "
+                    f"(output: {resolved_format}, profile: {output_profile.id if output_profile else None})"
+                )
+                _client_pre_registered = True
                 manager = proxy_server.stream_managers.get(channel_id)
                 if manager:
                     wait_start = time.time()
@@ -570,6 +589,10 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
         if output_profile:
             cmd = output_profile.build_command()
             if not proxy_server.ensure_output_profile(channel_id, output_profile.id, cmd):
+                if _client_pre_registered:
+                    mgr = proxy_server.client_managers.get(channel_id)
+                    if mgr:
+                        mgr.remove_client(client_id)
                 return JsonResponse(
                     {"error": "Failed to start output profile transcode"}, status=500
                 )
@@ -584,15 +607,16 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
         # When an output profile is active, append :p{id} to the format key so each
         # (format, profile) pair gets its own independent remux pipeline in Redis.
         resolved_format = f'{output_format}:p{output_profile.id}' if output_profile else output_format
-        client_manager.add_client(
-            client_id, client_ip, client_user_agent, user,
-            output_format=output_format,
-            output_profile_id=output_profile.id if output_profile else None,
-        )
-        logger.info(
-            f"[{client_id}] Client registered with channel {channel_id} "
-            f"(output: {resolved_format}, profile: {output_profile.id if output_profile else None})"
-        )
+        if not _client_pre_registered:
+            client_manager.add_client(
+                client_id, client_ip, client_user_agent, user,
+                output_format=output_format,
+                output_profile_id=output_profile.id if output_profile else None,
+            )
+            logger.info(
+                f"[{client_id}] Client registered with channel {channel_id} "
+                f"(output: {resolved_format}, profile: {output_profile.id if output_profile else None})"
+            )
 
         if output_format == 'fmp4':
             proxy_server.ensure_output_format(

core/utils.py

@@ -312,14 +312,64 @@ class TaskLockRenewer:
         return False
 
 
+def _gevent_ws_send(group_name, message):
+    """
+    Publishes a WebSocket group message synchronously through Redis.
+
+    gevent's monkey-patching removes select.epoll, which breaks asyncio event
+    loop creation in threadpool threads. This function replicates channels_redis
+    4.x group_send directly via a sync Redis client, avoiding asyncio entirely.
+
+    Matches channels_redis 4.x defaults: prefix="asgi", expiry=60,
+    group_expiry=86400, msgpack serializer with 12-byte random prefix.
+    """
+    try:
+        import msgpack
+        redis = RedisClient.get_buffer()  # decode_responses=False for binary values
+
+        prefix = "asgi"
+        group_expiry = 86400
+        channel_expiry = 60
+        rand_len = 12
+
+        group_key = f"{prefix}:group:{group_name}"
+        now = time.time()
+
+        redis.zremrangebyscore(group_key, 0, now - group_expiry)
+        raw = redis.zrange(group_key, 0, -1)
+        if not raw:
+            return
+
+        channels = [m.decode('utf-8') if isinstance(m, bytes) else m for m in raw]
+
+        # Group channels by non-local name (prefix up to and including "!") so
+        # specific channels sharing a prefix share one Redis sorted-set key.
+        nonlocal_map = {}
+        for ch in channels:
+            pos = ch.find("!")
+            nl = ch[:pos + 1] if pos >= 0 else ch
+            nonlocal_map.setdefault(nl, []).append(ch)
+
+        pipe = redis.pipeline(transaction=False)
+        for nl, chs in nonlocal_map.items():
+            channel_key = prefix + nl
+            msg = dict(message)
+            msg["__asgi_channel__"] = chs
+            serialized = os.urandom(rand_len) + msgpack.packb(msg)
+            pipe.zadd(channel_key, {serialized: now})
+            pipe.expire(channel_key, channel_expiry)
+        pipe.execute()
+    except Exception as e:
+        logger.warning(f"Failed to send WebSocket update: {e}")
+
+
 def send_websocket_update(group_name, event_type, data, collect_garbage=False):
     """
-    Standardized function to send WebSocket updates with proper memory management.
+    Sends a WebSocket group message.
 
-    In uWSGI + gevent deployments, async_to_sync creates an asyncio event loop
-    whose native EpollSelector blocks the entire OS thread, freezing all gevent
-    greenlets in the worker.  To avoid this, the actual send is offloaded to a
-    real OS thread from gevent's native threadpool when monkey-patching is active.
+    In gevent-patched uWSGI workers, asyncio event loop creation fails because
+    monkey-patching removes select.epoll. For those contexts a synchronous Redis
+    path is used instead, matching the channels_redis 4.x wire format.
     """
     channel_layer = get_channel_layer()
     message = {'type': event_type, 'data': data}
@@ -333,13 +383,13 @@ def send_websocket_update(group_name, event_type, data, collect_garbage=False):
     try:
         import gevent.monkey
         if gevent.monkey.is_module_patched('threading'):
-            from gevent import get_hub
-            get_hub().threadpool.spawn(_do_send)
+            import gevent
+            gevent.spawn(_gevent_ws_send, group_name, message)
             return
     except Exception:
         pass
 
-    # Not in a gevent-patched environment — call directly
+    # Not in a gevent-patched environment (Celery, tests) — use asyncio path
     _do_send()
 
     if collect_garbage:
@@ -635,6 +685,25 @@ def log_system_event(event_type, channel_id=None, channel_name=None, **details):
         logger.error(f"Failed to log system event {event_type}: {e}")
 
 
+def _send_async(channel_layer, group, message):
+    """Send a channel layer group message without blocking the gevent hub."""
+    def _do():
+        try:
+            async_to_sync(channel_layer.group_send)(group, message)
+        except Exception as e:
+            logger.warning(f"Failed WebSocket group_send to '{group}': {e}")
+
+    try:
+        import gevent.monkey
+        if gevent.monkey.is_module_patched("threading"):
+            import gevent
+            gevent.spawn(_gevent_ws_send, group, message)
+            return
+    except Exception:
+        pass
+    _do()
+
+
 def send_websocket_notification(notification):
     """
     Send a system notification to all connected WebSocket clients.
@@ -667,7 +736,8 @@ def send_websocket_notification(notification):
         else:
             notification_data = notification
 
-        async_to_sync(channel_layer.group_send)(
+        _send_async(
+            channel_layer,
             'updates',
             {
                 'type': 'update',
@@ -693,7 +763,8 @@ def send_notification_dismissed(notification_key):
     try:
         channel_layer = get_channel_layer()
 
-        async_to_sync(channel_layer.group_send)(
+        _send_async(
+            channel_layer,
             'updates',
             {
                 'type': 'update',

dispatcharr/gevent_patch.py

@@ -0,0 +1,54 @@
+"""
+Loaded via uWSGI's `import = dispatcharr.gevent_patch` directive.
+
+Two things happen here:
+
+1. gevent stdlib monkey-patching - replaces blocking socket/threading/os
+   primitives with gevent-cooperative versions. `gevent-early-monkey-patch = true`
+   in the ini should already have done this; calling it again is a safe no-op if
+   it did, and a necessary fallback if it didn't (e.g. older uWSGI build).
+
+2. psycogreen - installs a wait callback on psycopg2 so libpq yields to the
+   gevent hub during I/O instead of blocking the OS thread.
+
+Without (1), `async_to_sync(channel_layer.group_send)` in send_websocket_update
+would call epoll_wait() directly, freezing every greenlet on the worker (the
+90s pile-up observed on Worker 9). With (1), select.epoll is removed by
+monkey-patching, which breaks asyncio event loop creation in threadpool threads.
+send_websocket_update therefore uses a synchronous Redis path in gevent workers
+instead of asyncio - see _gevent_ws_send() in core/utils.py.
+
+Without (2), psycopg2 network calls pin the worker during slow/stalled queries.
+
+Celery and Daphne run in separate daemon processes and do not load this module.
+
+Note: this module runs before Django configures logging, so print() is used
+instead of logger so output reaches uWSGI's stdout.
+"""
+import sys
+
+from gevent import monkey
+
+if not monkey.is_module_patched("socket"):
+    print(
+        "[gevent_patch] WARNING: gevent-early-monkey-patch did not run - "
+        "applying monkey.patch_all() now.",
+        file=sys.stderr,
+        flush=True,
+    )
+    monkey.patch_all()
+else:
+    print("[gevent_patch] gevent stdlib monkey-patching already active.", flush=True)
+
+try:
+    from psycogreen.gevent import patch_psycopg
+    patch_psycopg()
+    print("[gevent_patch] psycogreen: psycopg2 patched for gevent.", flush=True)
+except ImportError:
+    print(
+        "[gevent_patch] WARNING: psycogreen not installed - "
+        "psycopg2 will block the gevent hub during DB I/O. "
+        "Run: uv pip install psycogreen",
+        file=sys.stderr,
+        flush=True,
+    )

dispatcharr/settings.py

@@ -111,9 +111,7 @@ XC_PROFILE_REFRESH_DELAY = float(os.environ.get('XC_PROFILE_REFRESH_DELAY', '2.5
 
 # Database optimization settings
 DATABASE_STATEMENT_TIMEOUT = 300  # Seconds before timing out long-running queries
-DATABASE_CONN_MAX_AGE = (
-    60  # Connection max age in seconds, helps with frequent reconnects
-)
+DATABASE_CONN_MAX_AGE = 0  # Close after each request; gevent makes per-greenlet connections
 
 # Disable atomic requests for performance-sensitive views
 ATOMIC_REQUESTS = False

docker/uwsgi.debug.ini

@@ -39,7 +39,13 @@ http-timeout = 600
 
 # Async mode (use gevent for high concurrency)
 gevent = 100
-async = 100
+
+# Patch the stdlib (socket, threading, time, ...) before any app code
+# loads so blocking calls yield to the gevent hub. Without this, a single
+# blocking psycopg2 / requests / DNS call freezes every greenlet on the
+# worker. The companion module greens psycopg2 specifically.
+gevent-early-monkey-patch = true
+import = dispatcharr.gevent_patch
 
 # Performance tuning
 thunder-lock = true
@@ -64,6 +70,7 @@ py-autoreload = 1
 honour-stdin = true
 
 # Environment variables
+env = GEVENT_SUPPORT=True
 env = PYTHONPATH=/app
 env = PYTHONUNBUFFERED=1
 env = PYDEVD_DISABLE_FILE_VALIDATION=1

docker/uwsgi.dev.ini

@@ -42,7 +42,13 @@ lazy-apps = true  # Improve memory efficiency
 
 # Async mode (use gevent for high concurrency)
 gevent = 100
-async = 100
+
+# Patch the stdlib (socket, threading, time, ...) before any app code
+# loads so blocking calls yield to the gevent hub. Without this, a single
+# blocking psycopg2 / requests / DNS call freezes every greenlet on the
+# worker. The companion module greens psycopg2 specifically.
+gevent-early-monkey-patch = true
+import = dispatcharr.gevent_patch
 
 # Performance tuning
 thunder-lock = true

docker/uwsgi.ini

@@ -46,6 +46,13 @@ gevent = 400  # Each unused greenlet costs ~2-4KB of memory
 # Higher values have minimal performance impact when idle, but provide capacity for traffic spikes
 # If memory usage becomes an issue, reduce this value
 
+# Patch the stdlib (socket, threading, time, ...) before any app code
+# loads so blocking calls yield to the gevent hub. Without this, a single
+# blocking psycopg2 / requests / DNS call freezes every greenlet on the
+# worker. The companion module greens psycopg2 specifically.
+gevent-early-monkey-patch = true
+import = dispatcharr.gevent_patch
+
 # Performance tuning
 thunder-lock = true
 log-4xx = true

docker/uwsgi.modular.ini

@@ -42,6 +42,13 @@ gevent = 400  # Each unused greenlet costs ~2-4KB of memory
 # Higher values have minimal performance impact when idle, but provide capacity for traffic spikes
 # If memory usage becomes an issue, reduce this value
 
+# Patch the stdlib (socket, threading, time, ...) before any app code
+# loads so blocking calls yield to the gevent hub. Without this, a single
+# blocking psycopg2 / requests / DNS call freezes every greenlet on the
+# worker. The companion module greens psycopg2 specifically.
+gevent-early-monkey-patch = true
+import = dispatcharr.gevent_patch
+
 # Performance tuning
 thunder-lock = true
 log-4xx = true

frontend/src/components/tables/StreamsTable.jsx

@@ -1355,6 +1355,7 @@ const StreamsTable = ({ onReady }) => {
     })();
     return () => {
       cancelled = true;
+      lastFilterOptionsParamsRef.current = null;
     };
   }, [buildFilterParams]);