Enhancement: Prevent 500 errors in live proxy preview when joining active channels on non-owner workers by ensuring proper client registration and cleanup. Update changelog to reflect this fix.

CHANGELOG.md

@@ -49,6 +49,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Live proxy failover works with the default VLC stream profile.** When `cvlc` could not open an upstream URL it often stayed running in dummy mode without writing TS data, so the stream manager never left its read loop and never exhausted retries to switch streams. The locked VLC profile now passes `--play-and-exit` (migration 0027 for existing installs), and `VLCLogParser` treats `unable to open the mrl` as a fatal input error so the connection is closed and the normal retry → failover cycle can proceed. (Fixes #1415)
 - **Live proxy no longer fails over after isolated provider disconnects.** Connection retries on the same URL used to accumulate indefinitely, so three brief drops spread across many hours of otherwise stable playback could exhaust retries and switch streams even when each reconnect succeeded. Retries now reset after 30 minutes without a failure (`RETRY_WINDOW_SECONDS`), so periodic provider drops (for example every few hours) each get a full retry budget on the current stream. Failover still triggers when three failures occur within that window.
 - **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`.
 
 ## [0.27.2] - 2026-06-30
 

apps/proxy/live_proxy/tests/test_live_db_cleanup.py

@@ -59,6 +59,7 @@ class StreamTsDbCleanupTests(SimpleTestCase):
         response = stream_ts(request, "channel-uuid")
 
         self.assertIsInstance(response, StreamingHttpResponse)
+        client_manager.add_client.assert_called_once()
         mock_close.assert_called_once()
 
 

apps/proxy/live_proxy/tests/test_stream_ts_client_registration.py

@@ -0,0 +1,353 @@
+"""Regression tests for stream_ts client registration ordering."""
+
+from unittest.mock import MagicMock, patch
+
+from django.http import JsonResponse, StreamingHttpResponse
+from django.test import RequestFactory, SimpleTestCase
+
+
+class StreamTsClientRegistrationTests(SimpleTestCase):
+    def setUp(self):
+        self.factory = RequestFactory()
+        self.channel_id = "channel-uuid"
+
+    def _channel(self, *, redirect=False):
+        channel = MagicMock()
+        channel.id = 1
+        channel.uuid = self.channel_id
+        channel.name = "Test Channel"
+        stream_profile = MagicMock()
+        stream_profile.is_redirect.return_value = redirect
+        channel.get_stream_profile.return_value = stream_profile
+        return channel
+
+    def _active_proxy_server(self, *, am_i_owner=False, client_manager=None):
+        client_manager = client_manager or MagicMock()
+        proxy_server = MagicMock()
+        proxy_server.redis_client = MagicMock()
+        proxy_server.redis_client.exists.return_value = True
+        proxy_server.redis_client.hgetall.return_value = {"state": "active"}
+        proxy_server.stream_buffers = {self.channel_id: MagicMock()}
+        proxy_server.client_managers = {self.channel_id: client_manager}
+        proxy_server.check_if_channel_exists.return_value = True
+        proxy_server.get_buffer.return_value = MagicMock()
+        proxy_server.am_i_owner.return_value = am_i_owner
+        proxy_server.ensure_output_profile.return_value = True
+        return proxy_server, client_manager
+
+    def _request(self):
+        request = self.factory.get(f"/proxy/ts/stream/{self.channel_id}/")
+        request.user = MagicMock(is_authenticated=False)
+        return request
+
+    @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")
+    @patch(
+        "apps.proxy.live_proxy.views.ChannelService.is_channel_unavailable_for_new_clients",
+        return_value=False,
+    )
+    @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_active_channel_registers_client_before_ensure_output_profile(
+        self,
+        mock_proxy_cls,
+        _network_ok,
+        mock_get_stream_object,
+        _unavailable,
+        mock_resolve_output_profile,
+        _output_format,
+        mock_create_generator,
+        _mock_close,
+    ):
+        profile = MagicMock()
+        profile.id = 2
+        profile.build_command.return_value = ["ffmpeg"]
+        mock_resolve_output_profile.return_value = profile
+
+        proxy_server, client_manager = self._active_proxy_server(am_i_owner=False)
+        call_order = []
+
+        def _add_client(*_args, **_kwargs):
+            call_order.append("add_client")
+            return 1
+
+        def _ensure_output_profile(*_args, **_kwargs):
+            call_order.append("ensure_output_profile")
+            return True
+
+        client_manager.add_client.side_effect = _add_client
+        proxy_server.ensure_output_profile.side_effect = _ensure_output_profile
+        mock_proxy_cls.get_instance.return_value = proxy_server
+        mock_get_stream_object.return_value = self._channel()
+        mock_create_generator.return_value = lambda: iter([b"chunk"])
+
+        from apps.proxy.live_proxy.views import stream_ts
+
+        response = stream_ts(self._request(), self.channel_id)
+
+        self.assertIsInstance(response, StreamingHttpResponse)
+        self.assertEqual(call_order, ["add_client", "ensure_output_profile"])
+
+    @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.ChannelService.is_channel_unavailable_for_new_clients",
+        return_value=False,
+    )
+    @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_active_channel_without_client_manager_returns_503(
+        self,
+        mock_proxy_cls,
+        _network_ok,
+        mock_get_stream_object,
+        _unavailable,
+        _output_profile,
+        _output_format,
+        mock_create_generator,
+        _mock_close,
+    ):
+        proxy_server, _client_manager = self._active_proxy_server()
+        proxy_server.client_managers = {}
+        proxy_server.initialize_channel.return_value = True
+        proxy_server.redis_client.hmget.return_value = (
+            b"http://example/stream",
+            b"stream-ua",
+            b"None",
+        )
+        mock_proxy_cls.get_instance.return_value = proxy_server
+        mock_get_stream_object.return_value = self._channel()
+        mock_create_generator.return_value = lambda: iter([b"chunk"])
+
+        from apps.proxy.live_proxy.views import stream_ts
+
+        response = stream_ts(self._request(), self.channel_id)
+
+        self.assertIsInstance(response, JsonResponse)
+        self.assertEqual(response.status_code, 503)
+        proxy_server.ensure_output_profile.assert_not_called()
+
+    @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")
+    @patch(
+        "apps.proxy.live_proxy.views.ChannelService.is_channel_unavailable_for_new_clients",
+        return_value=False,
+    )
+    @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_ensure_output_profile_failure_removes_registered_client(
+        self,
+        mock_proxy_cls,
+        _network_ok,
+        mock_get_stream_object,
+        _unavailable,
+        mock_resolve_output_profile,
+        _output_format,
+        mock_create_generator,
+        _mock_close,
+    ):
+        profile = MagicMock()
+        profile.id = 2
+        profile.build_command.return_value = ["ffmpeg"]
+        mock_resolve_output_profile.return_value = profile
+
+        proxy_server, client_manager = self._active_proxy_server(am_i_owner=False)
+        proxy_server.ensure_output_profile.return_value = False
+        mock_proxy_cls.get_instance.return_value = proxy_server
+        mock_get_stream_object.return_value = self._channel()
+        mock_create_generator.return_value = lambda: iter([b"chunk"])
+
+        from apps.proxy.live_proxy.views import stream_ts
+
+        response = stream_ts(self._request(), self.channel_id)
+
+        self.assertIsInstance(response, JsonResponse)
+        self.assertEqual(response.status_code, 500)
+        client_manager.add_client.assert_called_once()
+        client_manager.remove_client.assert_called_once()
+
+    @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")
+    @patch(
+        "apps.proxy.live_proxy.views.ChannelService.is_channel_unavailable_for_new_clients",
+        return_value=False,
+    )
+    @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_unhandled_exception_after_registration_removes_client(
+        self,
+        mock_proxy_cls,
+        _network_ok,
+        mock_get_stream_object,
+        _unavailable,
+        mock_resolve_output_profile,
+        _output_format,
+        mock_create_generator,
+        _mock_close,
+    ):
+        """A crash between pre-registration and the streaming response (e.g. a
+        Redis error inside ensure_output_profile/get_buffer) must not leave a
+        phantom client registered - the generic exception handler has to undo
+        the earlier add_client() call."""
+        profile = MagicMock()
+        profile.id = 2
+        profile.build_command.return_value = ["ffmpeg"]
+        mock_resolve_output_profile.return_value = profile
+
+        proxy_server, client_manager = self._active_proxy_server(am_i_owner=False)
+        proxy_server.ensure_output_profile.side_effect = RuntimeError("redis down")
+        mock_proxy_cls.get_instance.return_value = proxy_server
+        mock_get_stream_object.return_value = self._channel()
+        mock_create_generator.return_value = lambda: iter([b"chunk"])
+
+        from apps.proxy.live_proxy.views import stream_ts
+
+        response = stream_ts(self._request(), self.channel_id)
+
+        self.assertIsInstance(response, JsonResponse)
+        self.assertEqual(response.status_code, 500)
+        client_manager.add_client.assert_called_once()
+        client_manager.remove_client.assert_called_once()
+
+    @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.ChannelService.is_channel_unavailable_for_new_clients",
+        return_value=False,
+    )
+    @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_client_manager_removed_after_registration_cleans_up(
+        self,
+        mock_proxy_cls,
+        _network_ok,
+        mock_get_stream_object,
+        _unavailable,
+        _output_profile,
+        _output_format,
+        mock_create_generator,
+        _mock_close,
+    ):
+        proxy_server, client_manager = self._active_proxy_server(am_i_owner=False)
+
+        def _pop_manager_after_register(*_args, **_kwargs):
+            proxy_server.client_managers.pop(self.channel_id, None)
+            return 1
+
+        client_manager.add_client.side_effect = _pop_manager_after_register
+        mock_proxy_cls.get_instance.return_value = proxy_server
+        mock_get_stream_object.return_value = self._channel()
+        mock_create_generator.return_value = lambda: iter([b"chunk"])
+
+        from apps.proxy.live_proxy.views import stream_ts
+
+        response = stream_ts(self._request(), self.channel_id)
+
+        self.assertIsInstance(response, JsonResponse)
+        self.assertEqual(response.status_code, 503)
+        client_manager.remove_client.assert_not_called()
+        proxy_server.redis_client.srem.assert_called_once()
+        proxy_server.redis_client.delete.assert_called_once()
+
+    @patch("apps.proxy.live_proxy.views.close_old_connections")
+    @patch("apps.proxy.live_proxy.views.create_stream_generator")
+    @patch("apps.proxy.live_proxy.views.generate_stream_url")
+    @patch("apps.proxy.live_proxy.views.ChannelService.initialize_channel", return_value=True)
+    @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.ChannelService.is_channel_unavailable_for_new_clients",
+        return_value=False,
+    )
+    @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_owner_init_resolves_output_profile_once(
+        self,
+        mock_proxy_cls,
+        _network_ok,
+        mock_get_stream_object,
+        _unavailable,
+        mock_resolve_output_profile,
+        mock_resolve_output_format,
+        _mock_initialize,
+        mock_generate_stream_url,
+        mock_create_generator,
+        _mock_close,
+    ):
+        mock_generate_stream_url.return_value = (
+            "http://example/stream",
+            "ua",
+            False,
+            "None",
+            True,
+            None,
+        )
+
+        proxy_server, client_manager = self._active_proxy_server(am_i_owner=True)
+        proxy_server.redis_client.exists.return_value = False
+        proxy_server.redis_client.get.return_value = None
+        proxy_server.check_if_channel_exists.return_value = False
+        proxy_server.redis_client.hgetall.return_value = {}
+        mock_proxy_cls.get_instance.return_value = proxy_server
+        mock_get_stream_object.return_value = self._channel()
+        mock_create_generator.return_value = lambda: iter([b"chunk"])
+
+        from apps.proxy.live_proxy.views import stream_ts
+
+        response = stream_ts(self._request(), self.channel_id)
+
+        self.assertIsInstance(response, StreamingHttpResponse)
+        mock_resolve_output_profile.assert_called_once()
+        mock_resolve_output_format.assert_called_once()
+        client_manager.add_client.assert_called_once()
+
+    @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.ChannelService.is_channel_unavailable_for_new_clients",
+        return_value=False,
+    )
+    @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_existing_db_cleanup_test_still_registers_client(
+        self,
+        mock_proxy_cls,
+        _network_ok,
+        mock_get_stream_object,
+        _unavailable,
+        _output_profile,
+        _output_format,
+        mock_create_generator,
+        _mock_close,
+    ):
+        proxy_server, client_manager = self._active_proxy_server()
+        mock_proxy_cls.get_instance.return_value = proxy_server
+        mock_get_stream_object.return_value = self._channel()
+        mock_create_generator.return_value = lambda: iter([b"chunk"])
+
+        from apps.proxy.live_proxy.views import stream_ts
+
+        response = stream_ts(self._request(), self.channel_id)
+
+        self.assertIsInstance(response, StreamingHttpResponse)
+        client_manager.add_client.assert_called_once()
+        mock_create_generator.assert_called_once()

apps/proxy/live_proxy/views.py

@@ -53,6 +53,18 @@ def _channel_stopping_response():
     return response
 
 
+def _drop_pre_registered_client(proxy_server, channel_id, client_id):
+    """Undo an early add_client() when setup aborts before streaming starts."""
+    mgr = proxy_server.client_managers.get(channel_id)
+    if mgr:
+        mgr.remove_client(client_id)
+        return
+    if not proxy_server.redis_client:
+        return
+    proxy_server.redis_client.srem(RedisKeys.clients(channel_id), client_id)
+    proxy_server.redis_client.delete(RedisKeys.client_metadata(channel_id, client_id))
+
+
 def _resolve_output_format(user, force=None, request=None):
     """Return the output format string to use for this client."""
     _FORMAT_ALIASES = {
@@ -110,6 +122,9 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
     client_user_agent = None
     proxy_server = ProxyServer.get_instance()
     connection_allocated = False  # Track if connection slot was allocated via get_stream()
+    # Initialized before the try so the exception handler can always safely
+    # check/clean it up, regardless of where in the setup a failure occurs.
+    _client_pre_registered = False
 
     try:
         # Generate a unique client ID
@@ -208,7 +223,9 @@ 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
+        resolved_output_profile = None
+        resolved_output_format = None
+        output_options_resolved = False
 
         # Start initialization if needed
         if needs_initialization or not proxy_server.check_if_channel_exists(channel_id):
@@ -478,18 +495,28 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
             # longer than the grace period). The generator handles waiting
             # with keepalive packets via _wait_for_initialization().
             if proxy_server.am_i_owner(channel_id):
-                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,
+                resolved_output_profile = _resolve_output_profile(request, user)
+                resolved_output_format = _resolve_output_format(user, force_output_format, request)
+                output_options_resolved = True
+                resolved_format = (
+                    f'{resolved_output_format}:p{resolved_output_profile.id}'
+                    if resolved_output_profile else resolved_output_format
                 )
+                client_manager = proxy_server.client_managers[channel_id]
+                if not client_manager.add_client(
+                    client_id, client_ip, client_user_agent, user,
+                    output_format=resolved_output_format,
+                    output_profile_id=resolved_output_profile.id if resolved_output_profile else None,
+                ):
+                    logger.error(
+                        f"[{client_id}] Failed to register client with channel {channel_id} during init"
+                    )
+                    return JsonResponse(
+                        {"error": "Failed to register client"}, status=503
+                    )
                 logger.info(
                     f"[{client_id}] Client registered with channel {channel_id} "
-                    f"(output: {resolved_format}, profile: {output_profile.id if output_profile else None})"
+                    f"(output: {resolved_format}, profile: {resolved_output_profile.id if resolved_output_profile else None})"
                 )
                 _client_pre_registered = True
 
@@ -560,53 +587,84 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
 
         if ChannelService.is_channel_unavailable_for_new_clients(channel_id):
             if _client_pre_registered:
-                mgr = proxy_server.client_managers.get(channel_id)
-                if mgr:
-                    mgr.remove_client(client_id)
+                _drop_pre_registered_client(proxy_server, channel_id, client_id)
             logger.info(
                 f"[{client_id}] Channel {channel_id} became unavailable during setup, rejecting"
             )
             return _channel_stopping_response()
 
-        # Register client
-        output_profile = _resolve_output_profile(request, user)
-        if output_profile:
-            cmd = output_profile.build_command()
-            if not proxy_server.ensure_output_profile(channel_id, output_profile.id, cmd):
+        if not output_options_resolved:
+            resolved_output_profile = _resolve_output_profile(request, user)
+            resolved_output_format = _resolve_output_format(user, force_output_format, request)
+        # 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'{resolved_output_format}:p{resolved_output_profile.id}'
+            if resolved_output_profile else resolved_output_format
+        )
+
+        # Pre-register before slow setup (ensure_output_profile) so the non-owner
+        # cleanup thread does not tear down local resources while connecting.
+        if not _client_pre_registered:
+            client_manager = proxy_server.client_managers.get(channel_id)
+            if not client_manager:
+                logger.error(
+                    f"[{client_id}] Channel {channel_id} missing client_manager during setup"
+                )
+                return JsonResponse(
+                    {"error": "Channel resources unavailable"}, status=503
+                )
+            if not client_manager.add_client(
+                client_id, client_ip, client_user_agent, user,
+                output_format=resolved_output_format,
+                output_profile_id=resolved_output_profile.id if resolved_output_profile else None,
+            ):
+                logger.error(
+                    f"[{client_id}] Failed to register client with channel {channel_id}"
+                )
+                return JsonResponse(
+                    {"error": "Failed to register client"}, status=503
+                )
+            _client_pre_registered = True
+            logger.info(
+                f"[{client_id}] Client registered with channel {channel_id} "
+                f"(output: {resolved_format}, profile: {resolved_output_profile.id if resolved_output_profile else None})"
+            )
+
+        if resolved_output_profile:
+            cmd = resolved_output_profile.build_command()
+            if not proxy_server.ensure_output_profile(channel_id, resolved_output_profile.id, cmd):
                 if _client_pre_registered:
-                    mgr = proxy_server.client_managers.get(channel_id)
-                    if mgr:
-                        mgr.remove_client(client_id)
+                    _drop_pre_registered_client(proxy_server, channel_id, client_id)
                 return JsonResponse(
                     {"error": "Failed to start output profile transcode"}, status=500
                 )
 
         source_buffer = proxy_server.get_buffer(
             channel_id,
-            profile=output_profile.id if output_profile else None
+            profile=resolved_output_profile.id if resolved_output_profile else None
         )
-        client_manager = proxy_server.client_managers[channel_id]
-
-        output_format = _resolve_output_format(user, force_output_format, request)
-        # 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
-        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,
+        client_manager = proxy_server.client_managers.get(channel_id)
+        if not client_manager:
+            if _client_pre_registered:
+                _drop_pre_registered_client(proxy_server, channel_id, client_id)
+            logger.error(
+                f"[{client_id}] Channel {channel_id} client_manager removed during setup"
             )
-            logger.info(
-                f"[{client_id}] Client registered with channel {channel_id} "
-                f"(output: {resolved_format}, profile: {output_profile.id if output_profile else None})"
+            return JsonResponse(
+                {"error": "Channel resources unavailable"}, status=503
             )
 
-        if output_format == 'fmp4':
-            proxy_server.ensure_output_format(
+        if resolved_output_format == 'fmp4':
+            if not proxy_server.ensure_output_format(
                 channel_id, resolved_format,
-                source_buffer=source_buffer if output_profile else None,
-            )
+                source_buffer=source_buffer if resolved_output_profile else None,
+            ):
+                if _client_pre_registered:
+                    _drop_pre_registered_client(proxy_server, channel_id, client_id)
+                return JsonResponse(
+                    {"error": "Failed to start output format remux"}, status=500
+                )
             generate = create_fmp4_stream_generator(
                 channel_id, client_id, client_ip, client_user_agent, channel_initializing, user=user,
                 fmt=resolved_format,
@@ -635,6 +693,15 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
                     logger.warning(f"[{client_id}] Failed to release stream in exception handler")
             except Exception:
                 pass
+        # Client may have been pre-registered (before ensure_output_profile /
+        # get_buffer / generator setup) to protect against the non-owner
+        # cleanup thread. If setup then failed with an unhandled exception,
+        # remove it so it doesn't linger as a phantom connection.
+        if _client_pre_registered:
+            try:
+                _drop_pre_registered_client(proxy_server, channel_id, client_id)
+            except Exception:
+                logger.warning(f"[{client_id}] Failed to remove client during exception cleanup")
         return JsonResponse({"error": str(e)}, status=500)