Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-20 01:59:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

fix(proxy): reap ghost channel sessions stuck in initializing

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Josh Becker 2026-07-11 21:50:33 -04:00
parent 2f60dc91ae
commit efad028be1
No known key found for this signature in database
GPG key ID: FF7794A2AFB08013
5 changed files with 520 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

5
CHANGELOG.md
View file

@ -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

157
apps/proxy/live_proxy/server.py
View file

@ -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)

6
apps/proxy/live_proxy/services/channel_service.py
View file

@ -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

346
apps/proxy/live_proxy/tests/test_ghost_session_cleanup.py Normal file
View file

@ -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()

24
apps/proxy/live_proxy/views.py
View file

@ -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,