Coverage for src/lilbee/runtime/lock.py: 100%
348 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Cross-process locks: LanceDB write locking, the sync mark, and the server singleton.
3Write locking combines an in-process mutex with a cross-process file lock
4(filelock) so separate processes also coordinate writes. Read consistency is
5handled by LanceDB's built-in MVCC via ``read_consistency_interval`` in
6``lilbee.data.store``. The server lock makes ``lilbee serve`` a singleton per
7data dir.
8"""
10import asyncio
11import errno
12import json
13import logging
14import os
15import sqlite3
16import sys
17import threading
18import time
19import uuid
20from collections.abc import AsyncGenerator, Callable, Generator, Iterator
21from contextlib import (
22 AbstractContextManager,
23 asynccontextmanager,
24 closing,
25 contextmanager,
26)
27from dataclasses import dataclass
28from pathlib import Path
29from typing import Literal, NoReturn, Self, overload
31import psutil
32from filelock import AsyncFileLock, AsyncReadWriteLock, FileLock, ReadWriteLock
33from filelock import Timeout as FileLockTimeout
35from lilbee.core.config import Config, cfg
36from lilbee.runtime.cancellation import CancelSignal
38log = logging.getLogger(__name__)
40# Default timeout (seconds) for acquiring the write lock
41LOCK_TIMEOUT = 30.0
42# Grace (seconds) for a dying predecessor to release the server lock during a
43# restart handoff before a new `lilbee serve` gives up. This budgets the PROMPT
44# path only: a predecessor whose llama-swaps honor SIGTERM releases well inside
45# it. It deliberately does NOT cover SIGKILL escalation -- the teardown that
46# actually holds the lock (_release_engines -> stop_engine -> _stop_stale_swap in
47# lilbee.providers.fleet.swap_manager) can spend _ORPHAN_STOP_TIMEOUT_S plus the
48# kill and reap waits per group, i.e. tens of seconds across four groups on a
49# wedged machine. Sizing this to that worst case would make every ordinary
50# restart wait on a pathological one; instead the successor exits with
51# LOCK_REFUSAL_EXIT_CODE and the operator retries.
52SERVER_LOCK_TIMEOUT = 15.0
53_SERVER_LOCK_NAME = "server.lock"
54_SCOPE_LOCK_NAME = "server.scope.lock"
55_SCOPE_OWNER_NAME = "server.scope.owner.json"
56_SYNC_LOCK_NAME = "sync.lock"
57# One sync at a time plans and runs on a data root; the others wait for this lock.
58_SYNC_TURN_LOCK_NAME = "sync.turn.lock"
59# Who holds the turn, so a waiting sync can name it.
60_SYNC_TURN_OWNER_NAME = "sync.turn.owner.json"
61_TURN_OWNER_NOT_WRITTEN = "Could not record the sync owner in %s (%s)."
62_TURN_OWNER_NOT_REMOVED = "Could not delete %s (%s); it is safe to delete by hand."
63_STOP_WAITING = "Waiting for it to finish. Press Ctrl-C to stop waiting."
64_SYNC_WAIT_NAMED = (
65 "A sync started by {started_by} (process {pid}) is running on this library. " + _STOP_WAITING
66)
67_SYNC_WAIT_UNNAMED = "Another sync is running on this library. " + _STOP_WAITING
68_SYNC_PROBE_PREFIX = "sync.probe."
69# How long the probe waits out a momentary busy before it concludes the filesystem cannot lock.
70_PROBE_WAIT_S = 0.3
71# Longest attempt of a wait that takes the sync lock in slices; a cancelled wait returns
72# within it.
73_SYNC_TAKE_SLICE_S = 0.1
74# What the filesystem raises when the sync lock cannot be created or taken: OSError
75# from making the data root, sqlite3.Error from SQLite (a file that is not a
76# database, or a mount that refuses its locks). A busy lock arrives as filelock's
77# Timeout instead, which is itself an OSError, so it is caught first.
78_SYNC_LOCK_ERRORS = (OSError, sqlite3.Error)
79_SYNC_LOCK_REFUSED = "Cannot lock %s (%s); a reset refuses until it can."
80_SYNC_RUNNING = "A sync, an import, an add or a wiki build is running on this library."
81_RESET_WHEN_DONE = "Reset again when it finishes."
82_SYNC_LOCK_UNKNOWN = (
83 "Cannot tell whether a sync or import is running: {path} cannot be locked ({error}). "
84 "Stop every lilbee process, delete {path}, and {again}."
85)
86_SYNC_LOCK_NOT_CLOSED = "Could not close %s (%s)."
87_DATA_ROOT_UNLOCKABLE = (
88 "Cannot tell whether a sync or import is running: {path} cannot be locked ({error})."
89)
90_SYNC_REFUSED = "Cannot change the library: {path} cannot be locked ({error}). {advice}"
91_DELETE_SYNC_LOCK = "Stop every lilbee process, delete {path}, and try again."
92_FREE_OPEN_FILES = "Close some programs or raise the limit of open files, then try again."
93_MAKE_WRITABLE = "Make {path} writable, then try again."
94_SYNC_ADVICE_BY_ERRNO: dict[int | None, str] = {
95 errno.EMFILE: _FREE_OPEN_FILES,
96 errno.ENFILE: _FREE_OPEN_FILES,
97 errno.ENOSPC: "Free some disk space, then try again.",
98 errno.EACCES: _MAKE_WRITABLE,
99 errno.EPERM: _MAKE_WRITABLE,
100 errno.EROFS: _MAKE_WRITABLE,
101}
102_FIX_THE_CAUSE = "Fix that problem, then try again."
103_NO_LOCKING = (
104 "{path} is on a filesystem that does not support file locking. Move it to a local disk."
105)
106_NO_HARD_LINKS = "its filesystem does not support hard links; move it to {disk}"
107_LOCAL_DISK = "a local disk"
108# The disk to move to, by sys.platform; every other platform gets _LOCAL_DISK.
109_HARD_LINK_DISKS = {"darwin": "a disk formatted as APFS"}
110_PROBE_LEFT = "Could not delete the lock probe %s (%s); it is safe to delete by hand."
111# Minimum blocking wait granted to the in-process mutex even when the file lock
112# consumed the whole budget, so a deadline-edge acquire still gets a real attempt.
113_MUTEX_MIN_WAIT = 0.1
114# The timeout threading reads as "wait until the lock is free".
115_NO_DEADLINE = -1
118class LockTimeoutError(TimeoutError):
119 """Raised when a lock cannot be acquired within the timeout."""
122class LockingUnsupportedError(RuntimeError):
123 """Raised when the filesystem holding a lock file cannot lock files at all."""
126class ResetRefusedError(RuntimeError):
127 """Raised when a reset cannot safely run on the data root; nothing was deleted."""
130class LibraryUnlockableError(RuntimeError):
131 """Raised when a sync, an add or a wiki write cannot lock the data root."""
134class _NoHardLinksError(OSError):
135 """Raised by the lock probe on a filesystem that has no hard links."""
138class SyncMarkWaitCancelledError(Exception):
139 """Raised when a cancel ends a wait for the sync lock; the block did not run."""
142# How each error that refuses a writer tells the user to run the command again.
143_AGAIN = {ResetRefusedError: "reset again", LibraryUnlockableError: "try again"}
146# In-process write mutex: serializes writers within the same process
147_write_mutex = threading.Lock()
148# In-process mutex of the crawl page table: serializes page saves within the process.
149_crawl_pages_mutex = threading.Lock()
150_CRAWL_PAGES_LOCK_NAME = "crawl_pages.lock"
153def _lock_path(lancedb_dir: Path | None) -> Path:
154 return (lancedb_dir if lancedb_dir is not None else cfg.lancedb_dir) / ".lock"
157def server_lock_path(data_dir: Path) -> Path:
158 """Path of the one-server-per-data-dir lock file."""
159 return data_dir / _SERVER_LOCK_NAME
162def acquire_server_lock(data_dir: Path, timeout: float = SERVER_LOCK_TIMEOUT) -> FileLock | None:
163 """Hold the one-server-per-data-dir lock, or None when a live server owns it.
165 The lock is an OS file lock, so the kernel releases it the moment its holder
166 exits, however it died; a crashed or killed server leaves no stale state.
167 """
168 data_dir.mkdir(parents=True, exist_ok=True)
169 lock = FileLock(server_lock_path(data_dir))
170 try:
171 with _enolck_as_unsupported(data_dir):
172 lock.acquire(timeout=timeout)
173 except FileLockTimeout:
174 return None
175 return lock
178@dataclass(frozen=True)
179class ScopeOwner:
180 """The data dir the server holding a scope lock is serving, for the refusal message."""
182 data_dir: str
185@dataclass(frozen=True)
186class ScopeHold:
187 """A held scope lock plus its owner sidecar; release removes both."""
189 lock: FileLock
190 owner_path: Path
192 def release(self) -> None:
193 """Remove the owner sidecar, then free the scope for the next server."""
194 self.owner_path.unlink(missing_ok=True)
195 self.lock.release()
198def acquire_scope_lock(
199 scope_dir: Path, data_dir: Path, timeout: float = SERVER_LOCK_TIMEOUT
200) -> ScopeHold | None:
201 """Hold the one-server-per-scope lock, or None when a live server owns the scope.
203 The scope is a directory shared by several would-be servers (the Obsidian
204 plugin's shared root). Like the data-dir lock, the OS releases it the moment
205 the holder exits. The owner sidecar records which data dir the holder is
206 serving so a refused starter can name it in its message.
207 """
208 scope_dir.mkdir(parents=True, exist_ok=True)
209 lock = FileLock(scope_dir / _SCOPE_LOCK_NAME)
210 try:
211 with _enolck_as_unsupported(scope_dir):
212 lock.acquire(timeout=timeout)
213 except FileLockTimeout:
214 return None
215 owner_path = scope_dir / _SCOPE_OWNER_NAME
216 owner_path.write_text(json.dumps({"data_dir": str(data_dir)}), encoding="utf-8")
217 return ScopeHold(lock, owner_path)
220def read_scope_owner(scope_dir: Path) -> ScopeOwner | None:
221 """The scope's recorded owner, or None when absent or unreadable."""
222 try:
223 payload = json.loads((scope_dir / _SCOPE_OWNER_NAME).read_text(encoding="utf-8"))
224 return ScopeOwner(data_dir=str(payload["data_dir"]))
225 except (OSError, ValueError, KeyError, TypeError):
226 return None
229@overload
230def _open_sync_lock(data_root: Path, *, write: Literal[True]) -> ReadWriteLock: ...
231@overload
232def _open_sync_lock(
233 data_root: Path, *, write: Literal[False], refused: type[RuntimeError]
234) -> AsyncReadWriteLock | None: ...
235@overload
236def _open_sync_lock(
237 data_root: Path,
238 *,
239 write: Literal[False],
240 on_caller_thread: Literal[True],
241 refused: type[RuntimeError],
242) -> ReadWriteLock | None: ...
243@overload
244def _open_sync_lock(
245 data_root: Path, *, write: Literal[True], refused: type[RuntimeError]
246) -> ReadWriteLock: ...
247def _open_sync_lock(
248 data_root: Path,
249 *,
250 write: bool,
251 on_caller_thread: bool = False,
252 refused: type[RuntimeError] = ResetRefusedError,
253) -> ReadWriteLock | AsyncReadWriteLock | None:
254 """Open the data root's sync lock, or raise *refused* when it cannot be opened.
256 None is for a reader on a filesystem with no hard links, which runs unmarked.
257 The read side is taken on an event loop unless *on_caller_thread* is set.
258 """
259 if not _data_root_locks(data_root, refused, unmarked_reader=not write):
260 return None
261 path = data_root / _SYNC_LOCK_NAME
262 try:
263 kind = ReadWriteLock if write or on_caller_thread else AsyncReadWriteLock
264 return kind(path, is_singleton=False)
265 except _SYNC_LOCK_ERRORS as exc:
266 _sync_lock_unavailable(path, exc, refused, reader=not write)
269def _data_root_locks(
270 data_root: Path, refused: type[RuntimeError], *, unmarked_reader: bool
271) -> bool:
272 """True when *data_root* can lock files; False where an unmarked reader runs; else raise."""
273 try:
274 data_root.mkdir(parents=True, exist_ok=True)
275 _require_locking(data_root)
276 except _NoHardLinksError as exc:
277 if not unmarked_reader:
278 _sync_lock_unavailable(data_root, exc, refused, refusal=_DATA_ROOT_UNLOCKABLE)
279 # A reset cannot open the lock on this filesystem either, so it refuses.
280 log.warning(_SYNC_LOCK_REFUSED, data_root, exc)
281 return False
282 except _SYNC_LOCK_ERRORS as exc:
283 _sync_lock_unavailable(
284 data_root, exc, refused, reader=unmarked_reader, refusal=_DATA_ROOT_UNLOCKABLE
285 )
286 return True
289def refuse_where_a_sync_cannot_lock(data_root: Path) -> None:
290 """Raise the error a sync raises when it cannot open or take the sync lock of *data_root*.
292 The lock is not held on return, and a busy lock is not waited for.
293 """
294 lock = _open_sync_lock(
295 data_root, write=False, on_caller_thread=True, refused=LibraryUnlockableError
296 )
297 if lock is None:
298 return
299 try:
300 # SQLite reads the lock file at the take, so a file that is not a lock fails here.
301 lock.acquire_read(blocking=False)
302 except FileLockTimeout:
303 # A reset holds the lock, so the lock works; the sync waits for the reset.
304 pass
305 except _SYNC_LOCK_ERRORS as exc:
306 _sync_lock_unavailable(Path(lock.lock_file), exc, LibraryUnlockableError, reader=True)
307 finally:
308 _close_sync_lock(lock)
311def _close_sync_lock(lock: ReadWriteLock) -> None:
312 """Close *lock*; a close that fails only warns, so the caller keeps its own result."""
313 try:
314 lock.close()
315 except _SYNC_LOCK_ERRORS as exc:
316 log.warning(_SYNC_LOCK_NOT_CLOSED, lock.lock_file, exc)
319def _require_locking(data_root: Path) -> None:
320 """Raise ``LockingUnsupportedError`` when a lock nobody else holds still reports busy."""
321 # Unique per call: a probe shared with another thread or host would read its hold as busy.
322 probe_path = data_root / f"{_SYNC_PROBE_PREFIX}{uuid.uuid4().hex}"
323 try:
324 with closing(ReadWriteLock(probe_path, is_singleton=False)) as probe:
325 probe.acquire_write(timeout=_PROBE_WAIT_S)
326 except FileLockTimeout:
327 raise LockingUnsupportedError(_NO_LOCKING.format(path=data_root)) from None
328 except OSError as exc:
329 if not _hard_link_refused(exc):
330 raise
331 disk = _HARD_LINK_DISKS.get(sys.platform, _LOCAL_DISK)
332 raise _NoHardLinksError(_NO_HARD_LINKS.format(disk=disk)) from exc
333 finally:
334 _remove_probe(probe_path)
337def _hard_link_refused(error: OSError) -> bool:
338 """True when *error* is the lock's hard link, refused by a filesystem that has none."""
339 # Only a call on two paths sets filename2, and the link is the lock's only such call.
340 return error.errno == errno.ENOTSUP and error.filename2 is not None
343@contextmanager
344def _enolck_as_unsupported(directory: Path) -> Generator[None, None, None]:
345 """Raise ``LockingUnsupportedError`` for a file lock the filesystem refuses with ENOLCK."""
346 try:
347 yield
348 except OSError as exc:
349 if exc.errno != errno.ENOLCK:
350 raise
351 raise LockingUnsupportedError(_NO_LOCKING.format(path=directory)) from exc
354def _remove_probe(probe_path: Path) -> None:
355 """Delete a lock probe; a leftover probe is harmless, so a failure only warns."""
356 try:
357 probe_path.unlink(missing_ok=True)
358 except OSError as exc:
359 log.warning(_PROBE_LEFT, probe_path, exc)
362def _sync_lock_unavailable(
363 path: Path,
364 error: Exception,
365 refused: type[RuntimeError],
366 *,
367 reader: bool = False,
368 refusal: str = _SYNC_LOCK_UNKNOWN,
369) -> NoReturn:
370 """Raise *refused* with the text for a *reader* or a writer that cannot lock *path*."""
371 text = _SYNC_REFUSED if reader else refusal
372 advice = _sync_advice(path, error)
373 raise refused(
374 text.format(path=path, error=_cause_text(error), again=_AGAIN[refused], advice=advice)
375 ) from error
378def _cause_text(error: Exception) -> str:
379 """The operating system's words for *error*, without the file that it names."""
380 # The file can be the lock probe, whose name means nothing to the user.
381 if isinstance(error, OSError) and error.strerror is not None:
382 return f"[Errno {error.errno}] {error.strerror}"
383 return str(error)
386def _sync_advice(path: Path, error: Exception) -> str:
387 """What to do about *error*, which stopped a sync from locking *path*."""
388 # SQLite raises the bare DatabaseError for a file that is not one of its databases.
389 if path.name == _SYNC_LOCK_NAME and type(error) is sqlite3.DatabaseError:
390 return _DELETE_SYNC_LOCK.format(path=path)
391 code = error.errno if isinstance(error, OSError) else None
392 return _SYNC_ADVICE_BY_ERRNO.get(code, _FIX_THE_CAUSE).format(path=path)
395def _raise_cancel_behind(error: Exception) -> None:
396 """Raise the cancel that *error* replaced, when it replaced one."""
397 # filelock raises a failed take or close in place of the cancel that interrupted it.
398 if isinstance(error.__context__, asyncio.CancelledError):
399 raise error.__context__ from None
402class _SyncLockAttempt:
403 """One try at a side of the sync lock, *seconds* long at most; ``busy`` is how it failed."""
405 def __init__(self, seconds: float) -> None:
406 self.seconds = seconds
407 self.busy: FileLockTimeout | None = None
409 def __enter__(self) -> Self:
410 return self
412 def __exit__(self, _kind: object, error: BaseException | None, _traceback: object) -> bool:
413 if isinstance(error, _SYNC_LOCK_ERRORS):
414 _raise_cancel_behind(error)
415 # Only a busy lock is waited out.
416 if isinstance(error, FileLockTimeout):
417 self.busy = error
418 return self.busy is not None
421def _sync_lock_wait(cancel: CancelSignal | None) -> Iterator[_SyncLockAttempt]:
422 """Every wait for the sync lock or the sync turn: yield attempts until one takes it.
424 The caller takes the lock inside the ``with`` of each attempt, for its ``seconds``.
425 *cancel* is read after every attempt, the one that took the lock included: a set
426 *cancel* raises ``SyncMarkWaitCancelledError``, and the caller closes the lock.
427 """
428 while True:
429 attempt = _SyncLockAttempt(_SYNC_TAKE_SLICE_S)
430 yield attempt
431 if _is_set(cancel):
432 raise SyncMarkWaitCancelledError
433 if attempt.busy is None:
434 return
437def _is_set(cancel: CancelSignal | None) -> bool:
438 return cancel is not None and cancel.is_set()
441@asynccontextmanager
442async def sync_running(
443 data_root: Path, cancel: CancelSignal | None = None
444) -> AsyncGenerator[bool, None]:
445 """Mark a sync or import running on *data_root*, across processes; they share the mark.
447 The wait for a reset to finish runs off the event loop. A *cancel* that is set
448 during that wait, or when the mark is taken, raises ``asyncio.CancelledError``:
449 the block does not run and the mark is free. The block gets False when the
450 filesystem cannot hold the mark, and runs unmarked.
451 """
452 # Opening the lock reads and writes files, so it runs off the event loop too.
453 lock = await asyncio.to_thread(
454 _open_sync_lock, data_root, write=False, refused=LibraryUnlockableError
455 )
456 if lock is None:
457 yield False
458 return
459 try:
460 try:
461 for attempt in _sync_lock_wait(cancel):
462 with attempt:
463 await lock.acquire_read(timeout=attempt.seconds)
464 except SyncMarkWaitCancelledError:
465 raise asyncio.CancelledError from None
466 except _SYNC_LOCK_ERRORS as exc:
467 _sync_lock_unavailable(Path(lock.lock_file), exc, LibraryUnlockableError, reader=True)
468 yield True
469 finally:
470 try:
471 await lock.close()
472 except _SYNC_LOCK_ERRORS as exc:
473 # Logged only: the block keeps its own result, error or cancel.
474 log.warning(_SYNC_LOCK_NOT_CLOSED, lock.lock_file, exc)
475 # filelock keeps the worker thread of a lock whose close raised; wait until it ends.
476 await asyncio.to_thread(lock.executor.shutdown)
477 _raise_cancel_behind(exc)
480@dataclass(frozen=True)
481class SyncTurnOwner:
482 """The process that holds the sync turn of a data root, and what started its sync."""
484 started_by: str
485 pid: int
488def sync_wait_message(owner: SyncTurnOwner | None) -> str:
489 """What a sync that waits for the turn tells the user, naming *owner* when it is known."""
490 if owner is None:
491 return _SYNC_WAIT_UNNAMED
492 return _SYNC_WAIT_NAMED.format(started_by=owner.started_by, pid=owner.pid)
495def read_sync_turn_owner(data_root: Path) -> SyncTurnOwner | None:
496 """The recorded holder of *data_root*'s sync turn, or None when absent, unreadable or dead.
498 A holder killed with SIGKILL leaves its record, and the next holder replaces it only
499 after it takes the turn. A record whose process is gone is not the holder, so it reads
500 as no owner. ``psutil.pid_exists`` does the check on POSIX and on Windows.
501 """
502 try:
503 payload = json.loads((data_root / _SYNC_TURN_OWNER_NAME).read_text(encoding="utf-8"))
504 owner = SyncTurnOwner(started_by=str(payload["started_by"]), pid=int(payload["pid"]))
505 except (OSError, ValueError, KeyError, TypeError):
506 return None
507 return owner if psutil.pid_exists(owner.pid) else None
510def _record_turn_owner(data_root: Path, started_by: str) -> None:
511 """Write who holds the turn; the record only names the holder, so a failure only warns."""
512 path = data_root / _SYNC_TURN_OWNER_NAME
513 try:
514 path.write_text(
515 json.dumps({"started_by": started_by, "pid": os.getpid()}), encoding="utf-8"
516 )
517 except OSError as exc:
518 log.warning(_TURN_OWNER_NOT_WRITTEN, path, exc)
521def _clear_turn_owner(data_root: Path) -> None:
522 """Delete the holder record before the turn is freed; a failure only warns."""
523 path = data_root / _SYNC_TURN_OWNER_NAME
524 try:
525 path.unlink(missing_ok=True)
526 except OSError as exc:
527 log.warning(_TURN_OWNER_NOT_REMOVED, path, exc)
530@asynccontextmanager
531async def sync_turn(
532 data_root: Path,
533 cancel: CancelSignal | None,
534 *,
535 started_by: str,
536 on_wait: Callable[[SyncTurnOwner | None], None],
537) -> AsyncGenerator[None, None]:
538 """Hold *data_root*'s sync turn: one sync at a time plans and runs, across processes.
540 A sync that finds the turn held calls *on_wait* once with the holder, then waits
541 with no time limit. The OS frees the turn when its holder exits, however it died.
542 A *cancel* that is set during the wait raises ``asyncio.CancelledError``: the block
543 does not run and the turn is free.
544 """
545 lock = AsyncFileLock(data_root / _SYNC_TURN_LOCK_NAME)
546 try:
547 try:
548 told = False
549 for attempt in _sync_lock_wait(cancel):
550 with attempt:
551 await lock.acquire(timeout=attempt.seconds)
552 if attempt.busy is not None and not told:
553 told = True
554 on_wait(await asyncio.to_thread(read_sync_turn_owner, data_root))
555 except SyncMarkWaitCancelledError:
556 raise asyncio.CancelledError from None
557 except _SYNC_LOCK_ERRORS as exc:
558 _sync_lock_unavailable(Path(lock.lock_file), exc, LibraryUnlockableError, reader=True)
559 await asyncio.to_thread(_record_turn_owner, data_root, started_by)
560 yield
561 finally:
562 if lock.is_locked:
563 await asyncio.to_thread(_clear_turn_owner, data_root)
564 await lock.release()
567@contextmanager
568def source_keys_in_use(
569 data_root: Path, cancel: CancelSignal | None = None
570) -> Generator[None, None, None]:
571 """Hold the sync mark of *data_root* for a block that reads and writes source keys.
573 Waits while a reset holds syncs off. A *cancel* that is
574 set during that wait, or when the mark is taken, raises
575 ``SyncMarkWaitCancelledError``: the block does not run and the mark is free.
576 """
577 lock = _open_sync_lock(
578 data_root, write=False, on_caller_thread=True, refused=LibraryUnlockableError
579 )
580 if lock is None:
581 yield
582 return
583 try:
584 try:
585 for attempt in _sync_lock_wait(cancel):
586 with attempt:
587 lock.acquire_read(timeout=attempt.seconds)
588 except _SYNC_LOCK_ERRORS as exc:
589 if _is_set(cancel):
590 raise SyncMarkWaitCancelledError from None
591 _sync_lock_unavailable(Path(lock.lock_file), exc, LibraryUnlockableError, reader=True)
592 yield
593 finally:
594 _close_sync_lock(lock)
597@contextmanager
598def _no_sync_running(data_root: Path) -> Generator[None, None, None]:
599 """Keep syncs off *data_root* for the block; raise ``ResetRefusedError`` unless it can."""
600 try:
601 lock = _open_sync_lock(data_root, write=True)
602 except LockingUnsupportedError as exc:
603 raise ResetRefusedError(str(exc)) from exc
604 with closing(lock):
605 try:
606 lock.acquire_write(blocking=False)
607 except FileLockTimeout:
608 raise ResetRefusedError(f"{_SYNC_RUNNING} {_RESET_WHEN_DONE}") from None
609 except _SYNC_LOCK_ERRORS as exc:
610 _sync_lock_unavailable(Path(lock.lock_file), exc, ResetRefusedError)
611 yield
614def library_held_for_reset(config: Config) -> AbstractContextManager[None]:
615 """Keep syncs off *config*'s data root for the block of a reset, with no wait.
617 A sync that runs, or a sync lock that cannot be taken, raises ``ResetRefusedError``.
618 """
619 return _no_sync_running(config.data_root)
622@contextmanager
623def write_lock(
624 lancedb_dir: Path | None = None, timeout: float | None = LOCK_TIMEOUT
625) -> Generator[None, None, None]:
626 """Acquire the cross-process file lock then the in-process mutex.
628 The file lock lives next to the store's data, so cross-process writers
629 coordinate only when they lock the *same* directory: callers pass their
630 store's ``lancedb_dir`` (a per-instance ``Lilbee`` uses its own dir).
631 ``None`` falls back to the global ``cfg.lancedb_dir``.
633 The two stages share one budget: the time spent waiting on the file lock is
634 deducted before waiting on the mutex (plus a small ``_MUTEX_MIN_WAIT`` floor),
635 so a 30s request cannot stall for roughly twice that. A *timeout* of None
636 waits for both without a deadline.
637 """
638 with _held(_lock_path(lancedb_dir), _write_mutex, timeout, "write lock"):
639 yield
642@contextmanager
643def crawl_pages_lock(
644 lancedb_dir: Path, timeout: float | None = LOCK_TIMEOUT
645) -> Generator[None, None, None]:
646 """Hold the lock of the crawl page table in *lancedb_dir*: a file lock, then a mutex.
648 It is apart from ``write_lock``, so a page save never waits for a sync batch. A
649 holder may take ``write_lock`` inside it; a holder of ``write_lock`` never takes it.
650 """
651 with _held(lancedb_dir / _CRAWL_PAGES_LOCK_NAME, _crawl_pages_mutex, timeout, "crawl lock"):
652 yield
655@contextmanager
656def _held(
657 lock_path: Path, mutex: threading.Lock, timeout: float | None, name: str
658) -> Generator[None, None, None]:
659 """Hold the file lock at *lock_path*, then *mutex*, within one *timeout* for both.
661 A timeout on the mutex names the lock as *name*.
662 """
663 deadline = None if timeout is None else time.monotonic() + timeout
664 # The first write to a per-instance store can run before its data dir exists;
665 # the file lock cannot be created in a missing directory.
666 lock_path.parent.mkdir(parents=True, exist_ok=True)
667 flock = FileLock(lock_path)
668 try:
669 with _enolck_as_unsupported(lock_path.parent):
670 # None takes the lock's own timeout, which waits without a deadline.
671 flock.acquire(timeout=timeout)
672 except FileLockTimeout:
673 raise LockTimeoutError("Timed out waiting for exclusive file lock") from None
674 try:
675 # Floor the mutex budget so a file lock that wins right at the deadline
676 # still gets a brief blocking attempt instead of a zero-timeout poll that
677 # spuriously fails when another thread holds the mutex for an instant.
678 remaining = (
679 _NO_DEADLINE if deadline is None else max(_MUTEX_MIN_WAIT, deadline - time.monotonic())
680 )
681 acquired = mutex.acquire(timeout=remaining)
682 if not acquired:
683 raise LockTimeoutError(f"Timed out waiting for {name}")
684 try:
685 yield
686 finally:
687 mutex.release()
688 finally:
689 flock.release()