Enhance Redis chunk management in StreamBuffer and StreamGenerator

- Register Lua scripts for optimized chunk retrieval in StreamBuffer.
- Implement atomic Lua binary search to find the oldest available chunk in Redis.
- Update StreamGenerator to handle expired chunks more efficiently, reducing TOCTOU race conditions.

apps/proxy/ts_proxy/stream_buffer.py

@@ -45,6 +45,15 @@ class StreamBuffer:
         self._write_buffer = bytearray()
         self.target_chunk_size = ConfigHelper.get('BUFFER_CHUNK_SIZE', TS_PACKET_SIZE * 5644)  # ~1MB default
 
+        # Register Lua scripts once — subsequent calls use EVALSHA (just the
+        # SHA hash) instead of sending the full script text on every invocation.
+        if self.redis_client:
+            self._find_oldest_chunk_sha = self.redis_client.register_script(
+                self._FIND_OLDEST_CHUNK_LUA
+            )
+        else:
+            self._find_oldest_chunk_sha = None
+
         # Track timers for proper cleanup
         self.stopping = False
         self.fill_timers = []
@@ -328,6 +337,89 @@ class StreamBuffer:
 
         return chunks, client_index + chunk_count
 
+    # Lua script that runs an atomic binary search on the Redis server.
+    # Chunks expire in FIFO order (same TTL, sequential writes), so the
+    # alive range is contiguous: [oldest_surviving .. buffer_head].
+    # Binary search finds the boundary in O(log N) EXISTS calls with zero
+    # round-trips between steps and no TOCTOU races (Lua scripts are atomic).
+    #
+    # ARGV[1] = key prefix  (e.g. "ts_proxy:channel:<id>:buffer:chunk:")
+    # ARGV[2] = low index   (client_index + 1, first chunk the client needs)
+    # ARGV[3] = high index  (buffer head, most recent chunk)
+    #
+    # Returns: the index of the oldest existing chunk, or -1 if none exist.
+    _FIND_OLDEST_CHUNK_LUA = """
+    local prefix = ARGV[1]
+    local low    = tonumber(ARGV[2])
+    local high   = tonumber(ARGV[3])
+
+    if redis.call('EXISTS', prefix .. high) == 0 then
+        return -1
+    end
+
+    local result = high
+    while low <= high do
+        local mid = math.floor((low + high) / 2)
+        if redis.call('EXISTS', prefix .. mid) == 1 then
+            result = mid
+            high = mid - 1
+        else
+            low = mid + 1
+        end
+    end
+    return result
+    """
+
+    def find_oldest_available_chunk(self, client_index):
+        """Find the oldest (lowest-index) chunk that still exists in Redis.
+
+        Executes an atomic Lua binary search on the Redis server — one
+        round-trip, ~log2(N) EXISTS calls, no TOCTOU between steps.
+
+        The actual read attempt (get_optimized_client_data) is what
+        authoritatively detects expiration; this method is best-effort
+        positioning that self-corrects on the next iteration if the found
+        chunk also expires before the client can read it.
+
+        Args:
+            client_index: The client's current local_index (last consumed chunk).
+
+        Returns:
+            int or None: The local_index value the client should jump to
+                         (one before the first available chunk), or None if no
+                         chunks are available at all.
+        """
+        if not self.redis_client:
+            return None
+
+        low = client_index + 1   # First chunk the client needs
+        high = self.index        # Latest chunk written
+
+        if low > high:
+            return None
+
+        try:
+            # Uses EVALSHA under the hood — sends only the SHA hash,
+            # not the full script text, on every call after the first.
+            result = self._find_oldest_chunk_sha(
+                args=[
+                    RedisKeys.buffer_chunk_prefix(self.channel_id),
+                    low,
+                    high,
+                ],
+            )
+
+            if result == -1:
+                return None
+
+            # Return result - 1 so local_index points to one before the
+            # first available chunk (matching the "last consumed" convention).
+            return int(result) - 1
+
+        except Exception as e:
+            logger.error(f"Error running find_oldest_chunk Lua script for channel {self.channel_id}: {e}")
+            return None
+
     # Add a new method to safely create timers
     def schedule_timer(self, delay, callback, *args, **kwargs):
         """Schedule a timer and track it for proper cleanup"""

apps/proxy/ts_proxy/stream_generator.py

@@ -224,17 +224,35 @@ class StreamGenerator:
                 self.empty_reads += 1
                 self.consecutive_empty += 1
 
-                # Check if we're too far behind (chunks expired from Redis)
+                # We got no data despite being behind the buffer head.
+                # The read itself is the authoritative signal — no separate
+                # existence check needed, avoiding TOCTOU races with Redis TTL.
                 chunks_behind = self.buffer.index - self.local_index
-                if chunks_behind > 50:  # If more than 50 chunks behind, jump forward
-                    # Calculate new position: stay a few chunks behind current buffer
-                    initial_behind = ConfigHelper.initial_behind_chunks()
-                    new_index = max(self.local_index, self.buffer.index - initial_behind)
+                if chunks_behind > 0:
+                    # Next chunk has expired — find the oldest chunk still in Redis
+                    new_index = self.buffer.find_oldest_available_chunk(self.local_index)
 
-                    logger.warning(f"[{self.client_id}] Client too far behind ({chunks_behind} chunks), jumping from {self.local_index} to {new_index}")
-                    self.local_index = new_index
-                    self.consecutive_empty = 0  # Reset since we're repositioning
-                    continue  # Try again immediately with new position
+                    if new_index is not None:
+                        skipped = new_index - self.local_index
+                        logger.warning(
+                            f"[{self.client_id}] Next chunk expired (index {self.local_index + 1}), "
+                            f"jumping to oldest available: {new_index + 1} "
+                            f"(skipped {skipped} chunks, buffer head at {self.buffer.index})"
+                        )
+                        self.local_index = new_index
+                    else:
+                        # No chunks available at all — jump to near the buffer head
+                        initial_behind = ConfigHelper.initial_behind_chunks()
+                        new_index = max(self.local_index, self.buffer.index - initial_behind)
+                        logger.warning(
+                            f"[{self.client_id}] No chunks available in buffer, "
+                            f"jumping to near buffer head: {new_index} "
+                            f"(buffer head at {self.buffer.index})"
+                        )
+                        self.local_index = new_index
+
+                    self.consecutive_empty = 0
+                    continue  # Retry immediately with the new position
 
                 if self._should_send_keepalive(self.local_index):
                     keepalive_packet = create_ts_packet('keepalive')