Fixes timers not releasing on reconnect attempts

apps/proxy/ts_proxy/server.py

@@ -497,13 +497,12 @@ class ProxyServer:
         """Stop a channel with proper ownership handling"""
         try:
             logger.info(f"Stopping channel {channel_id}")
-
+            
             # First set a stopping key that clients will check
             if self.redis_client:
                 stop_key = f"ts_proxy:channel:{channel_id}:stopping"
-                # Set with 60 second TTL - enough time for clients to notice
                 self.redis_client.setex(stop_key, 10, "true")
-
+                
             # Only stop the actual stream manager if we're the owner
             if self.am_i_owner(channel_id):
                 logger.info(f"This worker ({self.worker_id}) is the owner - closing provider connection")
@@ -541,22 +540,41 @@ class ProxyServer:
                 self.release_ownership(channel_id)
                 logger.info(f"Released ownership of channel {channel_id}")
 
-            # Always clean up local resources
+            # Always clean up local resources - WITH SAFE CHECKS
             if channel_id in self.stream_managers:
                 del self.stream_managers[channel_id]
                 logger.info(f"Removed stream manager for channel {channel_id}")
-
+            
+            # Stop buffer and ensure all its timers are cancelled - SAFE CHECK HERE
             if channel_id in self.stream_buffers:
-                del self.stream_buffers[channel_id]
-                logger.info(f"Removed stream buffer for channel {channel_id}")
-
+                buffer = self.stream_buffers[channel_id]
+                # Call stop on buffer to properly shut it down
+                if hasattr(buffer, 'stop'):
+                    try:
+                        buffer.stop()
+                        logger.debug(f"Buffer for channel {channel_id} properly stopped")
+                    except Exception as e:
+                        logger.error(f"Error stopping buffer: {e}")
+                
+                # Save reference and check again before deleting
+                try:
+                    if channel_id in self.stream_buffers:  # Check again to prevent race conditions
+                        del self.stream_buffers[channel_id]
+                        logger.info(f"Removed stream buffer for channel {channel_id}")
+                except KeyError:
+                    logger.debug(f"Buffer for channel {channel_id} already removed")
+            
+            # Clean up client manager - SAFE CHECK HERE TOO
             if channel_id in self.client_managers:
-                del self.client_managers[channel_id]
-                logger.info(f"Removed client manager for channel {channel_id}")
-
+                try:
+                    del self.client_managers[channel_id]
+                    logger.info(f"Removed client manager for channel {channel_id}")
+                except KeyError:
+                    logger.debug(f"Client manager for channel {channel_id} already removed")
+                
             # Clean up Redis keys
             self._clean_redis_keys(channel_id)
-
+            
             return True
         except Exception as e:
             logger.error(f"Error stopping channel {channel_id}: {e}", exc_info=True)

apps/proxy/ts_proxy/stream_buffer.py

@@ -39,6 +39,10 @@ class StreamBuffer:
         self._write_buffer = bytearray()
         self.target_chunk_size = getattr(Config, 'BUFFER_CHUNK_SIZE', 188 * 5644)  # ~1MB default
 
+        # Track timers for proper cleanup
+        self.stopping = False
+        self.fill_timers = []
+
     def add_chunk(self, chunk):
         """Add data with optimized Redis storage and TS packet alignment"""
         if not chunk:
@@ -215,7 +219,26 @@ class StreamBuffer:
             return []
 
     def stop(self):
-        """Stop the buffer and flush any remaining data"""
+        """Stop the buffer and cancel all timers"""
+        # Set stopping flag first to prevent new timer creation
+        self.stopping = True
+        
+        # Cancel all pending timers
+        timers_cancelled = 0
+        for timer in list(self.fill_timers):
+            try:
+                if timer and timer.is_alive():
+                    timer.cancel()
+                    timers_cancelled += 1
+            except Exception as e:
+                logger.error(f"Error canceling timer: {e}")
+        
+        if timers_cancelled:
+            logger.info(f"Cancelled {timers_cancelled} buffer timers for channel {self.channel_id}")
+            
+        # Clear timer list
+        self.fill_timers.clear()
+
         try:
             # Flush any remaining data in the write buffer
             if hasattr(self, '_write_buffer') and len(self._write_buffer) > 0:
@@ -285,4 +308,16 @@ class StreamBuffer:
                 chunks.extend(more_chunks)
                 chunk_count += additional
 
-        return chunks, client_index + chunk_count
+        return chunks, client_index + chunk_count
+
+    # 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"""
+        if self.stopping:
+            return None
+            
+        timer = threading.Timer(delay, callback, args=args, kwargs=kwargs)
+        timer.daemon = True
+        timer.start()
+        self.fill_timers.append(timer)
+        return timer

apps/proxy/ts_proxy/stream_manager.py

@@ -45,6 +45,10 @@ class StreamManager:
         self.health_check_interval = Config.HEALTH_CHECK_INTERVAL
         self.chunk_size = getattr(Config, 'CHUNK_SIZE', 8192)
 
+        # Add to your __init__ method
+        self._buffer_check_timers = []
+        self.stopping = False
+
         logger.info(f"Initialized stream manager for channel {buffer.channel_id}")
 
     def _create_session(self):
@@ -221,8 +225,23 @@ class StreamManager:
                         break
 
                     timeout = min(2 ** self.retry_count, 30)
+                    
+                    # When a connection fails and reconnect is needed:
+                    self.reconnecting = True 
+                    
+                    # Cancel all existing buffer timers during reconnect
+                    for timer in list(self._buffer_check_timers):
+                        try:
+                            if timer and timer.is_alive():
+                                timer.cancel()
+                        except Exception as e:
+                            logger.error(f"Error canceling buffer timer: {e}")
+                    self._buffer_check_timers = []
+                    
                     logger.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count})")
                     time.sleep(timeout)
+                    
+                    self.reconnecting = False  # Reset flag after sleep
 
         except Exception as e:
             logger.error(f"Stream error: {e}", exc_info=True)
@@ -250,7 +269,21 @@ class StreamManager:
             logger.info(f"Stream manager stopped")
 
     def stop(self):
-        """Stop this stream"""
+        """Stop the stream manager and cancel all timers"""
+        # Add at the beginning of your stop method
+        self.stopping = True
+        
+        # Cancel all buffer check timers
+        for timer in list(self._buffer_check_timers):
+            try:
+                if timer and timer.is_alive():
+                    timer.cancel()
+            except Exception as e:
+                logger.error(f"Error canceling buffer check timer: {e}")
+        
+        self._buffer_check_timers.clear()
+        
+        # Rest of your existing stop method...
         # Set the flag first
         self.stop_requested = True
 
@@ -488,8 +521,14 @@ class StreamManager:
     def _check_buffer_and_set_state(self):
         """Check buffer size and set state to waiting_for_clients when ready"""
         try:
-            # This method will be called asynchronously to check buffer status
-            # and update state when enough chunks are available
+            # First check if we're stopping or reconnecting
+            if getattr(self, 'stopping', False) or getattr(self, 'reconnecting', False):
+                logger.debug(f"Buffer check aborted - channel {self.buffer.channel_id} is stopping or reconnecting")
+                return
+            
+            # Clean up completed timers
+            self._buffer_check_timers = [t for t in self._buffer_check_timers if t.is_alive()]
+            
             if hasattr(self.buffer, 'index') and hasattr(self.buffer, 'channel_id'):
                 current_buffer_index = self.buffer.index
                 initial_chunks_needed = getattr(Config, 'INITIAL_BEHIND_CHUNKS', 10)
@@ -502,13 +541,13 @@ class StreamManager:
                 else:
                     # Still waiting, log progress and schedule another check
                     logger.debug(f"Buffer filling for channel {channel_id}: {current_buffer_index}/{initial_chunks_needed} chunks")
-                    if current_buffer_index > 0 and current_buffer_index % 5 == 0:
-                        # Log less frequently to avoid spamming logs
-                        logger.info(f"Buffer filling for channel {channel_id}: {current_buffer_index}/{initial_chunks_needed} chunks")
-
-                    # Schedule another check
-                    timer = threading.Timer(0.5, self._check_buffer_and_set_state)
-                    timer.daemon = True
-                    timer.start()
+                    
+                    # Schedule another check - NOW WITH TRACKING
+                    if not getattr(self, 'stopping', False):
+                        timer = threading.Timer(0.5, self._check_buffer_and_set_state)
+                        timer.daemon = True
+                        timer.start()
+                        self._buffer_check_timers.append(timer)
         except Exception as e:
             logger.error(f"Error in buffer check: {e}")
+