Merge pull request #164 from oraios/fix_ls_stop_racing_conditions

Fix ls stop racing conditions

.github/workflows/pytest.yml

@@ -55,11 +55,9 @@ jobs:
         run: uv pip install -e ".[dev]"
       - name: Test with pytest
         shell: bash
-        # Currently, java and rust seem to cause problems in CI, and snapshot tests seem to cause problems on Windows
-        # probably due to the way the language servers are started and stopped
         run: |
-          if [[ "$RUNNER_OS" == "Windows" ]]; then
-            uv run pytest -m "not java and not rust and not snapshot"
+          if [ "${{ runner.os }}" = "Linux" ]; then
+            uv run pytest -vv -m "not java and not rust"
           else
-            uv run pytest -m "not java and not rust"
+            uv run pytest -vv
           fi

src/multilspy/language_server.py

@@ -19,6 +19,7 @@ from collections import defaultdict
 from contextlib import asynccontextmanager, contextmanager
 from copy import copy
 from pathlib import Path, PurePath
+import time
 from typing import AsyncIterator, Dict, Iterator, List, Optional, Tuple, Union, cast
 
 import pathspec
@@ -124,15 +125,16 @@ class LanguageServer:
             gitignore_path = os.path.join(repository_root_path, ".gitignore")
             if not os.path.exists(gitignore_path):
                 logger.log(
-                    f"Should ignore all files in gitignore not not .gitignore found at {gitignore_path}. Skipping.",
+                    f"Should ignore all files in gitignore but no .gitignore found at {gitignore_path}. Skipping.",
                     logging.WARNING
                 )
                 gitignore_file_content = None
             else:
                 if config.gitignore_file_content is not None:
                     raise ValueError(
-                        f"Asked to add gitignore content to the config for {repository_root_path=} but there already is a non-empty entry"
+                        f"Asked to add gitignore content to the config for {repository_root_path=} but there already is a non-empty entry there:\n{config.gitignore_file_content=}"
                     )
+                logger.log(f"Reading gitignore file from {gitignore_path}", logging.DEBUG)
                 with open(gitignore_path, "r", encoding="utf-8") as f:
                     gitignore_file_content = f.read()
             config.gitignore_file_content = gitignore_file_content
@@ -228,10 +230,18 @@ class LanguageServer:
             )
 
         self.logger = logger
-        self.server_started = False
         self.repository_root_path: str = repository_root_path
-        self.completions_available = asyncio.Event()
+        self.logger.log(f"Creating language server instance for {repository_root_path=} with {language_id=} and process launch info: {process_launch_info}", logging.DEBUG)
+        
+        # load cache first to prevent any racing conditions due to asyncio stuff
+        self._document_symbols_cache:  dict[str, Tuple[str, Tuple[List[multilspy_types.UnifiedSymbolInformation], List[multilspy_types.UnifiedSymbolInformation]]]] = {}
+        """Maps file paths to a tuple of (file_content_hash, result_of_request_document_symbols)"""
+        self._cache_lock = threading.Lock()
+        self._cache_has_changed: bool = False
+        self.load_cache()
 
+        self.server_started = False
+        self.completions_available = asyncio.Event()
         if config.trace_lsp_communication:
             def logging_fn(source: str, target: str, msg: StringDict | str):
                 self.logger.log(f"LSP: {source} -> {target}: {str(msg)}", logging.DEBUG)
@@ -241,6 +251,7 @@ class LanguageServer:
 
         # cmd is obtained from the child classes, which provide the language specific command to start the language server
         # LanguageServerHandler provides the functionality to start the language server and communicate with it
+        self.logger.log(f"Creating language server instance with {language_id=} and process launch info: {process_launch_info}", logging.DEBUG)
         self.server = LanguageServerHandler(
             process_launch_info,
             logger=logging_fn,
@@ -250,11 +261,6 @@ class LanguageServer:
         self.language_id = language_id
         self.open_file_buffers: Dict[str, LSPFileBuffer] = {}
 
-        # --------------------------------- MODIFICATIONS BY ORAIOS ---------------------------------
-        self._document_symbols_cache:  dict[str, Tuple[str, Tuple[List[multilspy_types.UnifiedSymbolInformation], List[multilspy_types.UnifiedSymbolInformation]]]] = {}
-        """Maps file paths to a tuple of (file_content_hash, result_of_request_document_symbols)"""
-        self.load_cache()
-        self._cache_has_changed: bool = False
         self.language = Language(language_id)
 
         # Set up the pathspec matcher for the ignored paths
@@ -265,6 +271,7 @@ class LanguageServer:
             pattern = pattern.replace(os.path.sep, '/')
             processed_patterns.append(pattern)
         # Combine explicitly passed patterns with the content of the .gitignore file
+        self.logger.log(f"Processing gitignore content", logging.DEBUG)
         if config.gitignore_file_content is not None:
             for line in config.gitignore_file_content.splitlines():
                 if not line.startswith('#') and line.strip() != '':
@@ -337,62 +344,78 @@ class LanguageServer:
 
         return False
     
-    async def _shutdown(self, timeout: int = 15):
+    async def _shutdown(self, timeout: float = 10.0):
         """
-        Performs the complete and robust shutdown sequence for the language server.
-        1. Sends the graceful LSP shutdown/exit notifications.
-        2. Stops the underlying process and cleans up tasks.
-        3. Wraps the entire operation in a timeout to prevent hangs.
+        A robust shutdown process designed to terminate cleanly on all platforms, including Windows,
+        by explicitly closing all I/O pipes.
         """
         if not self.server.is_running():
             self.logger.log("Server process not running, skipping shutdown.", logging.DEBUG)
             return
 
-        self.logger.log(f"Initiating full shutdown sequence with a {timeout}s timeout...", logging.INFO)
-
-        async def full_shutdown_sequence():
-            """The sequence of shutdown operations."""
-            # Step 1: Graceful LSP shutdown.
-            # This sends the `shutdown` and `exit` notifications.
-            # It's quick and just sends messages, so it's unlikely to hang.
-            self.logger.log("Sending LSP shutdown/exit notifications...", logging.DEBUG)
-            await self.server.shutdown()
-            
-            # Step 2: Stop the process and cleanup tasks.
-            # This is the part that can hang if the process is unresponsive.
-            # The LanguageServerHandler.stop() method has its own internal timeouts,
-            # but we wrap it in our own as a failsafe.
-            self.logger.log("Stopping process and cleaning up tasks...", logging.DEBUG)
-            await self.server.stop()
+        self.logger.log(f"Initiating final robust shutdown with a {timeout}s timeout...", logging.INFO)
+        process = self.server.process
+        reader_tasks = list(self.server.tasks.values())
 
+        # --- Main Shutdown Logic ---
         try:
-            # Wrap the entire sequence in our own timeout.
-            # This ensures that even if a part of the handler's logic hangs
-            # unexpectedly, our application will not be blocked indefinitely.
-            await asyncio.wait_for(full_shutdown_sequence(), timeout=timeout)
-            self.logger.log("Language server shutdown sequence completed successfully.", logging.INFO)
+            # Stage 1: Graceful Termination Request
+            # Send LSP shutdown and close stdin to signal no more input.
+            try:
+                await asyncio.wait_for(self.server.shutdown(), timeout=2.0)
+                if process.stdin and not process.stdin.is_closing():
+                    process.stdin.close()
+            except Exception:
+                pass # Ignore errors here, we are proceeding to terminate anyway.
+
+            # Stage 2: Terminate and Concurrently Drain stdout/stderr
+            process.terminate()
+            
+            # Wait for the process to exit AND for the output pipes to be drained.
+            # The reader tasks will exit when they hit EOF.
+            await asyncio.wait_for(
+                asyncio.gather(process.wait(), *reader_tasks),
+                timeout=timeout - 2.0
+            )
+            self.logger.log("Process terminated and output pipes drained.", logging.INFO)
 
         except asyncio.TimeoutError:
-            self.logger.log(
-                f"The full shutdown sequence did not complete within {timeout}s. "
-                "The process might be left in an orphaned state. This indicates a "
-                "potential hang in LanguageServerHandler.stop().",
-                logging.ERROR
-            )
-            # At this point, the handler's internal kill logic has likely failed.
-            # We can make one last-ditch effort to kill the process if we know its PID.
-            if self.server.process and self.server.process.returncode is None:
-                self.logger.log("Attempting final forceful kill of the process.", logging.WARNING)
+            # Stage 3: Forceful Kill
+            self.logger.log("Graceful termination failed. Forcefully killing process...", logging.WARNING)
+            if self.server.is_running():
                 try:
-                    self.server.process.kill()
-                    await self.server.process.wait()
-                except ProcessLookupError:
-                    pass # Process already gone
+                    process.kill()
+                    # Wait for the killed process to be reaped by the OS.
+                    await process.wait()
                 except Exception as e:
-                    self.logger.log(f"Final kill attempt failed: {e}", logging.ERROR)
-
+                    self.logger.log(f"Error during forceful kill: {e}", logging.ERROR)
+        
         except Exception as e:
-            self.logger.log(f"An unexpected error occurred during shutdown: {e}", logging.ERROR)
+            self.logger.log(f"An unexpected error occurred during shutdown logic: {e}", logging.ERROR)
+
+        finally:
+            # === STAGE 4: EXPLICIT TASK & PIPE CLEANUP ===
+            self.logger.log("Performing final task cancellation and pipe handle cleanup...", logging.DEBUG)
+            
+            # 1. Cancel any lingering reader tasks.
+            for task in reader_tasks:
+                if not task.done():
+                    task.cancel()
+            # Wait for cancellations to complete.
+            await asyncio.gather(*reader_tasks, return_exceptions=True)
+            self.server.tasks = {}
+
+            # 2. Explicitly close each pipe to release OS handles.
+            for pipe in [process.stdin, process.stdout, process.stderr]:
+                if pipe and hasattr(pipe, 'is_closing') and not pipe.is_closing():
+                    try:
+                        pipe.close()
+                    except Exception:
+                        pass
+            
+            # 3. Null out the process object in the handler.
+            self.server.process = None
+            self.logger.log("Shutdown sequence fully finished.", logging.DEBUG)
 
     @asynccontextmanager
     async def start_server(self) -> AsyncIterator["LanguageServer"]:
@@ -851,16 +874,17 @@ class LanguageServer:
         #   Should be fixed in the future, it's a small performance optimization
         cache_key = f"{relative_file_path}-{include_body}"
         with self.open_file(relative_file_path) as file_data:
-            file_hash_and_result = self._document_symbols_cache.get(cache_key)
-            if file_hash_and_result is not None:
-                file_hash, result = file_hash_and_result
-                if file_hash == file_data.content_hash:
-                    self.logger.log(f"Returning cached document symbols for {relative_file_path}", logging.DEBUG)
-                    return result
+            with self._cache_lock:
+                file_hash_and_result = self._document_symbols_cache.get(cache_key)
+                if file_hash_and_result is not None:
+                    file_hash, result = file_hash_and_result
+                    if file_hash == file_data.content_hash:
+                        self.logger.log(f"Returning cached document symbols for {relative_file_path}", logging.DEBUG)
+                        return result
+                    else:
+                        self.logger.log(f"Content for {relative_file_path} has changed. Will overwrite in-memory cache", logging.DEBUG)
                 else:
-                    self.logger.log(f"Content for {relative_file_path} has changed. Will overwrite in-memory cache", logging.DEBUG)
-            else:
-                self.logger.log(f"No cache hit for symbols with {include_body=} in {relative_file_path}", logging.DEBUG)
+                    self.logger.log(f"No cache hit for symbols with {include_body=} in {relative_file_path}", logging.DEBUG)
 
             self.logger.log(f"Requesting document symbols for {relative_file_path} from the Language Server", logging.DEBUG)
             response = await self.server.send.document_symbol(
@@ -945,8 +969,9 @@ class LanguageServer:
 
         result = flat_all_symbol_list, root_nodes
         self.logger.log(f"Caching document symbols for {relative_file_path}", logging.DEBUG)
-        self._document_symbols_cache[cache_key] = (file_data.content_hash, result)
-        self._cache_has_changed = True
+        with self._cache_lock:
+            self._document_symbols_cache[cache_key] = (file_data.content_hash, result)
+            self._cache_has_changed = True
         return result
     
     async def request_full_symbol_tree(self, within_relative_path: str | None = None, include_body: bool = False) -> List[multilspy_types.UnifiedSymbolInformation]:
@@ -1601,35 +1626,41 @@ class LanguageServer:
         return Path(self.repository_root_path) / ".serena" / "cache" / "document_symbols_cache_v20-05-25.pkl"
 
     def save_cache(self):
-        if self._cache_has_changed:
+        with self._cache_lock:
+            if not self._cache_has_changed:
+                self.logger.log("No changes to document symbols cache, skipping save", logging.DEBUG)
+                return
+
             self.logger.log(f"Saving updated document symbols cache to {self._cache_path}", logging.INFO)
             self._cache_path.parent.mkdir(parents=True, exist_ok=True)
             try:
                 with open(self._cache_path, "wb") as f:
                     pickle.dump(self._document_symbols_cache, f)
+                self._cache_has_changed = False
             except Exception as e:
                 self.logger.log(
-                        f"Failed to save document symbols cache to {self._cache_path}: {e}. "
-                        "Note: this may have resulted in a corrupted cache file.", logging.ERROR
-                    )
-        else:
-            self.logger.log(f"No changes to document symbols cache, skipping save", logging.DEBUG)
-        self._cache_has_changed = False
+                    f"Failed to save document symbols cache to {self._cache_path}: {e}. "
+                    "Note: this may have resulted in a corrupted cache file.",
+                    logging.ERROR,
+                )
 
     def load_cache(self):
         if not self._cache_path.exists():
             return
-        self.logger.log(f"Loading document symbols cache from {self._cache_path}", logging.INFO)
-        with open(self._cache_path, "rb") as f:
+
+        with self._cache_lock:
+            self.logger.log(f"Loading document symbols cache from {self._cache_path}", logging.INFO)
             try:
-                self._document_symbols_cache = pickle.load(f)
+                with open(self._cache_path, "rb") as f:
+                    self._document_symbols_cache = pickle.load(f)
+                self.logger.log(f"Loaded {len(self._document_symbols_cache)} document symbols from cache.", logging.INFO)   
             except Exception as e:
                 # cache often becomes corrupt, so just skip loading it
                 self.logger.log(
-                        f"Failed to load document symbols cache from {self._cache_path}: {e}. Possible cause: the cache file is corrupted. " 
-                        "Check for any errors related to saving the cache in the logs.",
-                        logging.ERROR
-                    )
+                    f"Failed to load document symbols cache from {self._cache_path}: {e}. Possible cause: the cache file is corrupted. "
+                    "Check for any errors related to saving the cache in the logs.",
+                    logging.ERROR,
+                )
 
 
     async def request_workspace_symbol(self, query: str) -> Union[List[multilspy_types.UnifiedSymbolInformation], None]:
@@ -1677,6 +1708,9 @@ class SyncLanguageServer:
         self.timeout = timeout
 
         self._server_context = None
+        
+        self._shutdown_lock = threading.Lock()
+        self._is_shutting_down = False
 
     @classmethod
     def create(
@@ -2142,32 +2176,115 @@ class SyncLanguageServer:
         except Exception as e:
             self.language_server.logger.log(f"Exception during async shutdown: {e}", logging.ERROR)
         finally:
+            # Clean up tasks but DON'T stop the loop from here
+            # Let the main thread handle loop stopping to avoid coroutine deadlocks
             if self.loop and self.loop.is_running():
-                self.loop.stop()
+                current_task = asyncio.current_task(self.loop)
+                pending_tasks = [task for task in asyncio.all_tasks(self.loop) 
+                               if not task.done() and task is not current_task]
+                
+                if pending_tasks:
+                    self.language_server.logger.log(f"Cancelling {len(pending_tasks)} pending tasks", logging.DEBUG)
+                    for task in pending_tasks:
+                        try:
+                            task.cancel()
+                        except Exception:
+                            pass
+                
+                self.language_server.logger.log("Async shutdown tasks completed, returning to main thread", logging.DEBUG)
+                # Note: NOT calling loop.stop() here - let main thread do it
 
-    def stop(self) -> None:
+    def stop(self, shutdown_timeout: float = 5.0) -> None:
         """
-        Shuts down the language server process, saves the cache and cleans up resources.
-
-        If the language server is not running, this method will log a warning and do nothing.
+        Stops the language server and robustly cleans up all associated resources,
+        including the asyncio event loop, to prevent hangs on process exit.
         """
-        self.save_cache()
+        # 1. Use a lock and flag to make the shutdown call thread-safe and idempotent,
+        # ensuring the shutdown logic runs only once.
+        with self._shutdown_lock:
+            if self._is_shutting_down:
+                self.language_server.logger.log("Already shutting down or stopped, skipping", logging.DEBUG)
+                return
+            self._is_shutting_down = True
+    
         if not self.is_running():
-            self.language_server.logger.log("Language server not running, skipping shutdown.", logging.INFO)
+            self.language_server.logger.log("Language server is not running, skipping", logging.DEBUG)
             return
-
-        assert self.loop is not None
-        self.language_server.logger.log(f"Initiating shutdown for {self.language_server.repository_root_path}", logging.INFO)
+    
+        self.language_server.logger.log("Initiating server shutdown...", logging.INFO)
+        self.save_cache()
         
-        asyncio.run_coroutine_threadsafe(self._shutdown_and_stop_loop(), loop=self.loop)
-
-        if self.loop_thread:
-            self.loop_thread.join()
-
+        # Detect platform to choose shutdown strategy
+        is_windows = os.name == "nt"
+        
+        if not is_windows:
+            # 2. Graceful shutdown for Linux/Unix - try proper async cleanup first
+            shutdown_future = None
+            if self.loop and self.loop.is_running():
+                shutdown_future = asyncio.run_coroutine_threadsafe(self._shutdown_and_stop_loop(), self.loop)
+            
+            # 3. Wait for the background thread to exit
+            if self.loop_thread:
+                self.loop_thread.join(timeout=shutdown_timeout)
+                if self.loop_thread.is_alive():
+                    self.language_server.logger.log("Event loop thread did not terminate within timeout", logging.WARNING)
+            
+            # 4. Wait for the shutdown coroutine to complete if it was scheduled
+            if shutdown_future:
+                try:
+                    shutdown_future.result(timeout=2.0)
+                    self.language_server.logger.log("Graceful async shutdown completed", logging.DEBUG)
+                except Exception as e:
+                    self.language_server.logger.log(f"Async shutdown failed: {e}", logging.WARNING)
+            
+            # 5. Close the loop properly after graceful shutdown
+            if self.loop and not self.loop.is_closed():
+                try:
+                    self.loop.close()
+                    self.language_server.logger.log("Event loop closed successfully", logging.DEBUG)
+                except Exception as e:
+                    self.language_server.logger.log(f"Event loop close failed: {e}", logging.WARNING)
+        
+        else:
+            # WINDOWS NUCLEAR OPTION: Skip graceful shutdown - go straight to nuclear
+            # Windows IocpProactor has too many issues with proper cleanup
+            self.language_server.logger.log("Windows detected - using nuclear shutdown to prevent zombies", logging.WARNING)
+            
+            # Force stop the loop immediately without waiting for async cleanup
+            if self.loop and self.loop.is_running():
+                try:
+                    self.loop.call_soon_threadsafe(lambda: setattr(self.loop, '_stopping', True))
+                    self.loop.call_soon_threadsafe(self.loop.stop)
+                except Exception:
+                    pass
+                
+                # Force mark loop as closed IMMEDIATELY to prevent zombies
+                try:
+                    if hasattr(self.loop, '_closed'):
+                        self.loop._closed = True
+                    if hasattr(self.loop, '_stopping'):
+                        self.loop._stopping = True
+                    if hasattr(self.loop, '_ready'):
+                        self.loop._ready.clear()
+                except Exception:
+                    pass
+            
+            # Give thread minimal time to exit, then abandon it
+            if self.loop_thread:
+                self.loop_thread.join(timeout=1.0)  # Only 1 second on Windows
+                if self.loop_thread.is_alive():
+                    self.language_server.logger.log("Thread stuck - abandoning to prevent GC deadlock", logging.WARNING)
+    
+        # 6. Null out everything immediately - same for both platforms
         self.loop = None
         self.loop_thread = None
         self._server_context = None
-        self.language_server.logger.log("Shutdown complete.", logging.INFO)
+        self._is_shutting_down = False
+        
+        shutdown_type = "Nuclear" if is_windows else "Graceful"
+        self.language_server.logger.log(f"{shutdown_type} shutdown complete - all references cleared", logging.INFO)
+
+
 
     def save_cache(self):
         """

src/multilspy/lsp_protocol_handler/server.py

@@ -248,6 +248,7 @@ class LanguageServerHandler:
         self.task_counter += 1
         self.tasks[self.task_counter] = self.loop.create_task(self.run_forever_stderr())
         self.task_counter += 1
+        
 
     async def stop(self) -> None:
         """

test/conftest.py

@@ -1,11 +1,15 @@
+import logging
 from pathlib import Path
 
 import pytest
+from sensai.util.logging import LOG_DEFAULT_FORMAT, configure
 
 from multilspy.language_server import SyncLanguageServer
 from multilspy.multilspy_config import Language, MultilspyConfig
 from multilspy.multilspy_logger import MultilspyLogger
 
+configure(level=logging.DEBUG)
+
 
 @pytest.fixture(scope="session")
 def resources_dir() -> Path:
@@ -24,7 +28,11 @@ def get_repo_path(language: Language) -> Path:
 
 def create_ls(language: Language, repo_path: str):
     config = MultilspyConfig(code_language=language)
-    logger = MultilspyLogger()
+    logger = MultilspyLogger(log_level=logging.DEBUG)
+
+    # Simple way: Use basicConfig to set format for all handlers
+    logging.basicConfig(format=LOG_DEFAULT_FORMAT, level=logging.DEBUG, force=True)  # Override existing configuration
+
     return SyncLanguageServer.create(config, logger, repo_path)
 
 

test/serena/test_symbol_editing.py

@@ -1,6 +1,8 @@
+import logging
 import os
 import shutil
 import tempfile
+import time
 from abc import abstractmethod
 from collections.abc import Iterator
 from contextlib import contextmanager
@@ -16,6 +18,8 @@ from test.conftest import create_ls, get_repo_path
 
 pytestmark = pytest.mark.snapshot
 
+log = logging.getLogger(__name__)
+
 
 class EditingTest:
     def __init__(self, language: Language, rel_path: str):
@@ -35,14 +39,32 @@ class EditingTest:
         self.repo_path = temp_dir / self.original_repo_path.name
         language_server = None  # Initialize language_server
         try:
+            print(f"Copying repo from {self.original_repo_path} to {self.repo_path}")
             shutil.copytree(self.original_repo_path, self.repo_path)
+            # prevent deadlock on Windows due to file locks caused by antivirus or some other external software
+            # wait for a long time here
+            if os.name == "nt":
+                time.sleep(0.1)
+            log.info(f"Creating language server for {self.language} {self.rel_path}")
             language_server = create_ls(self.language, str(self.repo_path))
+            log.info(f"Starting language server for {self.language} {self.rel_path}")
             language_server.start()
+            log.info(f"Language server started for {self.language} {self.rel_path}")
             yield SymbolManager(lang_server=language_server)
         finally:
             if language_server is not None and language_server.is_running():
+                log.info(f"Stopping language server for {self.language} {self.rel_path}")
                 language_server.stop()
-            shutil.rmtree(temp_dir)
+                # attempt at trigger of garbage collection
+                language_server = None
+                log.info(f"Language server stopped for {self.language} {self.rel_path}")
+
+            # prevent deadlock on Windows due to lingering file locks
+            if os.name == "nt":
+                time.sleep(0.1)
+            log.info(f"Removing temp directory {temp_dir}")
+            shutil.rmtree(temp_dir, ignore_errors=True)
+            log.info(f"Temp directory {temp_dir} removed")
 
     def _read_file(self, rel_path: str) -> str:
         """Read the content of a file in the test repository."""