From e78c24aad735952eb6f8939dc2935aa4fe20546d Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 3 Mar 2026 15:13:46 -0600 Subject: [PATCH] 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 | 92 +++++++++++++++++++++++++ apps/proxy/ts_proxy/stream_generator.py | 36 +++++++--- 2 files changed, 119 insertions(+), 9 deletions(-) diff --git a/apps/proxy/ts_proxy/stream_buffer.py b/apps/proxy/ts_proxy/stream_buffer.py index 85feb5dd..2bf99527 100644 --- a/apps/proxy/ts_proxy/stream_buffer.py +++ b/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::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""" diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 50404f1d..8f631d53 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/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')