fix(proxy): enhance stream switching logic and error handling

Updated the stream switching mechanism in the proxy server to ensure that the `stream_id` is persisted correctly and that the requesting worker waits for confirmation from the owning worker in multi-worker deployments. Improved error handling for stream switch failures, returning appropriate HTTP status codes (502/504) based on the confirmation status. Additionally, refined the metadata update process to handle cases where the requested URL is already in use, ensuring a successful response without unnecessary operations. This change addresses issues with stale metadata and enhances the robustness of the stream switching feature. (Fixes #1412)

CHANGELOG.md

@@ -41,6 +41,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **`/proxy/ts/change_stream` now persists `stream_id` and waits for the real switch outcome in multi-worker deployments.** When the request landed on a worker that didn't own the channel, the owning worker's `STREAM_SWITCH` pubsub handler performed the switch but wrote only `url`/`user_agent` back to the channel's Redis metadata, so `GET /proxy/ts/status` kept reporting the old `stream_id` indefinitely; the handler now persists the full switch metadata (`stream_id`, `m3u_profile`, stream name) via the same `_update_channel_metadata()` helper the direct owner path uses. `stream_name` from the initial `get_stream_info_for_switch()` lookup is now forwarded through the pubsub payload (and from `change_stream` / `next_stream` on the direct owner path) so the owner no longer re-queries the database when writing metadata. The endpoint also no longer returns an optimistic 200 on the non-owner path: the requesting worker waits (up to 15s) for the owner to confirm via the `switch_status` Redis key and answers 502 if the owner reports failure or 504 if no confirmation arrives (e.g. owner worker gone); `next_stream` gets the same treatment. Requesting a switch to the URL the channel is already playing is now treated as success (with a metadata refresh) instead of a silent no-op, and the `switch_status` key now expires (60s TTL) instead of persisting forever. (Fixes #1412)
 - **XC live streams, `/panel_api.php`, and `/xmltv.php` no longer crash when a visible channel has no channel number.** Since `channel_number` became nullable, auto-sync and compact numbering can leave a channel with `effective_channel_number=None` while it remains visible in output (`hidden_from_output=False`). The XC live-stream number map skipped those rows but still tried to emit them, causing `KeyError` on `get_live_streams` and a 500 on `/panel_api.php` (`xc_get_info`). Authenticated XMLTV export (`/xmltv.php`, profile EPG with a logged-in user) had the same gap and could fail chunk-cache rebuilds with `KeyError` on `channel_num_map[channel.id]`. Null-number channels are now deferred into the existing collision-free integer assignment pass (same second loop as fractional numbers) and receive the next available integer; `_xc_channel_entry` also falls back to `channel.id` if a row is still unmapped. HDHR lineup behaviour is unchanged (null-number channels remain omitted there).
 - **System events and Connect dispatch no longer close the DB connection mid-call outside worker contexts.** `log_system_event()` and `dispatch_event_system()` always called `close_old_connections()` in `finally` blocks, which broke Django `TestCase` transactions and could interrupt in-flight ORM work when Connect/plugin handlers ran synchronously. Closes are now limited to gevent uWSGI workers (and the Celery sync-dispatch path where gevent is patched but no hub runs). Celery workers still reset connections via existing `task_postrun` / `task_prerun` hooks on the standard PostgreSQL backend (no geventpool).
 - **Connect `trigger_event` no longer releases DB connections during plugin discovery.** Event dispatch calls `discover_plugins(use_cache=True)` to resolve plugin handlers; its unconditional `close_old_connections()` could tear down the caller's connection before `log_system_event()` or Schedules Direct refresh finished. Discovery from event dispatch now passes `release_connections=False`; boot-time discovery (`worker_process_init`, `post_migrate`, admin reload) still releases connections by default.

apps/proxy/live_proxy/server.py

@@ -249,26 +249,35 @@ class ProxyServer:
                                         user_agent = data.get("user_agent")
                                         event_stream_id = data.get("stream_id")
                                         event_m3u_profile_id = data.get("m3u_profile_id")
+                                        event_stream_name = data.get("stream_name")
 
                                         if new_url and channel_id in self.stream_managers:
                                             # Mark the switch as in-progress in Redis so other workers know to wait
+                                            status_key = RedisKeys.switch_status(channel_id)
                                             if self.redis_client:
-                                                status_key = RedisKeys.switch_status(channel_id)
-                                                self.redis_client.set(status_key, "switching")
+                                                self.redis_client.setex(status_key, 60, "switching")
 
-                                            # Perform the stream switch, forwarding stream_id and m3u_profile_id
                                             stream_manager = self.stream_managers[channel_id]
-                                            success = stream_manager.update_url(new_url, event_stream_id, event_m3u_profile_id)
+                                            if new_url == stream_manager.url:
+                                                # update_url() returns False for same URL; still success so metadata refreshes
+                                                logger.info(f"Channel {channel_id} already using requested URL, refreshing metadata only")
+                                                success = True
+                                            else:
+                                                success = stream_manager.update_url(new_url, event_stream_id, event_m3u_profile_id)
 
                                             if success:
                                                 logger.info(f"Stream switch initiated for channel {channel_id}")
 
-                                                # Confirm the URL in metadata now that the switch happened
                                                 if self.redis_client:
-                                                    metadata_key = RedisKeys.channel_metadata(channel_id)
-                                                    self.redis_client.hset(metadata_key, "url", new_url)
-                                                    if user_agent:
-                                                        self.redis_client.hset(metadata_key, "user_agent", user_agent)
+                                                    try:
+                                                        from .services.channel_service import ChannelService
+                                                        ChannelService._update_channel_metadata(
+                                                            channel_id, new_url, user_agent,
+                                                            event_stream_id, event_m3u_profile_id,
+                                                            event_stream_name,
+                                                        )
+                                                    except Exception as e:
+                                                        logger.error(f"Error updating switch metadata for channel {channel_id}: {e}", exc_info=True)
 
                                                 # Publish confirmation
                                                 switch_result = {
@@ -283,9 +292,8 @@ class ProxyServer:
                                                     json.dumps(switch_result)
                                                 )
 
-                                                # Update status
                                                 if self.redis_client:
-                                                    self.redis_client.set(status_key, "switched")
+                                                    self.redis_client.setex(status_key, 60, "switched")
                                             else:
                                                 logger.error(f"Failed to switch stream for channel {channel_id}")
 
@@ -309,6 +317,9 @@ class ProxyServer:
                                                     f"live:events:{channel_id}",
                                                     json.dumps(switch_result)
                                                 )
+
+                                                if self.redis_client:
+                                                    self.redis_client.setex(status_key, 60, "failed")
                                     elif event_type == EventType.CHANNEL_STOP:
                                         requester_worker_id = data.get("requester_worker_id")
                                         logger.info(

apps/proxy/live_proxy/services/channel_service.py

@@ -6,6 +6,7 @@ This separates business logic from HTTP handling in views.
 import logging
 import time
 import json
+import gevent
 from apps.channels.models import Channel, Stream
 from ..server import ProxyServer
 from ..redis_keys import RedisKeys
@@ -17,6 +18,9 @@ from .log_parsers import LogParserFactory
 
 logger = logging.getLogger("live_proxy")
 
+STREAM_SWITCH_CONFIRM_TIMEOUT = 15  # transcode teardown can take several seconds
+STREAM_SWITCH_POLL_INTERVAL = 0.1
+
 class ChannelService:
     """Service class for channel operations"""
 
@@ -349,7 +353,7 @@ class ChannelService:
         return success
 
     @staticmethod
-    def change_stream_url(channel_id, new_url=None, user_agent=None, target_stream_id=None, m3u_profile_id=None):
+    def change_stream_url(channel_id, new_url=None, user_agent=None, target_stream_id=None, m3u_profile_id=None, stream_name=None):
         """
         Change the URL of an existing stream.
 
@@ -367,7 +371,6 @@ class ChannelService:
 
         # If no direct URL is provided but a target stream is, get URL from target stream
         stream_id = None
-        stream_name = None
         if not new_url and target_stream_id:
             stream_info = get_stream_info_for_switch(channel_id, target_stream_id)
             if 'error' in stream_info:
@@ -440,9 +443,14 @@ class ChannelService:
             manager = proxy_server.stream_managers[channel_id]
             old_url = manager.url
 
-            # Update the stream
-            success = manager.update_url(new_url, stream_id, m3u_profile_id)
-            logger.info(f"Stream URL changed from {old_url} to {new_url}, result: {success}")
+            if new_url == old_url:
+                # update_url() returns False for same URL; still success so metadata refreshes
+                success = True
+                logger.info(f"Channel {channel_id} already using URL {new_url}, refreshing metadata only")
+            else:
+                # Update the stream
+                success = manager.update_url(new_url, stream_id, m3u_profile_id)
+                logger.info(f"Stream URL changed from {old_url} to {new_url}, result: {success}")
 
             # Update Redis metadata based on the actual outcome.
             # On success, write the new values. On failure, restore whatever URL
@@ -472,16 +480,57 @@ class ChannelService:
             # in the pubsub message, so there is no reason to pre-write metadata here.
             logger.debug(f"This worker is not the owner, publishing stream switch event for channel {channel_id}")
             if proxy_server.redis_client:
-                ChannelService._publish_stream_switch_event(channel_id, new_url, user_agent, stream_id, m3u_profile_id)
+                status_key = RedisKeys.switch_status(channel_id)
+                try:
+                    # Clear stale status from a prior switch
+                    proxy_server.redis_client.delete(status_key)
+                except Exception as e:
+                    logger.warning(f"Could not clear switch status for channel {channel_id}: {e}")
+
+                ChannelService._publish_stream_switch_event(
+                    channel_id, new_url, user_agent, stream_id, m3u_profile_id, stream_name
+                )
+
+                switch_status = None
+                deadline = time.time() + STREAM_SWITCH_CONFIRM_TIMEOUT
+                while time.time() < deadline:
+                    try:
+                        switch_status = proxy_server.redis_client.get(status_key)
+                    except Exception as e:
+                        logger.warning(f"Error polling switch status for channel {channel_id}: {e}")
+                        switch_status = None
+                        break
+                    if switch_status in ("switched", "failed"):
+                        break
+                    gevent.sleep(STREAM_SWITCH_POLL_INTERVAL)
+
                 result.update({
                     'direct_update': False,
                     'event_published': True,
                     'worker_id': proxy_server.worker_id
                 })
+
+                if switch_status == "switched":
+                    result['success'] = True
+                elif switch_status == "failed":
+                    result['success'] = False
+                    result['message'] = 'Owner worker failed to switch stream'
+                else:
+                    result['success'] = False
+                    result['confirmed'] = False
+                    result['message'] = (
+                        f'Stream switch was not confirmed by the channel owner '
+                        f'within {STREAM_SWITCH_CONFIRM_TIMEOUT}s'
+                    )
+                    logger.error(
+                        f"Stream switch for channel {channel_id} not confirmed within "
+                        f"{STREAM_SWITCH_CONFIRM_TIMEOUT}s (owner may be down or still switching)"
+                    )
             else:
                 result.update({
                     'direct_update': False,
                     'event_published': False,
+                    'success': False,
                     'error': 'Redis not available for pubsub'
                 })
 
@@ -865,7 +914,7 @@ class ChannelService:
             close_old_connections()
 
     @staticmethod
-    def _publish_stream_switch_event(channel_id, new_url, user_agent=None, stream_id=None, m3u_profile_id=None):
+    def _publish_stream_switch_event(channel_id, new_url, user_agent=None, stream_id=None, m3u_profile_id=None, stream_name=None):
         """Publish a stream switch event to Redis pubsub"""
         proxy_server = ProxyServer.get_instance()
 
@@ -879,6 +928,7 @@ class ChannelService:
             "user_agent": user_agent,
             "stream_id": stream_id,
             "m3u_profile_id": m3u_profile_id,
+            "stream_name": stream_name,
             "requester": proxy_server.worker_id,
             "timestamp": time.time()
         }

apps/proxy/live_proxy/views.py

@@ -771,6 +771,8 @@ def change_stream(request, channel_id):
         new_url = data.get("url")
         user_agent = data.get("user_agent")
         stream_id = data.get("stream_id")
+        m3u_profile_id = None
+        stream_name = None
 
         # If stream_id is provided, get the URL and user_agent from it
         if stream_id:
@@ -788,7 +790,7 @@ def change_stream(request, channel_id):
             new_url = stream_info["url"]
             user_agent = stream_info["user_agent"]
             m3u_profile_id = stream_info.get("m3u_profile_id")
-            # Stream ID will be passed to change_stream_url later
+            stream_name = stream_info.get("stream_name")
         elif not new_url:
             return JsonResponse(
                 {"error": "Either url or stream_id must be provided"}, status=400
@@ -801,7 +803,7 @@ def change_stream(request, channel_id):
         # Use the service layer instead of direct implementation
         # Pass stream_id to ensure proper connection tracking
         result = ChannelService.change_stream_url(
-            channel_id, new_url, user_agent, stream_id, m3u_profile_id
+            channel_id, new_url, user_agent, stream_id, m3u_profile_id, stream_name=stream_name
         )
 
         # Get the stream manager before updating URL
@@ -824,6 +826,20 @@ def change_stream(request, channel_id):
                 status=404,
             )
 
+        if result.get("success") is False:
+            error_data = {
+                "error": result.get("message", result.get("error", "Stream switch failed")),
+                "channel": channel_id,
+                "url": new_url,
+                "owner": result.get("direct_update", False),
+                "worker_id": proxy_server.worker_id,
+            }
+            if stream_id:
+                error_data["stream_id"] = stream_id
+            # confirmed=False means owner never responded (504); owner reported failure (502)
+            status_code = 504 if result.get("confirmed") is False else 502
+            return JsonResponse(error_data, status=status_code)
+
         # Format response based on whether it was a direct update or event-based
         response_data = {
             "message": "Stream changed successfully",
@@ -1068,6 +1084,7 @@ def next_stream(request, channel_id):
             stream_info["user_agent"],
             next_stream_id,
             stream_info.get("m3u_profile_id"),
+            stream_name=stream_info.get("stream_name"),
         )
 
         if result.get("status") == "error":
@@ -1081,6 +1098,18 @@ def next_stream(request, channel_id):
                 status=404,
             )
 
+        if result.get("success") is False:
+            return JsonResponse(
+                {
+                    "error": result.get("message", result.get("error", "Stream switch failed")),
+                    "current_stream_id": current_stream_id,
+                    "next_stream_id": next_stream_id,
+                    "owner": result.get("direct_update", False),
+                    "worker_id": proxy_server.worker_id,
+                },
+                status=504 if result.get("confirmed") is False else 502,
+            )
+
         # Format success response
         response_data = {
             "message": "Stream switched to next available",

apps/proxy/tests/test_stream_switch.py

@@ -0,0 +1,196 @@
+"""Tests for stream switch confirmation and metadata persistence."""
+
+import json
+from unittest.mock import MagicMock, patch
+
+from django.test import TestCase
+
+from apps.proxy.live_proxy.constants import ChannelMetadataField
+from apps.proxy.live_proxy.redis_keys import RedisKeys
+from apps.proxy.live_proxy.services import channel_service as cs_module
+from apps.proxy.live_proxy.services.channel_service import ChannelService
+
+
+class FakeRedis:
+    def __init__(self):
+        self.store = {}
+        self.hashes = {}
+        self.published = []
+
+    def get(self, key):
+        return self.store.get(key)
+
+    def set(self, key, value):
+        self.store[key] = str(value)
+
+    def setex(self, key, ttl, value):
+        self.store[key] = str(value)
+
+    def delete(self, *keys):
+        count = 0
+        for key in keys:
+            if self.store.pop(key, None) is not None:
+                count += 1
+            if self.hashes.pop(key, None) is not None:
+                count += 1
+        return count
+
+    def keys(self, pattern):
+        return []
+
+    def exists(self, key):
+        return key in self.store or key in self.hashes
+
+    def type(self, key):
+        return "hash" if key in self.hashes else "none"
+
+    def hset(self, key, field=None, value=None, mapping=None):
+        hash_value = self.hashes.setdefault(key, {})
+        if field is not None and value is not None:
+            hash_value[str(field)] = str(value)
+        for f, v in (mapping or {}).items():
+            hash_value[str(f)] = str(v)
+
+    def hget(self, key, field):
+        return self.hashes.get(key, {}).get(field)
+
+    def publish(self, channel, message):
+        self.published.append((channel, message))
+
+
+CHANNEL_ID = "ad8b11c0-4cd2-4bf5-a95b-153aee7f0671"
+NEW_URL = "http://provider.example/stream/144065.ts"
+
+
+def make_proxy_server(redis, owner):
+    proxy = MagicMock()
+    proxy.redis_client = redis
+    proxy.worker_id = "worker-under-test"
+    proxy.check_if_channel_exists.return_value = True
+    proxy.am_i_owner.return_value = owner
+    proxy.stream_managers = {}
+    proxy.stream_buffers = {}
+    return proxy
+
+
+class OwnerPathTests(TestCase):
+    def _run(self, manager_url="http://provider.example/stream/296622.ts"):
+        redis = FakeRedis()
+        proxy = make_proxy_server(redis, owner=True)
+
+        manager = MagicMock()
+        manager.url = manager_url
+        manager.update_url.return_value = True
+        proxy.stream_managers[CHANNEL_ID] = manager
+
+        with patch.object(cs_module.ProxyServer, "get_instance", return_value=proxy), \
+             patch("django.db.close_old_connections"):
+            result = ChannelService.change_stream_url(
+                CHANNEL_ID, NEW_URL, "test-agent",
+                target_stream_id=144065, m3u_profile_id=7,
+                stream_name="Alt Feed",
+            )
+        return result, redis, manager
+
+    def test_owner_switch_persists_stream_id_metadata(self):
+        result, redis, manager = self._run()
+
+        manager.update_url.assert_called_once_with(NEW_URL, 144065, 7)
+        self.assertTrue(result["success"])
+        self.assertTrue(result["direct_update"])
+
+        metadata = redis.hashes[RedisKeys.channel_metadata(CHANNEL_ID)]
+        self.assertEqual(metadata[ChannelMetadataField.URL], NEW_URL)
+        self.assertEqual(metadata[ChannelMetadataField.STREAM_ID], "144065")
+        self.assertEqual(metadata[ChannelMetadataField.M3U_PROFILE], "7")
+        self.assertEqual(metadata[ChannelMetadataField.STREAM_NAME], "Alt Feed")
+
+    def test_owner_same_url_is_success_and_repairs_metadata(self):
+        result, redis, manager = self._run(manager_url=NEW_URL)
+
+        manager.update_url.assert_not_called()
+        self.assertTrue(result["success"])
+
+        metadata = redis.hashes[RedisKeys.channel_metadata(CHANNEL_ID)]
+        self.assertEqual(metadata[ChannelMetadataField.STREAM_ID], "144065")
+
+
+class NonOwnerPathTests(TestCase):
+    def _run(self, owner_outcome):
+        redis = FakeRedis()
+        proxy = make_proxy_server(redis, owner=False)
+        status_key = RedisKeys.switch_status(CHANNEL_ID)
+
+        if owner_outcome is not None:
+            original_publish = redis.publish
+
+            def publish_and_confirm(channel, message):
+                original_publish(channel, message)
+                redis.store[status_key] = owner_outcome
+
+            redis.publish = publish_and_confirm
+
+        with patch.object(cs_module.ProxyServer, "get_instance", return_value=proxy), \
+             patch.object(cs_module, "STREAM_SWITCH_CONFIRM_TIMEOUT", 0.3), \
+             patch.object(cs_module, "STREAM_SWITCH_POLL_INTERVAL", 0.05):
+            result = ChannelService.change_stream_url(
+                CHANNEL_ID, NEW_URL, "test-agent",
+                target_stream_id=144065, m3u_profile_id=7,
+                stream_name="Alt Feed",
+            )
+        return result, redis
+
+    def test_pubsub_event_carries_stream_id(self):
+        result, redis = self._run(owner_outcome="switched")
+
+        self.assertEqual(len(redis.published), 1)
+        payload = json.loads(redis.published[0][1])
+        self.assertEqual(payload["stream_id"], 144065)
+        self.assertEqual(payload["m3u_profile_id"], 7)
+        self.assertEqual(payload["stream_name"], "Alt Feed")
+        self.assertEqual(payload["url"], NEW_URL)
+
+    def test_switch_confirmed_by_owner_reports_success(self):
+        result, _ = self._run(owner_outcome="switched")
+
+        self.assertTrue(result["success"])
+        self.assertFalse(result["direct_update"])
+        self.assertTrue(result["event_published"])
+
+    def test_switch_failed_by_owner_reports_failure(self):
+        result, _ = self._run(owner_outcome="failed")
+
+        self.assertFalse(result["success"])
+        self.assertIn("failed", result["message"].lower())
+
+    def test_no_confirmation_times_out_and_reports_failure(self):
+        result, _ = self._run(owner_outcome=None)
+
+        self.assertFalse(result["success"])
+        self.assertIs(result["confirmed"], False)
+        self.assertIn("not confirmed", result["message"])
+
+    def test_stale_status_key_is_cleared_before_publishing(self):
+        redis = FakeRedis()
+        proxy = make_proxy_server(redis, owner=False)
+        status_key = RedisKeys.switch_status(CHANNEL_ID)
+        redis.store[status_key] = "switched"
+
+        deleted_before_publish = []
+        original_publish = redis.publish
+
+        def tracking_publish(channel, message):
+            deleted_before_publish.append(status_key not in redis.store)
+            original_publish(channel, message)
+
+        redis.publish = tracking_publish
+
+        with patch.object(cs_module.ProxyServer, "get_instance", return_value=proxy), \
+             patch.object(cs_module, "STREAM_SWITCH_CONFIRM_TIMEOUT", 0.2), \
+             patch.object(cs_module, "STREAM_SWITCH_POLL_INTERVAL", 0.05):
+            result = ChannelService.change_stream_url(
+                CHANNEL_ID, NEW_URL, "test-agent", target_stream_id=144065,
+            )
+
+        self.assertEqual(deleted_before_publish, [True])
+        self.assertFalse(result["success"])