Upload folder using huggingface_hub (part 4)
bbae060 verified 3 days ago
58.6 kB
from __future__ import annotations

__lazy_modules__ = [
    "contextlib",
    "filelock._error",
    "filelock._identity",
    "filelock._marker",
    "filelock._soft",
    "filelock._util",
    "functools",
    "pathlib",
    "secrets",
    "tempfile",
]

import contextlib
import errno
import functools
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 filelock._api import BaseFileLock, ThreadLocalFileContext, _canonical, _raise_cleanup_errors
from filelock._error import SoftFileLockProtocolError
from filelock._identity import host_name, owner_is_stale, process_start_token
from filelock._marker import parse_marker
from filelock._soft import MALFORMED_LOCK_AGE_THRESHOLD, _parse_lock_holder
from filelock._soft_protocol import STRICT_SOFT_SENTINEL_RECORD
from filelock._util import foreign_directory_owner, write_all

if sys.platform == "win32":  # pragma: win32 cover
    from filelock._windows import _is_unlinked, _open_deletable_for_reading, _pin_directory, _unpin_directory

if TYPE_CHECKING:
    from collections.abc import Callable, Iterator

StrictSoftFileClaimState = Literal["intent", "held"]
_OccupantState = Literal["sentinel", "held", "exited", "damaged", "unparsable"]

_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
#: How long a scan waits out a claim held in Windows' delete-pending state before treating it as unreadable.
_CLAIM_READ_GRACE: Final[float] = 0.5
_CLAIM_READ_RETRY: Final[float] = 0.002
#: Rides out a peer replacing the lock path mid-inspection; a foreign node or unreadable record outlasting it is damage.
_SENTINEL_READ_GRACE: Final[float] = 0.5
_NODE_DESCRIPTIONS: Final[dict[int, str]] = {stat.S_IFDIR: "a directory", stat.S_IFLNK: "a symlink"}
#: Windows opens descriptors in text mode by default, which rewrites newlines and truncates a record at a control byte.
#: The claim and sentinel records are exact binary, so every record descriptor must be binary; POSIX ignores the flag.
_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})

# Termux/Android CPython ships without os.link (bionic long had only linkat), so the strict backend's whole hard-link
# mechanism is absent there. Probe once and gate every os.link reference on it, so importing filelock still works and
# only an actual StrictSoftFileLock acquire reports the missing capability.
_HAS_LINK: Final[bool] = hasattr(os, "link")

# Probe dir_fd capability once at import. A per-call ``os.unlink in os.supports_dir_fd`` check flips to False the moment
# a test mocks os.unlink, silently diverting the code to a different branch than the one under test.
_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
_MKDIR_SUPPORTS_DIR_FD: Final[bool] = os.mkdir in os.supports_dir_fd
_SCANDIR_SUPPORTS_FD: Final[bool] = os.scandir in os.supports_fd


def _probe_link_follow_symlinks() -> bool:
    # os.supports_follow_symlinks lists os.link on PyPy, but its linkat then rejects follow_symlinks=False with EINVAL,
    # and Windows raises NotImplementedError for the option outright. Link a throwaway file for real so the answer
    # reflects the runtime rather than its advertisement, and treat any failure as "not honored": the option only
    # hardens a source this process created with O_EXCL, so skipping it is safe, and a real environment fault surfaces
    # when the actual link runs.
    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]:
    # GraalPy's errno omits ENOTSUP, so importing the name outright breaks every runtime that ships without it. ENOTSUP
    # wins wherever it exists, leaving every runtime that names it unchanged. EOPNOTSUPP only stands in for the ones
    # that do not, and it approximates rather than matches: the two codes agree on Linux but differ on macOS/BSD and on
    # Windows. Where neither exists a runtime that cannot name "operation not supported" cannot raise it either, and
    # ENOSYS/EXDEV still classify the link failures it can raise.
    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
    #: 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()"
    #: Contending processes each publish and rescan several files, so jitter their retries. Seconds. Only a
    #: poll_interval below this cap jitters; the default equals it and keeps a fixed cadence.
    _poll_backoff_cap: float = 0.05
    _occupant_seen: tuple[tuple[float, int, int], float] | None = None
    #: Set by the finalizer, after which no retry of a failed cleanup can run.
    _finalizing: bool = False

    def _acquire(self) -> None:
        # Finish a doorway whose claim cleanup failed before publishing, so this owner never contends with leftovers.
        self._discard_pending_doorways(exited_only=True)
        lock_path = Path(self._lock_path)
        coordination_directory = Path(f"{lock_path}{_COORDINATION_SUFFIX}")
        claim_directory = coordination_directory / _CLAIM_DIRECTORY_NAME
        # A process that can write these directories can remove claims from them, so they take the lock's mode plus
        # search permission where the mode grants read.
        directory_mode: Final = (mode := self._open_mode()) | (mode & 0o444) >> 2
        _ensure_protocol_directory(self.lock_file, coordination_directory, directory_mode)
        _ensure_claim_directory(self.lock_file, claim_directory, directory_mode)
        if (
            sentinel_fd := _open_or_create_sentinel(
                self.lock_file, lock_path, self._open_mode(), self._occupant_unchanged_for
            )
        ) 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:
            if won := self._attempt_doorway(claim_directory):
                self._mark_descriptor_owned(sentinel_fd, sentinel_identity)
        except BaseException:
            # Once recorded as held, the acquire's rollback releases the claim instead.
            if not self.is_locked:
                self._discard_doorway()
            raise
        if not won:
            self._discard_doorway()

    def _discard_pending_doorways(self, *, exited_only: bool) -> None:
        if self._context.owner_claim_paths and not self.is_locked:
            self._discard_doorway()
        if not isinstance(context := self._context, ThreadLocalFileContext):
            return
        # Only this registry reaches an exited thread's thread-local doorway; a live thread finishes its own.
        own: Final = context.current
        with self._transition_lock:
            for pending, thread in self._thread_pending.copy().values():
                if exited_only and thread.is_alive():
                    continue
                context.current = pending
                try:
                    self._discard_doorway()
                finally:
                    context.current = own

    def _occupant_unchanged_for(self, occupant: os.stat_result) -> float:
        # SoftFileLock's marker rule: the mtime is only tested for change, timed on this process's monotonic clock.
        key, now = (occupant.st_mtime, occupant.st_ino, occupant.st_size), time.monotonic()
        if (seen := self._occupant_seen) is None or seen[0] != key:
            self._occupant_seen = key, now
            return 0.0
        return now - seen[1]

    def _attempt_doorway(self, claim_directory: Path) -> bool:
        if _read_existing_claims(self.lock_file, claim_directory):
            return False

        token = secrets.token_hex(_TOKEN_HEX_LENGTH // 2)
        intent_name = _claim_name("intent", token)
        intent_path = str(claim_directory / intent_name)
        # Record each claim before its link: once the link may have committed, any later failure must leave the name for
        # _discard_doorway. The token makes the name unique, so removing it when the link never happened is a no-op.
        self._context.owner_claim_paths = (intent_path,)
        try:
            with _pin_claim_directory(claim_directory):
                publication_cleanup_error = _publish_record(
                    intent_path, _claim_record(token), self._open_mode(), _open_claim_directory
                )
        except _PrivateRecordReclaimedError:
            return False
        except (NotImplementedError, OSError) as error:
            _raise_if_hard_links_unsupported(self.lock_file, error)
            if isinstance(error, OSError) and error.errno == EEXIST:
                # Another file holds the name, so it is not this owner's to remove.
                self._context.owner_claim_paths = ()
                return False
            raise
        if publication_cleanup_error is not None:  # pragma: needs dir-fd
            raise publication_cleanup_error

        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
        ):
            return False

        held_name = _claim_name("held", token)
        held_path = str(claim_directory / held_name)
        self._context.owner_claim_paths = (held_path, intent_path)
        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)
            if isinstance(error, OSError) and error.errno == EEXIST:
                self._context.owner_claim_paths = (intent_path,)
            raise
        if link_cleanup_error is not None:  # pragma: needs dir-fd
            raise link_cleanup_error

        claims = _read_existing_claims(self.lock_file, claim_directory)
        # Keep the intent claim for the whole hold rather than unlinking it now. The intent has existed, unchanged,
        # since this owner published it, so a contender's os.scandir is guaranteed to return it (POSIX only leaves the
        # visibility of entries created or removed *during* a scan unspecified). The freshly linked held claim carries
        # no such guarantee: a scan that races its creation can miss it. Were the intent removed here, that scan could
        # observe neither claim and let a larger-token contender win over this owner. The stable intent is the witness
        # that keeps the phase-five min-token decision computed over the true set. Release unlinks both.
        return {intent_name, held_name}.issubset(claim.name for claim in claims) and min(
            _claim_token_key(claim.name) for claim in claims
        ) == f"v1-{token}.claim"

    @property
    def claims(self) -> tuple[StrictSoftFileClaim, ...]:
        """Published claims that block acquisition."""
        # An inspection writes nothing, so it trusts no record's age rather than probe the filesystem clock.
        return _read_existing_claims(self.lock_file, self._claim_directory, probe_clock=False)

    @property
    def lock_path_occupant(self) -> str | None:
        """
        What sits at the lock path in place of the strict sentinel, or ``None`` when the sentinel or nothing is there.
        A soft-lock owner's marker blocks acquisition without appearing in :attr:`~filelock.StrictSoftFileLock.claims`;
        this describes it, naming the PID and host when the marker records them. A directory, a symlink, a marker whose
        owner on this host has exited, or a file it cannot parse that stays unchanged for two seconds makes acquisition
        raise :class:`~filelock.SoftFileLockProtocolError` instead and needs an operator to remove it.
        """
        try:
            fd, _, occupant = _inspect_lock_path(Path(_canonical(self.lock_file)))
        except FileNotFoundError:
            return None
        if fd is None:
            return occupant
        os.close(fd)
        return None

    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)
        with _pin_claim_directory(directory := self._claim_directory):
            _require_exact_name(directory, claim_name)
            cleanup_error = _unlink_in_directory(directory, claim_name)
        if cleanup_error is not None:  # pragma: needs dir-fd
            raise cleanup_error

    def _release_abandoned_holds(self) -> None:
        self._finalizing = True
        super()._release_abandoned_holds()
        # release() skips a doorway whose cleanup failed, as it is no hold, so a dropped instance would strand claims.
        with contextlib.suppress(Exception):
            # GraalPy defers finalizers to the host collector, so its tests cannot drive this cleanup.
            self._discard_pending_doorways(exited_only=False)  # pragma: needs collected-finalization

    def _reconcile_failed_acquire(self, canonical: str) -> None:
        # The acquisition is over, so the next one resolves the working directory again rather than reuse this one's.
        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 = self._unlink_owner_claims()
        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:  # ruff:ignore[blind-except]  # preserve claim and sentinel cleanup errors
            errors.append(close_error)
        if errors:
            _raise_recorded_errors("strict release cleanup failed", errors)

    def _discard_doorway(self) -> None:
        remaining, errors = self._unlink_owner_claims()
        # The pending descriptor holds no lock and a retry needs the claim paths alone. A dropped instance does not
        # retry, so close it now.
        if (fd := self._context.pending_lock_file_fd) is not None:
            self._context.pending_lock_file_fd = self._context.pending_lock_file_fd_identity = None
            try:
                self._close_released_fd(fd, default_suppresses=False)
            except BaseException as close_error:  # ruff:ignore[blind-except]  # preserve claim and sentinel cleanup errors
                errors.append(close_error)
        if remaining:
            # Stay pending, not owned: is_locked reports only a held claim that passed the final rescan. The next
            # acquire retries this cleanup.
            _raise_recorded_errors("strict doorway claim cleanup failed", errors)
        self._mark_descriptor_released()
        if errors:
            _raise_recorded_errors("strict doorway cleanup failed", errors)

    def _unlink_owner_claims(self) -> tuple[list[str], list[BaseException]]:
        remaining, errors = _unlink_owner_paths(self._context.owner_claim_paths)
        self._context.owner_claim_paths = tuple(remaining)
        if self._finalizing:
            # No retry runs after the finalizer and no caller sees its errors, so report success and let it close the
            # descriptor even when claims remain.
            return [], []  # pragma: needs collected-finalization
        return remaining, 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
    #: The owner's process start token, or ``None`` when the platform exposes no proven start time. A strict lock never
    #: reclaims a claim on its own, so this identifies the owner for tooling rather than driving any automatic break.
    start: int | None = None


class _PrivateRecordReclaimedError(Exception):
    pass


def _open_or_create_sentinel(
    lock_file: str, path: Path, mode: int, unchanged_for: Callable[[os.stat_result], float]
) -> int | None:
    try:
        return _open_sentinel(lock_file, path, unchanged_for)
    except FileNotFoundError:
        pass

    _reclaim_sentinel_private_records(lock_file, path)
    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:  # pragma: needs dir-fd
            raise publication_cleanup_error
    try:
        return _open_sentinel(lock_file, path, unchanged_for)
    except FileNotFoundError:
        return None


def _open_sentinel(lock_file: str, path: Path, unchanged_for: Callable[[os.stat_result], float]) -> int | None:
    # A live soft-lock owner blocks like any holder. Anything else raises: claims cannot show it and force_break()
    # cannot remove it, so waiting would wedge every strict contender with no diagnostic. A file no lock parses may be a
    # soft lock between its create and its write, so it raises only once unchanged as long as a soft contender waits.
    deadline: float | None = None
    while True:
        fd, state, occupant = _inspect_lock_path(path)
        if fd is not None or state == "held":
            return fd
        reason = f"expected the strict sentinel at the lock path, found {occupant}"
        if state == "unparsable":
            if unchanged_for(path.lstat()) < MALFORMED_LOCK_AGE_THRESHOLD:
                return None
            raise SoftFileLockProtocolError(
                lock_file, None, f"{reason}, unchanged for {MALFORMED_LOCK_AGE_THRESHOLD:g} seconds"
            )
        if deadline is None:
            deadline = time.monotonic() + _SENTINEL_READ_GRACE
        if state == "exited" or time.monotonic() >= deadline:
            raise SoftFileLockProtocolError(lock_file, None, reason)
        time.sleep(_CLAIM_READ_RETRY)


def _inspect_lock_path(path: Path) -> tuple[int | None, _OccupantState, str]:
    # "held" may be a soft-lock owner that still runs, "exited" is one this host proves gone, "unparsable" is content no
    # lock parses, and "damaged" is a node no lock writes.
    if not stat.S_ISREG(mode := path.lstat().st_mode):
        return None, "damaged", _NODE_DESCRIPTIONS.get(stat.S_IFMT(mode), "a special file")
    try:
        fd, record = _open_record((os.fspath(path.parent), None), path.name, _CLAIM_RECORD_LIMIT)
    except FileNotFoundError:
        raise
    except PermissionError as error:
        # Windows denies the read while a legacy holder keeps its marker open, and a holder running as another user
        # may leave it unreadable here, so a marker that cannot be read may still be held.
        return None, "held", f"a file it cannot read ({error.strerror or error})"
    except OSError as error:
        return None, "damaged", f"a file it cannot read ({error.strerror or error})"
    if record == _LEGACY_SENTINEL:
        return fd, "sentinel", "the strict sentinel"
    os.close(fd)
    return None, *_describe_soft_marker(record.decode("utf-8", errors="replace"))


def _describe_soft_marker(content: str) -> tuple[_OccupantState, str]:
    # A SoftFileLock, MarkerSoftFileLock or SoftFileLease owner holds the path legitimately until it releases, so its
    # marker is contention. One this host proves gone is named, never removed: strict never acts on an inferred death.
    if (record := parse_marker(content)) is not None:
        pid, hostname, start = record.pid, record.hostname, record.start
    elif (holder := _parse_lock_holder(content)) is not None:
        pid, hostname, start = holder
    else:
        return "unparsable", "a file that is neither the sentinel nor a soft-lock marker it can parse"
    if owner_is_stale(pid, hostname, start):
        return "exited", f"a soft-lock marker left by pid {pid} on {hostname}, which has exited"
    return "held", f"a soft-lock marker held by pid {pid} on {hostname}"


def _read_claims(
    lock_file: str, directory_ref: tuple[str, int | None], *, probe_clock: bool
) -> tuple[StrictSoftFileClaim, ...]:
    names: Final = _list_claim_names(lock_file, directory_ref, probe_clock=probe_clock)
    name_parts: dict[str, tuple[StrictSoftFileClaimState, str]] = {}
    for name in names:
        if (parts := _parse_claim_name(name)) is None:
            raise SoftFileLockProtocolError(lock_file, name, "unknown claim name or protocol version")
        name_parts[name] = parts
    records: Final = _read_claim_records(lock_file, directory_ref, names, probe_clock=probe_clock)
    return tuple(_parse_claim(lock_file, name, name_parts[name], records[name]) for name in names if name in records)


def _list_claim_names(lock_file: str, directory_ref: tuple[str, int | None], *, probe_clock: bool) -> list[str]:
    try:
        with _scandir_relative(directory_ref) as entries:
            return _public_claim_names(directory_ref, entries, probe_clock=probe_clock)
    except OSError as error:
        reason = f"cannot list claim directory: {error.strerror or type(error).__name__}"
        raise SoftFileLockProtocolError(lock_file, None, reason) from error


def _read_claim_records(
    lock_file: str, directory_ref: tuple[str, int | None], names: list[str], *, probe_clock: bool
) -> dict[str, bytes]:
    # A contended scan can list a claim that is not yet cleanly readable, in two ways that both resolve on a brief
    # retry. Windows fails an open with EACCES while a peer's unlink holds the claim; once the deletion is pending,
    # _attempt_claim_read skips the claim outright, as another process's handle can keep the name listed for seconds.
    # On NFS, a peer that unlinks its own claim leaves this client's cached filehandle stale, so the next open returns
    # ESTALE rather than a clean ENOENT until the lookup revalidates against the server. Retrying re-runs the path
    # lookup, which turns the vanished claim into ENOENT (skip) or reads it if it still exists. A genuinely unreadable
    # record (a locked-down file, an EIO fault) still fails closed. The grace starts at the first pending result rather
    # than before the first attempt, so a slow first read cannot use up the window and leave no retry. One deadline
    # covers the scan: a deadline per claim let claims that clear one after another stretch a scan, and the caller's
    # timeout with it, by a grace each.
    records: dict[str, bytes] = {}
    deadline: float | None = None
    while True:
        pending: dict[str, OSError] = {}
        for name in names:
            record, error = _attempt_claim_read(lock_file, directory_ref, name)
            if error is not None:
                pending[name] = error
            elif record is not None:
                records[name] = record
        if not pending:
            return records
        if deadline is None:
            deadline = time.monotonic() + _CLAIM_READ_GRACE
        elif time.monotonic() >= deadline:
            break
        names = list(pending)
        time.sleep(_CLAIM_READ_RETRY)
    # A stale handle that outlives revalidation is a claim the server no longer has (RFC 1813 NFS3ERR_STALE): skip it
    # like ENOENT. Skipping a vanished claim overcounts contention at worst; it cannot free a held lock.
    if denied := [name for name, error in pending.items() if error.errno != ESTALE]:
        # Windows can deny opens after a peer removes the claim from the directory.
        listed: Final = set(_list_claim_names(lock_file, directory_ref, probe_clock=probe_clock))
        if still_listed := [name for name in denied if name in listed]:
            error = pending[still_listed[0]]
            reason = f"cannot read claim: {error.strerror or str(error) or type(error).__name__}"
            raise SoftFileLockProtocolError(lock_file, still_listed[0], reason) from error
    return records


def _attempt_claim_read(
    lock_file: str, directory_ref: tuple[str, int | None], name: str
) -> tuple[bytes | None, OSError | None]:
    # Return the record, or (None, None) when the claim has already gone, or (None, error) for an open the caller may
    # retry: a Windows EACCES while a peer's unlink holds the claim, or an NFS ESTALE from a peer unlinking its own
    # claim out from under this client's cached filehandle. Any other OSError is a real fault and fails closed.
    try:
        return _read_record(directory_ref, name, _CLAIM_RECORD_LIMIT), None
    except FileNotFoundError:
        return None, None
    except PermissionError as error:
        if sys.platform == "win32" and _is_unlinked(str(Path(directory_ref[0], name))):  # pragma: win32 cover
            return None, None
        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_ref: tuple[str, int | None], entries: Iterator[os.DirEntry[str]], *, probe_clock: bool
) -> list[str]:
    names: list[str] = []
    now: Callable[[], float | None] = (
        functools.cache(
            functools.partial(
                _filesystem_now, directory_ref, _claim_name("held", secrets.token_hex(_TOKEN_HEX_LENGTH // 2))
            )
        )
        if probe_clock
        else lambda: None
    )
    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(directory_ref, entry.name, now)
    return sorted(names)


def _read_existing_claims(
    lock_file: str, directory: Path, *, probe_clock: bool = True
) -> tuple[StrictSoftFileClaim, ...]:
    if not _OPEN_SUPPORTS_DIR_FD:  # pragma: lacks dir-fd
        try:
            pinned = _pin_claim_directory(directory)
        except FileNotFoundError:
            return ()
        except OSError as error:
            reason = f"cannot open claim directory: {error.strerror or type(error).__name__}"
            raise SoftFileLockProtocolError(lock_file, None, reason) from error
        with pinned:
            return _read_claims(lock_file, (os.fspath(directory), None), probe_clock=probe_clock)
    return _read_claims_through_descriptor(lock_file, directory, probe_clock=probe_clock)  # pragma: needs dir-fd


def _read_claims_through_descriptor(  # pragma: needs dir-fd
    lock_file: str, directory: Path, *, probe_clock: bool
) -> tuple[StrictSoftFileClaim, ...]:
    try:
        directory_fd = _open_claim_directory(os.fspath(directory))
    except FileNotFoundError:
        return ()
    except OSError as error:
        reason = f"cannot open claim directory: {error.strerror or type(error).__name__}"
        raise SoftFileLockProtocolError(lock_file, None, reason) from error
    try:
        return _read_claims(lock_file, (os.fspath(directory), directory_fd), probe_clock=probe_clock)
    finally:
        # The scan wrote nothing through the descriptor, so a failed close has no result to report.
        with contextlib.suppress(OSError):
            os.close(directory_fd)


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,
    open_directory: Callable[[str], int] | None = None,
) -> 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 or _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:  # preserve publication and directory cleanup errors
        try:
            if directory_fd is not None:  # pragma: needs dir-fd
                os.close(directory_fd)
        except BaseException as close_error:  # ruff:ignore[blind-except]  # pragma: needs dir-fd  # preserve publication and directory cleanup errors
            _raise_cleanup_errors("strict publication directory cleanup failed", publication_error, close_error)
        raise
    if directory_fd is not None:  # pragma: needs dir-fd
        try:
            os.close(directory_fd)
        except BaseException as close_error:  # ruff:ignore[blind-except]  # a claim's caller recorded its path before publishing
            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:  # pragma: needs o-nofollow
        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:  # preserve publication and cleanup errors
        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:  # ruff:ignore[blind-except]  # returned for grouping with unlink failures
        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:  # ruff:ignore[blind-except]  # returned for grouping with close failures
        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_or_find_committed(directory_ref, *names, private_identity)
    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(lock_file: str, path: Path) -> None:
    directory_ref = os.fspath(path.parent), None
    now = functools.cache(functools.partial(_filesystem_now, directory_ref, path.name))
    try:
        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)
    except OSError as error:
        reason = f"cannot reclaim sentinel private records: {error.strerror or error}"
        raise SoftFileLockProtocolError(lock_file, None, reason) from error


def _reclaim_private_record(
    directory_ref: tuple[str, int | None], private_name: str, now: Callable[[], float | None]
) -> 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 (
        (current := now()) is None or current - 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 _filesystem_now(directory_ref: tuple[str, int | None], public_name: str) -> float | None:
    # A network filesystem stamps mtimes with the server's clock, which can run seconds off this client's, so read "now"
    # as the mtime of a probe written like a record (liblockfile uses a read's atime). The probe has a private-record
    # name, so a scan reaps one left behind. Without a stamp no age is trusted.
    probe = _private_record_name(public_name)
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_BINARY | getattr(os, "O_NOFOLLOW", 0)
    try:
        fd = _open_relative(directory_ref, probe, flags, 0o600)
    except OSError:
        return None
    try:
        write_all(fd, b"\n")
        return os.fstat(fd).st_mtime
    except OSError:
        return None
    finally:
        with contextlib.suppress(OSError):
            os.close(fd)
        with contextlib.suppress(OSError):
            _unlink_private_record_once(directory_ref, probe)


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:  # pragma: needs 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:  # pragma: needs dir-fd
            path_stat = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
        else:  # pragma: win32 cover
            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:  # pragma: needs 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))  # pragma: win32 cover


def _link_or_find_committed(
    directory_ref: tuple[str, int | None],
    source_name: str,
    destination_name: str,
    source_identity: tuple[int, int] | None = None,
) -> None:
    try:
        _link_relative(directory_ref, source_name, destination_name)
    except FileExistsError:
        # NFS retransmits a LINK whose reply was lost and reports EEXIST for the committed link; as in open(2) NOTES
        # and liblockfile, it succeeded when the destination names the source. Prefer the fstat-ed identity: a peer's
        # scan reaps a private record once it has two links, so its name may already resolve to nothing.
        if (identity := _relative_identity(directory_ref, destination_name)) is None or identity != (
            source_identity or _relative_identity(directory_ref, source_name)
        ):
            raise


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:
        # No os.link at all (Termux/Android): report it as unsupported like a filesystem that refuses hard links, so
        # the acquire path raises SoftFileLockProtocolError instead of a bare AttributeError.
        msg = "os.link is unavailable on this platform"
        raise NotImplementedError(msg)
    # The source is a private record this process created with O_CREAT | O_EXCL, so follow_symlinks guards nothing an
    # attacker can reach. Pass the option only when the runtime honors it: PyPy advertises it through
    # os.supports_follow_symlinks yet its linkat rejects it with EINVAL, so probe once rather than trust the set.
    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:  # pragma: lacks link-follow-symlinks
        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:  # pragma: needs dir-fd
        os.unlink(name, dir_fd=directory_fd)
    elif sys.platform == "win32":  # pragma: win32 cover
        _unlink_in_directory(Path(directory), name)
    else:
        Path(directory, name).unlink()  # pragma: win32 no cover


def _link_no_replace(directory: Path, source_name: str, destination_name: str) -> BaseException | None:
    if _LINK_SUPPORTS_DIR_FD:  # pragma: needs dir-fd
        directory_fd = _open_claim_directory(str(directory))
        try:
            _link_or_find_committed((str(directory), directory_fd), source_name, destination_name)
        except BaseException as link_error:  # preserve link and directory cleanup errors
            try:
                os.close(directory_fd)
            except BaseException as close_error:  # ruff:ignore[blind-except]  # preserve link and directory cleanup errors
                _raise_cleanup_errors("strict link directory cleanup failed", link_error, close_error)
            raise
        try:
            os.close(directory_fd)
        except BaseException as close_error:  # ruff:ignore[blind-except]  # the caller recorded the held path before linking
            return close_error
        return None
    with _pin_claim_directory(directory):  # pragma: win32 cover
        _link_or_find_committed((str(directory), None), source_name, destination_name)
    return None  # pragma: win32 cover


def _unlink_owner_path(path: str) -> BaseException | None:
    directory, name = os.path.split(path)
    with _pin_claim_directory(claim_directory := Path(directory or os.curdir)):
        return _unlink_in_directory(claim_directory, 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:  # ruff:ignore[blind-except]  # keep ownership when unlink did not commit
        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:  # pragma: needs dir-fd
        directory_fd = _open_claim_directory(str(directory))
        try:
            os.unlink(name, dir_fd=directory_fd)
        except BaseException as unlink_error:  # preserve unlink and directory cleanup errors
            try:
                os.close(directory_fd)
            except BaseException as close_error:  # ruff:ignore[blind-except]  # preserve unlink and directory cleanup errors
                _raise_cleanup_errors("strict unlink directory cleanup failed", unlink_error, close_error)
            raise
        try:
            os.close(directory_fd)
        except BaseException as close_error:  # ruff:ignore[blind-except]  # caller commits the removed path before raising
            return close_error
        return None
    if sys.platform != "win32":  # pragma: win32 no cover
        Path(directory / name).unlink()
        return None
    retry_delay = 0.001  # pragma: win32 cover
    for attempt in range(_UNLINK_MAX_RETRIES):  # pragma: win32 cover
        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  # pragma: no cover  # the final retry returns or raises


def _unlink_error(path: Path) -> OSError | None:  # pragma: win32 cover
    try:
        path.unlink()
    except OSError as error:
        return error
    return None


def _pin_claim_directory(directory: Path) -> contextlib.ExitStack[bool | None]:
    return _pin_directories(directory.parent, directory)


def _pin_directories(*directories: Path) -> contextlib.ExitStack[bool | None]:
    # Windows has no dir_fd, so each operation resolves every path component by name again. An open handle on a
    # directory blocks a rename of it, and with it a swap for a junction (a directory that points at another directory),
    # until the operation ends. Other platforms work through descriptors and hold nothing.
    stack: contextlib.ExitStack[bool | None] = contextlib.ExitStack()
    if sys.platform == "win32":  # pragma: win32 cover
        with stack:
            for directory in directories:
                stack.callback(_unpin_directory, _pin_directory(str(directory)))
            return stack.pop_all()
    return stack  # pragma: win32 no cover


def _open_directory(directory: str, dir_fd: int | None = None) -> int:  # pragma: needs dir-fd
    return os.open(directory, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), dir_fd=dir_fd)


def _open_claim_directory(directory: str) -> int:  # pragma: needs dir-fd
    # O_NOFOLLOW applies to the last path component. A process that can rename entries in the lock's directory can
    # swap the coordination directory for a symlink after any check of its name, so open the claim directory relative
    # to a descriptor of the coordination directory: a symlink at either level then fails the open.
    coordination_directory, name = os.path.split(directory)
    coordination_fd = _open_directory(coordination_directory)
    try:
        return _open_directory(name, coordination_fd)
    finally:
        # The walk wrote nothing through it, and an error here would leak the descriptor this function returns.
        with contextlib.suppress(OSError):
            os.close(coordination_fd)


def _scandir_relative(
    directory_ref: tuple[str, int | None],
) -> contextlib.AbstractContextManager[Iterator[os.DirEntry[str]]]:
    directory, directory_fd = directory_ref
    if directory_fd is not None and _SCANDIR_SUPPORTS_FD:  # pragma: needs dir-fd
        return os.scandir(directory_fd)
    return os.scandir(directory)  # pragma: lacks dir-fd


def _lstat_relative(directory_ref: tuple[str, int | None], name: str) -> os.stat_result:
    directory, directory_fd = directory_ref
    if directory_fd is not None and _STAT_SUPPORTS_DIR_FD:  # pragma: needs dir-fd
        return os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
    return Path(directory, name).lstat()


def _read_record(directory_ref: tuple[str, int | None], name: str, limit: int) -> bytes:
    fd, record = _open_record(directory_ref, name, limit, deletable=True)
    os.close(fd)
    return record


def _open_record(
    directory_ref: tuple[str, int | None], name: str, limit: int, *, deletable: bool = False
) -> tuple[int, bytes]:
    path = Path(directory_ref[0], name)
    path_stat = _lstat_relative(directory_ref, name)
    if not stat.S_ISREG(path_stat.st_mode):
        msg = f"{path} is not a regular file"
        raise OSError(msg)
    if deletable and sys.platform == "win32":  # pragma: win32 cover
        # A claim's owner must be able to unlink it while another process reads it. The sentinel stays for the lock's
        # lifetime, so it keeps os.open.
        fd = _open_deletable_for_reading(str(path))
    else:
        flags = os.O_RDONLY | _O_BINARY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
        fd = _open_relative(directory_ref, name, flags, 0)
    try:
        record = _read_opened_record(fd, path, path_stat, limit)
    except BaseException as read_error:  # preserve read and descriptor cleanup errors
        try:
            os.close(fd)
        except BaseException as close_error:  # ruff:ignore[blind-except]  # preserve read and descriptor cleanup errors
            _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_claim_directory(lock_file: str, directory: Path, mode: int) -> None:
    if not (_OPEN_SUPPORTS_DIR_FD and _MKDIR_SUPPORTS_DIR_FD):  # pragma: lacks dir-fd
        try:
            pinned = _pin_directories(directory.parent)
        except OSError as error:
            raise SoftFileLockProtocolError(lock_file, None, f"cannot open {directory.parent}") from error
        with pinned:
            _ensure_protocol_directory(lock_file, directory, mode)
    else:  # pragma: needs dir-fd
        # mkdir follows a symlink in every component but the last, so create against a descriptor of the real parent.
        try:
            coordination_fd = _open_directory(os.fspath(directory.parent))
        except OSError as error:
            raise SoftFileLockProtocolError(lock_file, None, f"cannot open {directory.parent}") from error
        try:
            _ensure_protocol_directory(lock_file, directory, mode, coordination_fd)
        finally:
            with contextlib.suppress(OSError):
                os.close(coordination_fd)


def _ensure_protocol_directory(lock_file: str, directory: Path, mode: int, parent_fd: int | None = None) -> None:
    try:
        if parent_fd is None:
            directory.mkdir(mode)
        else:  # pragma: needs dir-fd
            os.mkdir(directory.name, mode, dir_fd=parent_fd)
    except FileExistsError:
        try:
            found = _lstat_relative((os.fspath(directory.parent), parent_fd), directory.name)
            owner = foreign_directory_owner(
                found, directory.parent.stat() if parent_fd is None else os.fstat(parent_fd)
            )
        except OSError as error:
            raise SoftFileLockProtocolError(lock_file, None, f"cannot inspect {directory}") from error
        if (
            not stat.S_ISDIR(found.st_mode)
            or getattr(found, "st_file_attributes", 0) & stat.FILE_ATTRIBUTE_REPARSE_POINT
        ):
            raise SoftFileLockProtocolError(lock_file, None, f"{directory} is not a real directory") from None
        if owner is not None:  # pragma: needs foreign-sticky-tmp
            reason = f"{directory} belongs to uid {owner}, not to this process's account or the parent's owner"
            raise SoftFileLockProtocolError(lock_file, None, reason) from None
    _settle_directory_mode(lock_file, directory, mode, parent_fd)


def _settle_directory_mode(lock_file: str, directory: Path, mode: int, parent_fd: int | None) -> None:
    # The umask or a default ACL can leave the directory wider than the mode, now or in an earlier release, so narrow a
    # directory this account owns to the mode. The sticky bit (a directory flag that lets only a file's owner remove it)
    # then protects each claim from other accounts even where the mode admits them. Windows has neither mode bits nor a
    # sticky bit.
    if sys.platform != "win32":  # pragma: win32 no cover
        try:
            fd = os.open(
                directory.name if parent_fd is not None else os.fspath(directory),
                os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0),
                dir_fd=parent_fd,
            )
        except OSError as error:
            raise SoftFileLockProtocolError(lock_file, None, f"cannot open {directory}") from error
        try:
            found = os.fstat(fd)
            settled = found.st_mode & 0o7000 | found.st_mode & mode & 0o777 | stat.S_ISVTX
            if found.st_uid == os.geteuid() and settled != stat.S_IMODE(found.st_mode):
                os.fchmod(fd, settled)
        except OSError as error:
            raise SoftFileLockProtocolError(lock_file, None, f"cannot set the mode of {directory}") from error
        finally:
            os.close(fd)


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:
    directory_fd = _open_claim_directory(os.fspath(directory)) if _OPEN_SUPPORTS_DIR_FD else None
    try:
        with _scandir_relative((os.fspath(directory), directory_fd)) as entries:
            if not any(entry.name == name for entry in entries):
                raise FileNotFoundError(ENOENT, os.strerror(ENOENT), name)
    finally:
        if directory_fd is not None:  # pragma: needs dir-fd
            with contextlib.suppress(OSError):
                os.close(directory_fd)


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",
]