Class StrictSoftFileLock

src/filelock/_strict.py:95–260

Portable fail-closed lock based on immutable owner claims.

class StrictSoftFileLock(BaseFileLock):
    """Portable fail-closed lock based on immutable owner claims."""

    _preserve_lock_file_supported: bool = True
    _on_acquired_supported: bool = False
    #: Age cannot clear a strict claim: expiring one on a clock is the overlap the fail-closed contract exists to rule
    #: out, so only force_break() removes it.
    _lifetime_supported: bool = False
    _lifetime_unsupported_reason: str = "a strict claim is never broken by age, only by force_break()"
    #: The claim doorway publishes an intent and a held record per owner, so a shared instance must serialize them.
    _serialize_transitions: bool = True
    #: Contending processes each publish and rescan several files, so back their retries off across a jittered window
    #: rather than let them collide on every poll. Seconds; keeps a waiter responsive once it wins.
    _poll_backoff_cap: float = 0.05

    def _acquire(self) -> None:
        # Resolve once per acquisition, not per poll: a waiter on a relative path must keep publishing into the
        # directory it started waiting in even when another thread changes the working directory mid-wait.
        if (claim_root := self._context.claim_root) is None:
            claim_root = self._context.claim_root = _canonical(self.lock_file)
        lock_path = Path(claim_root)
        coordination_directory = Path(f"{lock_path}{_COORDINATION_SUFFIX}")
        claim_directory = coordination_directory / _CLAIM_DIRECTORY_NAME
        ensure_directory_exists(os.fspath(lock_path))
        _ensure_protocol_directory(self.lock_file, coordination_directory)
        _ensure_protocol_directory(self.lock_file, claim_directory)
        if (sentinel_fd := _open_or_create_sentinel(self.lock_file, lock_path, self._open_mode())) is None:
            return
        try:
            sentinel_identity = _file_identity(os.fstat(sentinel_fd))
        except BaseException as inspection_error:  # preserve inspection and descriptor cleanup errors
            try:
                os.close(sentinel_fd)
            except BaseException as close_error:  # ruff:ignore[blind-except]  # preserve inspection and descriptor cleanup errors
                _raise_cleanup_errors("strict sentinel inspection cleanup failed", inspection_error, close_error)
            raise
        self._mark_descriptor_pending(sentinel_fd, sentinel_identity)
        try:
            self._attempt_doorway(claim_directory, sentinel_fd, sentinel_identity)
        except BaseException:
            if self._context.pending_lock_file_fd == sentinel_fd:
                self._discard_doorway(sentinel_fd, sentinel_identity)
            raise

    def _attempt_doorway(self, claim_directory: Path, sentinel_fd: int, sentinel_identity: tuple[int, int]) -> None:
        if _read_existing_claims(self.lock_file, claim_directory):
            self._discard_doorway(sentinel_fd, sentinel_identity)
            return

        token = secrets.token_hex(_TOKEN_HEX_LENGTH // 2)
        intent_name = _claim_name("intent", token)
        intent_path = str(claim_directory / intent_name)
        try:
            publication_cleanup_error = _publish_record(intent_path, _claim_record(token), self._open_mode())
        except _PrivateRecordReclaimedError:
            self._discard_doorway(sentinel_fd, sentinel_identity)
            return
        except (NotImplementedError, OSError) as error: