venv/lib/python3.14/site-packages/filelock/_read_write.py

from __future__ import annotations

import logging
import os
import pathlib
import sqlite3
import sys
import threading
import time
from contextlib import contextmanager, suppress
from typing import TYPE_CHECKING, ClassVar, Final, Literal, TypeAlias, cast
from weakref import WeakValueDictionary

from ._api import (
    AcquireReturnProxy,
    _ensure_current_process,
    _fork_transition,
    _raise_chained_errors,
    _register_fork_class,
    _register_fork_object,
)
from ._error import Timeout

if TYPE_CHECKING:
    from collections.abc import Callable, Generator

    if sys.version_info >= (3, 11):
        from typing import Self
    else:
        from typing_extensions import Self

_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
_UNSAFE_FORK_EXIT_STATUS: Final[int] = 70


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()


class _ForkSafeConnection(sqlite3.Connection):
    _creator_pid: int
    _decrement_escrow: Callable[[sqlite3.Connection], None] | None

    def __new__(
        cls,
        *_args: _ConnectionParameter,
        **_kwargs: _ConnectionParameter,
    ) -> Self:
        connection = super().__new__(cls)
        connection._creator_pid = _GETPID()
        connection._decrement_escrow = 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)
                if (decrement := self._decrement_escrow) is not None:  # pragma: <3.12 cover  # pragma: needs fork
                    self._decrement_escrow = None
                    decrement(self)

    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):
            self.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()
        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

        normalized = pathlib.Path(lock_file).resolve()
        with cls._instances_lock:
            if normalized not in cls._instances:
                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
            else:
                instance = cls._instances[normalized]

            if instance.timeout != timeout 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.

    By default, ``is_singleton=True``: calling ``ReadWriteLock(path)`` with the same resolved path returns the same
    instance. The path is handed to :func:`sqlite3.connect` as given, so a ``.db`` extension is a convention rather
    than a requirement; the filesystem must be one the active SQLite VFS supports.

    :param lock_file: path to the SQLite database file used as the lock
    :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
    :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`` means block indefinitely
        :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(lock_file).resolve()
        _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._acquisition_thread_ids: set[int] = set()
        self._con: _ForkSafeConnection | None = None
        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 = -1, *, blocking: bool = True) -> 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: maximum wait time in seconds; ``-1`` means block indefinitely
        :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable

        :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

        """
        return self._acquire("read", timeout, blocking=blocking)

    def acquire_write(self, timeout: float = -1, *, blocking: bool = True) -> 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: maximum wait time in seconds; ``-1`` means block indefinitely
        :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable

        :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

        """
        return self._acquire("write", timeout, blocking=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

        """
        if timeout is None:
            timeout = self.timeout
        if blocking is None:
            blocking = self.blocking
        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

        """
        if timeout is None:
            timeout = self.timeout
        if blocking is None:
            blocking = self.blocking
        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``

        """
        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.

        """
        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 _release(self, *, force: bool, close: bool) -> None:
        with self._transaction_lock, self._internal_lock:
            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._lock_level -= 1
                return
            try:
                self._finish_connection()
            except sqlite3.Error:
                if self._connection_transaction_released:
                    self._clear_lock_state()
                raise
            self._clear_lock_state()
            if close:
                self._closed = True

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

    def __del__(self) -> None:
        if _GETPID() == getattr(self, "_creator_pid", None) and (connection := getattr(self, "_con", None)) is not None:
            with suppress(sqlite3.Error, RuntimeError):
                connection.close()

    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) -> AcquireReturnProxy:
        with _fork_transition():
            self._raise_if_unusable()
            operation_pid = _GETPID()
            thread_id = threading.get_ident()
            with self._internal_lock:
                if self._lock_level > 0:
                    return self._validate_reentrant(mode)
                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)
                try:
                    self._raise_if_unusable()
                    return self._do_acquire_inner(
                        mode,
                        timeout,
                        blocking=blocking,
                        operation_pid=operation_pid,
                        start_time=start_time,
                    )
                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,
        *,
        blocking: bool,
        operation_pid: int,
        start_time: float,
    ) -> AcquireReturnProxy:
        # Double-check: another thread may have acquired the lock while we waited on _transaction_lock.
        with self._internal_lock:
            if self._lock_level > 0:
                return self._validate_reentrant(mode)
        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_pid, start_time),
            )
            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 = threading.get_ident()
        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: tuple[int, float],
    ) -> None:
        with _sqlite_transition():
            operation_pid, start_time = operation
            connection = cast("_ForkSafeConnection", self._con)
            waited = time.perf_counter() - start_time
            timeout_ms = timeout_for_sqlite(timeout, blocking=blocking, already_waited=waited)
            self._raise_if_process_changed(operation_pid)
            connection.executescript(f"PRAGMA busy_timeout={timeout_ms}; PRAGMA journal_mode=MEMORY;").close()
            # 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
            #
            # Recompute the remaining timeout after the blocking journal_mode pragma.
            waited = time.perf_counter() - start_time
            recomputed = timeout_for_sqlite(timeout, blocking=blocking, already_waited=waited)
            self._raise_if_process_changed(operation_pid)
            statements = f"PRAGMA busy_timeout={recomputed}; " if recomputed != timeout_ms else ""
            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
                statements += " SELECT name FROM sqlite_schema LIMIT 1;"
            connection.executescript(statements).close()

    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)
            connection = _connect(
                os.fspath(self._canonical_path),
                factory=_ForkSafeConnection,
                timeout=sqlite_timeout,
            )
            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 _validate_reentrant(self, mode: Literal["read", "write"]) -> AcquireReturnProxy:
        if self._current_mode != mode:
            opposite = "write" if mode == "read" else "read"
            direction = "downgrade" if mode == "read" else "upgrade"
            msg = (
                f"Cannot acquire {mode} lock on {self.lock_file} (lock id: {id(self)}): "
                f"already holding a {opposite} lock ({direction} not allowed)"
            )
            raise RuntimeError(msg)
        if mode == "write" and (cur := threading.get_ident()) != 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)
        self._lock_level += 1
        return AcquireReturnProxy(lock=self)

    def _acquire_transaction_lock(self, *, blocking: bool, timeout: float) -> None:
        if not blocking:
            acquired = self._transaction_lock.acquire(blocking=False)
        elif timeout == -1:
            acquired = self._transaction_lock.acquire(blocking=True)
        else:
            acquired = self._transaction_lock.acquire(blocking=True, timeout=timeout)
        if not acquired:
            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()
    return sqlite3.connect(
        database,
        check_same_thread=False,
        factory=factory,
        cached_statements=0,
        timeout=timeout,
    )


@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: tuple[object, ...]) -> None:  # audit payloads are heterogeneous
    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_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",
]