Merge branch 'dev' of https://github.com/Dispatcharr/Dispatcharr into dev

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)

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

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

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]
@@ -132,6 +133,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:

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,12 +311,28 @@ 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
                     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
@@ -428,7 +449,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 +636,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
+    return generator.generate