🐛 fix(async): make cancellation atomic

docs/changelog/634.bugfix.rst

@@ -0,0 +1,6 @@
+Preserve parent lock ownership across ``fork()``. A child closes an inherited descriptor only while its identity still
+matches, so it neither unlocks nor unlinks the parent's lock. It clears inherited ownership state and singleton caches,
+then requires a new lock instance before acquiring. Fork-time construction replaces inherited cache mutexes and omits
+an instance whose construction crossed into the child from the cache. If ``fstat()`` cannot establish the identity,
+child cleanup skips the descriptor number because an earlier fork callback may have reused it. Descriptor tracking
+also covers third-party backends that fail after assigning a descriptor.

docs/changelog/640.bugfix.rst

@@ -0,0 +1,4 @@
+Keep executor-backed lock operations alive until they finish when a caller cancels. Serialize acquisitions on the same
+async lock so cancellation rollback cannot release a later caller's hold. Roll back acquisitions that finish after
+cancellation and surface release failures. SQLite read-write locks keep their state until rollback ends the transaction;
+callers can retry cleanup with ``release(force=True)`` or another acquisition.

pyproject.toml

@@ -193,6 +193,7 @@ run.concurrency = [
 ]
 run.parallel = true
 run.patch = [
+  "_exit",
   "subprocess",
 ]
 run.plugins = [

src/filelock/_async.py

@@ -0,0 +1,158 @@
+"""Separate caller cancellation from backend task and executor-future results."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import time
+from concurrent.futures import Future as ConcurrentFuture
+from dataclasses import dataclass
+from threading import Lock
+from typing import TYPE_CHECKING, Final, Generic, TypeVar, cast
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator, Awaitable, Callable
+
+_T = TypeVar("_T")
+
+
+class _AsyncTransitionUnavailableError(Exception):
+    pass
+
+
+@dataclass(frozen=True)
+class _BackendOutcome(Generic[_T]):
+    value: _T | None = None
+    error: BaseException | None = None
+
+
+class _AsyncTransitionGate:
+    def __init__(self) -> None:
+        self._tail_lock: Final[Lock] = Lock()
+        self._tail: ConcurrentFuture[None] | None = None
+
+    @contextlib.asynccontextmanager
+    async def hold(self) -> AsyncIterator[None]:
+        ticket: ConcurrentFuture[None] = ConcurrentFuture()
+        with self._tail_lock:
+            predecessor = self._tail
+            self._tail = ticket
+        if predecessor is not None:
+            try:
+                await _wait_until_done(asyncio.wrap_future(predecessor))
+            except asyncio.CancelledError:
+                predecessor.add_done_callback(lambda _predecessor: self._leave(ticket))
+                raise
+        try:
+            yield
+        finally:
+            self._leave(ticket)
+
+    @contextlib.asynccontextmanager
+    async def hold_for_acquire(
+        self,
+        *,
+        blocking: bool,
+        cancel_check: Callable[[], bool] | None,
+        deadline: float | None,
+        poll_interval: float,
+    ) -> AsyncIterator[None]:
+        ticket: ConcurrentFuture[None] = ConcurrentFuture()
+        with self._tail_lock:
+            predecessor = self._tail
+            self._tail = ticket
+        if predecessor is not None and not predecessor.done():
+            try:
+                await self._wait_for_predecessor(
+                    predecessor,
+                    blocking=blocking,
+                    cancel_check=cancel_check,
+                    deadline=deadline,
+                    poll_interval=poll_interval,
+                )
+            except BaseException:
+                predecessor.add_done_callback(lambda _predecessor: self._leave(ticket))
+                raise
+        try:
+            yield
+        finally:
+            self._leave(ticket)
+
+    @staticmethod
+    async def _wait_for_predecessor(
+        predecessor: ConcurrentFuture[None],
+        *,
+        blocking: bool,
+        cancel_check: Callable[[], bool] | None,
+        deadline: float | None,
+        poll_interval: float,
+    ) -> None:
+        if not blocking:
+            raise _AsyncTransitionUnavailableError
+        waiter = asyncio.wrap_future(predecessor)
+        while not predecessor.done():
+            if cancel_check is not None and cancel_check():
+                raise _AsyncTransitionUnavailableError
+            if deadline is not None:
+                if (remaining := deadline - time.perf_counter()) <= 0:
+                    raise _AsyncTransitionUnavailableError
+                wait_interval = min(poll_interval, remaining) if cancel_check is not None else remaining
+            else:
+                wait_interval = poll_interval if cancel_check is not None else None
+            await asyncio.wait((waiter,), timeout=wait_interval)
+
+    def _leave(self, ticket: ConcurrentFuture[None]) -> None:
+        with self._tail_lock:
+            if self._tail is ticket:
+                self._tail = None
+        ticket.set_result(None)
+
+
+async def _drain_future(future: asyncio.Future[_BackendOutcome[_T]]) -> _T:
+    while not future.done():
+        with contextlib.suppress(asyncio.CancelledError):
+            await _wait_until_done(future)
+    return _future_result(future)
+
+
+async def _wait_until_done(future: asyncio.Future[_T]) -> None:
+    if not future.done():
+        await asyncio.wait((future,))
+
+
+def _future_result(future: asyncio.Future[_BackendOutcome[_T]]) -> _T:
+    outcome = future.result()
+    if (error := outcome.error) is None:
+        return cast("_T", outcome.value)
+    context = error.__context__
+    try:
+        raise error  # noqa: TRY301  # the handler restores context changed across the async boundary
+    except BaseException:
+        error.__context__ = context
+        raise
+
+
+def _capture_call(func: Callable[[], _T]) -> _BackendOutcome[_T]:
+    try:
+        return _BackendOutcome(value=func())
+    except BaseException as error:  # noqa: BLE001  # backend control-flow exceptions are operation results
+        return _BackendOutcome(error=error)
+
+
+async def _capture_awaitable(awaitable: Awaitable[_T]) -> _BackendOutcome[_T]:
+    try:
+        return _BackendOutcome(value=await awaitable)
+    except BaseException as error:  # noqa: BLE001  # backend cancellation must remain distinct from caller cancellation
+        return _BackendOutcome(error=error)
+
+
+__all__ = [
+    "_AsyncTransitionGate",
+    "_AsyncTransitionUnavailableError",
+    "_BackendOutcome",
+    "_capture_awaitable",
+    "_capture_call",
+    "_drain_future",
+    "_future_result",
+    "_wait_until_done",
+]

src/filelock/_async_read_write.py

@@ -6,15 +6,23 @@
 import functools
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import asynccontextmanager
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ParamSpec, TypeVar
 
+from ._api import _append_exception_context, _raise_chained_errors
+from ._async import _BackendOutcome, _capture_call, _drain_future, _future_result, _wait_until_done
 from ._read_write import ReadWriteLock
 
 if TYPE_CHECKING:
     import os
     from collections.abc import AsyncGenerator, Callable
     from concurrent import futures
     from types import TracebackType
+    from typing import NoReturn
+
+    from ._api import AcquireReturnProxy
+
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
 
 
 class AsyncReadWriteLock:
@@ -95,10 +103,14 @@ async def read_lock(self, timeout: float | None = None, *, blocking: bool | None
         if blocking is None:
             blocking = self._lock.blocking
         await self.acquire_read(timeout, blocking=blocking)
+        body_error: BaseException | None = None
         try:
             yield
+        except BaseException as error:
+            body_error = error
+            raise
         finally:
-            await self.release()
+            await self._release_in_context(body_error)
 
     @asynccontextmanager
     async def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> AsyncGenerator[None]:
@@ -116,10 +128,22 @@ async def write_lock(self, timeout: float | None = None, *, blocking: bool | Non
         if blocking is None:
             blocking = self._lock.blocking
         await self.acquire_write(timeout, blocking=blocking)
+        body_error: BaseException | None = None
         try:
             yield
+        except BaseException as error:
+            body_error = error
+            raise
         finally:
+            await self._release_in_context(body_error)
+
+    async def _release_in_context(self, body_error: BaseException | None) -> None:
+        try:
             await self.release()
+        except BaseException as release_error:
+            if body_error is not None:
+                _append_exception_context(release_error, body_error)
+            raise
 
     async def acquire_read(self, timeout: float = -1, *, blocking: bool = True) -> AsyncAcquireReadWriteReturnProxy:
         """
@@ -135,8 +159,11 @@ async def acquire_read(self, timeout: float = -1, *, blocking: bool = True) -> A
         :raises RuntimeError: if a write lock is already held on this instance
         :raises Timeout: if the lock cannot be acquired within *timeout* seconds
 
+        If rollback fails, peers may remain blocked. A later acquisition first retries that cleanup;
+        ``release(force=True)`` retries it without acquiring.
+
         """
-        await self._run(self._lock.acquire_read, timeout, blocking=blocking)
+        await self._run_acquire(functools.partial(self._lock.acquire_read, timeout, blocking=blocking))
         return AsyncAcquireReadWriteReturnProxy(lock=self)
 
     async def acquire_write(self, timeout: float = -1, *, blocking: bool = True) -> AsyncAcquireReadWriteReturnProxy:
@@ -153,8 +180,11 @@ async def acquire_write(self, timeout: float = -1, *, blocking: bool = True) ->
         :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
 
+        If rollback fails, peers may remain blocked. A later acquisition first retries that cleanup;
+        ``release(force=True)`` retries it without acquiring.
+
         """
-        await self._run(self._lock.acquire_write, timeout, blocking=blocking)
+        await self._run_acquire(functools.partial(self._lock.acquire_write, timeout, blocking=blocking))
         return AsyncAcquireReadWriteReturnProxy(lock=self)
 
     async def release(self, *, force: bool = False) -> None:
@@ -163,7 +193,8 @@ async def release(self, *, force: bool = False) -> None:
 
         See :meth:`ReadWriteLock.release` for full semantics.
 
-        :param force: if ``True``, release the lock completely regardless of the current lock level
+        :param force: if ``True``, release the lock completely regardless of the current lock level and retry cleanup
+            from a failed acquisition
 
         :raises RuntimeError: if no lock is currently held and *force* is ``False``
 
@@ -177,14 +208,71 @@ async def close(self) -> None:
         After calling this method, the lock instance is no longer usable.
 
         """
-        await self._run(self._lock.close)
+        close_future = self._submit(self._lock.close)
+        try:
+            await _wait_until_done(close_future)
+        except asyncio.CancelledError as cancellation:
+            try:
+                await _drain_future(close_future)
+            except BaseException as error:  # noqa: BLE001  # reported with the cancellation below
+                self._raise_cancelled_error(cancellation, error)
+            self._shutdown_owned_executor()
+            raise
+        _future_result(close_future)
+        self._shutdown_owned_executor()
+
+    async def _run_acquire(self, acquire: Callable[[], AcquireReturnProxy]) -> None:
+        acquire_future = self._submit(acquire)
+        try:
+            await _wait_until_done(acquire_future)
+        except asyncio.CancelledError as cancellation:
+            try:
+                await _drain_future(acquire_future)
+            except asyncio.CancelledError as acquire_error:
+                self._raise_cancelled_error(cancellation, acquire_error)
+            except BaseException as error:  # noqa: BLE001  # reported with the cancellation below
+                self._raise_cancelled_error(cancellation, error)
+            try:
+                await _drain_future(self._submit(self._lock.release))
+            except BaseException as error:  # noqa: BLE001  # reported with the cancellation below
+                self._raise_cancelled_error(cancellation, error)
+            raise
+        _future_result(acquire_future)
+
+    async def _run(self, func: Callable[_P, _R], *args: _P.args, **kwargs: _P.kwargs) -> _R:
+        future = self._submit(func, *args, **kwargs)
+        try:
+            await _wait_until_done(future)
+        except asyncio.CancelledError as cancellation:
+            try:
+                await _drain_future(future)
+            except BaseException as error:  # noqa: BLE001  # reported with the cancellation below
+                self._raise_cancelled_error(cancellation, error)
+            raise
+        return _future_result(future)
+
+    def _submit(
+        self, func: Callable[_P, _R], *args: _P.args, **kwargs: _P.kwargs
+    ) -> asyncio.Future[_BackendOutcome[_R]]:
+        return (self._loop or asyncio.get_running_loop()).run_in_executor(
+            self._executor,
+            _capture_call,
+            functools.partial(func, *args, **kwargs),
+        )
+
+    @staticmethod
+    def _raise_cancelled_error(cancellation: asyncio.CancelledError, error: BaseException) -> NoReturn:
+        if (context := error.__context__) is not None and context is not cancellation:
+            if (cancellation_context := cancellation.__context__) is not None:
+                _append_exception_context(context, cancellation_context)
+            cancellation.__context__ = context
+        error.__context__ = cancellation
+        _raise_chained_errors(error)
+
+    def _shutdown_owned_executor(self) -> None:
         if self._owns_executor:
             self._executor.shutdown(wait=False)
 
-    async def _run(self, func: Callable[..., object], *args: object, **kwargs: object) -> object:
-        loop = self._loop or asyncio.get_running_loop()
-        return await loop.run_in_executor(self._executor, functools.partial(func, *args, **kwargs))
-
     def __del__(self) -> None:
         # Safety net when close() was never called: shut down the executor we own so its worker thread does not
         # outlive the lock. shutdown(wait=False) never blocks.