feat: Enhance file watcher and index service with improved logging and concurrency control

Author: johnhuang316

src/code_index_mcp/server.py

@@ -270,12 +270,12 @@ def get_file_watcher_status(ctx: Context) -> Dict[str, Any]:
         file_watcher_error = None
         if hasattr(ctx.request_context.lifespan_context, 'file_watcher_error'):
             file_watcher_error = ctx.request_context.lifespan_context.file_watcher_error
-        
+
         # Get file watcher service from context
         file_watcher_service = None
         if hasattr(ctx.request_context.lifespan_context, 'file_watcher_service'):
             file_watcher_service = ctx.request_context.lifespan_context.file_watcher_service
-        
+
         # If there's an error, return error status with recommendation
         if file_watcher_error:
             status = {
@@ -285,39 +285,39 @@ def get_file_watcher_status(ctx: Context) -> Dict[str, Any]:
                 "recommendation": "Use refresh_index tool for manual updates",
                 "manual_refresh_required": True
             }
-            
+
             # Add basic configuration if available
             if hasattr(ctx.request_context.lifespan_context, 'settings') and ctx.request_context.lifespan_context.settings:
                 file_watcher_config = ctx.request_context.lifespan_context.settings.get_file_watcher_config()
                 status["configuration"] = file_watcher_config
-            
+
             return status
-        
+
         # If no service and no error, it's not initialized
         if not file_watcher_service:
             return {
                 "available": True,
                 "active": False,
-                "status": "not_initialized", 
+                "status": "not_initialized",
                 "message": "File watcher service not initialized. Set project path to enable auto-refresh.",
                 "recommendation": "Use set_project_path tool to initialize file watcher"
             }
-        
+
         # Get status from file watcher service
         status = file_watcher_service.get_status()
-        
+
         # Add index service status
         index_service = IndexService(ctx)
         rebuild_status = index_service.get_rebuild_status()
         status["rebuild_status"] = rebuild_status
-        
+
         # Add configuration
         if hasattr(ctx.request_context.lifespan_context, 'settings') and ctx.request_context.lifespan_context.settings:
             file_watcher_config = ctx.request_context.lifespan_context.settings.get_file_watcher_config()
             status["configuration"] = file_watcher_config
-        
+
         return status
-        
+
     except Exception as e:
         return {"status": "error", "message": f"Failed to get file watcher status: {e}"}
 
@@ -334,9 +334,9 @@ def configure_file_watcher(
         # Get settings from context
         if not hasattr(ctx.request_context.lifespan_context, 'settings') or not ctx.request_context.lifespan_context.settings:
             return "Settings not available - project path not set"
-        
+
         settings = ctx.request_context.lifespan_context.settings
-        
+
         # Build updates dictionary
         updates = {}
         if enabled is not None:
@@ -345,17 +345,17 @@ def configure_file_watcher(
             updates["debounce_seconds"] = debounce_seconds
         if additional_exclude_patterns is not None:
             updates["additional_exclude_patterns"] = additional_exclude_patterns
-        
+
         if not updates:
             return "No configuration changes specified"
-        
+
         # Update configuration
         settings.update_file_watcher_config(updates)
-        
+
         # If file watcher is running, we would need to restart it for changes to take effect
         # For now, just return success message with note about restart
         return f"File watcher configuration updated: {updates}. Restart may be required for changes to take effect."
-        
+
     except Exception as e:
         return f"Failed to update file watcher configuration: {e}"
 
@@ -419,22 +419,6 @@ def set_project() -> list[types.PromptMessage]:
 
 def main():
     """Main function to run the MCP server."""
-    # Configure logging for debugging
-    import logging
-    logging.basicConfig(
-        level=logging.DEBUG,
-        format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
-        handlers=[
-            logging.StreamHandler(),
-            logging.FileHandler('mcp_server_debug.log', mode='w')
-        ]
-    )
-    
-    # Enable debug logging for file watcher and index services
-    logging.getLogger('code_index_mcp.services.file_watcher_service').setLevel(logging.DEBUG)
-    logging.getLogger('code_index_mcp.services.index_service').setLevel(logging.DEBUG)
-    
-    # Run the server. Tools are discovered automatically via decorators.
     mcp.run()
 
 if __name__ == '__main__':

src/code_index_mcp/services/file_watcher_service.py

@@ -111,7 +111,7 @@ def start_monitoring(self, rebuild_callback: Callable) -> bool:
 
             # Log detailed Observer setup
             watch_path = str(self.base_path)
-            self.logger.info(f"Scheduling Observer for path: {watch_path}")
+            self.logger.debug(f"Scheduling Observer for path: {watch_path}")
 
             self.observer.schedule(
                 self.event_handler,
@@ -120,20 +120,20 @@ def start_monitoring(self, rebuild_callback: Callable) -> bool:
             )
 
             # Log Observer start
-            self.logger.info("Starting Observer...")
+            self.logger.debug("Starting Observer...")
             self.observer.start()
             self.is_monitoring = True
             self.restart_attempts = 0
 
             # Log Observer thread info
             if hasattr(self.observer, '_thread'):
                 thread_info = f"Observer thread: {self.observer._thread}"
-                self.logger.info(thread_info)
+                self.logger.debug(thread_info)
 
             # Verify observer is actually running
             if self.observer.is_alive():
                 self.logger.info(
-                    "File watcher started successfully - Observer is alive",
+                    "File watcher started successfully",
                     extra={
                         "debounce_seconds": debounce_seconds,
                         "monitored_path": str(self.base_path),
@@ -143,14 +143,14 @@ def start_monitoring(self, rebuild_callback: Callable) -> bool:
 
                 # Add diagnostic test - create a test event to verify Observer works
                 import os
-                self.logger.info(f"Observer thread is alive: {self.observer.is_alive()}")
-                self.logger.info(f"Monitored path exists: {os.path.exists(str(self.base_path))}")
-                self.logger.info(f"Event handler is set: {self.event_handler is not None}")
+                self.logger.debug(f"Observer thread is alive: {self.observer.is_alive()}")
+                self.logger.debug(f"Monitored path exists: {os.path.exists(str(self.base_path))}")
+                self.logger.debug(f"Event handler is set: {self.event_handler is not None}")
 
                 # Log current directory for comparison
                 current_dir = os.getcwd()
-                self.logger.info(f"Current working directory: {current_dir}")
-                self.logger.info(f"Are paths same: {os.path.normpath(current_dir) == os.path.normpath(str(self.base_path))}")
+                self.logger.debug(f"Current working directory: {current_dir}")
+                self.logger.debug(f"Are paths same: {os.path.normpath(current_dir) == os.path.normpath(str(self.base_path))}")
 
                 return True
             else:
@@ -350,7 +350,7 @@ def on_any_event(self, event: FileSystemEvent) -> None:
 
         if should_process:
             process_msg = f"Processing file system event: {event.event_type} - {event.src_path}"
-            self.logger.info(process_msg)
+            self.logger.debug(process_msg)
             self.reset_debounce_timer()
         else:
             filter_msg = f"Event filtered out: {event.event_type} - {event.src_path}"
@@ -399,7 +399,7 @@ def should_process_event(self, event: FileSystemEvent) -> bool:
         if is_temp:
             return False
 
-        self.logger.info(f"Event will be processed: {event.src_path}")
+        self.logger.debug(f"Event will be processed: {event.src_path}")
         return True
 
     def is_excluded_path(self, path: Path) -> bool:
@@ -473,7 +473,7 @@ def reset_debounce_timer(self) -> None:
             self.logger.debug("Previous debounce timer cancelled")
 
         timer_msg = f"Starting debounce timer for {self.debounce_seconds} seconds"
-        self.logger.info(timer_msg)
+        self.logger.debug(timer_msg)
 
         self.debounce_timer = Timer(
             self.debounce_seconds,
@@ -491,12 +491,12 @@ def trigger_rebuild(self) -> None:
         if self.rebuild_callback:
             try:
                 callback_msg = "Calling rebuild callback..."
-                self.logger.info(callback_msg)
+                self.logger.debug(callback_msg)
 
                 result = self.rebuild_callback()
 
                 result_msg = f"Rebuild callback completed with result: {result}"
-                self.logger.info(result_msg)
+                self.logger.debug(result_msg)
             except Exception as e:
                 error_msg = f"Rebuild callback failed: {e}"
                 self.logger.error(error_msg)

src/code_index_mcp/services/index_service.py

@@ -8,6 +8,7 @@
 import fnmatch
 import json
 import logging
+import threading
 import time
 from typing import Dict, Any, Optional, Callable
 from mcp.server.fastmcp import Context
@@ -37,6 +38,8 @@ def __init__(self, ctx: Context):
         super().__init__(ctx)
         self.logger = logging.getLogger(__name__)
         self.is_rebuilding = False
+        self._rebuild_lock = threading.Lock()
+        self._executor = None
 
     def rebuild_index(self) -> str:
         """
@@ -178,30 +181,40 @@ def start_background_rebuild(self) -> bool:
         Start index rebuild in background without blocking searches.
         
         This method can be called from any thread (including file watcher threads).
-        Uses ThreadPoolExecutor to run rebuild in a separate thread.
+        Uses a persistent ThreadPoolExecutor with proper concurrency control.
         
         Returns:
             True if rebuild started, False if already in progress
         """
-        self.logger.info("start_background_rebuild called")
+        self.logger.debug("start_background_rebuild called")
 
-        if self.is_rebuilding:
-            self.logger.info("Rebuild already in progress, skipping")
-            return False
+        # Atomic check-and-set with proper locking
+        with self._rebuild_lock:
+            if self.is_rebuilding:
+                self.logger.debug("Rebuild already in progress, skipping")
+                return False
+            
+            # Atomically set rebuilding flag
+            self.is_rebuilding = True
 
         try:
             import concurrent.futures
-            import threading
 
-            self.logger.info("Starting background rebuild in thread pool")
+            # Initialize executor if needed
+            if self._executor is None:
+                self._executor = concurrent.futures.ThreadPoolExecutor(
+                    max_workers=1, 
+                    thread_name_prefix="index_rebuild"
+                )
+                self.logger.debug("Created persistent ThreadPoolExecutor for rebuilds")
+            
+            self.logger.debug("Starting background rebuild in thread pool")
 
             def run_sync_rebuild():
                 """Run the rebuild synchronously in a background thread."""
                 try:
-                    self.is_rebuilding = True
                     start_time = time.time()
-                    
-                    self.logger.info("Starting sync index rebuild...")
+                    self.logger.debug("Starting sync index rebuild...")
 
                     # Build new index using existing sync logic
                     file_count = self._index_project(self.base_path)
@@ -219,43 +232,64 @@ def run_sync_rebuild():
                     self.logger.error("Traceback: %s", traceback.format_exc())
                     return False
                 finally:
-                    self.is_rebuilding = False
-                    self.logger.info("Background rebuild finished, is_rebuilding set to False")
+                    # Thread-safe flag reset
+                    with self._rebuild_lock:
+                        self.is_rebuilding = False
+                    self.logger.debug("Background rebuild finished, is_rebuilding set to False")
+            
+            # Submit to persistent executor
+            future = self._executor.submit(run_sync_rebuild)
+            
+            # Add completion callback
+            def on_complete(fut):
+                try:
+                    result = fut.result()
+                    if result:
+                        self.logger.debug("Background rebuild completed successfully")
+                    else:
+                        self.logger.error("Background rebuild failed")
+                except Exception as e:
+                    self.logger.error(f"Background rebuild thread failed: {e}")
 
-            # Submit to thread pool and don't wait for completion
-            with concurrent.futures.ThreadPoolExecutor(max_workers=1, thread_name_prefix="rebuild") as executor:
-                future = executor.submit(run_sync_rebuild)
-                
-                # Add completion callback
-                def on_complete(fut):
-                    try:
-                        result = fut.result()
-                        if result:
-                            self.logger.info("Background rebuild completed successfully")
-                        else:
-                            self.logger.error("Background rebuild failed")
-                    except Exception as e:
-                        self.logger.error(f"Background rebuild thread failed: {e}")
-                
-                future.add_done_callback(on_complete)
+            future.add_done_callback(on_complete)
 
-            self.logger.info("Background rebuild scheduled successfully")
+            self.logger.debug("Background rebuild scheduled successfully")
             return True
 
         except Exception as e:
+            # Reset flag on error
+            with self._rebuild_lock:
+                self.is_rebuilding = False
+            
             self.logger.error(f"Failed to start background rebuild: {e}")
             import traceback
             self.logger.error("Traceback: %s", traceback.format_exc())
             return False
 
+    def shutdown(self) -> None:
+        """
+        Shutdown the index service and cleanup resources.
+        
+        This should be called when the service is no longer needed to ensure
+        proper cleanup of the ThreadPoolExecutor.
+        """
+        if self._executor is not None:
+            self.logger.debug("Shutting down ThreadPoolExecutor...")
+            self._executor.shutdown(wait=True)  # Wait for current rebuilds to complete
+            self._executor = None
+            self.logger.debug("ThreadPoolExecutor shutdown completed")
+    
     def get_rebuild_status(self) -> dict:
         """
         Get current rebuild status information.
         
         Returns:
             Dictionary with rebuild status
         """
+        with self._rebuild_lock:
+            is_rebuilding = self.is_rebuilding
+        
         return {
-            "is_rebuilding": self.is_rebuilding,
+            "is_rebuilding": is_rebuilding,
             "index_cache_size": len(self.index_cache.get('files', [])) if self.index_cache else 0
         }