From efad028be10874a58d9818b1eccd33f1b0507754 Mon Sep 17 00:00:00 2001 From: Josh Becker Date: Sat, 11 Jul 2026 21:50:33 -0400 Subject: [PATCH] fix(proxy): reap ghost channel sessions stuck in initializing A channel initialization that died between the early metadata write and stream manager creation left immortal 'initializing' metadata: no owner lock, no URL, no TTL. Every future play request attached to the dead session and got 'Error: Connection stalled'; failover was never tried; the orphan sweep never reaped it because it only checked whether the owner *worker* heartbeat was alive, not whether any worker actually held the channel ownership lock. The M3U profile connection slot also leaked (#947). - Detect ghost sessions (pre-active state, no ownership lock, past the init grace period) on play requests and reinitialize instead of attaching; check_if_channel_exists() now cleans them up too. - Mark failed initializations as 'error' (terminal) instead of leaving 'initializing' behind, and give early init metadata a TTL so any missed failure path expires on its own. - Extend the 30s orphan sweep to reap pre-active channels with no ownership lock and no clients after 2x the init grace period, even when the owning worker is still alive. - Cleaning a ghost releases the leaked M3U profile connection slot. Fixes the ghost-session variant of #669/#695 and the init-failure slot leak of #947. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 5 + apps/proxy/live_proxy/server.py | 157 +++++++- .../live_proxy/services/channel_service.py | 6 +- .../tests/test_ghost_session_cleanup.py | 346 ++++++++++++++++++ apps/proxy/live_proxy/views.py | 24 +- 5 files changed, 520 insertions(+), 18 deletions(-) create mode 100644 apps/proxy/live_proxy/tests/test_ghost_session_cleanup.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a8baaa9..9a9345da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -64,6 +64,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Live proxy failover now walks backup streams in channel order.** After a stable session on a backup stream, `tried_stream_ids` is cleared so rotation continues from the current position (stream 2 → 3 → 4 → 1) instead of jumping back to stream 1. `get_alternate_streams()` returns alternates in that rotated order, matching the manual next-stream API. - **Live proxy preview no longer 500s when joining an active channel on a non-owner worker.** `stream_ts()` now pre-registers the client before `ensure_output_profile()` (the same watchdog protection owners already had), so the non-owner cleanup thread no longer tears down `client_manager` while output-profile transcode is still starting. Failed setup paths remove the client again and return 503 instead of an unhandled `KeyError`. - **Backend test suite reliability.** `dispatcharr.settings_test` creates `test_dispatcharr` as UTF-8 (`template0`) so EPG programme indexes and Unicode XMLTV data round-trip correctly. Tests were updated for DOCTYPE-based HTML entity resolution, EPG name-normalization expectations, migration 0037 unit invocation, Schedules Direct mocks, and other areas; full app test modules are discoverable when listed explicitly (bare `manage.py test` still only runs `core.tests` and top-level `tests/`). +- **Ghost channel sessions stuck in `initializing` no longer block playback forever.** When an initialization failed between the early metadata write and stream manager creation (upstream hiccup, exception, worker race), the channel was left in `initializing` with no owner lock, no URL, and no TTL. Every future play request attached to the dead session, waited, and got `Error: Connection stalled`; the failover stream was never tried, and the session survived indefinitely because the orphan sweep only reaped channels whose owner *worker* had died (not sessions whose init had died on a still-alive worker). The M3U profile connection slot allocated for the failed init also leaked. Now: + - Play requests detect a pre-active channel with no ownership lock past the init grace period, tear it down (releasing the leaked profile slot), and reinitialize it fresh — so the first viewer after a failed init revives the channel and normal stream failover applies. + - Initialization failure paths mark the channel `error` (a terminal state that triggers cleanup and reinit) instead of leaving it `initializing`. + - The 30-second orphan sweep additionally reaps channels stuck in `initializing`/`connecting` with no ownership lock and no clients for longer than twice the init grace period, even when the owning worker is still alive. + - Early initialization metadata is written with a TTL so any missed failure path expires on its own instead of living forever. ## [0.27.2] - 2026-06-30 diff --git a/apps/proxy/live_proxy/server.py b/apps/proxy/live_proxy/server.py index 004a2d29..3eb1165a 100644 --- a/apps/proxy/live_proxy/server.py +++ b/apps/proxy/live_proxy/server.py @@ -25,7 +25,7 @@ from .client_manager import ClientManager from .output.fmp4.manager import FMP4RemuxManager from .output.profile.manager import OutputProfileManager, PROFILE_STATE_ACTIVE from .redis_keys import RedisKeys -from .constants import ChannelState, EventType, StreamType +from .constants import ChannelState, EventType, StreamType, REDIS_TTL_MEDIUM from .config_helper import ConfigHelper from .utils import get_logger @@ -585,17 +585,27 @@ class ProxyServer: active_states = [ChannelState.INITIALIZING, ChannelState.CONNECTING, ChannelState.WAITING_FOR_CLIENTS, ChannelState.ACTIVE, ChannelState.BUFFERING] if state in active_states: - logger.info(f"Channel {channel_id} already being initialized with state {state}") - # Create buffer and client manager only if we don't have them - if channel_id not in self.stream_buffers: - self.stream_buffers[channel_id] = StreamBuffer(channel_id, redis_client=RedisClient.get_buffer()) - if channel_id not in self.client_managers: - self.client_managers[channel_id] = ClientManager( - channel_id, - redis_client=self.redis_client, - worker_id=self.worker_id + if self.is_stale_pre_active(channel_id, metadata): + # Ghost session from a failed init: nobody owns it and it + # will never progress. Clean it up (releases the leaked M3U + # profile slot) and fall through to a fresh initialization. + logger.warning( + f"Channel {channel_id} metadata shows {state} but no worker " + f"owns it - discarding stale session and reinitializing" ) - return True + self._clean_redis_keys(channel_id) + else: + logger.info(f"Channel {channel_id} already being initialized with state {state}") + # Create buffer and client manager only if we don't have them + if channel_id not in self.stream_buffers: + self.stream_buffers[channel_id] = StreamBuffer(channel_id, redis_client=RedisClient.get_buffer()) + if channel_id not in self.client_managers: + self.client_managers[channel_id] = ClientManager( + channel_id, + redis_client=self.redis_client, + worker_id=self.worker_id + ) + return True # Create buffer and client manager instances (or reuse if they exist) if channel_id not in self.stream_buffers: @@ -621,6 +631,11 @@ class ProxyServer: if stream_id: initial_metadata["stream_id"] = str(stream_id) self.redis_client.hset(metadata_key, mapping=initial_metadata) + # A failed init must not leave immortal metadata behind: give the key + # a TTL now; successful initialization extends it and the registry + # refresh keeps it alive afterwards. + if self.redis_client.ttl(metadata_key) == -1: + self.redis_client.expire(metadata_key, REDIS_TTL_MEDIUM) logger.info(f"Set early initializing state for channel {channel_id}") # Get channel URL from Redis if available @@ -677,6 +692,7 @@ class ProxyServer: # or we can get it from Redis if not channel_url: logger.error(f"No URL available for channel {channel_id}") + self._mark_init_failure(channel_id, "No URL available during initialization") return False # Try to acquire ownership with Redis locking @@ -807,6 +823,9 @@ class ProxyServer: logger.error(f"Error initializing channel {channel_id}: {e}", exc_info=True) # Release ownership on failure self.release_ownership(channel_id) + # Leave the channel in a terminal state so future requests reinitialize + # instead of attaching to a half-created session that never progresses + self._mark_init_failure(channel_id, f"Initialization failed: {e}") return False def check_if_channel_exists(self, channel_id): @@ -838,6 +857,17 @@ class ProxyServer: # If the channel is in a valid state, check if the owner is still active if state in valid_states: + # Ghost detection: a pre-active state with no ownership lock means + # no worker is driving initialization. The owner worker may still be + # alive (its heartbeat passes below), but this channel will never + # progress - clean it up so it can be reinitialized. + if self.is_stale_pre_active(channel_id, metadata): + logger.warning( + f"Channel {channel_id} has stale {state} session with no owner lock - cleaning up" + ) + self._clean_zombie_channel(channel_id, metadata) + return False + # Check if owner still exists by checking heartbeat owner_heartbeat_key = f"live:worker:{owner}:heartbeat" owner_alive = self.redis_client.exists(owner_heartbeat_key) @@ -899,6 +929,97 @@ class ProxyServer: return False + def _stale_init_threshold(self): + """Age in seconds after which an ownerless pre-active channel is considered dead.""" + return max(ConfigHelper.channel_init_grace_period(), 30) + + def is_stale_pre_active(self, channel_id, metadata=None, max_age=None): + """ + Detect a ghost session: metadata shows a pre-active state (initializing/ + connecting) but no worker holds the ownership lock and this worker has no + stream manager for it. Nothing will ever advance such a channel — clients + that attach to it stall forever — so callers must treat it as dead. + """ + if not self.redis_client: + return False + + try: + if metadata is None: + metadata = self.redis_client.hgetall(RedisKeys.channel_metadata(channel_id)) + if not metadata: + return False + + state = metadata.get('state') + if state not in (ChannelState.INITIALIZING, ChannelState.CONNECTING): + return False + + # A local stream manager means this worker is driving the channel, + # even if the ownership lock momentarily lapsed before re-acquisition. + if channel_id in self.stream_managers or channel_id in self._live_stream_managers: + return False + + # Ownership lock present -> some worker is actively driving init. + if self.redis_client.exists(RedisKeys.channel_owner(channel_id)): + return False + + if max_age is None: + max_age = self._stale_init_threshold() + + newest = 0.0 + for field in ('state_changed_at', 'init_time', 'temp_init'): + raw = metadata.get(field) + if raw: + try: + newest = max(newest, float(raw)) + except (ValueError, TypeError): + pass + + if newest <= 0: + # Pre-active state with no owner and no usable timestamps - + # a half-written session that can only be a ghost. + return True + + return (time.time() - newest) > max_age + except Exception as e: + logger.error(f"Error checking stale pre-active state for channel {channel_id}: {e}") + return False + + def _mark_init_failure(self, channel_id, error_message): + """ + Transition a channel whose initialization failed to ERROR so future + requests reinitialize instead of attaching to a half-created session. + Only touches metadata when no other worker owns the channel. + """ + if not self.redis_client: + return + + try: + owner = self.get_channel_owner(channel_id) + if owner and owner != self.worker_id: + return + + metadata_key = RedisKeys.channel_metadata(channel_id) + if not self.redis_client.exists(metadata_key): + return + + state = self.redis_client.hget(metadata_key, 'state') + if state and state not in ChannelState.PRE_ACTIVE: + return + + now = str(time.time()) + self.redis_client.hset(metadata_key, mapping={ + 'state': ChannelState.ERROR, + 'state_changed_at': now, + 'error_message': error_message, + 'error_time': now, + }) + # Failed-init metadata must never become immortal + if self.redis_client.ttl(metadata_key) == -1: + self.redis_client.expire(metadata_key, REDIS_TTL_MEDIUM) + logger.info(f"Marked failed initialization for channel {channel_id}: {error_message}") + except Exception as e: + logger.error(f"Error marking init failure for channel {channel_id}: {e}") + def _clean_zombie_channel(self, channel_id, metadata=None): """Clean up a zombie channel (channel with Redis keys but no active owner)""" try: @@ -2167,6 +2288,20 @@ class ProxyServer: f"{real_count} live client(s) after ghost removal " f"- may need ownership takeover" ) + elif client_count == 0 and self.is_stale_pre_active( + channel_id, metadata, max_age=2 * self._stale_init_threshold() + ): + # Watchdog for ghost sessions: owner worker is alive but nobody + # holds the ownership lock and the channel has sat in a pre-active + # state past twice the init grace period. A failed initialization + # left it behind - reap it (also releases the M3U profile slot). + state = metadata.get('state', 'unknown') + logger.warning( + f"Reaping ghost channel {channel_id} stuck in {state} " + f"with no owner lock and no clients - cleaning up" + ) + self._stop_upstream_before_redis_cleanup(channel_id) + self._clean_redis_keys(channel_id) except Exception as e: logger.error(f"Error processing metadata key {key}: {e}", exc_info=True) diff --git a/apps/proxy/live_proxy/services/channel_service.py b/apps/proxy/live_proxy/services/channel_service.py index df4509e4..883766a1 100644 --- a/apps/proxy/live_proxy/services/channel_service.py +++ b/apps/proxy/live_proxy/services/channel_service.py @@ -10,7 +10,7 @@ import gevent from apps.channels.models import Channel, Stream from ..server import ProxyServer from ..redis_keys import RedisKeys -from ..constants import EventType, ChannelState, ChannelMetadataField +from ..constants import EventType, ChannelState, ChannelMetadataField, REDIS_TTL_MEDIUM from ..config_helper import ConfigHelper from ..url_utils import get_stream_info_for_switch from core.utils import log_system_event @@ -303,6 +303,10 @@ class ChannelService: "temp_init": str(time.time()) } proxy_server.redis_client.hset(metadata_key, mapping=initial_metadata) + # Don't let a failed init leave this key behind forever; the + # successful path replaces the TTL with the standard one. + if proxy_server.redis_client.ttl(metadata_key) == -1: + proxy_server.redis_client.expire(metadata_key, REDIS_TTL_MEDIUM) logger.info(f"Created initial metadata with stream_id {stream_id} for channel {channel_id}") # Verify the stream_id was set diff --git a/apps/proxy/live_proxy/tests/test_ghost_session_cleanup.py b/apps/proxy/live_proxy/tests/test_ghost_session_cleanup.py new file mode 100644 index 00000000..076462c9 --- /dev/null +++ b/apps/proxy/live_proxy/tests/test_ghost_session_cleanup.py @@ -0,0 +1,346 @@ +""" +Ghost session cleanup: a channel stuck in a pre-active state (initializing/ +connecting) with no ownership lock must be detected as dead, reaped by the +orphan watchdog, and never attached to by new clients. + +Regression tests for the failure mode where an initialization that dies +between the early metadata write and stream manager creation leaves immortal +'initializing' metadata that all future play requests attach to and stall on. +""" + +import time +from unittest.mock import MagicMock, patch + +from django.http import StreamingHttpResponse +from django.test import RequestFactory, SimpleTestCase + + +def make_proxy_server(redis_client): + """Build a ProxyServer without running __init__ (no threads, no Redis).""" + from apps.proxy.live_proxy.server import ProxyServer + + server = ProxyServer.__new__(ProxyServer) + server.redis_client = redis_client + server.stream_managers = {} + server.stream_buffers = {} + server.client_managers = {} + server.output_managers = {} + server.profile_managers = {} + server.profile_buffers = {} + server._channel_names = {} + server._stopping_channels = set() + server._stopping_since = {} + server._local_stop_locks = {} + server._live_stream_managers = {} + server.worker_id = "test-host:1" + return server + + +GRACE_PATCH = patch( + "apps.proxy.live_proxy.server.ConfigHelper.channel_init_grace_period", + return_value=60, +) + +CHANNEL_ID = "11111111-2222-3333-4444-555555555555" + + +class IsStalePreActiveTests(SimpleTestCase): + def setUp(self): + self.grace = GRACE_PATCH.start() + self.addCleanup(GRACE_PATCH.stop) + + def _ghost_metadata(self, state="initializing", age=3600): + return {"state": state, "init_time": str(time.time() - age)} + + def test_stale_when_no_owner_lock_and_old(self): + redis = MagicMock() + redis.exists.return_value = False # no ownership lock + server = make_proxy_server(redis) + + self.assertTrue(server.is_stale_pre_active(CHANNEL_ID, self._ghost_metadata())) + + def test_not_stale_when_owner_lock_exists(self): + redis = MagicMock() + redis.exists.return_value = True # ownership lock present + server = make_proxy_server(redis) + + self.assertFalse(server.is_stale_pre_active(CHANNEL_ID, self._ghost_metadata())) + + def test_not_stale_for_active_state(self): + redis = MagicMock() + redis.exists.return_value = False + server = make_proxy_server(redis) + + self.assertFalse( + server.is_stale_pre_active(CHANNEL_ID, self._ghost_metadata(state="active")) + ) + + def test_not_stale_when_init_is_recent(self): + redis = MagicMock() + redis.exists.return_value = False + server = make_proxy_server(redis) + + self.assertFalse( + server.is_stale_pre_active(CHANNEL_ID, self._ghost_metadata(age=5)) + ) + + def test_not_stale_with_local_stream_manager(self): + redis = MagicMock() + redis.exists.return_value = False + server = make_proxy_server(redis) + server.stream_managers[CHANNEL_ID] = object() + + self.assertFalse( + server.is_stale_pre_active(CHANNEL_ID, self._ghost_metadata()) + ) + + def test_stale_when_no_timestamps_at_all(self): + redis = MagicMock() + redis.exists.return_value = False + server = make_proxy_server(redis) + + self.assertTrue(server.is_stale_pre_active(CHANNEL_ID, {"state": "connecting"})) + + def test_respects_custom_max_age(self): + redis = MagicMock() + redis.exists.return_value = False + server = make_proxy_server(redis) + metadata = self._ghost_metadata(age=90) + + self.assertTrue(server.is_stale_pre_active(CHANNEL_ID, metadata, max_age=60)) + self.assertFalse(server.is_stale_pre_active(CHANNEL_ID, metadata, max_age=120)) + + +class MarkInitFailureTests(SimpleTestCase): + def test_marks_error_state_and_sets_ttl(self): + redis = MagicMock() + redis.get.return_value = None # no owner + redis.exists.return_value = True + redis.hget.return_value = "initializing" + redis.ttl.return_value = -1 + server = make_proxy_server(redis) + + server._mark_init_failure(CHANNEL_ID, "boom") + + redis.hset.assert_called_once() + mapping = redis.hset.call_args.kwargs["mapping"] + self.assertEqual(mapping["state"], "error") + self.assertEqual(mapping["error_message"], "boom") + redis.expire.assert_called_once() + + def test_does_not_touch_channel_owned_by_other_worker(self): + redis = MagicMock() + redis.get.return_value = "other-host:2" + server = make_proxy_server(redis) + + server._mark_init_failure(CHANNEL_ID, "boom") + + redis.hset.assert_not_called() + + def test_does_not_clobber_active_state(self): + redis = MagicMock() + redis.get.return_value = None + redis.exists.return_value = True + redis.hget.return_value = "active" + server = make_proxy_server(redis) + + server._mark_init_failure(CHANNEL_ID, "boom") + + redis.hset.assert_not_called() + + +class CheckIfChannelExistsGhostTests(SimpleTestCase): + def setUp(self): + self.grace = GRACE_PATCH.start() + self.addCleanup(GRACE_PATCH.stop) + + def test_ghost_channel_is_cleaned_and_reported_missing(self): + from apps.proxy.live_proxy.redis_keys import RedisKeys + + metadata_key = RedisKeys.channel_metadata(CHANNEL_ID) + owner_key = RedisKeys.channel_owner(CHANNEL_ID) + ghost_metadata = { + "state": "initializing", + "owner": "alive-worker:9", + "init_time": str(time.time() - 3600), + } + + redis = MagicMock() + redis.exists.side_effect = lambda key: key == metadata_key + redis.hgetall.return_value = ghost_metadata + server = make_proxy_server(redis) + + with patch.object(server, "_clean_zombie_channel") as mock_clean: + self.assertFalse(server.check_if_channel_exists(CHANNEL_ID)) + mock_clean.assert_called_once() + # Ensure the ownership lock was actually consulted + redis.exists.assert_any_call(owner_key) + + def test_initializing_channel_with_owner_lock_still_exists(self): + from apps.proxy.live_proxy.redis_keys import RedisKeys + + metadata_key = RedisKeys.channel_metadata(CHANNEL_ID) + owner_key = RedisKeys.channel_owner(CHANNEL_ID) + heartbeat_key = "live:worker:alive-worker:9:heartbeat" + metadata = { + "state": "initializing", + "owner": "alive-worker:9", + "init_time": str(time.time() - 3600), + } + + redis = MagicMock() + redis.exists.side_effect = lambda key: key in ( + metadata_key, + owner_key, + heartbeat_key, + ) + redis.hgetall.return_value = metadata + server = make_proxy_server(redis) + + with patch.object(server, "_clean_zombie_channel") as mock_clean: + self.assertTrue(server.check_if_channel_exists(CHANNEL_ID)) + mock_clean.assert_not_called() + + +class OrphanedMetadataReaperTests(SimpleTestCase): + def setUp(self): + self.grace = GRACE_PATCH.start() + self.addCleanup(GRACE_PATCH.stop) + + def _run_orphan_check(self, *, owner_lock_exists, age=3600, clients=0): + from apps.proxy.live_proxy.redis_keys import RedisKeys + + metadata_key = RedisKeys.channel_metadata(CHANNEL_ID) + owner_key = RedisKeys.channel_owner(CHANNEL_ID) + heartbeat_key = "live:worker:alive-worker:9:heartbeat" + ghost_metadata = { + "state": "initializing", + "owner": "alive-worker:9", + "init_time": str(time.time() - age), + } + + redis = MagicMock() + redis.keys.return_value = [metadata_key] + redis.hgetall.return_value = ghost_metadata + redis.scard.return_value = clients + + def exists(key): + if key == owner_key: + return owner_lock_exists + return key in (metadata_key, heartbeat_key) + + redis.exists.side_effect = exists + server = make_proxy_server(redis) + + with patch.object(server, "_stop_upstream_before_redis_cleanup") as mock_stop, \ + patch.object(server, "_clean_redis_keys") as mock_clean: + server._check_orphaned_metadata() + return mock_stop, mock_clean + + def test_reaps_stale_ghost_with_alive_owner_worker(self): + # Owner worker heartbeat is alive (worker survived, init died) - the + # reaper must still clean the channel because nobody holds the lock. + mock_stop, mock_clean = self._run_orphan_check(owner_lock_exists=False) + mock_stop.assert_called_once_with(CHANNEL_ID) + mock_clean.assert_called_once_with(CHANNEL_ID) + + def test_keeps_channel_with_owner_lock(self): + mock_stop, mock_clean = self._run_orphan_check(owner_lock_exists=True) + mock_stop.assert_not_called() + mock_clean.assert_not_called() + + def test_keeps_young_initializing_channel(self): + # Below 2x grace period the watchdog must not fire. + mock_stop, mock_clean = self._run_orphan_check( + owner_lock_exists=False, age=30 + ) + mock_stop.assert_not_called() + mock_clean.assert_not_called() + + def test_keeps_ghost_with_clients_until_they_stall_out(self): + mock_stop, mock_clean = self._run_orphan_check( + owner_lock_exists=False, clients=2 + ) + mock_stop.assert_not_called() + mock_clean.assert_not_called() + + +class StreamTsGhostRevivalTests(SimpleTestCase): + """A play request hitting ghost 'initializing' metadata must clean up and + reinitialize the channel instead of attaching and stalling.""" + + def setUp(self): + self.factory = RequestFactory() + + @patch("apps.proxy.live_proxy.views.close_old_connections") + @patch("apps.proxy.live_proxy.views.create_stream_generator") + @patch("apps.proxy.live_proxy.views._resolve_output_format", return_value="mpegts") + @patch("apps.proxy.live_proxy.views._resolve_output_profile", return_value=None) + @patch("apps.proxy.live_proxy.views.generate_stream_url") + @patch("apps.proxy.live_proxy.views.ChannelService") + @patch("apps.proxy.live_proxy.views.get_stream_object") + @patch("apps.proxy.live_proxy.views.network_access_allowed", return_value=True) + @patch("apps.proxy.live_proxy.views.ProxyServer") + def test_ghost_initializing_channel_is_reinitialized( + self, + mock_proxy_cls, + _network_ok, + mock_get_stream_object, + mock_channel_service, + mock_generate_url, + _output_profile, + _output_format, + mock_create_generator, + _mock_close, + ): + channel_id = "channel-uuid" + + channel = MagicMock() + channel.id = 1 + channel.uuid = channel_id + channel.name = "Test Channel" + channel.get_stream_profile.return_value.is_redirect.return_value = False + mock_get_stream_object.return_value = channel + + mock_channel_service.is_channel_unavailable_for_new_clients.return_value = False + mock_channel_service.initialize_channel.return_value = True + + mock_generate_url.return_value = ( + "http://upstream/stream.ts", "UA", False, "profile", True, None, + ) + + proxy_server = MagicMock() + proxy_server.redis_client.exists.return_value = True + proxy_server.redis_client.hgetall.return_value = { + "state": "initializing", + "owner": "dead-init-worker:9", + "init_time": str(time.time() - 3600), + } + proxy_server.redis_client.get.return_value = None + # The ghost detector fires for this channel + proxy_server.is_stale_pre_active.return_value = True + proxy_server.check_if_channel_exists.return_value = True + proxy_server.am_i_owner.return_value = True + proxy_server.stream_buffers = {channel_id: MagicMock()} + proxy_server.client_managers = {channel_id: MagicMock()} + proxy_server.get_buffer.return_value = MagicMock() + mock_proxy_cls.get_instance.return_value = proxy_server + + def _generate(): + yield b"chunk" + + mock_create_generator.return_value = lambda: _generate() + + request = self.factory.get(f"/proxy/live/{channel_id}/") + request.user = MagicMock(is_authenticated=False) + + from apps.proxy.live_proxy.views import stream_ts + + response = stream_ts(request, channel_id) + + self.assertIsInstance(response, StreamingHttpResponse) + # Ghost must be torn down before reinitialization + mock_channel_service.stop_channel.assert_called_once_with(channel_id) + mock_channel_service.initialize_channel.assert_called_once() + # And a fresh stream URL must have been requested + mock_generate_url.assert_called() diff --git a/apps/proxy/live_proxy/views.py b/apps/proxy/live_proxy/views.py index d42ffdc0..df691623 100644 --- a/apps/proxy/live_proxy/views.py +++ b/apps/proxy/live_proxy/views.py @@ -158,6 +158,7 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): needs_initialization = True channel_state = None channel_initializing = False + stale_ghost = False # Get current channel state from Redis if available if proxy_server.redis_client: @@ -186,10 +187,21 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): ChannelState.INITIALIZING, ChannelState.CONNECTING, ]: - channel_initializing = True - logger.debug( - f"[{client_id}] Channel {channel_id} is still initializing, client will wait" - ) + if proxy_server.is_stale_pre_active(channel_id, metadata): + # Ghost session from a failed init: no worker owns it, + # so waiting on it would stall forever. Clean up and + # reinitialize instead of attaching. + stale_ghost = True + needs_initialization = True + logger.warning( + f"[{client_id}] Channel {channel_id} stuck in {channel_state} " + f"with no owner - cleaning up and reinitializing" + ) + else: + channel_initializing = True + logger.debug( + f"[{client_id}] Channel {channel_id} is still initializing, client will wait" + ) elif channel_state == ChannelState.STOPPING: logger.info( f"[{client_id}] Channel {channel_id} is stopping, rejecting request" @@ -236,8 +248,8 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): return _channel_stopping_response() logger.info(f"[{client_id}] Starting channel {channel_id} initialization") - # Force cleanup of any previous instance if in terminal state - if channel_state in [ + # Force cleanup of any previous instance if in terminal state or ghosted + if stale_ghost or channel_state in [ ChannelState.ERROR, ChannelState.STOPPING, ChannelState.STOPPED,