Class FileLock

ci/util/file_lock_rw.py:122–222

Inter-process file lock wrapper backed by SQLite database. Provides a context manager for acquiring/releasing locks safely. Automatically detects and breaks stale locks (dead processes). Supports both read and write lock modes.

class FileLock:
    """
    Inter-process file lock wrapper backed by SQLite database.

    Provides a context manager for acquiring/releasing locks safely.
    Automatically detects and breaks stale locks (dead processes).
    Supports both read and write lock modes.
    """

    def __init__(
        self,
        lock_file_path: Path,
        timeout: float | None = None,
        operation: str = "lock",
        mode: str = "write",
    ):
        """
        Initialize file lock.

        Args:
            lock_file_path: Path to the lock file (used as lock name)
            timeout: Optional timeout in seconds (None = block indefinitely)
            operation: Description of the operation (for debugging)
            mode: Lock mode - 'read' or 'write' (default: 'write')
        """
        self.lock_file_path = Path(lock_file_path)
        self.timeout = timeout
        self.operation = operation
        self._mode = mode
        self._lock_name = str(self.lock_file_path)
        self._db: LockDatabase = get_lock_database(self.lock_file_path)
        self._acquired = False

    def __enter__(self) -> "FileLock":
        """Acquire the lock with stale lock detection and retry."""
        from ci.util.global_interrupt_handler import is_interrupted

        my_pid = os.getpid()
        hostname = platform.node()
        start_time = time.time()
        stale_check_done = False

        while True:
            if is_interrupted():
                raise KeyboardInterrupt()

            # Try to acquire
            acquired = self._db.try_acquire(
                self._lock_name, my_pid, hostname, self.operation, mode=self._mode
            )

            if acquired:
                self._acquired = True
                return self

            # Check for stale locks (once per acquisition attempt cycle)
            if not stale_check_done:
                if self._db.is_lock_stale(self._lock_name):