diff --git a/apps/proxy/config.py b/apps/proxy/config.py index 879736ce..a5d39452 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -47,4 +47,5 @@ class TSConfig(BaseConfig): CLEANUP_CHECK_INTERVAL = 1 # How often to check for disconnected clients (seconds) CHANNEL_INIT_GRACE_PERIOD = 5 # How long to wait for first client after initialization (seconds) CLIENT_HEARTBEAT_INTERVAL = 1 # How often to send client heartbeats (seconds) + GHOST_CLIENT_MULTIPLIER = 5.0 # How many heartbeat intervals before client considered ghost (5 would mean 5 secondsif heartbeat interval is 1) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 7bb4d5db..12ceb045 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -651,25 +651,62 @@ class ClientManager: with self.lock: if not self.clients or not self.redis_client: continue + + # 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: + client_key = f"ts_proxy:channel:{self.channel_id}:clients:{client_id}" - # Use pipeline for efficiency + # Check if client exists in Redis at all + exists = self.redis_client.exists(client_key) + if not exists: + # Client entry has expired in Redis but still in our local set + logging.warning(f"Found ghost client {client_id} - expired in Redis but still in local set") + clients_to_remove.add(client_id) + continue + + # Check for stale activity using last_active field + last_active = self.redis_client.hget(client_key, "last_active") + if last_active: + last_active_time = float(last_active.decode('utf-8')) + time_since_activity = current_time - last_active_time + + # If client hasn't been active for too long, mark for removal + # Use configurable threshold for detection + ghost_threshold = getattr(Config, 'GHOST_CLIENT_MULTIPLIER', 5.0) + if time_since_activity > self.heartbeat_interval * ghost_threshold: + logging.warning(f"Detected ghost client {client_id} - last active {time_since_activity:.1f}s ago") + 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: + logging.info(f"Removed {len(clients_to_remove)} ghost clients from channel {self.channel_id}") + + # Now send heartbeats only for remaining clients pipe = self.redis_client.pipeline() current_time = time.time() - # For each client, update its TTL and timestamp for client_id in self.clients: + # Skip clients we just marked for removal + if client_id in clients_to_remove: + continue + # Skip if we just sent a heartbeat recently if client_id in self.last_heartbeat_time: time_since_last = current_time - self.last_heartbeat_time[client_id] if time_since_last < self.heartbeat_interval * 0.8: continue - # FIXED: Update client hash instead of separate activity key + # Only update clients that remain client_key = f"ts_proxy:channel:{self.channel_id}:clients:{client_id}" - - # Only update the last_active field in the hash, preserving other fields pipe.hset(client_key, "last_active", str(current_time)) - pipe.expire(client_key, self.client_ttl) # Refresh TTL + pipe.expire(client_key, self.client_ttl) # Keep client in the set with TTL pipe.sadd(self.client_set_key, client_id) @@ -681,9 +718,10 @@ class ClientManager: # Execute all commands atomically pipe.execute() - # Notify channel owner of client activity - self._notify_owner_of_activity() - + # Only notify if we have real clients + if self.clients and not all(c in clients_to_remove for c in self.clients): + self._notify_owner_of_activity() + except Exception as e: logging.error(f"Error in client heartbeat thread: {e}") diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index b051d4f4..23b41e1c 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -293,7 +293,16 @@ def stream_ts(request, channel_id): packets_in_chunk = len(chunk) // ts_packet_size bytes_sent += len(chunk) chunks_sent += 1 - yield chunk + + # CRITICAL FIX: Detect client disconnection when yielding data + try: + yield chunk + except Exception as e: + logger.info(f"[{client_id}] Client disconnected while yielding data: {e}") + # Explicit client removal when we detect a broken pipe + if channel_id in proxy_server.client_managers: + proxy_server.client_managers[channel_id].remove_client(client_id) + break # Exit the generator # Pacing logic packets_sent_in_batch += packets_in_chunk @@ -351,6 +360,21 @@ def stream_ts(request, channel_id): stream_status = "healthy" if (stream_manager and stream_manager.healthy) else "unknown" logger.debug(f"[{client_id}] Waiting for chunks beyond {local_index} (buffer at {buffer.index}, stream: {stream_status})") + # CRITICAL FIX: Check for client disconnect during wait periods + # Django/WSGI might not immediately detect disconnections, but we can check periodically + if consecutive_empty > 10: # After some number of empty reads + if hasattr(request, 'META') and request.META.get('wsgi.input'): + try: + # Try to check if the connection is still alive + available = request.META['wsgi.input'].read(0) + if available is None: # Connection closed + logger.info(f"[{client_id}] Detected client disconnect during wait") + break + except Exception: + # Error reading from connection, likely closed + logger.info(f"[{client_id}] Connection error, client likely disconnected") + break + # Disconnect after long inactivity # For non-owner workers, we're more lenient with timeout if time.time() - last_yield_time > Config.STREAM_TIMEOUT: @@ -362,6 +386,12 @@ def stream_ts(request, channel_id): logger.warning(f"[{client_id}] Non-owner worker with no data for {Config.STREAM_TIMEOUT}s, disconnecting") break + # ADD THIS: Check if worker has more recent chunks but still stuck + # This can indicate the client is disconnected but we're not detecting it + if consecutive_empty > 100 and buffer.index > local_index + 50: + logger.warning(f"[{client_id}] Possible ghost client: buffer has advanced {buffer.index - local_index} chunks ahead but client stuck at {local_index}") + break + except Exception as e: logger.error(f"[{client_id}] Stream error: {e}", exc_info=True) finally: