Function list_active_locks

ci/util/cache_lock.py:164–202

List all active locks in cache directory with metadata. Args: cache_dir: Global cache directory Returns: List of LockInfo objects

(cache_dir: Path)

def list_active_locks(cache_dir: Path) -> list[LockInfo]:
    """
    List all active locks in cache directory with metadata.

    Args:
        cache_dir: Global cache directory

    Returns:
        List of LockInfo objects
    """
    if not cache_dir.exists():
        return []

    locks: list[LockInfo] = []

    # Query database for all locks
    db = get_lock_database()
    all_db_locks = db.list_all_locks()

    from ci.util.file_lock_rw_util import is_process_alive

    for lock in all_db_locks:
        # Filter to locks relevant to this cache_dir
        lock_name = lock["lock_name"]
        if str(cache_dir) in lock_name:
            pid = lock["owner_pid"]
            stale = not is_process_alive(pid)
            locks.append(
                LockInfo(
                    path=lock_name,
                    pid=pid,
                    operation=lock["operation"],
                    timestamp=str(lock["acquired_at"]),
                    is_stale=stale,
                    hostname=lock["hostname"],
                )
            )

    return locks


def is_artifact_locked(cache_dir: Path, url: str) -> bool: