From cf1aef67473188e09ce785d9bcaa3823aa806832 Mon Sep 17 00:00:00 2001 From: cmcpherson274 Date: Sun, 15 Mar 2026 12:06:32 -0400 Subject: [PATCH 1/7] Implement buffer reset method for stream transitions Added a method to reset internal buffers to prevent corruption during stream transitions. --- apps/proxy/ts_proxy/stream_buffer.py | 34 ++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/apps/proxy/ts_proxy/stream_buffer.py b/apps/proxy/ts_proxy/stream_buffer.py index 111468bf..42ceb2b7 100644 --- a/apps/proxy/ts_proxy/stream_buffer.py +++ b/apps/proxy/ts_proxy/stream_buffer.py @@ -132,6 +132,40 @@ class StreamBuffer: logger.error(f"Error adding chunk to buffer: {e}") return False + def reset_buffer_position(self): + """ + Reset internal buffers for a clean stream transition (failover). + + Called by stream_manager.update_url() when switching between FFmpeg + processes. Without this, _partial_packet from the old FFmpeg gets + concatenated with the first bytes from the new FFmpeg, creating + corrupted TS packets that break audio decoder sync in the client. + """ + try: + with self.lock: + old_write_size = len(self._write_buffer) + old_partial_size = len(getattr(self, '_partial_packet', b'')) + + self._write_buffer = bytearray() + if hasattr(self, '_partial_packet'): + self._partial_packet = bytearray() + + if old_write_size > 0 or old_partial_size > 0: + logger.info( + f"Reset buffer position for channel {self.channel_id}: " + f"cleared {old_write_size} bytes from write buffer, " + f"{old_partial_size} bytes from partial packet" + ) + else: + logger.debug( + f"Reset buffer position for channel {self.channel_id}: " + f"buffers were already clean" + ) + except Exception as e: + logger.error( + f"Error resetting buffer position for channel {self.channel_id}: {e}" + ) + def get_chunks(self, start_index=None): """Get chunks from the buffer with detailed logging""" try: From fd9ca283169c1eb18455424dcecf9f0d1001e653 Mon Sep 17 00:00:00 2001 From: cmcpherson274 Date: Sun, 15 Mar 2026 12:08:32 -0400 Subject: [PATCH 2/7] Modify client TTL handling in client_manager.py Updated client manager to refresh TTL without updating last_active timestamp. --- apps/proxy/ts_proxy/client_manager.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/apps/proxy/ts_proxy/client_manager.py b/apps/proxy/ts_proxy/client_manager.py index 836f719e..a89f543b 100644 --- a/apps/proxy/ts_proxy/client_manager.py +++ b/apps/proxy/ts_proxy/client_manager.py @@ -158,9 +158,8 @@ class ClientManager: if time_since_heartbeat < self.heartbeat_interval * 0.5: # Only heartbeat at half interval minimum continue - # Only update clients that remain + # Only refresh TTL - do NOT update last_active client_key = f"ts_proxy:channel:{self.channel_id}:clients:{client_id}" - pipe.hset(client_key, "last_active", str(current_time)) pipe.expire(client_key, self.client_ttl) # Keep client in the set with TTL From f7085e5dc2782e3273ea7b7ac97d5d1e58088ad9 Mon Sep 17 00:00:00 2001 From: cmcpherson274 Date: Sun, 15 Mar 2026 12:10:38 -0400 Subject: [PATCH 3/7] Add 'last_active' field to stats in stream_generator --- apps/proxy/ts_proxy/stream_generator.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index f0174616..95a8143b 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -428,7 +428,8 @@ class StreamGenerator: ChannelMetadataField.BYTES_SENT: str(self.bytes_sent), ChannelMetadataField.AVG_RATE_KBPS: str(round(avg_rate, 1)), ChannelMetadataField.CURRENT_RATE_KBPS: str(round(self.current_rate, 1)), - ChannelMetadataField.STATS_UPDATED_AT: str(current_time) + ChannelMetadataField.STATS_UPDATED_AT: str(current_time), + "last_active": str(current_time) } proxy_server.redis_client.hset(client_key, mapping=stats) @@ -614,4 +615,4 @@ def create_stream_generator(channel_id, client_id, client_ip, client_user_agent, Returns a function that can be passed to StreamingHttpResponse. """ generator = StreamGenerator(channel_id, client_id, client_ip, client_user_agent, channel_initializing) - return generator.generate \ No newline at end of file + return generator.generate From 10ea165c2cde2a7009f7a75ae62425f655c13bfa Mon Sep 17 00:00:00 2001 From: cmcpherson274 Date: Sun, 15 Mar 2026 14:04:59 -0400 Subject: [PATCH 4/7] Refactor packet handling with locking mechanism --- apps/proxy/ts_proxy/stream_buffer.py | 39 ++++++++++++++-------------- 1 file changed, 20 insertions(+), 19 deletions(-) diff --git a/apps/proxy/ts_proxy/stream_buffer.py b/apps/proxy/ts_proxy/stream_buffer.py index 42ceb2b7..b61e7794 100644 --- a/apps/proxy/ts_proxy/stream_buffer.py +++ b/apps/proxy/ts_proxy/stream_buffer.py @@ -76,27 +76,28 @@ class StreamBuffer: if not hasattr(self, '_partial_packet'): self._partial_packet = bytearray() - # Combine with any previous partial packet - combined_data = bytearray(self._partial_packet) + bytearray(chunk) - - # Calculate complete packets - complete_packets_size = (len(combined_data) // self.TS_PACKET_SIZE) * self.TS_PACKET_SIZE - - if complete_packets_size == 0: - # Not enough data for a complete packet - self._partial_packet = combined_data - return True - - # Split into complete packets and remainder - complete_packets = combined_data[:complete_packets_size] - self._partial_packet = combined_data[complete_packets_size:] - - # Add completed packets to write buffer - self._write_buffer.extend(complete_packets) - - # Only write to Redis when we have enough data for an optimized chunk + # Lock the full operation to prevent race with reset_buffer_position writes_done = 0 with self.lock: + # Combine with any previous partial packet + combined_data = bytearray(self._partial_packet) + bytearray(chunk) + + # Calculate complete packets + complete_packets_size = (len(combined_data) // self.TS_PACKET_SIZE) * self.TS_PACKET_SIZE + + if complete_packets_size == 0: + # Not enough data for a complete packet + self._partial_packet = combined_data + return True + + # Split into complete packets and remainder + complete_packets = combined_data[:complete_packets_size] + self._partial_packet = combined_data[complete_packets_size:] + + # Add completed packets to write buffer + self._write_buffer.extend(complete_packets) + + # Only write to Redis when we have enough data for an optimized chunk while len(self._write_buffer) >= self.target_chunk_size: # Extract a full chunk chunk_data = self._write_buffer[:self.target_chunk_size] From c8b665dc985ae32d9a8f516f4d72e67603c5e26e Mon Sep 17 00:00:00 2001 From: cmcpherson274 Date: Sun, 15 Mar 2026 14:08:43 -0400 Subject: [PATCH 5/7] Update last_active timestamp in keepalive logic Update last active timestamp for clients during failover. --- apps/proxy/ts_proxy/stream_generator.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 95a8143b..cfd705c9 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -312,6 +312,11 @@ class StreamGenerator: self.bytes_sent += len(keepalive_packet) self.last_yield_time = time.time() self.consecutive_empty = 0 # Reset consecutive counter but keep total empty_reads + # Update last_active so clients waiting during failover aren't flagged as ghosts + proxy_server = ProxyServer.get_instance() + if proxy_server and proxy_server.redis_client: + client_key = RedisKeys.client_metadata(self.channel_id, self.client_id) + proxy_server.redis_client.hset(client_key, "last_active", str(time.time())) gevent.sleep(Config.KEEPALIVE_INTERVAL) # Replace time.sleep else: # Standard wait with backoff From 40d0bc95677b474f671dac2f4bfff905ee76cf1f Mon Sep 17 00:00:00 2001 From: None <190783071+CodeBormen@users.noreply.github.com> Date: Sun, 15 Mar 2026 14:22:57 -0500 Subject: [PATCH 6/7] fix: cap keepalive mode duration to prevent indefinite client hold - Add MAX_KEEPALIVE_DURATION = 300s to TSConfig - Track keepalive_start_time in _stream_data_generator; disconnect after cap is exceeded with a warning log - Reset timer on real data so independent stalls don't accumulate - Fix pre-existing test helper missing throttle attributes in NonOwnerWorkerKeepaliveTests --- .../channels/tests/test_ts_proxy_keepalive.py | 6 + .../tests/test_ts_proxy_keepalive_duration.py | 195 ++++++++++++++++++ apps/proxy/config.py | 1 + apps/proxy/ts_proxy/stream_generator.py | 16 ++ 4 files changed, 218 insertions(+) create mode 100644 apps/channels/tests/test_ts_proxy_keepalive_duration.py diff --git a/apps/channels/tests/test_ts_proxy_keepalive.py b/apps/channels/tests/test_ts_proxy_keepalive.py index db388b3d..9e14fdfd 100644 --- a/apps/channels/tests/test_ts_proxy_keepalive.py +++ b/apps/channels/tests/test_ts_proxy_keepalive.py @@ -85,6 +85,12 @@ class NonOwnerWorkerKeepaliveTests(TestCase): gen.stream_manager = None # non-owner worker gen.consecutive_empty = consecutive_empty + + # Throttle attributes added by a later PR; required by _should_send_keepalive. + gen._last_health_check_time = 0.0 + gen._last_health_check_result = False + gen._health_check_interval = 2.0 + gen.proxy_server = None return gen def _mock_proxy_server(self, last_data_value): diff --git a/apps/channels/tests/test_ts_proxy_keepalive_duration.py b/apps/channels/tests/test_ts_proxy_keepalive_duration.py new file mode 100644 index 00000000..74494131 --- /dev/null +++ b/apps/channels/tests/test_ts_proxy_keepalive_duration.py @@ -0,0 +1,195 @@ +""" +Unit tests for the keepalive duration cap in StreamGenerator._stream_data_generator. + +Verifies that a client held in keepalive mode is disconnected after +MAX_KEEPALIVE_DURATION seconds, and that the timer resets when real data resumes. +""" + +import time +from unittest.mock import MagicMock, patch, call +from django.test import TestCase + + +def _make_generator(consecutive_empty=10, local_index=10, buffer_index=10): + """Minimal StreamGenerator stub for testing _stream_data_generator logic.""" + from apps.proxy.ts_proxy.stream_generator import StreamGenerator + + gen = StreamGenerator.__new__(StreamGenerator) + gen.channel_id = "00000000-0000-0000-0000-000000000099" + gen.client_id = "test-client-duration" + gen.consecutive_empty = consecutive_empty + gen.empty_reads = 0 + gen.local_index = local_index + gen.bytes_sent = 0 + gen.chunks_sent = 0 + gen.last_yield_time = time.time() + gen.stream_start_time = time.time() + gen.last_stats_time = time.time() + gen.last_stats_bytes = 0 + gen.current_rate = 0.0 + gen.last_ttl_refresh = time.time() + gen.ttl_refresh_interval = 3 + gen.is_owner_worker = False + gen.stream_manager = None + gen._last_health_check_time = 0.0 + gen._last_health_check_result = False + gen._health_check_interval = 2.0 + gen.proxy_server = None + + buffer = MagicMock() + buffer.index = buffer_index + buffer.get_optimized_client_data.return_value = ([], local_index) + buffer.find_oldest_available_chunk.return_value = None + gen.buffer = buffer + + return gen + + +class KeepaliveDurationCapTests(TestCase): + """MAX_KEEPALIVE_DURATION cap disconnects clients stuck in keepalive mode.""" + + def _run_generator_to_break(self, gen, max_iterations=20): + """Drive _stream_data_generator until it breaks or hits iteration limit.""" + iterations = 0 + for _ in gen._stream_data_generator(): + iterations += 1 + if iterations >= max_iterations: + break + return iterations + + def test_cap_fires_after_max_duration_exceeded(self): + """Generator exits when keepalive has run longer than MAX_KEEPALIVE_DURATION.""" + gen = _make_generator() + + with patch.object(gen, '_check_resources', return_value=True), \ + patch.object(gen, '_should_send_keepalive', return_value=True), \ + patch.object(gen, '_is_ghost_client', return_value=False), \ + patch.object(gen, '_is_timeout', return_value=False), \ + patch('apps.proxy.ts_proxy.stream_generator.create_ts_packet', return_value=b'\x00' * 188), \ + patch('apps.proxy.ts_proxy.stream_generator.ProxyServer') as MockPS, \ + patch('apps.proxy.ts_proxy.stream_generator.Config') as MockConfig, \ + patch('apps.proxy.ts_proxy.stream_generator.gevent') as mock_gevent, \ + patch('apps.proxy.ts_proxy.stream_generator.time') as mock_time: + + MockPS.get_instance.return_value = None + MockConfig.KEEPALIVE_INTERVAL = 0 + MockConfig.MAX_KEEPALIVE_DURATION = 30 + + # First call: keepalive_start_time not yet set (returns current) + # Second call: inside the cap check — simulate time elapsed > 30s + mock_time.time.side_effect = [ + 1000.0, # keepalive_start_time assignment + 1031.0, # cap check: 31s elapsed > 30s limit + ] + + packets = list(gen._stream_data_generator()) + + # No packets should be yielded — cap fires before yield + self.assertEqual(len(packets), 0) + + def test_cap_does_not_fire_before_max_duration(self): + """Generator yields keepalive packets while within MAX_KEEPALIVE_DURATION.""" + gen = _make_generator() + + call_count = 0 + + def time_side_effect(): + nonlocal call_count + call_count += 1 + # keepalive_start_time set at t=1000; cap checks always see <30s elapsed + if call_count == 1: + return 1000.0 # keepalive_start_time + return 1010.0 # always 10s elapsed — under the 30s cap + + with patch.object(gen, '_check_resources', side_effect=[True, True, False]), \ + patch.object(gen, '_should_send_keepalive', return_value=True), \ + patch.object(gen, '_is_ghost_client', return_value=False), \ + patch.object(gen, '_is_timeout', return_value=False), \ + patch('apps.proxy.ts_proxy.stream_generator.create_ts_packet', return_value=b'\x00' * 188), \ + patch('apps.proxy.ts_proxy.stream_generator.ProxyServer') as MockPS, \ + patch('apps.proxy.ts_proxy.stream_generator.Config') as MockConfig, \ + patch('apps.proxy.ts_proxy.stream_generator.gevent'), \ + patch('apps.proxy.ts_proxy.stream_generator.time') as mock_time: + + MockPS.get_instance.return_value = None + MockConfig.KEEPALIVE_INTERVAL = 0 + MockConfig.MAX_KEEPALIVE_DURATION = 30 + mock_time.time.side_effect = time_side_effect + + packets = list(gen._stream_data_generator()) + + # Two iterations with _check_resources=True should yield two keepalive packets + self.assertGreater(len(packets), 0) + + def test_timer_resets_when_real_data_resumes(self): + """keepalive_start_time is cleared to None when real chunks are received.""" + gen = _make_generator() + + chunk = b'\x47' * 188 + real_chunks = ([chunk], gen.local_index + 1) + no_chunks = ([], gen.local_index) + + # Sequence: no data (keepalive), then real data, then stop + gen.buffer.get_optimized_client_data.side_effect = [ + no_chunks, # iteration 1: keepalive + real_chunks, # iteration 2: real data — should reset timer + no_chunks, # iteration 3: keepalive again — timer restarts fresh + ] + + captured_start_times = [] + + original_gen = gen + + with patch.object(gen, '_check_resources', side_effect=[True, True, True, False]), \ + patch.object(gen, '_should_send_keepalive', return_value=True), \ + patch.object(gen, '_is_ghost_client', return_value=False), \ + patch.object(gen, '_is_timeout', return_value=False), \ + patch.object(gen, '_process_chunks', return_value=iter([chunk])), \ + patch('apps.proxy.ts_proxy.stream_generator.create_ts_packet', return_value=b'\x00' * 188), \ + patch('apps.proxy.ts_proxy.stream_generator.ProxyServer') as MockPS, \ + patch('apps.proxy.ts_proxy.stream_generator.Config') as MockConfig, \ + patch('apps.proxy.ts_proxy.stream_generator.gevent'), \ + patch('apps.proxy.ts_proxy.stream_generator.time') as mock_time: + + MockPS.get_instance.return_value = None + MockConfig.KEEPALIVE_INTERVAL = 0 + MockConfig.MAX_KEEPALIVE_DURATION = 300 + mock_time.time.return_value = 1000.0 + + list(gen._stream_data_generator()) + + # Test passes if no exception and generator completes normally — + # if the timer were NOT reset, the second keepalive block would + # carry over the old start time rather than starting fresh. + + def test_cap_uses_config_value(self): + """Cap threshold reads MAX_KEEPALIVE_DURATION from Config, not a hardcoded value.""" + gen = _make_generator() + + with patch.object(gen, '_check_resources', return_value=True), \ + patch.object(gen, '_should_send_keepalive', return_value=True), \ + patch.object(gen, '_is_ghost_client', return_value=False), \ + patch.object(gen, '_is_timeout', return_value=False), \ + patch('apps.proxy.ts_proxy.stream_generator.create_ts_packet', return_value=b'\x00' * 188), \ + patch('apps.proxy.ts_proxy.stream_generator.ProxyServer') as MockPS, \ + patch('apps.proxy.ts_proxy.stream_generator.Config') as MockConfig, \ + patch('apps.proxy.ts_proxy.stream_generator.gevent'), \ + patch('apps.proxy.ts_proxy.stream_generator.time') as mock_time: + + MockPS.get_instance.return_value = None + MockConfig.KEEPALIVE_INTERVAL = 0 + # Set a custom cap of 60s + MockConfig.MAX_KEEPALIVE_DURATION = 60 + + mock_time.time.side_effect = [ + 1000.0, # keepalive_start_time + 1050.0, # cap check: 50s elapsed — under 60s, should NOT fire + 1000.0, # last_yield_time update + 1070.0, # cap check on next iteration: 70s elapsed — fires + ] + + packets = list(gen._stream_data_generator()) + + # First iteration: 50s < 60s cap — one keepalive yielded + # Second iteration: 70s > 60s cap — generator exits + self.assertEqual(len(packets), 1) diff --git a/apps/proxy/config.py b/apps/proxy/config.py index 68e27a6f..0424dee1 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -108,6 +108,7 @@ class TSConfig(BaseConfig): MIN_STABLE_TIME_BEFORE_RECONNECT = 30 # Minimum seconds a stream must be stable to try reconnect FAILOVER_GRACE_PERIOD = 20 # Extra time (seconds) to allow for stream switching before disconnecting clients URL_SWITCH_TIMEOUT = 20 # Max time allowed for a stream switch operation + MAX_KEEPALIVE_DURATION = 300 # Keepalive packets prevent _is_timeout() from firing, so without this a permanently failed stream holds clients open indefinitely. diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index cfd705c9..2e16008b 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -255,6 +255,10 @@ class StreamGenerator: def _stream_data_generator(self): """Generate stream data chunks based on buffer contents.""" + # Keepalive packets refresh last_yield_time, so _is_timeout() never fires + # during sustained stream failure. This timer enforces a wall-clock cap. + keepalive_start_time = None + # Main streaming loop while True: # Check if resources still exist @@ -265,6 +269,7 @@ class StreamGenerator: chunks, next_index = self.buffer.get_optimized_client_data(self.local_index) if chunks: + keepalive_start_time = None # Each recovery restarts the cap independently. yield from self._process_chunks(chunks, next_index) self.local_index = next_index self.last_yield_time = time.time() @@ -306,6 +311,17 @@ class StreamGenerator: continue # Retry immediately with the new position if self._should_send_keepalive(self.local_index): + if keepalive_start_time is None: + keepalive_start_time = time.time() + + max_keepalive = getattr(Config, 'MAX_KEEPALIVE_DURATION', 300) + if time.time() - keepalive_start_time > max_keepalive: + logger.warning( + f"[{self.client_id}] Keepalive duration exceeded {max_keepalive}s " + f"with no stream recovery, disconnecting" + ) + break + keepalive_packet = create_ts_packet('keepalive') logger.debug(f"[{self.client_id}] Sending keepalive packet while waiting at buffer head") yield keepalive_packet From 92f98a24c78bf5157e89f9d8fcddaeb2fd92731b Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 15 Mar 2026 20:45:55 -0500 Subject: [PATCH 7/7] fix: increase ghost client multiplier to improve client detection timing --- apps/proxy/config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/proxy/config.py b/apps/proxy/config.py index 0424dee1..4de92fa0 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -99,7 +99,7 @@ class TSConfig(BaseConfig): CLIENT_RECORD_TTL = 60 # How long client records persist in Redis (seconds). Client will be considered MIA after this time. CLEANUP_CHECK_INTERVAL = 1 # How often to check for disconnected clients (seconds) CLIENT_HEARTBEAT_INTERVAL = 5 # How often to send client heartbeats (seconds) - GHOST_CLIENT_MULTIPLIER = 6.0 # How many heartbeat intervals before client considered ghost (6 would mean 36 seconds if heartbeat interval is 6) + GHOST_CLIENT_MULTIPLIER = 10.0 # How many heartbeat intervals before client considered ghost (10 = 50s, must exceed STREAM_TIMEOUT + FAILOVER_GRACE_PERIOD = 40s) CLIENT_WAIT_TIMEOUT = 30 # Seconds to wait for client to connect # Stream health and recovery settings