File size: 3,042 Bytes
791c076 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """Filesystem primitives shared by DataForge transaction apply/revert paths."""
from __future__ import annotations
import hashlib
import os
import secrets
import time
from collections.abc import Iterator
from contextlib import contextmanager, suppress
from datetime import UTC, datetime
from pathlib import Path
class SourceLockError(RuntimeError):
"""Raised when a source file lock cannot be acquired."""
def fsync_parent_directory(path: Path) -> None:
"""Best-effort fsync of a path's parent directory after atomic replacement."""
parent = path.resolve().parent
try:
fd = os.open(parent, os.O_RDONLY)
except OSError:
return
try:
os.fsync(fd)
finally:
os.close(fd)
def atomic_write_bytes(path: Path, payload: bytes) -> None:
"""Write bytes to ``path`` through an atomic same-directory replacement."""
resolved = path.resolve()
resolved.parent.mkdir(parents=True, exist_ok=True)
temp_path = resolved.with_name(f".{resolved.name}.{secrets.token_hex(8)}.tmp")
try:
with temp_path.open("xb") as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
os.replace(temp_path, resolved)
fsync_parent_directory(resolved)
finally:
if temp_path.exists():
temp_path.unlink()
def lock_path_for(source_path: Path) -> Path:
"""Return the filesystem lock path for a source file."""
digest = hashlib.sha256(str(source_path.resolve()).encode("utf-8")).hexdigest()[:24]
return source_path.resolve().parent / ".dataforge" / "locks" / f"{digest}.lock"
@contextmanager
def source_path_lock(
source_path: Path,
*,
timeout_seconds: float = 5.0,
stale_after_seconds: float = 300.0,
) -> Iterator[None]:
"""Acquire an exclusive lock for a source path using an atomic lock file."""
lock_path = lock_path_for(source_path)
lock_path.parent.mkdir(parents=True, exist_ok=True)
deadline = time.monotonic() + timeout_seconds
while True:
try:
fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
try:
payload = f"{os.getpid()} {datetime.now(UTC).isoformat()}\n".encode()
os.write(fd, payload)
finally:
os.close(fd)
break
except FileExistsError as exc:
try:
age = time.time() - lock_path.stat().st_mtime
except OSError:
age = 0.0
if age > stale_after_seconds:
try:
lock_path.unlink()
continue
except OSError:
pass
if time.monotonic() >= deadline:
raise SourceLockError(
f"Timed out waiting for DataForge source lock: {source_path.resolve()}"
) from exc
time.sleep(0.05)
try:
yield
finally:
with suppress(FileNotFoundError):
lock_path.unlink()
|