feat(robot): add advisory file locking to SQLite cache

Add shared/exclusive advisory file locking (fcntl.flock) to prevent
`prune` from deleting a cache database while another process (language
server, analyze run) is using it.

Author: d-biehl

packages/analyze/src/robotcode/analyze/cache/cli.py

@@ -6,7 +6,14 @@
 from robotcode.plugin import Application, OutputFormat, pass_application
 from robotcode.robot.config.loader import load_robot_config_from_path
 from robotcode.robot.config.utils import get_config_files
-from robotcode.robot.diagnostics.data_cache import CACHE_DIR_NAME, CacheSection, SqliteDataCache, build_cache_dir
+from robotcode.robot.diagnostics.data_cache import (
+    _LOCK_FILE_NAME,
+    CACHE_DIR_NAME,
+    CacheSection,
+    SqliteDataCache,
+    build_cache_dir,
+    exclusive_cache_lock,
+)
 
 from ..config import AnalyzeConfig
 
@@ -371,11 +378,12 @@ def _resolve_cache_root(
 
 
 @cache_group.command(name="prune")
+@click.option("--force", is_flag=True, help="Force prune even if cache is in use by another process.")
 @click.argument(
     "paths", nargs=-1, type=click.Path(exists=True, dir_okay=True, file_okay=True, readable=True, path_type=Path)
 )
 @pass_application
-def cache_prune(app: Application, paths: Tuple[Path, ...]) -> None:
+def cache_prune(app: Application, force: bool, paths: Tuple[Path, ...]) -> None:
     """\
     Remove the entire cache directory.
 
@@ -390,5 +398,19 @@ def cache_prune(app: Application, paths: Tuple[Path, ...]) -> None:
         app.echo(f"Cache directory does not exist: {cache_root}")
         return
 
+    if not force:
+        locked_dirs = []
+        for lock_file in cache_root.rglob(_LOCK_FILE_NAME):
+            with exclusive_cache_lock(lock_file.parent) as acquired:
+                if not acquired:
+                    locked_dirs.append(lock_file.parent)
+
+        if locked_dirs:
+            app.echo("Cannot prune: cache is in use by another process.")
+            for d in locked_dirs:
+                app.echo(f"  Locked: {d}")
+            app.echo("Use --force to override.")
+            return
+
     shutil.rmtree(cache_root)
     app.echo(f"Removed {cache_root}")

packages/robot/src/robotcode/robot/diagnostics/data_cache.py

@@ -1,10 +1,12 @@
+import os
 import pickle
 import sqlite3
 import sys
+from contextlib import contextmanager
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
-from typing import Any, Generic, List, Optional, Tuple, Type, TypeVar, Union, cast
+from typing import Any, Generic, Iterator, List, Optional, Tuple, Type, TypeVar, Union, cast
 
 from ..utils import get_robot_version_str
 
@@ -80,6 +82,74 @@ def data(self) -> _D:
 _TABLE_NAMES = [s.value for s in CacheSection]
 
 CACHE_DIR_NAME = ".robotcode_cache"
+_LOCK_FILE_NAME = "cache.lock"
+
+
+def _acquire_shared_lock(cache_dir: Path) -> Optional[int]:
+    """Acquire a shared advisory lock on the cache directory.
+
+    Returns the file descriptor on Unix, None on Windows
+    (Windows already prevents deletion of open files).
+    """
+    if sys.platform == "win32":
+        return None
+
+    import fcntl
+
+    lock_path = cache_dir / _LOCK_FILE_NAME
+    fd = os.open(str(lock_path), os.O_RDWR | os.O_CREAT, 0o666)
+    try:
+        fcntl.flock(fd, fcntl.LOCK_SH)
+    except Exception:
+        os.close(fd)
+        raise
+    return fd
+
+
+def _release_lock(fd: Optional[int]) -> None:
+    """Release an advisory lock acquired via _acquire_shared_lock."""
+    if fd is None:
+        return
+    try:
+        if sys.platform != "win32":
+            import fcntl
+
+            fcntl.flock(fd, fcntl.LOCK_UN)
+    finally:
+        os.close(fd)
+
+
+@contextmanager
+def exclusive_cache_lock(cache_dir: Path) -> Iterator[bool]:
+    """Context manager that tries to acquire an exclusive lock on a cache directory.
+
+    Yields True if the lock was acquired (cache is not in use),
+    False if another process holds a shared lock (cache is in use).
+    The lock is held until the context manager exits.
+    """
+    if sys.platform == "win32":
+        yield True
+        return
+
+    import fcntl
+
+    lock_path = cache_dir / _LOCK_FILE_NAME
+    if not lock_path.exists():
+        yield True
+        return
+
+    fd = os.open(str(lock_path), os.O_RDWR)
+    try:
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        yield True
+    except OSError:
+        yield False
+    finally:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+        except OSError:
+            pass
+        os.close(fd)
 
 
 def build_cache_dir(base_path: Path) -> Path:
@@ -109,6 +179,8 @@ def __init__(self, cache_dir: Path, app_version: str = "") -> None:
                 "utf-8",
             )
 
+        self._lock_fd = _acquire_shared_lock(cache_dir)
+
         db_path = cache_dir / "cache.db"
         self._conn = sqlite3.connect(str(db_path), check_same_thread=False)
         self._conn.execute("PRAGMA journal_mode=WAL")
@@ -177,6 +249,8 @@ def save_entry(
 
     def close(self) -> None:
         self._conn.close()
+        _release_lock(self._lock_fd)
+        self._lock_fd = None
 
     @property
     def db_path(self) -> Path:

tests/robotcode/robot/diagnostics/test_data_cache_locking.py

@@ -0,0 +1,142 @@
+"""Tests for SqliteDataCache advisory file locking."""
+
+import sys
+from pathlib import Path
+
+import pytest
+
+from robotcode.robot.diagnostics.data_cache import (
+    _LOCK_FILE_NAME,
+    CacheSection,
+    SqliteDataCache,
+    exclusive_cache_lock,
+)
+
+pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="Advisory file locking is Unix-only")
+
+
+class TestSharedLocking:
+    def test_lock_file_created_on_open(self, tmp_path: Path) -> None:
+        cache_dir = tmp_path / "cache"
+        cache = SqliteDataCache(cache_dir)
+        assert (cache_dir / _LOCK_FILE_NAME).exists()
+        cache.close()
+
+    def test_two_caches_can_open_same_dir(self, tmp_path: Path) -> None:
+        cache_dir = tmp_path / "cache"
+        cache1 = SqliteDataCache(cache_dir)
+        cache2 = SqliteDataCache(cache_dir)
+
+        # Both can read and write
+        cache1.save_entry(CacheSection.LIBRARY, "e1", None, "data1")
+        cache2.save_entry(CacheSection.LIBRARY, "e2", None, "data2")
+
+        entry1 = cache2.read_entry(CacheSection.LIBRARY, "e1", str, str)
+        assert entry1 is not None
+        assert entry1.data == "data1"
+
+        cache1.close()
+        cache2.close()
+
+    def test_lock_released_on_close(self, tmp_path: Path) -> None:
+        cache_dir = tmp_path / "cache"
+        cache = SqliteDataCache(cache_dir)
+        cache.close()
+
+        with exclusive_cache_lock(cache_dir) as acquired:
+            assert acquired, "Exclusive lock should succeed after cache is closed"
+
+
+class TestExclusiveLocking:
+    def test_exclusive_blocked_by_shared(self, tmp_path: Path) -> None:
+        cache_dir = tmp_path / "cache"
+        cache = SqliteDataCache(cache_dir)
+
+        with exclusive_cache_lock(cache_dir) as acquired:
+            assert not acquired, "Exclusive lock should fail while cache is open"
+
+        cache.close()
+
+    def test_exclusive_blocked_by_multiple_shared(self, tmp_path: Path) -> None:
+        cache_dir = tmp_path / "cache"
+        cache1 = SqliteDataCache(cache_dir)
+        cache2 = SqliteDataCache(cache_dir)
+
+        with exclusive_cache_lock(cache_dir) as acquired:
+            assert not acquired
+
+        cache1.close()
+
+        # Still blocked by cache2
+        with exclusive_cache_lock(cache_dir) as acquired:
+            assert not acquired
+
+        cache2.close()
+
+        # Now should succeed
+        with exclusive_cache_lock(cache_dir) as acquired:
+            assert acquired
+
+    def test_exclusive_succeeds_when_no_cache_open(self, tmp_path: Path) -> None:
+        cache_dir = tmp_path / "cache"
+        cache = SqliteDataCache(cache_dir)
+        cache.close()
+
+        with exclusive_cache_lock(cache_dir) as acquired:
+            assert acquired
+
+    def test_exclusive_succeeds_when_no_lock_file(self, tmp_path: Path) -> None:
+        cache_dir = tmp_path / "cache"
+        cache_dir.mkdir(parents=True)
+        # No lock file exists
+
+        with exclusive_cache_lock(cache_dir) as acquired:
+            assert acquired
+
+    def test_exclusive_lock_held_during_context(self, tmp_path: Path) -> None:
+        cache_dir = tmp_path / "cache"
+        cache = SqliteDataCache(cache_dir)
+        cache.close()
+
+        with exclusive_cache_lock(cache_dir) as acquired:
+            assert acquired
+            # While we hold exclusive, a new shared lock should block
+            # (opening a new SqliteDataCache would try LOCK_SH which
+            # is compatible with LOCK_EX held by the same process on Linux,
+            # so we test with a raw flock instead)
+            import fcntl
+            import os
+
+            fd = os.open(str(cache_dir / _LOCK_FILE_NAME), os.O_RDWR)
+            try:
+                with pytest.raises(BlockingIOError):
+                    fcntl.flock(fd, fcntl.LOCK_SH | fcntl.LOCK_NB)
+            finally:
+                os.close(fd)
+
+
+class TestLockingWithPrune:
+    def test_prune_safe_when_cache_closed(self, tmp_path: Path) -> None:
+        import shutil
+
+        cache_dir = tmp_path / "cache"
+        cache = SqliteDataCache(cache_dir)
+        cache.save_entry(CacheSection.LIBRARY, "e1", None, "data")
+        cache.close()
+
+        with exclusive_cache_lock(cache_dir) as acquired:
+            assert acquired
+            shutil.rmtree(cache_dir)
+
+        assert not cache_dir.exists()
+
+    def test_prune_blocked_when_cache_open(self, tmp_path: Path) -> None:
+        cache_dir = tmp_path / "cache"
+        cache = SqliteDataCache(cache_dir)
+
+        with exclusive_cache_lock(cache_dir) as acquired:
+            assert not acquired
+            # Cache dir should still exist
+            assert cache_dir.exists()
+
+        cache.close()