diff --git a/CHANGELOG.md b/CHANGELOG.md index 008d1205..1a8baaa9 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/apps/proxy/live_proxy/server.py b/apps/proxy/live_proxy/server.py index 97c5b68b..004a2d29 100644 --- a/apps/proxy/live_proxy/server.py +++ b/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( diff --git a/apps/proxy/live_proxy/services/channel_service.py b/apps/proxy/live_proxy/services/channel_service.py index e18e07af..df4509e4 100644 --- a/apps/proxy/live_proxy/services/channel_service.py +++ b/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() } diff --git a/apps/proxy/live_proxy/views.py b/apps/proxy/live_proxy/views.py index cf766505..d42ffdc0 100644 --- a/apps/proxy/live_proxy/views.py +++ b/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", diff --git a/apps/proxy/tests/test_stream_switch.py b/apps/proxy/tests/test_stream_switch.py new file mode 100644 index 00000000..5fd7b7d9 --- /dev/null +++ b/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"])