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

fix(proxy): enhance channel teardown and client management

Browse source 
- Improved channel teardown process to prevent lingering upstream connections and ensure proper resource cleanup.
- Enhanced client management by implementing checks for ghost clients and ensuring accurate client disconnection handling.
- Updated logging to provide clearer insights during channel initialization and teardown, including handling of unavailable channels.
- Refined stream manager behavior to manage ownership transitions more effectively and prevent blocking during shutdown.
This commit is contained in:
SergeantPanda 2026-06-14 21:41:50 -05:00
parent 9393f72cc1
commit 67239d921d
10 changed files with 945 additions and 302 deletions
Download patch file Download diff file Expand all files Collapse all files

4
CHANGELOG.md
View file

@ -43,9 +43,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
- **Live proxy channels could remain running with no clients after disconnect.** `stop_channel` called `log_system_event('channel_stop')` synchronously before local cleanup and Redis key deletion; Connect integrations or DB event writes on that path could block teardown indefinitely while the cleanup thread kept refreshing metadata TTL for a stale `stream_buffers` entry. Teardown now stops ffmpeg/output managers, releases ownership, removes local buffers and managers, deletes Redis keys, and only then logs the stop event asynchronously. Metadata TTL refresh skips channels mid-shutdown and non-owned channels; a stuck-stop watchdog forces cleanup if `stop_channel` does not return within ~30s. Stream buffers also ignore late `add_chunk()` writes once shutdown begins (`buffer.stopping` set at the start of `stream_manager.stop()`).
- **Live proxy channels could remain running with no clients after disconnect.** `stop_channel` called `log_system_event('channel_stop')` synchronously before local cleanup and Redis key deletion; Connect integrations or DB event writes on that path could block teardown indefinitely while the cleanup thread kept refreshing metadata TTL for a stale `stream_buffers` entry. Teardown now signals shutdown (`buffer.stopping`), runs model `release_stream()` and Redis key deletion, releases ownership, stops ffmpeg/output managers and local buffers, and only then logs the stop event asynchronously. Metadata TTL refresh skips channels mid-shutdown and non-owned channels; a stuck-stop watchdog forces cleanup if `stop_channel` does not return within ~10s (or 2× `channel_shutdown_delay`, whichever is greater). Stream buffers ignore late `add_chunk()` writes once shutdown begins (`buffer.stopping` set at teardown start).
- **Client reconnect during channel teardown could wedge the channel across uWSGI workers.** The disconnect path called `proxy_server.stop_channel()` directly, so other workers were not notified and reconnecting clients could attach while ownership was released on the owner worker. The cleanup thread then re-acquired expired ownership and looped on orphaned clients while the upstream connection stayed open. All lifecycle stops now go through coordinated Redis teardown (`channel_stopping` flag, metadata `stopping` state, and `CHANNEL_STOP` pubsub). New stream requests are rejected with `503 Retry-After` during pending shutdown or active teardown; `initialize_channel()` and ownership re-acquisition refuse to proceed in those states; non-owner workers clean local resources only on pubsub; and orphaned-client sweeps force cleanup when teardown is already active. (Fixes #1342)
- **Orphaned upstream threads kept writing Redis after teardown.** Partial cleanup could remove a channel from `stream_managers` while the `stream-{uuid}` OS thread kept running, so orphan metadata sweeps only deleted Redis keys and chunks/metadata were immediately recreated (bare `total_bytes` hash with TTL -1). Stream managers now stop upstream when ownership is lost, orphan cleanup stops local ffmpeg/stream threads even when registry entries are gone (`_live_stream_managers` tracks managers until the thread exits), and forced recovery stops processes before Redis deletion instead of dropping dict entries without calling `stop()`.
- **Rapid channel switching could leave bare `buffer:index` keys in Redis.** `buffer.stop()` flushed a final partial chunk via `INCR` during teardown, creating a persistent index key (TTL -1) after chunk keys expired; under load overlapping disconnect and cleanup-thread stops could race on `_stop_local_stream_activity`, blocking on ffmpeg stderr join before `_clean_redis_keys` ever ran. Teardown no longer writes to Redis from `buffer.stop()`, disconnect uses coordinated stop only (no duplicate upstream stop), model `release_stream()` runs before Redis keys are scanned/deleted (so `profile_connections` counters are not left stuck), ownership is released after Redis cleanup but before the blocking local ffmpeg stop, and a `finally` block guarantees Redis cleanup if local teardown raises.
- **Zombie ffmpeg could survive owner-lock expiry and orphan Redis sweeps.** When the 30s owner lock lapsed under single-worker load, disconnect handling treated the worker as non-owner so coordinated stop never ran, while ffmpeg kept writing and the orphan sweeper only deleted Redis keys (recreating them immediately). Disconnect now re-acquires ownership when local upstream is still active, stops locally when the last client leaves without a lock, orphan cleanup stops local ffmpeg before Redis deletion via `_has_local_upstream_activity`, re-init stops lingering upstream before starting a duplicate thread, and `is_channel_teardown_active` includes channels mid-`stop_channel` on this worker so rapid reconnect gets 503 during teardown.
- **Stale `channel_stream` Redis keys after a channel stopped could skip connection accounting on retune.** On `dev`, `get_stream()` reused any existing `channel_stream` / `stream_profile` assignment without reserving a new slot. If those keys were left behind after a stop, the next tune-in could reach the provider without incrementing Redis counters. `get_stream()` now releases stale assignments when proxy metadata shows the channel is inactive, then reserves fresh slots.
- **EPG auto-match reliability fixes.**
- Memory could spike to multiple GB on large EPG sources when building a full in-memory catalog before fuzzy matching; single-channel matching now streams rows and bounds ML work to a small candidate set.

28
apps/channels/models.py
View file

@ -626,11 +626,11 @@ class Channel(models.Model):
def _release_stale_stream_assignment(self, redis_client, stream_id: int) -> None:
"""Release pool counters and remove stale channel/stream assignment keys."""
profile_id = None
profile_id_bytes = redis_client.get(f"stream_profile:{stream_id}")
if profile_id_bytes:
try:
profile_id = int(profile_id_bytes)
release_profile_slot(profile_id, redis_client)
except (ValueError, TypeError):
logger.debug(
"Invalid profile ID for stale assignment on stream %s: %s",
@ -638,6 +638,32 @@ class Channel(models.Model):
profile_id_bytes,
)
if profile_id is None:
metadata_key = RedisKeys.channel_metadata(str(self.uuid))
meta_profile_id = redis_client.hget(
metadata_key, ChannelMetadataField.M3U_PROFILE
)
if meta_profile_id:
try:
profile_id = int(meta_profile_id)
except (ValueError, TypeError):
logger.debug(
"Invalid profile ID in metadata for stale assignment on "
"channel %s: %s",
self.uuid,
meta_profile_id,
)
if profile_id is not None:
release_profile_slot(profile_id, redis_client)
else:
logger.warning(
"Channel %s: releasing stale stream %s assignment without profile "
"info - profile_connections may leak",
self.uuid,
stream_id,
)
redis_client.delete(f"channel_stream:{self.id}")
redis_client.delete(f"stream_profile:{stream_id}")

452
apps/channels/tests/test_ts_proxy_teardown.py
View file

@ -5,6 +5,7 @@ from unittest.mock import MagicMock, patch
from django.test import TestCase
from apps.proxy.live_proxy.constants import ChannelMetadataField, ChannelState
from apps.proxy.live_proxy.input.buffer import StreamBuffer
from apps.proxy.live_proxy.input.manager import StreamManager
from apps.proxy.live_proxy.redis_keys import RedisKeys
from apps.proxy.live_proxy.server import ProxyServer
@ -14,6 +15,29 @@ from apps.proxy.live_proxy.services.channel_service import ChannelService
CHANNEL_ID = "00000000-0000-0000-0000-000000000099"
def _configure_ownership_pipeline(
redis,
*,
stop_exists=0,
metadata_exists=1,
client_count=0,
owner=None,
disconnect=None,
state=None,
):
pipe = MagicMock()
redis.pipeline.return_value = pipe
pipe.execute.return_value = (
stop_exists,
metadata_exists,
client_count,
owner,
disconnect,
state,
)
return pipe
def _mock_proxy_server(redis_client=None):
server = MagicMock()
server.redis_client = redis_client or MagicMock()
@ -79,11 +103,6 @@ class LocalStreamActivityTests(TestCase):
server.stop_all_output_profiles = MagicMock()
return server
def test_has_local_stream_activity_from_live_registry(self):
server = self._make_server()
server._live_stream_managers[CHANNEL_ID] = MagicMock()
self.assertTrue(server._has_local_stream_activity(CHANNEL_ID))
@patch.object(ProxyServer, "_join_stream_thread")
def test_stop_local_stream_activity_stops_live_manager(self, mock_join):
server = self._make_server()
@ -114,9 +133,9 @@ class OrphanMetadataCleanupTests(TestCase):
@patch.object(ProxyServer, "_clean_redis_keys")
@patch.object(ProxyServer, "_stop_local_stream_activity")
@patch.object(ProxyServer, "_has_local_stream_activity", return_value=True)
@patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
def test_orphan_metadata_stops_local_processes_before_redis(
self, mock_has_local, mock_stop_local, mock_clean_redis
self, mock_has_upstream, mock_stop_local, mock_clean_redis
):
server = self._make_server()
metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
@ -130,15 +149,16 @@ class OrphanMetadataCleanupTests(TestCase):
server._check_orphaned_metadata()
mock_has_local.assert_called_with(CHANNEL_ID)
mock_has_upstream.assert_called_with(CHANNEL_ID)
mock_stop_local.assert_called_once_with(CHANNEL_ID)
mock_clean_redis.assert_called_once_with(CHANNEL_ID)
@patch.object(ProxyServer, "_clean_redis_keys")
@patch.object(ProxyServer, "_broadcast_upstream_stop")
@patch.object(ProxyServer, "_stop_local_stream_activity")
@patch.object(ProxyServer, "_has_local_stream_activity", return_value=False)
def test_orphan_metadata_remote_channel_only_cleans_redis(
self, mock_has_local, mock_stop_local, mock_clean_redis
@patch.object(ProxyServer, "_has_local_upstream_activity", return_value=False)
def test_orphan_metadata_remote_channel_broadcasts_stop(
self, mock_has_upstream, mock_stop_local, mock_broadcast, mock_clean_redis
):
server = self._make_server()
metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
@ -149,6 +169,7 @@ class OrphanMetadataCleanupTests(TestCase):
server._check_orphaned_metadata()
mock_broadcast.assert_called_once_with(CHANNEL_ID)
mock_stop_local.assert_not_called()
mock_clean_redis.assert_called_once_with(CHANNEL_ID)
@ -171,9 +192,9 @@ class OrphanChannelCleanupTests(TestCase):
@patch.object(ProxyServer, "_clean_redis_keys")
@patch.object(ProxyServer, "_stop_local_stream_activity")
@patch.object(ProxyServer, "_has_local_stream_activity", return_value=True)
@patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
def test_orphan_channel_stops_local_before_redis(
self, mock_has_local, mock_stop_local, mock_clean_redis
self, mock_has_upstream, mock_stop_local, mock_clean_redis
):
server = self._make_server()
metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
@ -189,10 +210,10 @@ class OrphanChannelCleanupTests(TestCase):
class StreamManagerOwnershipTests(TestCase):
def test_still_owner_false_when_different_worker(self):
buffer = MagicMock()
buffer.redis_client.get.side_effect = lambda key: (
None if "stopping" in key else b"worker-b"
_configure_ownership_pipeline(
buffer.redis_client,
owner=b"worker-b",
)
buffer.redis_client.exists.return_value = False
manager = StreamManager(
CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
)
@ -201,8 +222,12 @@ class StreamManagerOwnershipTests(TestCase):
def test_still_owner_true_when_owner_lock_expired_but_not_stopping(self):
buffer = MagicMock()
buffer.redis_client.get.return_value = None
buffer.redis_client.exists.return_value = False
_configure_ownership_pipeline(
buffer.redis_client,
client_count=0,
owner=None,
state=ChannelState.CONNECTING.encode(),
)
manager = StreamManager(
CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
)
@ -211,7 +236,10 @@ class StreamManagerOwnershipTests(TestCase):
def test_still_owner_false_when_channel_stopping_key_set(self):
buffer = MagicMock()
buffer.redis_client.exists.return_value = True
_configure_ownership_pipeline(
buffer.redis_client,
stop_exists=1,
)
manager = StreamManager(
CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
)
@ -220,8 +248,10 @@ class StreamManagerOwnershipTests(TestCase):
def test_update_bytes_skipped_after_ownership_lost(self):
buffer = MagicMock()
buffer.redis_client.get.return_value = b"other-worker"
buffer.redis_client.exists.return_value = False
_configure_ownership_pipeline(
buffer.redis_client,
owner=b"other-worker",
)
manager = StreamManager(
CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
)
@ -230,3 +260,383 @@ class StreamManagerOwnershipTests(TestCase):
manager._update_bytes_processed(500)
buffer.redis_client.hincrby.assert_not_called()
class StreamBufferStopTests(TestCase):
def test_stop_discards_local_data_without_redis_writes(self):
redis = MagicMock()
buffer = StreamBuffer(channel_id=CHANNEL_ID, redis_client=redis)
buffer._write_buffer = bytearray(b"x" * 376)
buffer.stop()
self.assertTrue(buffer.stopping)
self.assertEqual(len(buffer._write_buffer), 0)
redis.incr.assert_not_called()
redis.setex.assert_not_called()
redis.delete.assert_not_called()
class StopChannelTeardownTests(TestCase):
def _make_server(self):
with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
server = ProxyServer()
server.worker_id = "testhost:1"
server.stream_managers = {}
server.stream_buffers = {}
server.client_managers = {}
server.profile_managers = {}
server.profile_buffers = {}
server._live_stream_managers = {}
server._stopping_channels = set()
server._stopping_since = {}
server.redis_client = MagicMock()
server.redis_client.exists.return_value = False
server.am_i_owner = MagicMock(return_value=False)
server._collect_channel_stop_event_data = MagicMock(return_value=None)
server.release_ownership = MagicMock()
return server
@patch.object(ProxyServer, "_spawn_channel_stop_event")
@patch.object(ProxyServer, "_clean_redis_keys")
@patch.object(ProxyServer, "_stop_local_stream_activity")
def test_stop_channel_cleans_redis_before_blocking_local_stop(
self, mock_stop_local, mock_clean_redis, mock_spawn_event
):
server = self._make_server()
call_order = []
def stop_local(channel_id):
call_order.append("local")
def clean_redis(channel_id):
call_order.append("redis")
mock_stop_local.side_effect = stop_local
mock_clean_redis.side_effect = clean_redis
server.stop_channel(CHANNEL_ID)
self.assertEqual(call_order, ["redis", "local"])
mock_spawn_event.assert_called_once_with(None)
@patch.object(ProxyServer, "_spawn_channel_stop_event")
@patch.object(ProxyServer, "_clean_redis_keys")
@patch.object(ProxyServer, "_stop_local_stream_activity", side_effect=RuntimeError("boom"))
def test_stop_channel_cleans_redis_in_finally_when_local_stop_fails(
self, mock_stop_local, mock_clean_redis, mock_spawn_event
):
server = self._make_server()
result = server.stop_channel(CHANNEL_ID)
self.assertFalse(result)
mock_clean_redis.assert_called_once_with(CHANNEL_ID)
mock_spawn_event.assert_not_called()
self.assertNotIn(CHANNEL_ID, server._stopping_channels)
@patch.object(ProxyServer, "_spawn_channel_stop_event")
@patch.object(ProxyServer, "_clean_redis_keys")
@patch.object(ProxyServer, "_stop_local_stream_activity")
def test_stop_channel_owner_releases_after_redis_cleanup(
self, mock_stop_local, mock_clean_redis, mock_spawn_event
):
server = self._make_server()
server.am_i_owner.return_value = True
stop_data = {"channel_id": CHANNEL_ID}
server._collect_channel_stop_event_data.return_value = stop_data
call_order = []
def stop_local(channel_id):
call_order.append("local")
def release(channel_id):
call_order.append("release")
def clean_redis(channel_id):
call_order.append("redis")
mock_stop_local.side_effect = stop_local
server.release_ownership.side_effect = release
mock_clean_redis.side_effect = clean_redis
server.stop_channel(CHANNEL_ID)
self.assertEqual(call_order, ["redis", "release", "local"])
server._collect_channel_stop_event_data.assert_called_once_with(CHANNEL_ID)
mock_spawn_event.assert_called_once_with(stop_data)
class CleanRedisKeysOrderTests(TestCase):
@patch("apps.proxy.live_proxy.server.Stream.objects.get")
@patch("apps.proxy.live_proxy.server.Channel.objects.get")
def test_clean_redis_keys_releases_profile_slot_before_live_keys_deleted(
self, mock_channel_get, mock_stream_get
):
from apps.channels.models import Channel, Stream
with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
server = ProxyServer()
server.redis_client = MagicMock()
call_order = []
channel = MagicMock()
channel.release_stream.return_value = True
def channel_get(uuid):
call_order.append("release")
return channel
mock_channel_get.side_effect = channel_get
mock_stream_get.side_effect = Stream.DoesNotExist
def scan(*args, **kwargs):
call_order.append("redis")
return (0, [b"live:channel:foo:buffer:index"])
server.redis_client.scan.side_effect = scan
server._clean_redis_keys(CHANNEL_ID)
self.assertEqual(call_order, ["release", "redis"])
channel.release_stream.assert_called_once()
server.redis_client.delete.assert_called_once()
class LocalUpstreamActivityTests(TestCase):
def _make_server(self):
with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
server = ProxyServer()
server.worker_id = "testhost:1"
server.stream_managers = {}
server.stream_buffers = {}
server.client_managers = {}
server._live_stream_managers = {}
return server
def test_upstream_activity_excludes_reader_only_client_manager(self):
server = self._make_server()
server.client_managers[CHANNEL_ID] = MagicMock()
server.stream_buffers[CHANNEL_ID] = MagicMock()
self.assertFalse(server._has_local_upstream_activity(CHANNEL_ID))
def test_upstream_activity_from_live_registry(self):
server = self._make_server()
server._live_stream_managers[CHANNEL_ID] = MagicMock()
self.assertTrue(server._has_local_upstream_activity(CHANNEL_ID))
class ClientDisconnectOwnershipTests(TestCase):
def test_last_client_triggers_stop_when_upstream_active_without_owner_lock(self):
from apps.proxy.live_proxy.client_manager import ClientManager
proxy_server = MagicMock()
proxy_server.worker_id = "testhost:1"
proxy_server.am_i_owner.return_value = False
proxy_server._has_local_upstream_activity.return_value = True
proxy_server.extend_ownership.return_value = False
redis = MagicMock()
redis.hget.return_value = b"viewer"
redis.scard.return_value = 0
manager = ClientManager(
channel_id=CHANNEL_ID,
redis_client=redis,
worker_id="testhost:1",
)
manager.proxy_server = proxy_server
manager.clients = {"client-1"}
manager.client_set_key = RedisKeys.clients(CHANNEL_ID)
manager._notify_owner_of_activity = MagicMock()
manager._trigger_stats_update = MagicMock()
manager.get_total_client_count = MagicMock(return_value=0)
proxy_server._spawn_on_hub = MagicMock()
manager.remove_client("client-1")
proxy_server._spawn_on_hub.assert_called_once_with(
proxy_server.handle_client_disconnect, CHANNEL_ID
)
class TeardownActiveLocalStopTests(TestCase):
@patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance")
def test_teardown_active_when_local_stop_in_progress(self, mock_get_instance):
server = MagicMock()
server._stopping_channels = {CHANNEL_ID}
server.redis_client = MagicMock()
server.redis_client.exists.return_value = False
mock_get_instance.return_value = server
self.assertTrue(ChannelService.is_channel_teardown_active(CHANNEL_ID))
class HandleClientDisconnectUpstreamFirstTests(TestCase):
def _make_server(self):
with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
server = ProxyServer()
server.worker_id = "testhost:1"
server.client_managers = {CHANNEL_ID: MagicMock()}
server.stream_managers = {CHANNEL_ID: MagicMock()}
server._live_stream_managers = {}
server.profile_managers = {}
server.output_managers = {}
server.redis_client = MagicMock()
server.redis_client.scard.return_value = 0
server._stopping_channels = set()
return server
@patch.object(ProxyServer, "_coordinated_stop_channel")
@patch.object(ProxyServer, "_stop_upstream_before_redis_cleanup")
@patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
def test_last_client_uses_coordinated_stop_only(
self, _mock_upstream, mock_stop_upstream, mock_coordinated
):
server = self._make_server()
with patch(
"apps.proxy.live_proxy.server.ConfigHelper.channel_shutdown_delay",
return_value=0,
):
server.handle_client_disconnect(CHANNEL_ID)
mock_stop_upstream.assert_not_called()
mock_coordinated.assert_called_once_with(CHANNEL_ID)
class InitWaitAbortTests(TestCase):
def _make_generator(self):
from apps.proxy.live_proxy.output.ts.generator import StreamGenerator
return StreamGenerator(
CHANNEL_ID,
"client-1",
"127.0.0.1",
"test-agent",
channel_initializing=True,
)
def test_abort_when_client_removed_locally(self):
generator = self._make_generator()
server = MagicMock()
client_manager = MagicMock()
client_manager.clients = set()
server.client_managers = {CHANNEL_ID: client_manager}
server.redis_client = MagicMock()
self.assertEqual(generator._init_wait_abort_reason(server, time.time()), "client_gone")
def test_abort_when_connect_stalled_without_buffer(self):
from apps.proxy.config import TSConfig as Config
generator = self._make_generator()
server = MagicMock()
client_manager = MagicMock()
client_manager.clients = {"client-1"}
server.client_managers = {CHANNEL_ID: client_manager}
buffer = MagicMock()
buffer.index = 0
server.stream_buffers = {CHANNEL_ID: buffer}
server.redis_client = MagicMock()
server.redis_client.hget.return_value = b"connecting"
server.redis_client.get.return_value = None
started = time.time() - (getattr(Config, "CONNECTION_TIMEOUT", 10) + 1)
self.assertEqual(generator._init_wait_abort_reason(server, started), "stalled")
class UpstreamStopBroadcastTests(TestCase):
def _make_server(self):
with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
server = ProxyServer()
server.worker_id = "testhost:1"
server.stream_managers = {}
server._live_stream_managers = {}
server.redis_client = MagicMock()
return server
@patch.object(ProxyServer, "_stop_local_stream_activity")
@patch.object(ProxyServer, "_broadcast_upstream_stop")
@patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
def test_local_upstream_stops_locally_without_broadcast(
self, _mock_has, mock_broadcast, mock_stop_local
):
server = self._make_server()
server._stop_upstream_before_redis_cleanup(CHANNEL_ID)
mock_stop_local.assert_called_once_with(CHANNEL_ID)
mock_broadcast.assert_not_called()
@patch.object(ProxyServer, "_stop_local_stream_activity")
@patch.object(ProxyServer, "_broadcast_upstream_stop")
@patch.object(ProxyServer, "_has_local_upstream_activity", return_value=False)
def test_orphan_cleanup_broadcasts_when_no_local_upstream(
self, _mock_has, mock_broadcast, mock_stop_local
):
server = self._make_server()
server._stop_upstream_before_redis_cleanup(CHANNEL_ID)
mock_broadcast.assert_called_once_with(CHANNEL_ID)
mock_stop_local.assert_not_called()
class StreamManagerStillOwnerTests(TestCase):
def _make_manager(self, redis_client):
buffer = MagicMock()
buffer.redis_client = redis_client
buffer.channel_id = CHANNEL_ID
manager = StreamManager(
CHANNEL_ID,
"http://example/stream.ts",
buffer,
worker_id="testhost:1",
)
return manager
def test_stops_when_metadata_removed(self):
redis = MagicMock()
_configure_ownership_pipeline(redis, metadata_exists=0)
manager = self._make_manager(redis)
self.assertFalse(manager._still_owner())
def test_keeps_running_during_connecting_before_client_registered(self):
redis = MagicMock()
_configure_ownership_pipeline(
redis,
client_count=0,
owner=b"testhost:1",
state=ChannelState.CONNECTING.encode(),
)
manager = self._make_manager(redis)
self.assertTrue(manager._still_owner())
def test_stops_after_disconnect_when_shutdown_delay_is_zero(self):
redis = MagicMock()
_configure_ownership_pipeline(
redis,
client_count=0,
owner=b"testhost:1",
disconnect=b"1700000000.0",
state=ChannelState.ACTIVE.encode(),
)
manager = self._make_manager(redis)
with patch(
"apps.proxy.live_proxy.input.manager.ConfigHelper.channel_shutdown_delay",
return_value=0,
):
self.assertFalse(manager._still_owner())
def test_keeps_running_during_shutdown_delay(self):
redis = MagicMock()
_configure_ownership_pipeline(
redis,
client_count=0,
owner=b"testhost:1",
disconnect=str(time.time()).encode(),
state=ChannelState.ACTIVE.encode(),
)
manager = self._make_manager(redis)
with patch(
"apps.proxy.live_proxy.input.manager.ConfigHelper.channel_shutdown_delay",
return_value=5,
):
self.assertTrue(manager._still_owner())

116
apps/proxy/live_proxy/client_manager.py
View file

@ -102,6 +102,7 @@ class ClientManager:
break
# Send heartbeat for all local clients
clients_to_remove = set()
with self.lock:
# Skip this cycle if we have no local clients
if not self.clients:
@ -109,7 +110,6 @@ class ClientManager:
# IMPROVED GHOST DETECTION: Check for stale clients before sending heartbeats
current_time = time.time()
clients_to_remove = set()
# First identify clients that should be removed
for client_id in self.clients:
@ -132,12 +132,11 @@ class ClientManager:
logger.debug(f"Client {client_id} inactive for {current_time - last_active_time:.1f}s, removing as ghost")
clients_to_remove.add(client_id)
# Remove ghost clients in a separate step
for client_id in clients_to_remove:
self.remove_client(client_id)
if clients_to_remove:
logger.info(f"Removed {len(clients_to_remove)} ghost clients from channel {self.channel_id}")
logger.info(
f"Identified {len(clients_to_remove)} ghost clients "
f"for channel {self.channel_id}"
)
# Now send heartbeats only for remaining clients
pipe = self.redis_client.pipeline()
@ -172,6 +171,9 @@ class ClientManager:
if self.clients and not all(c in clients_to_remove for c in self.clients):
self._notify_owner_of_activity()
for client_id in clients_to_remove:
self.remove_client(client_id)
except Exception as e:
logger.error(f"Error in client heartbeat thread: {e}")
@ -305,6 +307,9 @@ class ClientManager:
def remove_client(self, client_id):
"""Remove a client from this channel and Redis"""
client_username = "unknown"
remaining = None
with self.lock:
if client_id in self.clients:
self.clients.remove(client_id)
@ -315,61 +320,80 @@ class ClientManager:
self.last_active_time = time.time()
if self.redis_client:
# Get client data before removing the data
client_key = f"live:channel:{self.channel_id}:clients:{client_id}"
client_username = self.redis_client.hget(client_key, "username") or "unknown"
if isinstance(client_username, bytes):
client_username = client_username.decode("utf-8")
# Remove from channel's client set
self.redis_client.srem(self.client_set_key, client_id)
client_key = f"live:channel:{self.channel_id}:clients:{client_id}"
self.redis_client.delete(client_key)
# Check if this was the last client
remaining = self.redis_client.scard(self.client_set_key) or 0
if remaining == 0:
logger.warning(f"Last client removed: {client_id} - channel may shut down soon")
# Trigger disconnect time tracking even if we're not the owner
disconnect_key = RedisKeys.last_client_disconnect(self.channel_id)
self.redis_client.setex(disconnect_key, 60, str(time.time()))
local_count = len(self.clients)
self._notify_owner_of_activity()
if not self.redis_client:
return local_count
# Check if we're the owner - if so, handle locally; if not, publish event
am_i_owner = self.proxy_server and self.proxy_server.am_i_owner(self.channel_id)
if remaining == 0:
logger.warning(f"Last client removed: {client_id} - channel may shut down soon")
disconnect_key = RedisKeys.last_client_disconnect(self.channel_id)
self.redis_client.setex(disconnect_key, 60, str(time.time()))
if am_i_owner:
# We're the owner - handle the disconnect directly
logger.debug(f"Owner handling CLIENT_DISCONNECTED for client {client_id} locally (not publishing)")
if (remaining == 0
or self.proxy_server.output_managers.get(self.channel_id)
or self.proxy_server.profile_managers.get(self.channel_id)):
import gevent
gevent.spawn(self.proxy_server.handle_client_disconnect, self.channel_id)
else:
# We're not the owner - publish event so owner can handle it
logger.debug(f"Non-owner publishing CLIENT_DISCONNECTED event for client {client_id} on channel {self.channel_id} from worker {self.worker_id}")
event_data = json.dumps({
"event": EventType.CLIENT_DISCONNECTED,
"channel_id": self.channel_id,
"client_id": client_id,
"worker_id": self.worker_id or "unknown",
"timestamp": time.time(),
"remaining_clients": remaining,
"username": client_username
})
self.redis_client.publish(RedisKeys.events_channel(self.channel_id), event_data)
self._notify_owner_of_activity()
# Trigger channel stats update via WebSocket
self._trigger_stats_update()
am_i_owner = self.proxy_server and self.proxy_server.am_i_owner(self.channel_id)
total_clients = self.get_total_client_count()
logger.info(f"Client disconnected: {client_id} (local: {len(self.clients)}, total: {total_clients})")
# Owner lock TTL can expire while local ffmpeg is still running on this worker.
if (not am_i_owner and self.proxy_server
and self.proxy_server._has_local_upstream_activity(self.channel_id)):
if self.proxy_server.extend_ownership(self.channel_id):
am_i_owner = True
return len(self.clients)
schedule_disconnect = False
if am_i_owner:
logger.debug(
f"Owner handling CLIENT_DISCONNECTED for client {client_id} locally (not publishing)"
)
if (remaining == 0
or self.proxy_server.output_managers.get(self.channel_id)
or self.proxy_server.profile_managers.get(self.channel_id)):
schedule_disconnect = True
elif (remaining == 0 and self.proxy_server
and self.proxy_server._has_local_upstream_activity(self.channel_id)):
logger.warning(
f"Last client removed while local upstream still active for "
f"{self.channel_id} without owner lock - stopping channel"
)
schedule_disconnect = True
else:
logger.debug(
f"Non-owner publishing CLIENT_DISCONNECTED event for client {client_id} "
f"on channel {self.channel_id} from worker {self.worker_id}"
)
event_data = json.dumps({
"event": EventType.CLIENT_DISCONNECTED,
"channel_id": self.channel_id,
"client_id": client_id,
"worker_id": self.worker_id or "unknown",
"timestamp": time.time(),
"remaining_clients": remaining,
"username": client_username
})
self.redis_client.publish(RedisKeys.events_channel(self.channel_id), event_data)
if schedule_disconnect and self.proxy_server:
self.proxy_server._spawn_on_hub(
self.proxy_server.handle_client_disconnect, self.channel_id
)
self._trigger_stats_update()
total_clients = self.get_total_client_count()
logger.info(
f"Client disconnected: {client_id} (local: {local_count}, total: {total_clients})"
)
return local_count
def get_client_count(self):
"""Get local client count"""

42
apps/proxy/live_proxy/input/buffer.py
View file

@ -309,38 +309,16 @@ class StreamBuffer:
self.fill_timers.clear()
try:
# Flush any remaining data in the write buffer
if hasattr(self, '_write_buffer') and len(self._write_buffer) > 0:
# Ensure remaining data is aligned to TS packets
complete_size = (len(self._write_buffer) // 188) * 188
if complete_size > 0:
final_chunk = self._write_buffer[:complete_size]
# Write final chunk to Redis
with self.lock:
if self.redis_client:
try:
chunk_index = self.redis_client.incr(self.buffer_index_key)
chunk_key = f"{self.buffer_prefix}{chunk_index}"
self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(final_chunk))
self.index = chunk_index
logger.info(f"Flushed final chunk of {len(final_chunk)} bytes to Redis")
except Exception as e:
logger.error(f"Error flushing final chunk: {e}")
# Clear buffers
self._write_buffer = bytearray()
if hasattr(self, '_partial_packet'):
self._partial_packet = bytearray()
# Clean up the chunk timestamps sorted set
if self.redis_client and self.chunk_timestamps_key:
try:
self.redis_client.delete(self.chunk_timestamps_key)
except Exception as e:
logger.error(f"Error deleting chunk timestamps key: {e}")
with self.lock:
if hasattr(self, '_write_buffer') and len(self._write_buffer) > 0:
discarded = len(self._write_buffer)
self._write_buffer = bytearray()
if hasattr(self, '_partial_packet'):
self._partial_packet = bytearray()
logger.debug(
f"Discarded {discarded} bytes from local write buffer "
f"for channel {self.channel_id}"
)
except Exception as e:
logger.error(f"Error during buffer stop: {e}")

160
apps/proxy/live_proxy/input/manager.py
View file

@ -112,6 +112,11 @@ class StreamManager:
# Add tracking for data throughput
self.bytes_processed = 0
self.last_bytes_update = time.time()
# Cached result of the pipelined Redis ownership audit (hot read path).
self._ownership_cache_valid_until = 0.0
self._ownership_cached = True
self._OWNERSHIP_CHECK_INTERVAL = 1.0
self.bytes_update_interval = 5 # Update Redis every 5 seconds
# Add stderr reader thread property
@ -182,7 +187,90 @@ class StreamManager:
logger.warning(f"Timeout waiting for existing processes to close for channel {self.channel_id} after {timeout}s")
return False
def _still_owner(self):
def _invalidate_ownership_cache(self):
self._ownership_cache_valid_until = 0.0
@staticmethod
def _decode_redis_value(value):
if value is None:
return None
if isinstance(value, bytes):
return value.decode()
return value
def _disconnect_shutdown_ready(self, disconnect_value):
"""True when last-client disconnect has passed the configured shutdown delay."""
if not disconnect_value:
return False
shutdown_delay = ConfigHelper.channel_shutdown_delay()
if shutdown_delay <= 0:
return True
disconnect_value = self._decode_redis_value(disconnect_value)
try:
disconnect_time = float(disconnect_value)
except (ValueError, TypeError):
return False
return (time.time() - disconnect_time) >= shutdown_delay
def _evaluate_ownership_from_redis(self, redis_client):
"""Single pipelined Redis round-trip for the full ownership audit."""
stop_key = RedisKeys.channel_stopping(self.channel_id)
metadata_key = RedisKeys.channel_metadata(self.channel_id)
clients_key = RedisKeys.clients(self.channel_id)
owner_key = RedisKeys.channel_owner(self.channel_id)
disconnect_key = RedisKeys.last_client_disconnect(self.channel_id)
pipe = redis_client.pipeline(transaction=False)
pipe.exists(stop_key)
pipe.exists(metadata_key)
pipe.scard(clients_key)
pipe.get(owner_key)
pipe.get(disconnect_key)
pipe.hget(metadata_key, ChannelMetadataField.STATE)
(
stop_exists,
metadata_exists,
client_count,
current_owner,
disconnect_value,
state_raw,
) = pipe.execute()
if stop_exists:
return False
if not metadata_exists:
logger.warning(
f"Channel {self.channel_id} metadata removed from Redis - stopping upstream"
)
return False
client_count = client_count or 0
state = self._decode_redis_value(state_raw)
current_owner = self._decode_redis_value(current_owner)
if current_owner and current_owner != self.worker_id:
return False
if not current_owner:
if client_count == 0 and state not in ChannelState.PRE_ACTIVE:
logger.warning(
f"Channel {self.channel_id} has no owner and no clients - stopping upstream"
)
return False
return True
if client_count == 0 and self._disconnect_shutdown_ready(disconnect_value):
logger.info(
f"Channel {self.channel_id} disconnect shutdown ready - stopping upstream"
)
return False
return True
def _still_owner(self, *, force=False):
"""Return True while this worker should keep the upstream connection open."""
if self.stopping or self.stop_requested:
return False
@ -194,26 +282,49 @@ class StreamManager:
if not redis_client:
return True
try:
stop_key = RedisKeys.channel_stopping(self.channel_id)
if redis_client.exists(stop_key):
return False
now = time.time()
if not force and now < self._ownership_cache_valid_until:
return self._ownership_cached
owner_key = RedisKeys.channel_owner(self.channel_id)
current_owner = redis_client.get(owner_key)
if not current_owner:
# Owner lock TTL may lapse briefly between cleanup-thread renewals.
# Keep running unless coordinated teardown has started (checked above).
return True
if isinstance(current_owner, bytes):
current_owner = current_owner.decode()
return current_owner == self.worker_id
try:
result = self._evaluate_ownership_from_redis(redis_client)
self._ownership_cached = result
self._ownership_cache_valid_until = now + self._OWNERSHIP_CHECK_INTERVAL
return result
except Exception as e:
logger.debug(f"Ownership check failed for channel {self.channel_id}: {e}")
return True
def _upstream_may_continue(self):
"""
Per-chunk gate for the hot read path.
Local stop flags are checked every chunk. The coordinated-teardown Redis
flag is checked every chunk (one EXISTS). The full ownership audit is
pipelined and cached for ~1s — enough for loop boundaries while avoiding
6+ Redis round-trips per chunk during steady streaming.
"""
if self.stopping or self.stop_requested or not self.running:
return False
if self.buffer is not None and self.buffer.stopping:
return False
redis_client = getattr(self.buffer, 'redis_client', None)
if redis_client and self.worker_id:
try:
if redis_client.exists(RedisKeys.channel_stopping(self.channel_id)):
self._ownership_cached = False
self._ownership_cache_valid_until = 0.0
return False
except Exception as e:
logger.debug(
f"Channel stopping check failed for {self.channel_id}: {e}"
)
return self._still_owner()
def _ensure_owner_or_stop(self):
if self._still_owner():
if self._still_owner(force=True):
return True
logger.warning(
@ -539,6 +650,7 @@ class StreamManager:
logger.debug(f"Closing existing HTTP connections before establishing transcode connection for channel {self.channel_id}")
self._close_connection()
close_old_connections()
channel = get_stream_object(self.channel_id)
# Use FFmpeg specifically for HLS streams
@ -1071,7 +1183,7 @@ class StreamManager:
def _update_bytes_processed(self, chunk_size):
"""Update the total bytes processed in Redis metadata"""
if not self._still_owner():
if not self._upstream_may_continue():
return
try:
@ -1146,17 +1258,11 @@ class StreamManager:
"""Stop the stream manager and cancel all timers"""
logger.info(f"Stopping stream manager for channel {self.channel_id}")
# Add at the beginning of your stop method
self.stopping = True
self._invalidate_ownership_cache()
if self.buffer is not None:
self.buffer.stopping = True
# Release stream resources if we're the owner
if self.current_stream_id and hasattr(self, 'worker_id') and self.worker_id:
if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
owner_key = RedisKeys.channel_owner(self.channel_id)
current_owner = self.buffer.redis_client.get(owner_key)
# Cancel all buffer check timers
for timer in list(self._buffer_check_timers):
try:
@ -1545,7 +1651,8 @@ class StreamManager:
if hasattr(self, 'stderr_reader_thread') and self.stderr_reader_thread and self.stderr_reader_thread.is_alive():
try:
logger.debug(f"Waiting for stderr reader thread to terminate for channel {self.channel_id}")
self.stderr_reader_thread.join(timeout=2.0)
stderr_join_timeout = 0.25 if self.stopping else 2.0
self.stderr_reader_thread.join(timeout=stderr_join_timeout)
if self.stderr_reader_thread.is_alive():
logger.warning(f"Stderr reader thread did not terminate within timeout for channel {self.channel_id}")
except Exception as e:
@ -1638,7 +1745,7 @@ class StreamManager:
self.connected = False
return False
if not self._still_owner():
if not self._upstream_may_continue():
self.stop()
return False
@ -1649,8 +1756,7 @@ class StreamManager:
# Add directly to buffer without TS-specific processing
success = self.buffer.add_chunk(chunk)
# Update last data timestamp in Redis if successful
if success and self._still_owner() and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
if success and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
last_data_key = RedisKeys.last_data(self.buffer.channel_id)
self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60)

107
apps/proxy/live_proxy/output/ts/generator.py
View file

@ -139,12 +139,75 @@ class StreamGenerator:
finally:
self._cleanup()
def _init_wait_abort_reason(self, proxy_server, initialization_start):
"""Return a reason string when init wait should end early, else None."""
from ...services.channel_service import ChannelService
if ChannelService.is_channel_teardown_active(self.channel_id):
return "stopping"
client_mgr = proxy_server.client_managers.get(self.channel_id)
if client_mgr is not None:
if self.client_id not in client_mgr.clients:
return "client_gone"
elif proxy_server.redis_client:
client_key = RedisKeys.client_metadata(self.channel_id, self.client_id)
if not proxy_server.redis_client.exists(client_key):
total = proxy_server.redis_client.scard(RedisKeys.clients(self.channel_id)) or 0
if total == 0:
return "client_gone"
elapsed = time.time() - initialization_start
connection_timeout = getattr(Config, 'CONNECTION_TIMEOUT', 10)
if elapsed < connection_timeout or not proxy_server.redis_client:
return None
metadata_key = RedisKeys.channel_metadata(self.channel_id)
state_raw = proxy_server.redis_client.hget(metadata_key, 'state')
if not state_raw:
return None
state = state_raw.decode() if isinstance(state_raw, bytes) else state_raw
if state not in ('connecting', 'initializing', 'buffering'):
return None
buffer = proxy_server.stream_buffers.get(self.channel_id)
if buffer is not None and buffer.index > 0:
return None
if proxy_server.redis_client.get(RedisKeys.last_data(self.channel_id)):
return None
return "stalled"
def _wait_for_initialization(self):
initialization_start = time.time()
max_init_wait = ConfigHelper.client_wait_timeout()
proxy_server = ProxyServer.get_instance()
while time.time() - initialization_start < max_init_wait:
abort_reason = self._init_wait_abort_reason(proxy_server, initialization_start)
if abort_reason == "stopping":
logger.error(
f"[{self.client_id}] Channel {self.channel_id} teardown active during initialization"
)
yield create_ts_packet('error', "Error: Channel is stopping")
return False
if abort_reason == "client_gone":
logger.info(
f"[{self.client_id}] Client disconnected during initialization wait, aborting"
)
yield create_ts_packet('error', "Error: Client disconnected")
return False
if abort_reason == "stalled":
logger.warning(
f"[{self.client_id}] Channel {self.channel_id} stalled in connecting state "
f"with no buffer data after "
f"{getattr(Config, 'CONNECTION_TIMEOUT', 10)}s, aborting init wait"
)
yield create_ts_packet('error', "Error: Connection stalled")
return False
if proxy_server.redis_client:
metadata_key = RedisKeys.channel_metadata(self.channel_id)
metadata = proxy_server.redis_client.hgetall(metadata_key)
@ -537,35 +600,31 @@ class StreamGenerator:
stream_released = False
if proxy_server.redis_client:
try:
metadata_key = RedisKeys.channel_metadata(self.channel_id)
metadata = proxy_server.redis_client.hgetall(metadata_key)
if metadata:
stream_id_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID)
if stream_id_bytes:
# Check if we're the last client
if self.channel_id in proxy_server.client_managers:
client_count = proxy_server.client_managers[self.channel_id].get_total_client_count()
# Only the last client or owner should release the stream
if client_count <= 1 and proxy_server.am_i_owner(self.channel_id):
try:
# Try Channel first (normal flow), fall back to Stream (preview flow)
try:
obj = Channel.objects.get(uuid=self.channel_id)
except (Channel.DoesNotExist, Exception):
obj = Stream.objects.get(stream_hash=self.channel_id)
stream_released = obj.release_stream()
if stream_released:
logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}")
else:
logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}")
except Exception as e:
logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}")
if self.channel_id in proxy_server.client_managers:
client_count = proxy_server.client_managers[self.channel_id].get_total_client_count()
# Pool slots are global; the last client on any worker must release.
if client_count <= 1:
try:
try:
obj = Channel.objects.get(uuid=self.channel_id)
except (Channel.DoesNotExist, Exception):
obj = Stream.objects.get(stream_hash=self.channel_id)
stream_released = obj.release_stream()
if stream_released:
logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}")
else:
logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}")
except Exception as e:
logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}")
except Exception as e:
logger.error(f"[{self.client_id}] Error checking stream data for release: {e}")
if self.channel_id in proxy_server.client_managers:
client_manager = proxy_server.client_managers[self.channel_id]
local_clients = client_manager.remove_client(self.client_id)
if self.client_id in client_manager.clients:
local_clients = client_manager.remove_client(self.client_id)
else:
local_clients = client_manager.get_client_count()
total_clients = client_manager.get_total_client_count()
logger.info(f"[{self.client_id}] Disconnected after {elapsed:.2f}s (local: {local_clients}, total: {total_clients})")

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

@ -72,6 +72,7 @@ class ProxyServer:
self._channel_names = {}
self._stopping_channels = set() # channels with an active stop_channel call in progress
self._stopping_since = {} # channel_id -> time.time() when stop_channel began
self._local_stop_locks = {}
# Managers kept until the stream OS thread exits (may outlive stream_managers dict)
self._live_stream_managers = {}
@ -141,6 +142,18 @@ class ProxyServer:
logger.error(f"Redis command error: {e}")
return None
def _spawn_on_hub(self, fn, *args, **kwargs):
"""Schedule fn on the gevent hub from any thread without blocking the caller."""
import gevent
try:
hub = gevent.get_hub()
hub.loop.run_callback_threadsafe(
lambda: gevent.spawn(fn, *args, **kwargs)
)
except Exception as e:
logger.error(f"Failed to schedule {getattr(fn, '__name__', fn)} on hub: {e}")
def _start_event_listener(self):
"""Listen for events from other workers"""
if not self.redis_client:
@ -507,9 +520,9 @@ class ProxyServer:
current = self.redis_client.get(lock_key)
if current is None:
# Key expired, re-acquire if we have the stream_manager, but never
# Key expired, re-acquire if we still run local upstream, but never
# during coordinated teardown (multi-worker reconnect-during-stop race).
if channel_id in self.stream_managers:
if channel_id in self.stream_managers or channel_id in self._live_stream_managers:
if self._channel_unavailable_for_new_clients(channel_id):
logger.info(
f"Refusing ownership re-acquisition for {channel_id}; "
@ -546,6 +559,12 @@ class ProxyServer:
)
return False
if self._has_local_upstream_activity(channel_id):
logger.warning(
f"Stopping lingering local upstream before initializing channel {channel_id}"
)
self._stop_local_stream_activity(channel_id)
if self.redis_client:
metadata_key = RedisKeys.channel_metadata(channel_id)
if self.redis_client.exists(metadata_key):
@ -879,26 +898,7 @@ class ProxyServer:
owner = metadata.get('owner', 'unknown')
logger.info(f"Zombie channel details - state: {state}, owner: {owner}")
# Clean up Redis keys
self._clean_redis_keys(channel_id)
# Force release resources in the Channel model
try:
channel = Channel.objects.get(uuid=channel_id)
if not channel.release_stream():
logger.warning(f"Failed to release stream for zombie channel {channel_id}")
else:
logger.info(f"Released stream allocation for zombie channel {channel_id}")
except Exception as e:
try:
stream = Stream.objects.get(stream_hash=channel_id)
if not stream.release_stream():
logger.warning(f"Failed to release stream for zombie channel {channel_id}")
else:
logger.info(f"Released stream allocation for zombie channel {channel_id}")
except Exception as e:
logger.error(f"Error releasing stream for zombie channel {channel_id}: {e}")
return True
except Exception as e:
logger.error(f"Error cleaning zombie channel {channel_id}: {e}", exc_info=True)
@ -914,8 +914,10 @@ class ProxyServer:
# to manage for this channel at all.
if (channel_id not in self.client_managers
and channel_id not in self.stream_managers
and channel_id not in self._live_stream_managers
and channel_id not in self.profile_managers
and channel_id not in self.output_managers):
and channel_id not in self.output_managers
and not self._has_local_upstream_activity(channel_id)):
return
try:
@ -987,12 +989,12 @@ class ProxyServer:
if total == 0:
logger.debug(f"No clients left after disconnect event - stopping channel {channel_id}")
disconnect_key = RedisKeys.last_client_disconnect(channel_id)
self.redis_client.setex(disconnect_key, 60, str(time.time()))
shutdown_delay = ConfigHelper.channel_shutdown_delay()
disconnect_key = RedisKeys.last_client_disconnect(channel_id)
if shutdown_delay > 0:
self.redis_client.setex(disconnect_key, 60, str(time.time()))
logger.info(f"Waiting {shutdown_delay}s before stopping channel...")
gevent.sleep(shutdown_delay)
@ -1002,6 +1004,13 @@ class ProxyServer:
self.redis_client.delete(disconnect_key)
return
if shutdown_delay <= 0:
self.redis_client.setex(disconnect_key, 60, str(time.time()))
# Coordinated stop runs local teardown + Redis cleanup once.
# Do not call _stop_upstream_before_redis_cleanup here — it races
# with stop_channel and can leave stop_channel blocked on stderr join
# before _clean_redis_keys ever runs (orphaned buffer:index keys).
self._coordinated_stop_channel(channel_id)
except Exception as e:
logger.error(f"Error handling client disconnect for channel {channel_id}: {e}")
@ -1371,18 +1380,41 @@ class ProxyServer:
return thread
return None
def _has_local_stream_activity(self, channel_id):
def _has_local_upstream_activity(self, channel_id):
"""True when this worker runs a local ffmpeg / stream-{uuid} OS thread."""
if channel_id in self.stream_managers:
return True
if channel_id in self._live_stream_managers:
return True
if channel_id in self.stream_buffers:
return True
if channel_id in self.client_managers:
return True
thread = self._get_stream_thread(channel_id)
return thread is not None and thread.is_alive()
def _broadcast_upstream_stop(self, channel_id):
"""Ask every uWSGI worker to stop local ffmpeg for this channel."""
from .services.channel_service import ChannelService
if not self.redis_client:
return
try:
ChannelService.mark_channel_stopping(channel_id, broadcast=True)
logger.info(
f"Broadcast upstream stop for channel {channel_id} to all workers"
)
except Exception as e:
logger.error(
f"Error broadcasting upstream stop for channel {channel_id}: {e}"
)
def _stop_upstream_before_redis_cleanup(self, channel_id):
"""Stop local ffmpeg before deleting Redis keys (prevents delete/recreate loops)."""
if self._has_local_upstream_activity(channel_id):
logger.info(
f"Channel {channel_id} has local upstream activity - stopping processes"
)
self._stop_local_stream_activity(channel_id)
return
self._broadcast_upstream_stop(channel_id)
def _join_stream_thread(self, channel_id, timeout=2.0):
thread = self._get_stream_thread(channel_id)
if thread and thread.is_alive():
@ -1402,8 +1434,41 @@ class ProxyServer:
manager = self._live_stream_managers.pop(channel_id, None)
return manager
def _signal_upstream_shutdown(self, channel_id):
"""Halt upstream Redis writes immediately without blocking on ffmpeg join."""
if channel_id in self.stream_buffers:
self.stream_buffers[channel_id].stopping = True
for registry in (self.stream_managers, self._live_stream_managers):
manager = registry.get(channel_id)
if manager is None:
continue
manager.stopping = True
manager.stop_requested = True
if getattr(manager, 'buffer', None) is not None:
manager.buffer.stopping = True
def _get_local_stop_lock(self, channel_id):
lock = self._local_stop_locks.get(channel_id)
if lock is None:
lock = threading.Lock()
self._local_stop_locks[channel_id] = lock
return lock
def _stop_local_stream_activity(self, channel_id):
"""Stop local ffmpeg/stream threads regardless of registry state."""
lock = self._get_local_stop_lock(channel_id)
if not lock.acquire(blocking=False):
logger.debug(
f"Local stop already in progress for channel {channel_id}, skipping duplicate"
)
return
try:
self._stop_local_stream_activity_locked(channel_id)
finally:
lock.release()
self._local_stop_locks.pop(channel_id, None)
def _stop_local_stream_activity_locked(self, channel_id):
stream_manager = self._resolve_stream_manager(channel_id)
if stream_manager is not None:
if hasattr(stream_manager, 'stop'):
@ -1445,17 +1510,13 @@ class ProxyServer:
self.profile_buffers.pop(channel_id, None)
self._live_stream_managers.pop(channel_id, None)
def _cleanup_stopped_channel_local(self, channel_id):
"""Remove local managers/buffers for a channel that is shutting down."""
self._stop_local_stream_activity(channel_id)
def _recover_stuck_channel_stops(self):
"""Force cleanup when stop_channel never finishes (e.g. blocked on logging)."""
if not self._stopping_channels:
return
now = time.time()
stuck_threshold = max(30, ConfigHelper.channel_shutdown_delay() * 2)
stuck_threshold = max(10, ConfigHelper.channel_shutdown_delay() * 2)
for channel_id in list(self._stopping_channels):
started = self._stopping_since.get(channel_id, now)
@ -1483,10 +1544,10 @@ class ProxyServer:
self._stopping_channels.add(channel_id)
self._stopping_since[channel_id] = time.time()
stop_event_data = None
redis_cleaned = False
try:
logger.info(f"Stopping channel {channel_id}")
# First set a stopping key that clients will check
if self.redis_client:
stop_key = RedisKeys.channel_stopping(channel_id)
if self.redis_client.exists(stop_key):
@ -1494,54 +1555,26 @@ class ProxyServer:
else:
self.redis_client.setex(stop_key, 60, "true")
# Only stop the actual stream manager if we're the owner
if self.am_i_owner(channel_id):
logger.info(f"This worker ({self.worker_id}) is the owner - closing provider connection")
if channel_id in self.stream_managers:
stream_manager = self.stream_managers[channel_id]
# Signal thread to stop and close resources
if hasattr(stream_manager, 'stop'):
stream_manager.stop()
else:
stream_manager.running = False
if hasattr(stream_manager, '_close_socket'):
stream_manager._close_socket()
# Wait for stream thread to finish
stream_thread_name = f"stream-{channel_id}"
stream_thread = None
for thread in threading.enumerate():
if thread.name == stream_thread_name:
stream_thread = thread
break
if stream_thread and stream_thread.is_alive():
logger.info(f"Waiting for stream thread to terminate")
try:
# Very short timeout to prevent hanging the app
stream_thread.join(timeout=2.0)
if stream_thread.is_alive():
logger.warning(f"Stream thread did not terminate within timeout")
except RuntimeError:
logger.debug(f"Could not join stream thread (may be current thread)")
# Stop all output format managers before releasing ownership
self.stop_all_output_formats(channel_id)
# Stop all output profile transcodes before releasing ownership
self.stop_all_output_profiles(channel_id)
was_owner = self.am_i_owner(channel_id)
if was_owner:
logger.info(
f"This worker ({self.worker_id}) is the owner - closing provider connection"
)
stop_event_data = self._collect_channel_stop_event_data(channel_id)
# Release ownership
# Stop new chunk writes before Redis cleanup; do not block on ffmpeg join yet.
self._signal_upstream_shutdown(channel_id)
# Release profile slots and delete Redis keys before any blocking local stop.
# A concurrent disconnect + cleanup-thread stop used to wedge here behind a
# 2s stderr join, never reaching scan/delete (bare buffer:index with TTL -1).
self._clean_redis_keys(channel_id)
redis_cleaned = True
if was_owner:
self.release_ownership(channel_id)
# Always clean up local resources before Redis/logging so teardown
# cannot be blocked by connect integrations or DB event writes.
self._cleanup_stopped_channel_local(channel_id)
self._clean_redis_keys(channel_id)
self._stop_local_stream_activity(channel_id)
self._spawn_channel_stop_event(stop_event_data)
return True
@ -1549,6 +1582,13 @@ class ProxyServer:
logger.error(f"Error stopping channel {channel_id}: {e}")
return False
finally:
if not redis_cleaned:
try:
self._clean_redis_keys(channel_id)
except Exception as e:
logger.error(
f"Error cleaning Redis keys for channel {channel_id} during finally: {e}"
)
self._stopping_channels.discard(channel_id)
self._stopping_since.pop(channel_id, None)
@ -1595,7 +1635,11 @@ class ProxyServer:
self._recover_stuck_channel_stops()
# Create a unified list of all channels we have locally
all_local_channels = set(self.stream_managers.keys()) | set(self.client_managers.keys())
all_local_channels = (
set(self.stream_managers.keys())
| set(self.client_managers.keys())
| set(self._live_stream_managers.keys())
)
# Single loop through all channels - process each exactly once
for channel_id in list(all_local_channels):
@ -1764,7 +1808,8 @@ class ProxyServer:
# === NON-OWNER CHANNEL HANDLING ===
# Safety: if we have a stream_manager, we ARE the real owner
# but the Redis key may have expired. Try to re-acquire.
if channel_id in self.stream_managers:
if (channel_id in self.stream_managers
or channel_id in self._live_stream_managers):
# Ownership was explicitly released by an active stop_channel call -
# don't fight the shutdown by trying to re-acquire.
if channel_id in self._stopping_channels:
@ -1930,11 +1975,7 @@ class ProxyServer:
# Orphaned channel with no clients - clean it up
logger.info(f"Cleaning up orphaned channel {channel_id}")
if self._has_local_stream_activity(channel_id):
logger.info(
f"Orphaned channel {channel_id} has local stream activity - stopping processes"
)
self._stop_local_stream_activity(channel_id)
self._stop_upstream_before_redis_cleanup(channel_id)
self._clean_redis_keys(channel_id)
except Exception as e:
logger.error(f"Error processing channel key {key}: {e}")
@ -1964,8 +2005,7 @@ class ProxyServer:
if not metadata:
# Empty metadata - clean it up
logger.warning(f"Found empty metadata for channel {channel_id} - cleaning up")
if self._has_local_stream_activity(channel_id):
self._stop_local_stream_activity(channel_id)
self._stop_upstream_before_redis_cleanup(channel_id)
self._clean_redis_keys(channel_id)
continue
@ -1987,11 +2027,7 @@ class ProxyServer:
state = metadata.get('state', 'unknown')
logger.warning(f"Found orphaned metadata for channel {channel_id} (state: {state}, owner: {owner}, clients: {client_count}) - cleaning up")
if self._has_local_stream_activity(channel_id):
logger.info(
f"Channel {channel_id} has local stream activity - stopping processes"
)
self._stop_local_stream_activity(channel_id)
self._stop_upstream_before_redis_cleanup(channel_id)
self._clean_redis_keys(channel_id)
elif not owner_alive and client_count > 0:
# SCARD may include ghost entries from a dead worker's
@ -2008,8 +2044,7 @@ class ProxyServer:
f"owner: {owner}) had {client_count} ghost client(s) "
f"- cleaning up"
)
if self._has_local_stream_activity(channel_id):
self._stop_local_stream_activity(channel_id)
self._stop_upstream_before_redis_cleanup(channel_id)
self._clean_redis_keys(channel_id)
else:
if self._channel_teardown_active(channel_id):
@ -2018,8 +2053,7 @@ class ProxyServer:
f"{real_count} client(s) during teardown; forcing cleanup"
)
self._force_clear_channel_clients(channel_id)
if self._has_local_stream_activity(channel_id):
self._stop_local_stream_activity(channel_id)
self._stop_upstream_before_redis_cleanup(channel_id)
self._clean_redis_keys(channel_id)
else:
logger.warning(
@ -2036,7 +2070,11 @@ class ProxyServer:
def _clean_redis_keys(self, channel_id):
"""Clean up all Redis keys for a channel more efficiently"""
# Release the channel, stream, and profile keys from the channel
total_deleted = 0
# Release the M3U profile slot while channel_stream / metadata still exist.
# Scanning live:channel keys first deletes metadata and breaks release_stream()
# fallback, leaving profile_connections counters stuck (e.g. profile_connections:70 = 1).
try:
channel = Channel.objects.get(uuid=channel_id)
if not channel.release_stream():
@ -2049,36 +2087,29 @@ class ProxyServer:
except (Stream.DoesNotExist, Exception):
logger.debug(f"No Channel or Stream found for {channel_id}")
if not self.redis_client:
return 0
if self.redis_client:
try:
patterns = [
f"live:channel:{channel_id}:*",
RedisKeys.events_channel(channel_id),
]
try:
# Define key patterns to scan for
patterns = [
f"live:channel:{channel_id}:*", # All channel keys
RedisKeys.events_channel(channel_id) # Event channel
]
for pattern in patterns:
cursor = 0
while True:
cursor, keys = self.redis_client.scan(cursor, match=pattern, count=100)
if keys:
self.redis_client.delete(*keys)
total_deleted += len(keys)
total_deleted = 0
if cursor == 0:
break
for pattern in patterns:
cursor = 0
while True:
cursor, keys = self.redis_client.scan(cursor, match=pattern, count=100)
if keys:
self.redis_client.delete(*keys)
total_deleted += len(keys)
logger.info(f"Cleaned up {total_deleted} Redis keys for channel {channel_id}")
except Exception as e:
logger.error(f"Error cleaning Redis keys for channel {channel_id}: {e}")
# Exit when cursor returns to 0
if cursor == 0:
break
logger.info(f"Cleaned up {total_deleted} Redis keys for channel {channel_id}")
return total_deleted
except Exception as e:
logger.error(f"Error cleaning Redis keys for channel {channel_id}: {e}")
return 0
return total_deleted
def refresh_channel_registry(self):
"""Refresh TTL for active channels using standard keys"""

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

@ -73,6 +73,9 @@ class ChannelService:
def is_channel_teardown_active(channel_id):
"""True when a coordinated channel stop is in progress (visible to all workers)."""
proxy_server = ProxyServer.get_instance()
if channel_id in proxy_server._stopping_channels:
return True
if not proxy_server.redis_client:
return False
@ -345,41 +348,19 @@ class ChannelService:
except Exception as e:
logger.error(f"Error fetching channel state: {e}")
# Mark stopping in Redis and notify all workers before local teardown
# Mark stopping in Redis and notify all workers before local teardown.
# stop_channel() releases profile slots via _clean_redis_keys() before Redis deletion.
if proxy_server.redis_client:
ChannelService.mark_channel_stopping(channel_id, broadcast=True)
logger.info(f"Marked channel {channel_id} stopping and broadcast stop to all workers")
local_result = proxy_server.stop_channel(channel_id)
else:
local_result = proxy_server.stop_channel(channel_id)
# Release the channel in the channel model if applicable
try:
channel = Channel.objects.get(uuid=channel_id)
model_released = channel.release_stream()
if model_released:
logger.info(f"Released channel {channel_id} stream allocation")
else:
logger.warning(f"Channel {channel_id}: release_stream found no keys to clean")
except (Channel.DoesNotExist, Exception):
logger.warning(f"Could not find Channel model for UUID {channel_id}, attempting stream hash")
try:
stream = Stream.objects.get(stream_hash=channel_id)
model_released = stream.release_stream()
if model_released:
logger.info(f"Released stream {channel_id} stream allocation")
else:
logger.warning(f"Stream {channel_id}: release_stream found no keys to clean")
except (Stream.DoesNotExist, Exception) as e:
logger.error(f"No Channel or Stream found for {channel_id}: {e}")
model_released = False
local_result = proxy_server.stop_channel(channel_id)
return {
'status': 'success',
'message': 'Channel stop request sent',
'channel_id': channel_id,
'previous_state': channel_info,
'model_released': model_released,
'model_released': bool(local_result),
'local_stop_result': local_result
}

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

@ -211,6 +211,12 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
# Start initialization if needed
if needs_initialization or not proxy_server.check_if_channel_exists(channel_id):
if ChannelService.is_channel_unavailable_for_new_clients(channel_id):
logger.info(
f"[{client_id}] Channel {channel_id} became unavailable before init, rejecting"
)
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 [
@ -433,6 +439,16 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
) # 502 Bad Gateway
# Initialize channel with the stream's user agent (not the client's)
if ChannelService.is_channel_unavailable_for_new_clients(channel_id):
if connection_allocated:
if not channel.release_stream():
logger.warning(f"[{client_id}] Failed to release stream before teardown reject")
connection_allocated = False
logger.info(
f"[{client_id}] Channel {channel_id} unavailable before init call, rejecting"
)
return _channel_stopping_response()
success = ChannelService.initialize_channel(
channel_id,
stream_url,
@ -541,6 +557,16 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
f"[{client_id}] Successfully initialized channel {channel_id} locally"
)
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)
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: