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

__lazy_modules__ = [
    "collections",
    "filelock._error",
    "filelock._identity",
    "hashlib",
    "itertools",
    "pathlib",
    "secrets",
    "tempfile",
]

import hashlib
import logging
import os
import pathlib
import secrets
import sqlite3
import stat
import sys
import tempfile
import threading
import time
from collections import deque
from contextlib import contextmanager, suppress
from dataclasses import dataclass, field
from itertools import count
from typing import TYPE_CHECKING, ClassVar, Final, Literal, NamedTuple, TypeAlias, cast
from weakref import WeakValueDictionary

from filelock._api import (
    AcquireReturnProxy,
    _acquire_thread_lock,
    _canonical,
    _ensure_current_process,
    _fork_transition,
    _raise_chained_errors,
    _register_fork_class,
    _register_fork_object,
    _register_owned_descriptor,
    _resolve_lazy_imports_before_shutdown,
    _resolve_read_write_timeout,
    _seconds,
    _timeout_behavior,
    _unregister_owned_descriptor,
)
from filelock._error import Timeout
from filelock._identity import host_name, owner_is_stale, process_start_token

if sys.platform == "win32":  # pragma: win32 cover
    from filelock._windows import (
        _directory_is_private,
        _identity_at,
        _identity_of_fd,
        _make_private_directory,
        _open_non_reparse_fd,
        _pinned,
    )

if TYPE_CHECKING:
    from collections.abc import Callable, Generator
    from typing import Self

    from _typeshed import Unused

_LOGGER: Final[logging.Logger] = logging.getLogger("filelock")
_GETPID: Final[Callable[[], int]] = os.getpid
_IS_PYPY: Final[bool] = sys.implementation.name == "pypy"
_NEEDS_CONNECTION_ESCROW: Final[bool] = (
    hasattr(os, "register_at_fork") and sys.implementation.name == "cpython" and sys.version_info < (3, 12)
)
_ConnectionParameter: TypeAlias = (
    str | bytes | os.PathLike[str] | os.PathLike[bytes] | float | int | type[sqlite3.Connection] | None
)
_DatabaseIdentity: TypeAlias = tuple[int, int]

# sqlite3_busy_timeout() accepts a C int, max 2_147_483_647 on 32-bit. Use a lower value to be safe (~23 days).
_MAX_SQLITE_TIMEOUT_MS: Final[int] = 2_000_000_000 - 1
# How long a wait that its caller may abandon runs before it checks whether it should stop.
_CANCEL_POLL_INTERVAL: Final[float] = 0.1
_UNSAFE_FORK_EXIT_STATUS: Final[int] = 70
#: The longest a contended acquisition waits for the database inside one fork transition, in seconds.
_MAX_TRANSITION_WAIT: Final[float] = 0.05
# O_NONBLOCK keeps an open from blocking on a FIFO planted at the path; the regular-file check then rejects it.
_DB_OPEN_FLAGS: Final[int] = os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
_DIRECTORY_FLAGS: Final[int] = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
# Header bytes 18 and 19 hold the file format versions, 2 in WAL mode: https://www.sqlite.org/fileformat.html
_SQLITE_MAGIC: Final[bytes] = b"SQLite format 3\x00"
_FORMAT_VERSIONS: Final[slice] = slice(18, 20)
_HEADER_SIZE: Final[int] = 20
_WAL_FORMAT_VERSION: Final[int] = 2


class _SQLiteTransitionContext(threading.local):
    depth: int = 0


_SQLITE_TRANSITION_CONTEXT: Final = _SQLiteTransitionContext()


class _ConnectionEscrow:
    def __init__(self) -> None:
        self._lock = threading.RLock()
        self._functions: tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None = None

    def functions(
        self,
    ) -> tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None:
        if not _NEEDS_CONNECTION_ESCROW:
            return None  # pragma: >=3.12 cover
        with self._lock:  # pragma: <3.12 cover  # pragma: needs fork
            if self._functions is None:
                import ctypes  # ruff:ignore[import-outside-top-level]  # keep optional ctypes and its audited dlsym out of ordinary imports

                function_type = ctypes.PYFUNCTYPE(None, ctypes.py_object)
                increment_address = ctypes.cast(ctypes.pythonapi.Py_IncRef, ctypes.c_void_p).value
                decrement_address = ctypes.cast(ctypes.pythonapi.Py_DecRef, ctypes.c_void_p).value
                if increment_address is None or decrement_address is None:  # pragma: no cover - resolved CPython API
                    msg = "CPython reference functions have no address"
                    raise RuntimeError(msg)
                self._functions = (
                    cast("Callable[[sqlite3.Connection], None]", function_type(increment_address)),
                    cast("Callable[[sqlite3.Connection], None]", function_type(decrement_address)),
                )
            return self._functions

    def _reset_after_fork_in_child(self) -> None:  # pragma: forked child
        self._lock = threading.RLock()


_CONNECTION_ESCROW: Final = _ConnectionEscrow()


class _ForkedDatabaseRegistry:
    def __init__(self) -> None:
        self._lock = threading.RLock()
        self._paths: set[pathlib.Path] = set()
        self._identities: set[_DatabaseIdentity] = set()
        self._sqlite_used = False
        self._all_paths_poisoned = False

    def raise_if_poisoned(self, path: pathlib.Path) -> None:
        identity = self.identity(path)
        with self._lock:
            all_paths_poisoned = self._all_paths_poisoned
            poisoned = (
                all_paths_poisoned or path in self._paths or (identity is not None and identity in self._identities)
            )
        if poisoned:  # pragma: needs fork
            msg = (
                "ReadWriteLock is unavailable in a PyPy fork child; exec or exit before using it"
                if all_paths_poisoned
                else f"SQLite database {path!s} was active across fork(); exec or exit before using it in the child"
            )
            raise RuntimeError(msg)

    def poison_after_fork(self, path: pathlib.Path, identity: _DatabaseIdentity | None) -> None:
        self._paths.add(path)
        if identity is not None:
            self._identities.add(identity)

    def note_sqlite_use(self) -> None:
        if _IS_PYPY:
            with self._lock:
                self._sqlite_used = True

    def _reset_after_fork_in_child(self) -> None:  # pragma: forked child
        self._lock = threading.RLock()
        self._all_paths_poisoned = self._all_paths_poisoned or (_IS_PYPY and self._sqlite_used)
        self._sqlite_used = False

    @staticmethod
    def identity(path: pathlib.Path) -> _DatabaseIdentity | None:
        try:
            stat_result = path.stat()
        except OSError:
            return None
        return stat_result.st_dev, stat_result.st_ino


_FORKED_DATABASES: Final = _ForkedDatabaseRegistry()

_LINK_PREFIX: Final[str] = ".filelock-"
_LINK_NAME: Final[str] = "lock.db"
_SWEPT_DATABASES: Final[set[_DatabaseIdentity]] = set()


# POSIX locks belong to the process and the inode, so closing any descriptor on the inode drops the locks every other
# connection of this process holds there. SQLite defers its own closes for this reason (section 2.2 of
# https://www.sqlite.org/howtocorrupt.html); the probes stay open until the last connection on their inode closes.
class _ProbeDescriptors:
    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._inodes: dict[_DatabaseIdentity, _InodeProbes] = {}
        self._releases: deque[_Probe] = deque()
        self._leases = count()

    def adopt(self, database: str, claimed: list[_Probe]) -> _Probe | None:  # pragma: win32 no cover
        # Reuse a probe no live connection holds: each connect would otherwise keep another descriptor open until the
        # inode's last close.
        try:
            stat_result: Final = os.lstat(database)
        except OSError:
            return None
        identity: Final = stat_result.st_dev, stat_result.st_ino
        with self._lock:
            if not stat.S_ISREG(stat_result.st_mode) or not (probes := self._inodes.get(identity)) or not probes.idle:
                return None
            claimed.append(adopted := _Probe(identity, probes.idle.pop(), next(self._leases)))
            probes.leases.add(adopted.lease)
        return adopted

    def retain(self, fd: int, claimed: list[_Probe]) -> _Probe:  # pragma: win32 no cover
        stat_result: Final = os.fstat(fd)
        identity: Final = stat_result.st_dev, stat_result.st_ino
        token: Final = _register_owned_descriptor(fd, identity)
        with self._lock:
            (probes := self._inodes.setdefault(identity, _InodeProbes())).descriptors.append((fd, token))
            claimed.append(retained := _Probe(identity, fd, next(self._leases)))
            probes.leases.add(retained.lease)
        return retained

    def release(self, probe: _Probe) -> None:  # pragma: win32 no cover
        # A finalizer may run this on a thread already inside the lock, so it never waits; the lock holder drains.
        self._releases.append(probe)
        self.drain()

    def drain(self) -> None:  # pragma: win32 no cover
        while self._releases and self._lock.acquire(blocking=False):
            try:
                # Dequeued only once applied, so an exception leaves the rest of the release for the next drain.
                while self._releases:
                    self._release_locked(self._releases[0])
                    self._releases.popleft()
            finally:
                self._lock.release()

    def _release_locked(self, probe: _Probe) -> None:  # pragma: win32 no cover
        if (probes := self._inodes.get(probe.identity)) is None:
            return
        if probe.lease in probes.leases:
            probes.leases.remove(probe.lease)
            if probes.leases:
                probes.idle.append(probe.fd)
        if probes.leases:
            return
        # No connect may adopt a probe about to close, even one an interrupt leaves listed.
        probes.idle.clear()
        # Closing under the lock keeps a concurrent connect from taking a lock on this inode before the close.
        while probes.descriptors:
            fd, token = probes.descriptors[-1]
            if token is not None:  # pragma: needs fork
                _unregister_owned_descriptor(token)
            # Forgotten before the close: after an interrupt, a leaked descriptor beats closing a reused number.
            probes.descriptors.pop()
            # The descriptor is gone even when close reports an error, and no lock of this process depends on it.
            with suppress(OSError):
                os.close(fd)
        del self._inodes[probe.identity]

    def _reset_after_fork_in_child(self) -> None:  # pragma: forked child
        # The child holds no POSIX locks, and the owned-descriptor registry closes the inherited probes.
        self._lock = threading.Lock()
        self._inodes = {}
        self._releases = deque()


class _Operation(NamedTuple):
    pid: int
    start_time: float
    #: Asked between bounded waits, so an abandoned acquire stops waiting; ``None`` waits in one call.
    cancel_check: Callable[[], bool] | None


class _Probe(NamedTuple):
    identity: _DatabaseIdentity
    fd: int
    # One per connection, so a release applied again after an interrupt kept it queued counts once.
    lease: int


@dataclass(slots=True)
class _InodeProbes:
    leases: set[int] = field(default_factory=set)
    descriptors: list[tuple[int, int | None]] = field(default_factory=list)
    idle: list[int] = field(default_factory=list)


_PROBE_DESCRIPTORS: Final = _ProbeDescriptors()


class _ForkSafeConnection(sqlite3.Connection):
    _creator_pid: int
    _decrement_escrow: Callable[[sqlite3.Connection], None] | None
    _database_directory: _LinkDirectory | None
    database_identity: _DatabaseIdentity | None = None
    _probe: _Probe | None = None

    def __new__(
        cls,
        *_args: _ConnectionParameter,
        **_kwargs: _ConnectionParameter,
    ) -> Self:
        connection = super().__new__(cls)
        connection._creator_pid = _GETPID()
        connection._decrement_escrow = None
        connection._database_directory = None
        return connection

    def close(self) -> None:
        with _sqlite_transition():
            if _GETPID() != self._creator_pid:  # pragma: needs fork
                return
            with _fork_transition():
                sqlite3.Connection.close(self)
                try:
                    if (directory := self._database_directory) is not None:  # pragma: needs posix-hard-link
                        self._database_directory = None
                        _remove_link_directory(directory)
                finally:
                    if (probe := self._probe) is not None:  # pragma: win32 no cover
                        self._probe = self.database_identity = None
                        _PROBE_DESCRIPTORS.release(probe)
                if (decrement := self._decrement_escrow) is not None:  # pragma: <3.12 cover  # pragma: needs fork
                    self._decrement_escrow = None
                    decrement(self)

    def retain_link(self, directory: _LinkDirectory | None) -> None:  # pragma: win32 cover
        """Keep the private link directory, if there is one, until SQLite no longer uses it."""
        self._database_directory = directory

    def retain_until_close(self, directory: _LinkDirectory | None, probe: _Probe) -> None:  # pragma: win32 no cover
        """Keep the private link and this connection's probe descriptor until SQLite no longer uses them."""
        self._database_directory = directory
        self._probe = probe
        self.database_identity = probe.identity

    def acquire_escrow(  # pragma: <3.12 cover  # pragma: needs fork
        self,
        functions: tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None,
    ) -> None:
        # The caller only reaches here holding the escrow functions; it skips the call entirely without them.
        if functions is not None:  # pragma: no branch
            increment, decrement = functions
            increment(self)
            self._decrement_escrow = decrement

    def __del__(self) -> None:
        with suppress(sqlite3.Error, RuntimeError, OSError):
            self.close()


@dataclass(slots=True)
class _OwnedConnection:
    """
    The connection of one lock, closed when the lock is collected.
    CPython runs ``__del__`` before it clears weak references. A finalizer on the lock itself would close the
    connection while the singleton cache still hands the lock out, and a thread given the lock would nest into a hold
    with no database lock behind it. The lock alone refers to this holder, so the holder is collected once the cache no
    longer finds the lock.
    """

    connection: _ForkSafeConnection | None = None

    def __del__(self) -> None:
        if (connection := self.connection) is not None:
            with suppress(sqlite3.Error, RuntimeError):
                connection.close()


class _ReadWriteLockMeta(type):
    """
    Resolve singleton instances for ``is_singleton=True`` construction.
    This logic lives here rather than in ReadWriteLock.get_lock so ``ReadWriteLock(path)`` returns cached instances
    without a 2-arg ``super()`` call that type checkers cannot verify.
    """

    _instances: WeakValueDictionary[pathlib.Path, ReadWriteLock]
    _instances_lock: threading.RLock
    _instances_pid: int
    _instances_under_construction: set[pathlib.Path]

    def __call__(
        cls,
        lock_file: str | os.PathLike[str],
        timeout: float = -1,
        *,
        blocking: bool = True,
        is_singleton: bool = True,
    ) -> ReadWriteLock:
        _ensure_current_process()
        _resolve_lazy_imports_before_shutdown()
        timeout = _seconds("timeout", timeout)
        if cls._instances_pid != _GETPID():
            cls._reset_class_after_fork()
        construction_pid = _GETPID()
        if not is_singleton:
            instance = super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton)
            if _GETPID() != construction_pid:  # pragma: forked child
                msg = "ReadWriteLock construction cannot continue after fork"
                raise RuntimeError(msg)
            return instance

        # Resolve only the parent, so a symlink at the lock path stays its own key instead of aliasing its target.
        normalized = pathlib.Path(_canonical(lock_file))
        with cls._instances_lock:
            # One lookup, since another thread can drop the last reference between a membership check and a read.
            if (instance := cls._instances.get(normalized)) is None:
                if normalized in cls._instances_under_construction:  # pragma: no cover - exercised in an audit callback
                    msg = f"Singleton lock construction is already active for {lock_file!s}"
                    raise RuntimeError(msg)
                construction_registry = cls._instances_under_construction
                construction_registry.add(normalized)
                try:
                    instance = super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton)
                finally:
                    if _GETPID() == construction_pid:
                        construction_registry.discard(normalized)
                if _GETPID() != construction_pid:
                    msg = "ReadWriteLock construction cannot continue after fork"
                    raise RuntimeError(msg)
                cls._instances[normalized] = instance

            if (
                _timeout_behavior(instance.timeout, any_negative=False)
                != _timeout_behavior(timeout, any_negative=False)
                or instance.blocking != blocking
            ):
                msg = (
                    f"Singleton lock created with timeout={instance.timeout}, blocking={instance.blocking},"
                    f" cannot be changed to timeout={timeout}, blocking={blocking}"
                )
                raise ValueError(msg)
            return instance

    def _reset_class_after_fork(cls) -> None:  # pragma: forked child
        cls._instances = WeakValueDictionary()
        cls._instances_lock = threading.RLock()
        cls._instances_pid = _GETPID()
        cls._instances_under_construction = set()


class ReadWriteLock(metaclass=_ReadWriteLockMeta):
    """
    Cross-process read-write lock backed by SQLite.
    Allows concurrent shared readers or a single exclusive writer. The lock is reentrant within the same mode (multiple
    ``acquire_read`` calls nest, as do multiple ``acquire_write`` calls from the same thread), but upgrading from read
    to write or downgrading from write to read raises :class:`RuntimeError`. Write locks are pinned to the thread that
    acquired them, and each thread sharing a read lock can release only the levels it acquired.
    An instance coordinates processes, not threads: a thread asking for the opposite mode of another thread's hold
    raises :class:`RuntimeError` instead of waiting. Threads that must wait for each other each need their own
    instance (``is_singleton=False``).
    By default, ``is_singleton=True``: calling ``ReadWriteLock(path)`` with the same resolved path returns the same
    instance. A ``.db`` extension is a convention rather than a requirement; the filesystem must be one the active
    SQLite VFS supports. Give the lock a file of its own: a WAL-mode database, or one with a schema larger than a lock
    needs, raises :class:`ValueError`. A database directory that other accounts can write must also be writable by
    this one and support hard links; the protection against other accounts ends once one of them can replace the
    database directory or a directory above it.
    :param lock_file: path to the SQLite database file used as the lock
    :param timeout: maximum wait time in seconds; ``-1`` waits up to SQLite's busy-timeout cap of about 23 days
    :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
    :param is_singleton: if ``True``, reuse existing instances for the same resolved path
    .. versionadded:: 3.21.0
    """

    _instances: WeakValueDictionary[pathlib.Path, ReadWriteLock] = WeakValueDictionary()
    _instances_lock = threading.RLock()
    _instances_pid = _GETPID()
    _instances_under_construction: ClassVar[set[pathlib.Path]] = set()

    def __init_subclass__(cls) -> None:
        super().__init_subclass__()
        cls._instances = WeakValueDictionary()
        cls._instances_lock = threading.RLock()
        cls._instances_pid = _GETPID()
        cls._instances_under_construction = set()
        _register_fork_class(cls)

    @classmethod
    def get_lock(
        cls, lock_file: str | os.PathLike[str], timeout: float = -1, *, blocking: bool = True
    ) -> ReadWriteLock:
        """
        Return the singleton :class:`ReadWriteLock` for *lock_file*.
        :param lock_file: path to the SQLite database file used as the lock
        :param timeout: maximum wait time in seconds; ``-1`` waits up to SQLite's busy-timeout cap of about 23 days
        :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
        :returns: the singleton lock instance
        :raises ValueError: if an instance already exists for this path with different *timeout* or *blocking* values
        """
        return cls(lock_file, timeout, blocking=blocking)

    def __init__(
        self,
        lock_file: str | os.PathLike[str],
        timeout: float = -1,
        *,
        blocking: bool = True,
        is_singleton: bool = True,  # ruff:ignore[unused-method-argument]  # consumed by _ReadWriteLockMeta.__call__
    ) -> None:
        self.lock_file = os.fspath(lock_file)
        self._canonical_path = pathlib.Path(_canonical(lock_file))
        _FORKED_DATABASES.raise_if_poisoned(self._canonical_path)
        self.timeout = timeout
        self.blocking = blocking
        self._transaction_lock = threading.Lock()  # serializes the (possibly blocking) SQLite transaction work
        self._internal_lock = threading.Lock()  # protects _lock_level / _current_mode updates and rollback
        self._lock_level = 0
        self._current_mode: Literal["read", "write"] | None = None
        self._write_thread_id: int | None = None
        self._read_depths: dict[int, int] = {}
        self._acquisition_thread_ids: set[int] = set()
        self._owned_connection = _OwnedConnection()
        self._connection_transaction_released = True
        self._connection_identity: _DatabaseIdentity | None = None
        self._closed = False
        self._creator_pid = _GETPID()
        self._fork_invalidated = False
        _register_fork_object(self)
        with _fork_transition(), _sqlite_transition():
            validation_connection = self._open_connection(sqlite_timeout=5.0)
            validation_connection.close()

    def acquire_read(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy:
        """
        Acquire a shared read lock.
        If this instance already holds a read lock, the lock level is incremented (reentrant). Attempting to acquire a
        read lock while holding a write lock raises :class:`RuntimeError` (downgrade not allowed).
        :param timeout: seconds to wait; ``None`` uses the instance setting; ``-1`` waits up to SQLite's
            busy-timeout cap of about 23 days
        :param blocking: if ``False``, raise :class:`~filelock.Timeout` on contention;
            ``None`` uses the instance setting
        :returns: a proxy that can be used as a context manager to release the lock
        :raises RuntimeError: if a write lock is already held on this instance
        :raises Timeout: if the lock cannot be acquired within *timeout* seconds
        :raises TypeError: if *timeout* is a :class:`bool` or not a real number
        :raises ValueError: if a blocking call gets a ``nan`` *timeout* or a negative one other than ``-1``
        """
        return self._acquire(
            "read",
            self.timeout if timeout is None else timeout,
            blocking=self.blocking if blocking is None else blocking,
        )

    def acquire_write(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy:
        """
        Acquire an exclusive write lock.
        If this instance already holds a write lock from the same thread, the lock level is incremented (reentrant).
        Attempting to acquire a write lock while holding a read lock raises :class:`RuntimeError` (upgrade not allowed).
        Write locks are pinned to the acquiring thread: a different thread trying to re-enter also raises
        :class:`RuntimeError`.
        :param timeout: seconds to wait; ``None`` uses the instance setting; ``-1`` waits up to SQLite's
            busy-timeout cap of about 23 days
        :param blocking: if ``False``, raise :class:`~filelock.Timeout` on contention;
            ``None`` uses the instance setting
        :returns: a proxy that can be used as a context manager to release the lock
        :raises RuntimeError: if a read lock is already held, or a write lock is held by a different thread
        :raises Timeout: if the lock cannot be acquired within *timeout* seconds
        :raises TypeError: if *timeout* is a :class:`bool` or not a real number
        :raises ValueError: if a blocking call gets a ``nan`` *timeout* or a negative one other than ``-1``
        """
        return self._acquire(
            "write",
            self.timeout if timeout is None else timeout,
            blocking=self.blocking if blocking is None else blocking,
        )

    @contextmanager
    def read_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]:
        """
        Context manager that acquires and releases a shared read lock.
        Falls back to instance defaults for *timeout* and *blocking* when ``None``.
        :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
        :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
        """
        self.acquire_read(timeout, blocking=blocking)
        try:
            yield
        finally:
            self.release()

    @contextmanager
    def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]:
        """
        Context manager that acquires and releases an exclusive write lock.
        Falls back to instance defaults for *timeout* and *blocking* when ``None``.
        :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
        :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
        """
        self.acquire_write(timeout, blocking=blocking)
        try:
            yield
        finally:
            self.release()

    def release(self, *, force: bool = False) -> None:
        """
        Release one level of the current lock.
        When the lock level reaches zero the underlying SQLite transaction is rolled back, releasing the database lock.
        :param force: if ``True``, release the lock completely regardless of the current lock level
        :raises RuntimeError: if no lock is currently held and *force* is ``False``, or if the calling thread holds
            no level of the current lock and *force* is ``False``
        """
        with _fork_transition():
            _ensure_current_process()
            if self._inherited:  # pragma: needs fork
                return
            self._raise_if_acquiring("release")
            self._release(force=force, close=False)

    def close(self) -> None:
        """
        Release the lock (if held) and close the underlying SQLite connection.
        After calling this method, the lock instance is no longer usable. Unlike :meth:`release`, it ends every
        thread's hold on this instance, whichever thread calls it.
        """
        with _fork_transition():
            _ensure_current_process()
            if self._inherited:  # pragma: needs fork
                return
            self._raise_if_acquiring("close")
            self._release(force=True, close=True)

    def _acquire_level_of(
        self,
        holder: int,
        mode: Literal["read", "write"],
        timeout: float,
        *,
        blocking: bool,
        cancel_check: Callable[[], bool],
    ) -> AcquireReturnProxy:
        # Records the level under *holder* instead of the calling thread: an async wrapper's executor thread does not
        # identify the tasks it holds for. Gives up with Timeout once *cancel_check* is true.
        return self._acquire(mode, timeout, blocking=blocking, holder=holder, cancel_check=cancel_check)

    def _release_level_of(self, thread_id: int) -> None:
        with _fork_transition():
            _ensure_current_process()
            self._raise_if_acquiring("release")
            self._release(force=False, close=False, thread_id=thread_id)

    def _holds_level_of(self, thread_id: int) -> bool:
        # Read without the internal lock. The async wrapper asks from its event loop, which must not wait out a release
        # on another thread. A stale answer is what a caller asking a moment earlier would have had. A level whose
        # transaction ended is one an interrupted release left behind; it counts as gone, as _holds_level() treats it.
        if thread_id != self._write_thread_id and thread_id not in self._read_depths:
            return False
        try:
            return not self._transaction_ended()
        except sqlite3.ProgrammingError:
            # The connection closed between the read of _con and the in_transaction call, so the transaction is over.
            return False

    def _release(self, *, force: bool, close: bool, thread_id: int | None = None) -> None:
        thread_id = threading.get_ident() if thread_id is None else thread_id
        with self._transaction_lock, self._internal_lock:
            if not force:
                self._raise_if_not_holder(thread_id)
            if self._lock_level == 0:
                if force and self._con is None:
                    if close:
                        self._closed = True
                    return
                if not force:
                    msg = f"Cannot release a lock on {self.lock_file} (lock id: {id(self)}) that is not held"
                    raise RuntimeError(msg)
            if not force and self._lock_level > 1:
                self._leave_one_level(thread_id)
                return
            try:
                self._finish_connection()
            except BaseException:
                # Once the rollback ran, the level is gone whatever stopped the cleanup, a signal handler's exception
                # included. Kept, it would let the next acquire nest into a transaction that no longer exists.
                if self._transaction_ended():
                    self._clear_lock_state()
                raise
            self._clear_lock_state()
            if close:
                self._closed = True

    def _transaction_ended(self) -> bool:
        if self._connection_transaction_released or (connection := self._con) is None:
            return True
        # An interrupt can land between the rollback and the flag that records it.
        return not connection.in_transaction

    def _raise_if_not_holder(self, thread_id: int) -> None:
        # The connection runs with check_same_thread=False, so a release from another thread would end the owner's
        # transaction under it and let a second writer in.
        if self._current_mode == "write" and thread_id != self._write_thread_id:
            msg = (
                f"Cannot release write lock on {self.lock_file} (lock id: {id(self)}) "
                f"from thread {thread_id} while it is held by thread {self._write_thread_id}"
            )
            raise RuntimeError(msg)
        # Readers share that one transaction too, so a thread that holds no read level would end another's hold.
        if self._current_mode == "read" and thread_id not in self._read_depths:
            msg = (
                f"Cannot release read lock on {self.lock_file} (lock id: {id(self)}) "
                f"from thread {thread_id}, which does not hold it"
            )
            raise RuntimeError(msg)

    def _leave_one_level(self, thread_id: int) -> None:
        self._lock_level -= 1
        if self._current_mode == "read" and (depth := self._read_depths.pop(thread_id) - 1):
            self._read_depths[thread_id] = depth

    def _clear_lock_state(self) -> None:
        self._lock_level = 0
        self._current_mode = None
        self._write_thread_id = None
        self._read_depths = {}

    @property
    def _con(self) -> _ForkSafeConnection | None:
        return self._owned_connection.connection

    @_con.setter
    def _con(self, connection: _ForkSafeConnection | None) -> None:
        self._owned_connection.connection = connection

    def _reset_after_fork_in_child(self) -> None:  # pragma: forked child
        if self._con is not None:
            _FORKED_DATABASES.poison_after_fork(self._canonical_path, self._connection_identity)
        self._con = None
        self._connection_transaction_released = True
        self._connection_identity = None
        self._transaction_lock = threading.Lock()
        self._internal_lock = threading.Lock()
        self._clear_lock_state()
        self._acquisition_thread_ids = set()
        self._fork_invalidated = True

    @property
    def _inherited(self) -> bool:
        return self._fork_invalidated or _GETPID() != self._creator_pid

    def _raise_if_unusable(self) -> None:
        _ensure_current_process()
        if self._inherited:  # pragma: needs fork
            msg = f"ReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance"
            raise RuntimeError(msg)
        if self._closed:
            msg = "Cannot operate on a closed database."
            raise sqlite3.ProgrammingError(msg)

    def _acquire(
        self,
        mode: Literal["read", "write"],
        timeout: float,
        *,
        blocking: bool,
        holder: int | None = None,
        cancel_check: Callable[[], bool] | None = None,
    ) -> AcquireReturnProxy:
        timeout = _resolve_read_write_timeout(timeout, blocking=blocking)
        start_time: Final = time.monotonic()
        remaining = timeout  # -1 waits up to SQLite's busy-timeout cap
        while True:
            # A fork waits for the open transitions and admits no new one meanwhile. A wait for the database inside one
            # stalls the forking thread and, behind it, each lock call in the process. With no other thread to fork,
            # the wait runs in one piece, which keeps SQLite's writer preference. threading does not count a
            # thread started through _thread, so a fork from one of those still waits.
            final = (
                not blocking
                or 0 <= remaining <= _MAX_TRANSITION_WAIT
                or not hasattr(os, "register_at_fork")
                or threading.active_count() == 1
            )
            try:
                return self._attempt_acquire(
                    mode,
                    remaining if final else _MAX_TRANSITION_WAIT,
                    blocking=blocking,
                    holder=holder,
                    cancel_check=cancel_check,
                )
            except Timeout:
                if final or (cancel_check is not None and cancel_check()):
                    raise
            # Only a runtime with fork handlers waits in more than one attempt.
            if timeout != -1:  # pragma: needs register-at-fork
                remaining = max(timeout - (time.monotonic() - start_time), 0)

    def _attempt_acquire(
        self,
        mode: Literal["read", "write"],
        timeout: float,
        *,
        blocking: bool,
        holder: int | None = None,
        cancel_check: Callable[[], bool] | None = None,
    ) -> AcquireReturnProxy:
        with _fork_transition():
            self._raise_if_unusable()
            operation_pid = _GETPID()
            thread_id = threading.get_ident()
            holder = thread_id if holder is None else holder
            with self._internal_lock:
                if self._holds_level():
                    return self._validate_reentrant(mode, holder)
                if thread_id in self._acquisition_thread_ids:  # pragma: no cover - exercised in an audit callback
                    msg = f"Cannot acquire ReadWriteLock on {self.lock_file} while acquisition is active in this thread"
                    raise RuntimeError(msg)
                self._acquisition_thread_ids.add(thread_id)
            try:
                start_time = time.perf_counter()
                self._acquire_transaction_lock(blocking=blocking, timeout=timeout, cancel_check=cancel_check)
                try:
                    self._raise_if_unusable()
                    return self._do_acquire_inner(
                        mode,
                        timeout,
                        holder,
                        blocking=blocking,
                        operation=_Operation(operation_pid, start_time, cancel_check),
                    )
                finally:
                    self._transaction_lock.release()
            finally:
                with self._internal_lock:
                    self._acquisition_thread_ids.discard(thread_id)

    def _do_acquire_inner(
        self,
        mode: Literal["read", "write"],
        timeout: float,
        holder: int,
        *,
        blocking: bool,
        operation: _Operation,
    ) -> AcquireReturnProxy:
        operation_pid, start_time, _ = operation
        # Double-check: another thread may have acquired the lock while we waited on _transaction_lock.
        with self._internal_lock:
            if self._holds_level():
                return self._validate_reentrant(mode, holder)
        if self._con is not None:
            self._finish_connection()
        try:
            self._open_for_acquisition(
                timeout,
                blocking=blocking,
                operation_pid=operation_pid,
                start_time=start_time,
            )
            self._configure_and_begin(mode, timeout, blocking=blocking, operation=operation)
            self._raise_if_process_changed(operation_pid)
        except BaseException as error:
            acquisition_error: BaseException
            if isinstance(error, sqlite3.OperationalError) and "database is locked" in str(error):
                acquisition_error = Timeout(self.lock_file)
            else:
                acquisition_error = error
            try:
                self._finish_connection()
            except sqlite3.Error as cleanup_error:
                _raise_chained_errors(acquisition_error, cleanup_error)
            if acquisition_error is not error:
                raise acquisition_error from None
            raise
        with self._internal_lock:
            self._raise_if_process_changed(operation_pid)
            self._current_mode = mode
            self._lock_level = 1
            if mode == "write":
                self._write_thread_id = holder
            else:
                self._read_depths = {holder: 1}
        return AcquireReturnProxy(lock=self)

    def _open_for_acquisition(self, timeout: float, *, blocking: bool, operation_pid: int, start_time: float) -> None:
        with _sqlite_transition():
            sqlite_timeout = (
                timeout_for_sqlite(
                    timeout,
                    blocking=blocking,
                    already_waited=time.perf_counter() - start_time,
                )
                / 1000
            )
            connection = self._open_connection(sqlite_timeout=sqlite_timeout)
            self._con, self._connection_transaction_released, self._connection_identity = (
                connection,
                False,
                _FORKED_DATABASES.identity(self._canonical_path),
            )
            self._raise_if_process_changed(operation_pid)

    def _configure_and_begin(
        self,
        mode: Literal["read", "write"],
        timeout: float,
        *,
        blocking: bool,
        operation: _Operation,
    ) -> None:
        with _sqlite_transition():
            connection = cast("_ForkSafeConnection", self._con)
            # SQLite opened the file the name held at connect time, and its first statement would replay a journal
            # beside that file, so the name is checked before any statement runs.
            self._raise_if_database_replaced(connection)
            with _bounded_schema(connection, self.lock_file):
                self._raise_if_process_changed(operation.pid)
                # Use legacy journal mode (not WAL) because WAL does not block readers while a concurrent EXCLUSIVE
                # write transaction is active, which makes read-write locking impossible without modifying table data.
                # MEMORY is safe here since no writes happen, so a crash cannot corrupt the DB.
                # See https://sqlite.org/lang_transaction.html#deferred_immediate_and_exclusive_transactions
                self._execute_waiting(
                    connection, "PRAGMA journal_mode=MEMORY;", timeout, blocking=blocking, operation=operation
                )
                self._raise_if_process_changed(operation.pid)
                statements = "BEGIN EXCLUSIVE TRANSACTION;" if mode == "write" else "BEGIN TRANSACTION;"
                if mode == "read":
                    # SQLite takes the SHARED lock only when a statement reads; BEGIN alone stays deferred.
                    # https://www.sqlite.org/lockingv3.html#transaction_control
                    # sqlite_schema is an alias added in SQLite 3.33.0; sqlite_master works on every version.
                    statements += " SELECT name FROM sqlite_master LIMIT 1;"
                self._execute_waiting(connection, statements, timeout, blocking=blocking, operation=operation)
            self._raise_if_database_replaced(connection)

    @staticmethod
    def _execute_waiting(
        connection: _ForkSafeConnection, statements: str, timeout: float, *, blocking: bool, operation: _Operation
    ) -> None:
        # SQLite's busy wait cannot be interrupted. An acquire its caller may abandon waits in bounded slices and asks
        # between them whether to stop.
        slice_ms = int(_CANCEL_POLL_INTERVAL * 1000)
        while True:
            busy_ms = timeout_for_sqlite(
                timeout, blocking=blocking, already_waited=time.perf_counter() - operation.start_time
            )
            cancel_check = operation.cancel_check if busy_ms > slice_ms else None
            try:
                connection.executescript(
                    f"PRAGMA busy_timeout={busy_ms if cancel_check is None else slice_ms}; {statements}"
                ).close()
            except sqlite3.OperationalError as error:
                if cancel_check is None or "database is locked" not in str(error) or cancel_check():
                    raise
                # A read's BEGIN succeeds before its SELECT waits, so the retry has to start a fresh transaction.
                if connection.in_transaction:
                    connection.rollback()
            else:
                return

    def _raise_if_database_replaced(self, connection: _ForkSafeConnection) -> None:
        # Windows pins the name: SQLite opens the database without delete sharing, so it cannot be unlinked or renamed.
        if (identity := connection.database_identity) is None:  # pragma: win32 cover
            return
        try:  # pragma: win32 no cover
            current = self._canonical_path.stat(follow_symlinks=False)
        except FileNotFoundError:  # pragma: win32 no cover
            current = None
        # The lock lives on the inode, so one taken after the path moved on excludes no later opener.
        if current is None or (current.st_dev, current.st_ino) != identity:  # pragma: win32 no cover
            msg = f"lock database {self.lock_file!r} was unlinked or replaced while acquiring; leave it in place"
            raise OSError(msg)

    def _open_connection(self, *, sqlite_timeout: float) -> _ForkSafeConnection:
        with _sqlite_transition():
            creator_pid = _GETPID()
            functions = _CONNECTION_ESCROW.functions()
            if _GETPID() != creator_pid:  # pragma: forked child
                msg = "SQLite connection construction cannot continue after fork"
                raise RuntimeError(msg)
            try:
                connection = _connect(
                    os.fspath(self._canonical_path),
                    factory=_ForkSafeConnection,
                    timeout=sqlite_timeout,
                )
            except BlockingIOError:
                raise Timeout(self.lock_file) from None
            if functions is not None:  # pragma: <3.12 cover  # pragma: needs fork
                connection.acquire_escrow(functions)
            if _GETPID() != creator_pid:  # pragma: forked child
                _FORKED_DATABASES.poison_after_fork(
                    self._canonical_path,
                    _FORKED_DATABASES.identity(self._canonical_path),
                )
                msg = "SQLite connection construction cannot continue after fork"
                raise RuntimeError(msg)
            return connection

    def _finish_connection(self) -> None:
        with _sqlite_transition():
            if (connection := self._con) is None:
                return
            rollback_error: sqlite3.Error | None = None
            if not self._connection_transaction_released:
                if connection.in_transaction:
                    try:
                        connection.rollback()
                    except sqlite3.Error as error:
                        if connection.in_transaction:
                            raise
                        self._connection_transaction_released = True
                        rollback_error = error
                    else:
                        self._connection_transaction_released = True
                else:
                    self._connection_transaction_released = True
            try:
                connection.close()
            except sqlite3.Error as close_error:
                if rollback_error is not None:
                    _raise_chained_errors(rollback_error, close_error)
                raise
            self._con = None
            self._connection_transaction_released = True
            self._connection_identity = None
            if rollback_error is not None:
                raise rollback_error

    def _holds_level(self) -> bool:
        # A release interrupted after its rollback, but before it recorded that, leaves a level with no transaction.
        if self._lock_level > 0 and self._transaction_ended():
            self._clear_lock_state()
        return self._lock_level > 0

    def _validate_reentrant(self, mode: Literal["read", "write"], cur: int) -> AcquireReturnProxy:
        if self._current_mode != mode:
            opposite = "write" if mode == "read" else "read"
            if cur == self._write_thread_id or cur in self._read_depths:
                direction = "downgrade" if mode == "read" else "upgrade"
                reason = f"already holding a {opposite} lock ({direction} not allowed)"
            else:
                reason = (
                    f"another thread holds a {opposite} lock on this instance, which coordinates processes rather "
                    "than threads; give each thread its own instance with is_singleton=False"
                )
            msg = f"Cannot acquire {mode} lock on {self.lock_file} (lock id: {id(self)}): {reason}"
            raise RuntimeError(msg)
        if mode == "write" and cur != self._write_thread_id:
            msg = (
                f"Cannot acquire write lock on {self.lock_file} (lock id: {id(self)}) "
                f"from thread {cur} while it is held by thread {self._write_thread_id}"
            )
            raise RuntimeError(msg)
        if mode == "read":
            self._read_depths[cur] = self._read_depths.get(cur, 0) + 1
        self._lock_level += 1
        return AcquireReturnProxy(lock=self)

    def _acquire_transaction_lock(
        self, *, blocking: bool, timeout: float, cancel_check: Callable[[], bool] | None
    ) -> None:
        if not _acquire_thread_lock(
            self._transaction_lock,
            blocking=blocking,
            timeout=timeout,
            poll_interval=_CANCEL_POLL_INTERVAL,
            cancel_check=cancel_check,
        ):
            raise Timeout(self.lock_file) from None

    def _raise_if_acquiring(self, operation: Literal["acquire", "close", "release"]) -> None:
        with self._internal_lock:
            active_in_current_thread = threading.get_ident() in self._acquisition_thread_ids
        if active_in_current_thread:  # pragma: no cover - exercised in an audit callback
            msg = f"Cannot {operation} ReadWriteLock on {self.lock_file} while acquisition is active in this thread"
            raise RuntimeError(msg)

    def _raise_if_process_changed(self, operation_pid: int) -> None:
        if _GETPID() != operation_pid or self._inherited:  # pragma: forked child
            msg = f"ReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance"
            raise RuntimeError(msg)


def _connect(database: str, *, factory: type[_ForkSafeConnection], timeout: float) -> _ForkSafeConnection:
    _FORKED_DATABASES.note_sqlite_use()
    # A symlink at the path would make SQLite open, lock, or create its target, so connect through a descriptor that
    # refuses one.
    if sys.platform == "win32":  # pragma: win32 cover
        fd: Final = _open_lock_database(database, oplock_wait=timeout)
        link: _LinkDirectory | None = None
        # A Windows lock belongs to the handle that took it, so closing this one leaves SQLite's locks in place.
        try:
            _raise_if_wal(fd, database)
            target = database
            if not _directory_is_private(os.fspath(pathlib.Path(database).parent)):
                link = _link_beside_on_windows(database, _identity_of_fd(fd))
                target = os.fspath(link.database)
            try:
                connection = sqlite3.connect(
                    target, check_same_thread=False, factory=factory, cached_statements=0, timeout=timeout
                )
            except BaseException:
                if link is not None:
                    _remove_link_directory(link)
                raise
            connection.retain_link(link)
            return connection
        finally:
            os.close(fd)
    return _connect_through_descriptor(  # pragma: win32 no cover
        database, factory=factory, timeout=timeout, owner=os.geteuid()
    )


def _connect_through_descriptor(  # pragma: win32 no cover
    database: str, *, factory: type[_ForkSafeConnection], timeout: float, owner: int
) -> _ForkSafeConnection:
    # Listed before its lease is taken and cleared once a connection owns it, so any exception between, an interrupt
    # included, releases the lease instead of pinning the inode's probes.
    claimed: Final[list[_Probe]] = []
    link: _LinkDirectory | None = None
    try:
        if (probe := _PROBE_DESCRIPTORS.adopt(database, claimed)) is None:
            probe = _PROBE_DESCRIPTORS.retain(_open_lock_database(database), claimed)
        identity, fd, _ = probe
        # Releases a finalizer queued while the table was locked.
        _PROBE_DESCRIPTORS.drain()
        _raise_if_wal(fd, database)
        target = database
        # SQLite opens the name it is given, following symlinks, and uses the -journal, -wal and -shm names beside it,
        # so the database name is safe only where no process of another account can create names. A descriptor path
        # such as /proc/self/fd/<n> is no better: SQLite reads it back into the name the kernel reports.
        if not _names_are_private(pathlib.Path(database).parent, owner):  # pragma: needs posix-hard-link
            link = _link_beside(database, identity, owner)
            target = os.fspath(link.database)
        connection = sqlite3.connect(
            target, check_same_thread=False, factory=factory, cached_statements=0, timeout=timeout
        )
        connection.retain_until_close(link, probe)
        claimed.clear()
        return connection
    finally:
        for unowned in claimed:
            _PROBE_DESCRIPTORS.release(unowned)
        if claimed and link is not None:  # pragma: needs posix-hard-link
            # A connection that already took the link removes it too when it closes.
            _remove_link_directory(link)


def _names_are_private(directory: pathlib.Path, owner: int) -> bool:  # pragma: win32 no cover
    status: Final = os.lstat(directory)
    # The sticky bit still lets others create names, so a directory anyone else may write to never qualifies.
    return status.st_uid in {owner, 0} and not status.st_mode & (stat.S_IWGRP | stat.S_IWOTH)


class _LinkDirectory(NamedTuple):  # pragma: needs posix-hard-link
    path: pathlib.Path
    identity: _DatabaseIdentity
    # Made through a descriptor of path, which nobody else can list, so a directory another process swaps in under
    # path cannot hold this name and SQLite fails to open instead of following the swap.
    name: str

    @property
    def database(self) -> pathlib.Path:
        return self.path / self.name / _LINK_NAME


def _link_beside(  # pragma: needs posix-hard-link
    database: str, identity: _DatabaseIdentity, owner: int
) -> _LinkDirectory:
    # Beside the database because a link cannot cross filesystems, and TMPDIR is often on another one.
    parent: Final = pathlib.Path(database).parent
    _sweep_links_of_the_dead(parent, identity, owner)
    path: Final = pathlib.Path(tempfile.mkdtemp(prefix=_link_prefix(), dir=parent))
    fd: Final = os.open(path, _DIRECTORY_FLAGS)
    try:
        made: Final = os.fstat(fd)
        # Whoever can rename in the parent can put another directory under the new name before it opens.
        if made.st_uid != owner or made.st_mode & 0o077:
            msg = f"link directory {os.fspath(path)!r} was replaced while opening"
            raise OSError(msg)
        link: Final = _LinkDirectory(path, (made.st_dev, made.st_ino), secrets.token_hex(8))
        try:
            os.mkdir(link.name, 0o700, dir_fd=fd)
            os.link(database, link.database, follow_symlinks=False)
            linked: Final = link.database.stat(follow_symlinks=False)
        except BaseException:
            _remove_link_directory(link)
            raise
    finally:
        os.close(fd)
    if (linked.st_dev, linked.st_ino) != identity:
        _remove_link_directory(link)
        msg = f"lock database changed while opening: {database!r}"
        raise OSError(msg)
    return link


def _sweep_links_of_the_dead(
    parent: pathlib.Path, identity: _DatabaseIdentity, owner: int
) -> None:  # pragma: needs posix-hard-link
    # A process killed while it held a link leaves the directory, and an extra link to the database, behind for good.
    # Two threads sweeping at once only both try the same removals, which a lock would cost a fork-safety reset for.
    if identity in _SWEPT_DATABASES:
        return
    _SWEPT_DATABASES.add(identity)
    try:
        with os.scandir(parent) as entries:
            names: Final = [
                entry.name
                for entry in entries
                if entry.name.startswith(_LINK_PREFIX) and entry.is_dir(follow_symlinks=False)
            ]
    except OSError:
        return
    for name in names:
        # A peer sweeping alongside, or a directory this user cannot remove, must not fail the connect.
        with suppress(OSError):
            _remove_dead_links(parent / name, identity, owner)


def _remove_dead_links(
    path: pathlib.Path, identity: _DatabaseIdentity, owner: int
) -> None:  # pragma: needs posix-hard-link
    if sys.platform == "win32":  # pragma: win32 cover
        _remove_dead_links_on_windows(path, identity)
        return
    # Any directory can carry a dead holder's name after a rename in the parent, so only hard links to this database
    # go, and the directory only once that leaves it empty.
    fd: Final = os.open(path, _DIRECTORY_FLAGS)
    try:
        # Only a directory this account made carries a holder of its own; a name another account chose is never
        # parsed, so no PID is probed on its behalf.
        if os.fstat(fd).st_uid != owner or not _owner_is_gone(path.name):
            return
        with os.scandir(fd) as entries:
            names: Final = [entry.name for entry in entries]
        for entry in names:
            # Releases up to 4.2.0 linked the database right inside the directory.
            if entry == _LINK_NAME:
                _unlink_link_to(identity, fd)
                continue
            try:
                nested = os.open(entry, _DIRECTORY_FLAGS, dir_fd=fd)
            except OSError:
                continue
            try:
                _unlink_link_to(identity, nested)
            finally:
                os.close(nested)
            with suppress(OSError):
                os.rmdir(entry, dir_fd=fd)
    finally:
        os.close(fd)
    with suppress(OSError):
        path.rmdir()


def _unlink_link_to(identity: _DatabaseIdentity, dir_fd: int) -> None:  # pragma: needs posix-hard-link
    try:
        linked: Final = os.stat(_LINK_NAME, dir_fd=dir_fd, follow_symlinks=False)
    except OSError:
        return
    if (linked.st_dev, linked.st_ino) == identity:
        os.unlink(_LINK_NAME, dir_fd=dir_fd)


def _owner_is_gone(name: str) -> bool:  # pragma: needs posix-hard-link
    match name.removeprefix(_LINK_PREFIX).split("-"):
        case [pid, start, digest, _] if (
            pid.isascii() and pid.isdigit() and (not start or (start.isascii() and start.isdigit()))
        ):
            if not 1 <= (owner_pid := int(pid)) <= 2**31 - 1:
                return False
            host: Final = host_name()
            return digest == _host_digest(host) and owner_is_stale(owner_pid, host, int(start) if start else None)
        case _:
            return False


def _link_prefix() -> str:  # pragma: needs posix-hard-link
    # The owner rides in the name, which mkdtemp creates in one step, so no directory ever exists without it.
    start: Final = process_start_token(pid := os.getpid())
    return f"{_LINK_PREFIX}{pid}-{'' if start is None else start}-{_host_digest(host_name())}-"


def _host_digest(host: str) -> str:  # pragma: needs posix-hard-link
    # SHA-256 as host_name does: some runtimes, GraalPy's among them, have no BLAKE2 with a custom digest size.
    return hashlib.sha256(host.encode()).hexdigest()[:16]


def _remove_link_directory(link: _LinkDirectory) -> None:  # pragma: needs posix-hard-link
    if sys.platform == "win32":  # pragma: win32 cover
        _remove_link_directory_on_windows(link)
        return
    # A rename in the parent may have put any directory under this name, so entries go through a descriptor checked
    # to be the directory made, and the name goes only through rmdir, which refuses a non-empty directory.
    try:
        fd: Final = os.open(link.path, _DIRECTORY_FLAGS)
    except OSError:
        return
    try:
        opened: Final = os.fstat(fd)
        if (opened.st_dev, opened.st_ino) != link.identity:
            return
        for suffix in ("", "-journal", "-wal", "-shm"):
            with suppress(FileNotFoundError):
                os.unlink(f"{link.name}/{_LINK_NAME}{suffix}", dir_fd=fd)
        with suppress(FileNotFoundError):
            os.rmdir(link.name, dir_fd=fd)
    finally:
        os.close(fd)
    # The link is gone, so a failure here leaves one stray empty directory and is not worth failing a release for.
    with suppress(OSError):
        link.path.rmdir()


if sys.platform == "win32":  # pragma: win32 cover

    def _link_beside_on_windows(database: str, identity: _DatabaseIdentity) -> _LinkDirectory:
        parent: Final = pathlib.Path(database).parent
        # No uid exists here: a directory is a dead holder's only if its ACL admits nobody but the trusted accounts.
        _sweep_links_of_the_dead(parent, identity, 0)
        path: Final = parent / f"{_link_prefix()}{secrets.token_hex(8)}"
        _make_private_directory(os.fspath(path))
        link: _LinkDirectory | None = None
        try:
            # The handle keeps a rename in the parent from swapping the directory while the link is made inside it.
            with _pinned(os.fspath(path)) as directory_identity:
                link = _LinkDirectory(path, directory_identity, secrets.token_hex(8))
                _fill_link_directory(link, database, identity)
        except BaseException:
            if link is not None:  # pragma: no branch  # a pinned directory always gets its link
                _remove_link_directory(link)
            raise
        return link

    def _fill_link_directory(link: _LinkDirectory, database: str, identity: _DatabaseIdentity) -> None:
        if not _directory_is_private(os.fspath(link.path)):
            msg = f"link directory {os.fspath(link.path)!r} was replaced while opening"
            raise OSError(msg)
        _make_private_directory(os.fspath(link.path / link.name))
        os.link(database, link.database)
        # The open database handle pins its path against a rename, so only a file index that is not unique can differ.
        if _identity_at(os.fspath(link.database)) != identity:  # pragma: no cover
            msg = f"lock database changed while opening: {database!r}"
            raise OSError(msg)

    def _remove_dead_links_on_windows(path: pathlib.Path, identity: _DatabaseIdentity) -> None:
        with _pinned(os.fspath(path)):
            if not _directory_is_private(os.fspath(path)) or not _owner_is_gone(path.name):
                return
            with os.scandir(path) as entries:
                nested: Final = [pathlib.Path(entry.path) for entry in entries]
            for directory in nested:
                _unlink_link_to_on_windows(identity, directory / _LINK_NAME)
                with suppress(OSError):
                    directory.rmdir()
        with suppress(OSError):
            path.rmdir()

    def _unlink_link_to_on_windows(identity: _DatabaseIdentity, path: pathlib.Path) -> None:
        with suppress(OSError):
            if _identity_at(os.fspath(path)) == identity:
                path.unlink()

    def _remove_link_directory_on_windows(link: _LinkDirectory) -> None:
        # A removal never fails a release: SQLite or a scanner may still hold a name, and the next sweep retries.
        with suppress(OSError):
            with _pinned(os.fspath(link.path)) as directory_identity:
                if directory_identity != link.identity:
                    return
                for suffix in ("", "-journal", "-wal", "-shm"):
                    with suppress(FileNotFoundError):
                        pathlib.Path(f"{link.database}{suffix}").unlink()
                with suppress(FileNotFoundError):
                    pathlib.Path(link.database.parent).rmdir()
            link.path.rmdir()


def _open_lock_database(database: str, *, oplock_wait: float = 0) -> int:
    if sys.platform == "win32":  # pragma: win32 cover
        # The handle refuses a reparse point and omits delete sharing, so the name cannot be swapped while it is open.
        if (fd := _open_non_reparse_fd(database, 0o600, oplock_wait=oplock_wait)) is None:
            msg = f"lock database {database!r} is held without sharing or pending deletion"
            raise PermissionError(msg)
        return fd
    fd = os.open(database, _DB_OPEN_FLAGS, 0o600)  # pragma: win32 no cover
    if not stat.S_ISREG(os.fstat(fd).st_mode):  # pragma: win32 no cover
        os.close(fd)
        msg = f"refusing a non-regular lock database: {database!r}"
        raise OSError(msg)
    return fd  # pragma: win32 no cover


def _raise_if_wal(fd: int, database: str) -> None:
    # SQLite derives the -wal name from the path it connects through, a descriptor name on POSIX, and WAL never blocks
    # readers behind a writer, so the lock cannot work on such a database.
    # pread: an adopted probe is shared, so its file offset is not this connect's to move.
    header: Final = os.pread(fd, _HEADER_SIZE, 0) if hasattr(os, "pread") else os.read(fd, _HEADER_SIZE)
    if header.startswith(_SQLITE_MAGIC) and _WAL_FORMAT_VERSION in header[_FORMAT_VERSIONS]:
        msg = f"refusing a WAL-mode lock database: {database!r}; give ReadWriteLock a file of its own"
        raise ValueError(msg)


@contextmanager
def _bounded_schema(connection: sqlite3.Connection, database: str) -> Generator[None]:
    # The first statement loads the file's complete schema, which no timeout bounds, and a lock database has none.
    # Waiting for a lock runs no virtual machine steps, so a step budget bounds only that load.
    calls: Final = count()
    max_calls: Final = 10
    connection.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, 4096)
    connection.set_progress_handler(lambda: next(calls) >= max_calls, 1000)
    try:
        yield
    except sqlite3.DatabaseError as error:
        # An interrupted schema load reports a malformed schema rather than the interrupt.
        if isinstance(error, sqlite3.DataError) or next(calls) > max_calls:
            msg = (
                f"refusing a lock database whose schema is too large to load: {database!r}; "
                "give ReadWriteLock a file of its own"
            )
            raise ValueError(msg) from error
        raise
    finally:
        connection.set_progress_handler(None, 0)


@contextmanager
def _sqlite_transition() -> Generator[None]:
    _SQLITE_TRANSITION_CONTEXT.depth += 1
    try:
        yield
    finally:
        _SQLITE_TRANSITION_CONTEXT.depth -= 1


def _abort_forked_sqlite_transition() -> None:  # pragma: forked child
    if _SQLITE_TRANSITION_CONTEXT.depth:
        os._exit(_UNSAFE_FORK_EXIT_STATUS)  # inherited SQLite handles cannot be used or closed safely


def _track_sqlite_use(event: str, _args: Unused) -> None:
    if event == "sqlite3.connect":
        _FORKED_DATABASES.note_sqlite_use()


def timeout_for_sqlite(timeout: float, *, blocking: bool, already_waited: float) -> int:
    if blocking is False:
        return 0

    if timeout == -1:
        return _MAX_SQLITE_TIMEOUT_MS

    if timeout < 0:
        msg = "timeout must be a non-negative number or -1"
        raise ValueError(msg)

    timeout_ms = int((max(timeout - already_waited, 0) if timeout > 0 else timeout) * 1000)
    if timeout_ms > _MAX_SQLITE_TIMEOUT_MS or timeout_ms < 0:
        _LOGGER.warning("timeout %s is too large for SQLite, using %s ms instead", timeout, _MAX_SQLITE_TIMEOUT_MS)
        return _MAX_SQLITE_TIMEOUT_MS
    return timeout_ms


_register_fork_object(_CONNECTION_ESCROW)
_register_fork_object(_FORKED_DATABASES)
_register_fork_object(_PROBE_DESCRIPTORS)
_register_fork_class(ReadWriteLock)
if _IS_PYPY:
    sys.addaudithook(_track_sqlite_use)  # pragma: pypy cover
if hasattr(os, "register_at_fork"):  # pragma: needs fork
    os.register_at_fork(after_in_child=_abort_forked_sqlite_transition)

__all__ = [
    "ReadWriteLock",
    "timeout_for_sqlite",
]