82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
"""Cross-process mutation lock for one user's knowledge-base library."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from contextlib import contextmanager, nullcontext
|
|
from pathlib import Path
|
|
from typing import Iterator, Optional
|
|
from uuid import UUID
|
|
|
|
from .file_store import FileLockBusy, interprocess_file_lock
|
|
|
|
|
|
class KbBusyError(RuntimeError):
|
|
"""The target knowledge base is being mutated by another process."""
|
|
|
|
|
|
def _user_root(workspace_dir: Path, user_id: UUID) -> Path:
|
|
return Path(workspace_dir) / "users" / str(user_id)
|
|
|
|
|
|
def kb_lock_path(user_root: Path, kb_name: str) -> Path:
|
|
digest = hashlib.sha256(kb_name.encode("utf-8")).hexdigest()[:32]
|
|
return Path(user_root) / ".kb" / ".locks" / f"{digest}.lock"
|
|
|
|
|
|
@contextmanager
|
|
def kb_mutation_lock(
|
|
workspace_dir: Path,
|
|
user_id: UUID,
|
|
kb_name: str,
|
|
*,
|
|
timeout_seconds: Optional[float] = 0,
|
|
) -> Iterator[None]:
|
|
try:
|
|
with interprocess_file_lock(
|
|
kb_lock_path(_user_root(workspace_dir, user_id), kb_name),
|
|
timeout_seconds=timeout_seconds,
|
|
):
|
|
yield
|
|
except FileLockBusy as e:
|
|
raise KbBusyError(kb_name) from e
|
|
|
|
|
|
@contextmanager
|
|
def kb_mutation_lock_for_path(
|
|
target: Path,
|
|
user_root: Optional[Path],
|
|
*,
|
|
timeout_seconds: Optional[float] = 0,
|
|
) -> Iterator[None]:
|
|
"""Lock the containing ``.kb/<name>`` library; no-op outside ``.kb``."""
|
|
if user_root is None:
|
|
with nullcontext():
|
|
yield
|
|
return
|
|
try:
|
|
rel = Path(target).resolve().relative_to(Path(user_root).resolve())
|
|
except (OSError, ValueError):
|
|
with nullcontext():
|
|
yield
|
|
return
|
|
if len(rel.parts) < 3 or rel.parts[0] != ".kb" or rel.parts[1].startswith("."):
|
|
with nullcontext():
|
|
yield
|
|
return
|
|
try:
|
|
with interprocess_file_lock(
|
|
kb_lock_path(Path(user_root), rel.parts[1]),
|
|
timeout_seconds=timeout_seconds,
|
|
):
|
|
yield
|
|
except FileLockBusy as e:
|
|
raise KbBusyError(rel.parts[1]) from e
|
|
|
|
|
|
def kb_is_locked(workspace_dir: Path, user_id: UUID, kb_name: str) -> bool:
|
|
try:
|
|
with kb_mutation_lock(workspace_dir, user_id, kb_name):
|
|
return False
|
|
except KbBusyError:
|
|
return True
|