| from __future__ import annotations |
|
|
| import contextlib |
| import errno |
| import os |
| import secrets |
| import stat |
| import sys |
| import tempfile |
| import time |
| from dataclasses import dataclass |
| from errno import EACCES, EEXIST, ENOENT, ENOSYS, EPERM, ESTALE, EXDEV |
| from pathlib import Path |
| from typing import TYPE_CHECKING, Final, Literal, cast |
|
|
| from ._api import BaseFileLock, _canonical, _raise_cleanup_errors |
| from ._error import SoftFileLockProtocolError |
| from ._identity import host_name, process_start_token |
| from ._soft_protocol import STRICT_SOFT_SENTINEL_RECORD |
| from ._util import ensure_directory_exists, write_all |
|
|
| if TYPE_CHECKING: |
| from collections.abc import Iterator |
|
|
| StrictSoftFileClaimState = Literal["intent", "held"] |
|
|
| _CLAIM_STATES: Final[frozenset[str]] = frozenset({"intent", "held"}) |
| _COORDINATION_SUFFIX: Final[str] = ".filelock" |
| _CLAIM_DIRECTORY_NAME: Final[str] = "claims" |
| _CLAIM_MAGIC: Final[str] = "filelock-strict-v1" |
| _CLAIM_RECORD_LIMIT: Final[int] = 1024 |
| _CLAIM_NAME_PART_COUNT: Final[int] = 3 |
| _TOKEN_HEX_LENGTH: Final[int] = 32 |
| _PRIVATE_RECORD_MARKER: Final[str] = ".private-v1-" |
| _PRIVATE_RECORD_SUFFIX: Final[str] = ".tmp" |
| _PRIVATE_RECORD_RANDOM_HEX_LENGTH: Final[int] = 32 |
| _PRIVATE_RECORD_GRACE: Final[float] = 2.0 |
| _UNLINK_MAX_RETRIES: Final[int] = 10 |
| |
| _CLAIM_READ_GRACE: Final[float] = 0.5 |
| _CLAIM_READ_RETRY: Final[float] = 0.002 |
| |
| |
| _O_BINARY: Final[int] = getattr(os, "O_BINARY", 0) |
| _LEGACY_SENTINEL: Final[bytes] = STRICT_SOFT_SENTINEL_RECORD.encode() |
| _WINDOWS_HARD_LINK_UNSUPPORTED: Final[frozenset[int]] = frozenset({1, 17, 50}) |
|
|
| |
| |
| |
| _HAS_LINK: Final[bool] = hasattr(os, "link") |
|
|
| |
| |
| _OPEN_SUPPORTS_DIR_FD: Final[bool] = os.open in os.supports_dir_fd |
| _UNLINK_SUPPORTS_DIR_FD: Final[bool] = os.unlink in os.supports_dir_fd |
| _STAT_SUPPORTS_DIR_FD: Final[bool] = os.stat in os.supports_dir_fd |
| _LINK_SUPPORTS_DIR_FD: Final[bool] = _HAS_LINK and os.link in os.supports_dir_fd |
|
|
|
|
| def _probe_link_follow_symlinks() -> bool: |
| |
| |
| |
| |
| |
| if not _HAS_LINK: |
| return False |
| try: |
| with tempfile.TemporaryDirectory() as directory: |
| source = Path(directory, "probe-source") |
| source.touch() |
| os.link(source, Path(directory, "probe-link"), follow_symlinks=False) |
| except (OSError, NotImplementedError, ValueError): |
| return False |
| return True |
|
|
|
|
| _LINK_HONORS_FOLLOW_SYMLINKS: Final[bool] = _probe_link_follow_symlinks() |
|
|
|
|
| def _probe_hard_link_unsupported_errnos() -> frozenset[int]: |
| |
| |
| |
| |
| |
| not_supported = getattr(errno, "ENOTSUP", getattr(errno, "EOPNOTSUPP", None)) |
| return frozenset({ENOSYS, EXDEV} if not_supported is None else {ENOSYS, EXDEV, not_supported}) |
|
|
|
|
| _HARD_LINK_UNSUPPORTED_ERRNOS: Final[frozenset[int]] = _probe_hard_link_unsupported_errnos() |
|
|
|
|
| class StrictSoftFileLock(BaseFileLock): |
| """Portable fail-closed lock based on immutable owner claims.""" |
|
|
| _preserve_lock_file_supported: bool = True |
| _on_acquired_supported: bool = False |
| |
| |
| _lifetime_supported: bool = False |
| _lifetime_unsupported_reason: str = "a strict claim is never broken by age, only by force_break()" |
| |
| _serialize_transitions: bool = True |
| |
| |
| _poll_backoff_cap: float = 0.05 |
|
|
| def _acquire(self) -> None: |
| |
| |
| 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: |
| try: |
| os.close(sentinel_fd) |
| except BaseException as close_error: |
| _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: |
| _raise_if_hard_links_unsupported(self.lock_file, error) |
| if isinstance(error, OSError) and error.errno == EEXIST: |
| self._discard_doorway(sentinel_fd, sentinel_identity) |
| return |
| raise |
| if publication_cleanup_error is not None: |
| raise publication_cleanup_error |
| self._context.owner_claim_paths = (intent_path,) |
|
|
| claims = _read_existing_claims(self.lock_file, claim_directory) |
| if ( |
| not claims |
| or any(claim.state == "held" for claim in claims) |
| or min(claim.name for claim in claims) != intent_name |
| ): |
| self._discard_doorway(sentinel_fd, sentinel_identity) |
| return |
|
|
| held_name = _claim_name("held", token) |
| held_path = str(claim_directory / held_name) |
| try: |
| link_cleanup_error = _link_no_replace(claim_directory, intent_name, held_name) |
| except (NotImplementedError, OSError) as error: |
| _raise_if_hard_links_unsupported(self.lock_file, error) |
| raise |
| self._context.owner_claim_paths = (held_path, intent_path) |
| if link_cleanup_error is not None: |
| self._context.owner_claim_paths = () |
| raise link_cleanup_error |
|
|
| claims = _read_existing_claims(self.lock_file, claim_directory) |
| if ( |
| not {intent_name, held_name}.issubset(claim.name for claim in claims) |
| or min(_claim_token_key(claim.name) for claim in claims) != f"v1-{token}.claim" |
| ): |
| self._discard_doorway(sentinel_fd, sentinel_identity) |
| return |
| |
| |
| |
| |
| |
| |
| self._mark_descriptor_owned(sentinel_fd, sentinel_identity) |
|
|
| @property |
| def claims(self) -> tuple[StrictSoftFileClaim, ...]: |
| """Published claims that block acquisition.""" |
| return _read_existing_claims(self.lock_file, self._claim_directory) |
|
|
| def force_break(self, claim_name: str) -> None: |
| """Remove one named claim, allowing overlap if its owner still holds the protected resource.""" |
| _validate_force_break_name(claim_name) |
| _require_exact_name(self._claim_directory, claim_name) |
| if ( |
| cleanup_error := _unlink_in_directory(self._claim_directory, claim_name) |
| ) is not None: |
| raise cleanup_error |
|
|
| def _rollback_failed_acquire(self, acquisition_error: BaseException) -> None: |
| |
| |
| |
| |
| if self.is_locked: |
| return |
| super()._rollback_failed_acquire(acquisition_error) |
|
|
| def _reconcile_failed_acquire(self, canonical: str) -> None: |
| |
| if not self.is_locked: |
| self._context.claim_root = None |
| super()._reconcile_failed_acquire(canonical) |
|
|
| def _release(self) -> None: |
| fd = cast("int", self._context.lock_file_fd) |
| self._context.claim_root = None |
| remaining, errors = _unlink_owner_paths(self._context.owner_claim_paths) |
| self._context.owner_claim_paths = tuple(remaining) |
| if remaining: |
| _raise_recorded_errors("strict claim release failed", errors) |
| self._mark_descriptor_released() |
| try: |
| self._close_released_fd(fd, default_suppresses=False) |
| except BaseException as close_error: |
| errors.append(close_error) |
| if errors: |
| _raise_recorded_errors("strict release cleanup failed", errors) |
|
|
| def _discard_doorway(self, fd: int, identity: tuple[int, int]) -> None: |
| remaining, errors = _unlink_owner_paths(self._context.owner_claim_paths) |
| self._context.owner_claim_paths = tuple(remaining) |
| if remaining: |
| self._mark_descriptor_owned(fd, identity) |
| _raise_recorded_errors("strict doorway claim cleanup failed", errors) |
| self._mark_descriptor_released() |
| try: |
| self._close_released_fd(fd, default_suppresses=False) |
| except BaseException as close_error: |
| errors.append(close_error) |
| if errors: |
| _raise_recorded_errors("strict doorway cleanup failed", errors) |
|
|
| @property |
| def _claim_directory(self) -> Path: |
| if self._context.owner_claim_paths: |
| return Path(self._context.owner_claim_paths[0]).parent |
| return Path(f"{_canonical(self.lock_file)}{_COORDINATION_SUFFIX}") / _CLAIM_DIRECTORY_NAME |
|
|
|
|
| @dataclass(frozen=True) |
| class StrictSoftFileClaim: |
| """One parsed strict soft-lock claim.""" |
|
|
| name: str |
| state: StrictSoftFileClaimState |
| token: str |
| pid: int |
| hostname: str |
| |
| |
| start: int | None = None |
|
|
|
|
| class _PrivateRecordReclaimedError(Exception): |
| pass |
|
|
|
|
| def _open_or_create_sentinel(lock_file: str, path: Path, mode: int) -> int | None: |
| try: |
| return _open_sentinel(path) |
| except FileNotFoundError: |
| pass |
| except OSError: |
| return None |
|
|
| _reclaim_sentinel_private_records(path, time.time()) |
| try: |
| publication_cleanup_error = _publish_record(os.fspath(path), _LEGACY_SENTINEL, mode) |
| except _PrivateRecordReclaimedError: |
| return None |
| except (NotImplementedError, OSError) as error: |
| _raise_if_hard_links_unsupported(lock_file, error) |
| if not isinstance(error, OSError) or error.errno != EEXIST: |
| raise |
| else: |
| if publication_cleanup_error is not None: |
| raise publication_cleanup_error |
| try: |
| return _open_sentinel(path) |
| except OSError: |
| return None |
|
|
|
|
| def _open_sentinel(path: Path) -> int | None: |
| fd, record = _open_record(path, len(_LEGACY_SENTINEL)) |
| if record == _LEGACY_SENTINEL: |
| return fd |
| os.close(fd) |
| return None |
|
|
|
|
| def _read_claims(lock_file: str, directory: Path) -> tuple[StrictSoftFileClaim, ...]: |
| try: |
| with os.scandir(directory) as entries: |
| names = _public_claim_names(directory, entries) |
| except OSError as error: |
| reason = f"cannot list claim directory: {error.strerror or type(error).__name__}" |
| raise SoftFileLockProtocolError(lock_file, None, reason) from error |
|
|
| claims: list[StrictSoftFileClaim] = [] |
| for name in names: |
| if (name_parts := _parse_claim_name(name)) is None: |
| raise SoftFileLockProtocolError(lock_file, name, "unknown claim name or protocol version") |
| if (record := _read_claim_record(lock_file, directory, name)) is not None: |
| claims.append(_parse_claim(lock_file, name, name_parts, record)) |
| return tuple(claims) |
|
|
|
|
| def _read_claim_record(lock_file: str, directory: Path, name: str) -> bytes | None: |
| |
| |
| |
| |
| |
| |
| deadline = time.monotonic() + _CLAIM_READ_GRACE |
| delaying = False |
| while True: |
| if delaying: |
| time.sleep(_CLAIM_READ_RETRY) |
| record, pending = _attempt_claim_read(lock_file, directory, name) |
| if pending is None: |
| return record |
| if time.monotonic() >= deadline: |
| if pending.errno == ESTALE: |
| |
| |
| |
| return None |
| reason = f"cannot read claim: {pending.strerror or str(pending) or type(pending).__name__}" |
| raise SoftFileLockProtocolError(lock_file, name, reason) from pending |
| delaying = True |
|
|
|
|
| def _attempt_claim_read(lock_file: str, directory: Path, name: str) -> tuple[bytes | None, OSError | None]: |
| |
| |
| |
| try: |
| return _read_record(directory / name, _CLAIM_RECORD_LIMIT), None |
| except FileNotFoundError: |
| return None, None |
| except PermissionError as error: |
| return None, error |
| except OSError as error: |
| if error.errno == ESTALE: |
| return None, error |
| reason = f"cannot read claim: {error.strerror or str(error) or type(error).__name__}" |
| raise SoftFileLockProtocolError(lock_file, name, reason) from error |
|
|
|
|
| def _public_claim_names(directory: Path, entries: Iterator[os.DirEntry[str]]) -> list[str]: |
| names: list[str] = [] |
| now = time.time() |
| for entry in entries: |
| if not entry.name.startswith("."): |
| names.append(entry.name) |
| elif (public_name := _private_public_name(entry.name)) is not None and _parse_claim_name( |
| public_name |
| ) is not None: |
| _reclaim_private_record((os.fspath(directory), None), entry.name, now) |
| return sorted(names) |
|
|
|
|
| def _read_existing_claims(lock_file: str, directory: Path) -> tuple[StrictSoftFileClaim, ...]: |
| if not directory.exists(): |
| return () |
| return _read_claims(lock_file, directory) |
|
|
|
|
| def _parse_claim( |
| lock_file: str, |
| name: str, |
| name_parts: tuple[StrictSoftFileClaimState, str], |
| record: bytes, |
| ) -> StrictSoftFileClaim: |
| try: |
| magic, token, pid_text, hostname_hex, start_text, trailing = record.decode("ascii").split("\n") |
| pid = int(pid_text) |
| start = int(start_text) if start_text else None |
| hostname = bytes.fromhex(hostname_hex).decode("utf-8") |
| except (UnicodeDecodeError, ValueError) as error: |
| raise SoftFileLockProtocolError(lock_file, name, "malformed claim record") from error |
| if not all(( |
| not trailing, |
| magic == _CLAIM_MAGIC, |
| token == name_parts[1], |
| 1 <= pid <= 2**31 - 1, |
| str(pid) == pid_text, |
| start is None or (start >= 0 and str(start) == start_text), |
| hostname.encode().hex() == hostname_hex |
| and hostname.isprintable() |
| and not any(character.isspace() for character in hostname), |
| )): |
| raise SoftFileLockProtocolError(lock_file, name, "malformed claim record") |
| return StrictSoftFileClaim(name=name, state=name_parts[0], token=token, pid=pid, hostname=hostname, start=start) |
|
|
|
|
| def _parse_claim_name(name: str) -> tuple[StrictSoftFileClaimState, str] | None: |
| if not name.endswith(".claim"): |
| return None |
| parts = name.removesuffix(".claim").split("-") |
| if len(parts) != _CLAIM_NAME_PART_COUNT or parts[0] not in _CLAIM_STATES or parts[1] != "v1": |
| return None |
| token = parts[2] |
| if len(token) != _TOKEN_HEX_LENGTH or any(character not in "0123456789abcdef" for character in token): |
| return None |
| return cast("StrictSoftFileClaimState", parts[0]), token |
|
|
|
|
| def _claim_name(state: StrictSoftFileClaimState, token: str) -> str: |
| return f"{state}-v1-{token}.claim" |
|
|
|
|
| def _claim_token_key(name: str) -> str: |
| return name.removeprefix("held-").removeprefix("intent-") |
|
|
|
|
| def _claim_record(token: str) -> bytes: |
| hostname_hex = host_name().encode().hex() |
| start = process_start_token(os.getpid()) |
| start_text = "" if start is None else str(start) |
| return f"{_CLAIM_MAGIC}\n{token}\n{os.getpid()}\n{hostname_hex}\n{start_text}\n".encode("ascii") |
|
|
|
|
| def _publish_record( |
| public_path: str, |
| record: bytes, |
| mode: int, |
| ) -> BaseException | None: |
| directory, public_name = os.path.split(public_path) |
| directory = directory or os.curdir |
| private_name = _private_record_name(public_name) |
| directory_fd = _open_directory(directory) if _OPEN_SUPPORTS_DIR_FD else None |
| directory_ref = directory, directory_fd |
| try: |
| _publish_record_in_directory(directory_ref, (private_name, public_name), mode, record) |
| except BaseException as publication_error: |
| try: |
| if directory_fd is not None: |
| os.close(directory_fd) |
| except BaseException as close_error: |
| _raise_cleanup_errors("strict publication directory cleanup failed", publication_error, close_error) |
| raise |
| if directory_fd is not None: |
| try: |
| os.close(directory_fd) |
| except BaseException as close_error: |
| return close_error |
| return None |
|
|
|
|
| def _publish_record_in_directory( |
| directory_ref: tuple[str, int | None], |
| names: tuple[str, str], |
| mode: int, |
| record: bytes, |
| ) -> None: |
| flags = os.O_RDWR | os.O_CREAT | os.O_EXCL | _O_BINARY |
| if (o_nofollow := getattr(os, "O_NOFOLLOW", None)) is not None: |
| flags |= o_nofollow |
| private_fd = _open_relative(directory_ref, names[0], flags, mode) |
| private_identity: tuple[int, int] | None = None |
| try: |
| private_identity = _file_identity(os.fstat(private_fd)) |
| write_all(private_fd, record) |
| _link_private_record(directory_ref, names, private_identity) |
| except BaseException as publication_error: |
| close_error, unlink_error = _close_and_unlink_private_record( |
| directory_ref, |
| names[0], |
| private_fd, |
| private_identity, |
| ) |
| if close_error is not None or unlink_error is not None: |
| _raise_cleanup_errors( |
| "strict record publication cleanup failed", |
| publication_error, |
| close_error, |
| unlink_error, |
| ) |
| raise |
| close_error, unlink_error = _close_and_unlink_private_record( |
| directory_ref, |
| names[0], |
| private_fd, |
| private_identity, |
| ) |
| if close_error is not None or unlink_error is not None: |
| _raise_record_finalization_errors(close_error, unlink_error) |
|
|
|
|
| def _close_and_unlink_private_record( |
| directory_ref: tuple[str, int | None], |
| private_name: str, |
| private_fd: int, |
| private_identity: tuple[int, int] | None, |
| ) -> tuple[BaseException | None, BaseException | None]: |
| close_error: BaseException | None = None |
| try: |
| os.close(private_fd) |
| except BaseException as error: |
| close_error = error |
| unlink_error: BaseException | None = None |
| try: |
| if private_identity is None: |
| _unlink_relative(directory_ref, private_name) |
| else: |
| _unlink_relative_if_identity(directory_ref, private_name, private_identity) |
| except FileNotFoundError: |
| pass |
| except BaseException as error: |
| unlink_error = error |
| return close_error, unlink_error |
|
|
|
|
| def _link_private_record( |
| directory_ref: tuple[str, int | None], |
| names: tuple[str, str], |
| private_identity: tuple[int, int], |
| ) -> None: |
| try: |
| _link_relative(directory_ref, *names) |
| except FileNotFoundError as error: |
| if _relative_identity(directory_ref, names[0]) is not None: |
| raise |
| msg = "private publication record was reclaimed" |
| raise _PrivateRecordReclaimedError(msg) from error |
| if _relative_identity(directory_ref, names[1]) == private_identity: |
| return |
| msg_0 = "private publication record was replaced" |
| raise _PrivateRecordReclaimedError(msg_0) |
|
|
|
|
| def _raise_record_finalization_errors( |
| close_error: BaseException | None, |
| unlink_error: BaseException | None, |
| ) -> None: |
| errors = [error for error in (close_error, unlink_error) if error is not None] |
| if len(errors) > 1: |
| _raise_cleanup_errors("strict record finalization failed", errors[0], *errors[1:]) |
| raise errors[0] |
|
|
|
|
| def _private_record_name(public_name: str) -> str: |
| return f".{public_name}{_PRIVATE_RECORD_MARKER}{secrets.token_hex(_PRIVATE_RECORD_RANDOM_HEX_LENGTH // 2)}.tmp" |
|
|
|
|
| def _private_public_name(private_name: str) -> str | None: |
| if not private_name.startswith(".") or not private_name.endswith(_PRIVATE_RECORD_SUFFIX): |
| return None |
| public_name, marker, random_hex = private_name[1 : -len(_PRIVATE_RECORD_SUFFIX)].rpartition(_PRIVATE_RECORD_MARKER) |
| if ( |
| marker != _PRIVATE_RECORD_MARKER |
| or len(random_hex) != _PRIVATE_RECORD_RANDOM_HEX_LENGTH |
| or any(character not in "0123456789abcdef" for character in random_hex) |
| ): |
| return None |
| return public_name |
|
|
|
|
| def _reclaim_sentinel_private_records(path: Path, now: float) -> None: |
| directory_ref = os.fspath(path.parent), None |
| with os.scandir(path.parent) as entries: |
| for entry in entries: |
| if _private_public_name(entry.name) == path.name: |
| _reclaim_private_record(directory_ref, entry.name, now) |
|
|
|
|
| def _reclaim_private_record(directory_ref: tuple[str, int | None], private_name: str, now: float) -> None: |
| directory, directory_fd = directory_ref |
| try: |
| private_stat = ( |
| os.stat(private_name, dir_fd=directory_fd, follow_symlinks=False) |
| if directory_fd is not None and _STAT_SUPPORTS_DIR_FD |
| else Path(directory, private_name).lstat() |
| ) |
| except FileNotFoundError: |
| return |
| if not stat.S_ISREG(private_stat.st_mode): |
| msg = f"{Path(directory, private_name)} is not a regular private record" |
| raise OSError(msg) |
| if private_stat.st_nlink == 1 and now - private_stat.st_mtime < _PRIVATE_RECORD_GRACE: |
| return |
| try: |
| _unlink_private_record_once(directory_ref, private_name) |
| except FileNotFoundError: |
| pass |
| except OSError as error: |
| if error.errno not in {EACCES, EPERM}: |
| raise |
|
|
|
|
| def _unlink_private_record_once(directory_ref: tuple[str, int | None], private_name: str) -> None: |
| directory, directory_fd = directory_ref |
| if ( |
| directory_fd is not None and _UNLINK_SUPPORTS_DIR_FD |
| ): |
| os.unlink(private_name, dir_fd=directory_fd) |
| else: |
| Path(directory, private_name).unlink() |
|
|
|
|
| def _relative_identity(directory_ref: tuple[str, int | None], name: str) -> tuple[int, int] | None: |
| directory, directory_fd = directory_ref |
| try: |
| if directory_fd is not None and _STAT_SUPPORTS_DIR_FD: |
| path_stat = os.stat(name, dir_fd=directory_fd, follow_symlinks=False) |
| else: |
| path_stat = Path(directory, name).lstat() |
| except FileNotFoundError: |
| return None |
| return _file_identity(path_stat) |
|
|
|
|
| def _unlink_relative_if_identity( |
| directory_ref: tuple[str, int | None], |
| name: str, |
| identity: tuple[int, int], |
| ) -> None: |
| if _relative_identity(directory_ref, name) == identity: |
| with contextlib.suppress(FileNotFoundError): |
| _unlink_relative(directory_ref, name) |
|
|
|
|
| def _open_relative(directory_ref: tuple[str, int | None], name: str, flags: int, mode: int) -> int: |
| directory, directory_fd = directory_ref |
| return ( |
| os.open(name, flags, mode, dir_fd=directory_fd) |
| if directory_fd is not None |
| else os.open(Path(directory, name), flags, mode) |
| ) |
|
|
|
|
| def _link_relative(directory_ref: tuple[str, int | None], source_name: str, destination_name: str) -> None: |
| directory, directory_fd = directory_ref |
| if directory_fd is not None and _LINK_SUPPORTS_DIR_FD: |
| _link_no_follow(source_name, destination_name, src_dir_fd=directory_fd, dst_dir_fd=directory_fd) |
| return |
| _link_no_follow(Path(directory, source_name), Path(directory, destination_name)) |
|
|
|
|
| def _link_no_follow( |
| source: str | Path, |
| destination: str | Path, |
| *, |
| src_dir_fd: int | None = None, |
| dst_dir_fd: int | None = None, |
| ) -> None: |
| if not _HAS_LINK: |
| |
| |
| msg = "os.link is unavailable on this platform" |
| raise NotImplementedError(msg) |
| |
| |
| |
| if _LINK_HONORS_FOLLOW_SYMLINKS: |
| os.link(source, destination, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd, follow_symlinks=False) |
| else: |
| os.link(source, destination, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd) |
|
|
|
|
| def _unlink_relative(directory_ref: tuple[str, int | None], name: str) -> None: |
| directory, directory_fd = directory_ref |
| if directory_fd is not None and _UNLINK_SUPPORTS_DIR_FD: |
| os.unlink(name, dir_fd=directory_fd) |
| elif sys.platform == "win32": |
| _unlink_in_directory(Path(directory), name) |
| else: |
| Path(directory, name).unlink() |
|
|
|
|
| def _link_no_replace(directory: Path, source_name: str, destination_name: str) -> BaseException | None: |
| if _LINK_SUPPORTS_DIR_FD: |
| directory_fd = _open_directory(str(directory)) |
| try: |
| _link_no_follow(source_name, destination_name, src_dir_fd=directory_fd, dst_dir_fd=directory_fd) |
| except BaseException as link_error: |
| try: |
| os.close(directory_fd) |
| except BaseException as close_error: |
| _raise_cleanup_errors("strict link directory cleanup failed", link_error, close_error) |
| raise |
| try: |
| os.close(directory_fd) |
| except BaseException as close_error: |
| return close_error |
| return None |
| _link_no_follow(directory / source_name, directory / destination_name) |
| return None |
|
|
|
|
| def _unlink_owner_path(path: str) -> BaseException | None: |
| directory, name = os.path.split(path) |
| return _unlink_in_directory(Path(directory or os.curdir), name) |
|
|
|
|
| def _unlink_owner_paths(paths: tuple[str, ...]) -> tuple[list[str], list[BaseException]]: |
| results = [(path, _unlink_owner_path_result(path)) for path in paths] |
| return [path for path, (removed, _) in results if not removed], [ |
| error for _, (_, error) in results if error is not None |
| ] |
|
|
|
|
| def _unlink_owner_path_result(path: str) -> tuple[bool, BaseException | None]: |
| try: |
| cleanup_error = _unlink_owner_path(path) |
| except FileNotFoundError: |
| return True, None |
| except BaseException as error: |
| return False, error |
| return True, cleanup_error |
|
|
|
|
| def _raise_recorded_errors(message: str, errors: list[BaseException]) -> None: |
| if len(errors) > 1: |
| _raise_cleanup_errors(message, errors[0], *errors[1:]) |
| raise errors[0] |
|
|
|
|
| def _unlink_in_directory(directory: Path, name: str) -> BaseException | None: |
| if _UNLINK_SUPPORTS_DIR_FD: |
| directory_fd = _open_directory(str(directory)) |
| try: |
| os.unlink(name, dir_fd=directory_fd) |
| except BaseException as unlink_error: |
| try: |
| os.close(directory_fd) |
| except BaseException as close_error: |
| _raise_cleanup_errors("strict unlink directory cleanup failed", unlink_error, close_error) |
| raise |
| try: |
| os.close(directory_fd) |
| except BaseException as close_error: |
| return close_error |
| return None |
| if sys.platform != "win32": |
| Path(directory / name).unlink() |
| return None |
| retry_delay = 0.001 |
| for attempt in range(_UNLINK_MAX_RETRIES): |
| if (error := _unlink_error(directory / name)) is None: |
| return None |
| if error.errno not in {EACCES, EPERM} or attempt == _UNLINK_MAX_RETRIES - 1: |
| raise error |
| time.sleep(retry_delay) |
| retry_delay *= 2 |
| return None |
|
|
|
|
| def _unlink_error(path: Path) -> OSError | None: |
| try: |
| path.unlink() |
| except OSError as error: |
| return error |
| return None |
|
|
|
|
| def _open_directory(directory: str) -> int: |
| return os.open( |
| directory, |
| os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), |
| ) |
|
|
|
|
| def _read_record(path: Path, limit: int) -> bytes: |
| fd, record = _open_record(path, limit) |
| os.close(fd) |
| return record |
|
|
|
|
| def _open_record(path: Path, limit: int) -> tuple[int, bytes]: |
| path_stat = path.lstat() |
| if not stat.S_ISREG(path_stat.st_mode): |
| msg = f"{path} is not a regular file" |
| raise OSError(msg) |
| flags = os.O_RDONLY | _O_BINARY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0) |
| fd = os.open(path, flags) |
| try: |
| record = _read_opened_record(fd, path, path_stat, limit) |
| except BaseException as read_error: |
| try: |
| os.close(fd) |
| except BaseException as close_error: |
| _raise_cleanup_errors("strict record read cleanup failed", read_error, close_error) |
| raise |
| return fd, record |
|
|
|
|
| def _read_opened_record(fd: int, path: Path, path_stat: os.stat_result, limit: int) -> bytes: |
| opened_stat = os.fstat(fd) |
| if not stat.S_ISREG(opened_stat.st_mode): |
| msg = f"{path} is not a regular file" |
| raise OSError(msg) |
| if _file_identity(opened_stat) != _file_identity(path_stat): |
| msg = f"{path} changed while opening" |
| raise OSError(msg) |
| record = os.read(fd, limit + 1) |
| if len(record) > limit: |
| msg = f"{path} exceeds {limit} bytes" |
| raise OSError(msg) |
| return record |
|
|
|
|
| def _ensure_protocol_directory(lock_file: str, directory: Path) -> None: |
| try: |
| directory.mkdir() |
| except FileExistsError: |
| try: |
| mode = directory.lstat().st_mode |
| except OSError as error: |
| raise SoftFileLockProtocolError(lock_file, None, f"cannot inspect {directory}") from error |
| if stat.S_ISDIR(mode) and not stat.S_ISLNK(mode): |
| return |
| raise SoftFileLockProtocolError(lock_file, None, f"{directory} is not a real directory") from None |
|
|
|
|
| def _validate_force_break_name(name: str) -> None: |
| invalid_component = not name or name.startswith(".") or name in {os.curdir, os.pardir} |
| if invalid_component or any(separator in name for separator in ("/", "\\", "\x00")): |
| msg = "claim_name must be one public claim basename" |
| raise ValueError(msg) |
|
|
|
|
| def _require_exact_name(directory: Path, name: str) -> None: |
| with os.scandir(directory) as entries: |
| if not any(entry.name == name for entry in entries): |
| raise FileNotFoundError(ENOENT, os.strerror(ENOENT), name) |
|
|
|
|
| def _raise_if_hard_links_unsupported(lock_file: str, error: NotImplementedError | OSError) -> None: |
| unsupported = ( |
| isinstance(error, NotImplementedError) |
| or error.errno in _HARD_LINK_UNSUPPORTED_ERRNOS |
| or (getattr(error, "winerror", None) in _WINDOWS_HARD_LINK_UNSUPPORTED) |
| ) |
| if unsupported: |
| reason = "filesystem does not support atomic no-replace hard-link publication" |
| raise SoftFileLockProtocolError(lock_file, None, reason) from error |
|
|
|
|
| def _file_identity(st: os.stat_result) -> tuple[int, int]: |
| return st.st_dev, st.st_ino |
|
|
|
|
| __all__ = [ |
| "StrictSoftFileClaim", |
| "StrictSoftFileClaimState", |
| "StrictSoftFileLock", |
| ] |
|
|