diff --git a/docs/changelog/844.bugfix.rst b/docs/changelog/844.bugfix.rst
new file mode 100644
index 00000000..e3fed168
--- /dev/null
+++ b/docs/changelog/844.bugfix.rst
@@ -0,0 +1,4 @@
+Canceling an :class:`~filelock.AsyncReadWriteLock` or :class:`~filelock.AsyncSoftReadWriteLock` acquire that waits for
+another process now stops the wait and raises the cancellation. It used to wait until that process released, or raise
+:class:`~filelock.Timeout`. A ``thread_local=True`` :class:`~filelock.AsyncFileLock` that threads with their own event
+loops share no longer deadlocks when one thread waits for the lock while another holds it.
diff --git a/src/filelock/_api.py b/src/filelock/_api.py
index ddc37774..d0b77584 100644
--- a/src/filelock/_api.py
+++ b/src/filelock/_api.py
@@ -59,6 +59,7 @@
 
 if TYPE_CHECKING:
     from collections.abc import Generator, Mapping
+    from threading import Lock
     from types import FrameType, TracebackType
     from typing import Protocol
 
@@ -1719,6 +1720,24 @@ def _resolve_read_write_timeout(timeout: float, *, blocking: bool) -> float:
     return timeout
 
 
+def _acquire_thread_lock(
+    lock: Lock, *, blocking: bool, timeout: float, poll_interval: float, cancel_check: Callable[[], bool] | None
+) -> bool:
+    """Take a read-write lock's in-process lock; with *cancel_check*, ask it every *poll_interval* while waiting."""
+    if not blocking:
+        return lock.acquire(blocking=False)
+    if cancel_check is None:
+        # Lock.acquire raises OverflowError past TIMEOUT_MAX, which is 49.7 days on Windows.
+        return lock.acquire() if timeout == -1 else lock.acquire(timeout=min(timeout, TIMEOUT_MAX))
+    deadline = None if timeout == -1 else time.perf_counter() + timeout
+    while not lock.acquire(
+        timeout=poll_interval if deadline is None else min(poll_interval, max(deadline - time.perf_counter(), 0))
+    ):
+        if cancel_check() or (deadline is not None and time.perf_counter() >= deadline):
+            return False
+    return True
+
+
 def _resolve_timeout(timeout: float, *, blocking: bool) -> float:
     """Validate an acquire ``timeout``; every negative value waits without a limit."""
     if math.isnan(timeout := _seconds("timeout", timeout)) and blocking:
@@ -2232,6 +2251,7 @@ def _audit_fork_safety(  # pragma: no cover - CPython disables tracing while Pyt
     "FileLockMeta",
     "LockOptions",
     "ThreadLocalFileContext",
+    "_acquire_thread_lock",
     "_append_exception_context",
     "_canonical",
     "_check_timeout_max",
diff --git a/src/filelock/_async_read_write.py b/src/filelock/_async_read_write.py
index 555cb808..59204573 100644
--- a/src/filelock/_async_read_write.py
+++ b/src/filelock/_async_read_write.py
@@ -2,12 +2,13 @@
 
 from __future__ import annotations
 
-__lazy_modules__ = ["asyncio", "concurrent.futures", "filelock._async", "functools"]
+__lazy_modules__ = ["asyncio", "concurrent.futures", "filelock._async", "filelock._error", "functools", "threading"]
 
 import asyncio
 import functools
 import os
 import sqlite3
+import threading
 from concurrent.futures import ThreadPoolExecutor
 from concurrent.futures import thread as executor_threads
 from contextlib import asynccontextmanager
@@ -29,10 +30,10 @@
     _task_owners_for,
     _wait_until_done,
 )
+from filelock._error import Timeout
 from filelock._read_write import ReadWriteLock
 
 if TYPE_CHECKING:
-    import threading
     from collections.abc import AsyncGenerator, Callable, MutableMapping
     from concurrent import futures
     from types import TracebackType
@@ -311,13 +312,19 @@ async def close(self) -> None:
         if self._owns_executor:
             await asyncio.to_thread(functools.partial(self._executor.shutdown, wait=True))
 
-    async def _run_acquire(self, acquire: Callable[[], AcquireReturnProxy]) -> None:
-        acquire_future = self._submit(acquire)
+    async def _run_acquire(self, acquire: Callable[..., AcquireReturnProxy]) -> None:
+        abandoned = threading.Event()
+        acquire_future = self._submit(acquire, cancel_check=abandoned.is_set)
         try:
             await _wait_until_done(acquire_future)
         except asyncio.CancelledError as cancellation:
+            # Stops a backend wait on another process's hold, which would otherwise outlast the caller by as long.
+            abandoned.set()
             try:
                 await _drain_future(acquire_future)
+            except Timeout:
+                # Nothing was taken, so the caller's cancellation is the whole outcome.
+                raise cancellation from None
             except BaseException as error:  # ruff:ignore[blind-except]  # reported with the cancellation below
                 _raise_cancelled_error(cancellation, error)
             try:
diff --git a/src/filelock/_read_write.py b/src/filelock/_read_write.py
index f498d5db..5b01da2d 100644
--- a/src/filelock/_read_write.py
+++ b/src/filelock/_read_write.py
@@ -32,6 +32,7 @@
 
 from filelock._api import (
     AcquireReturnProxy,
+    _acquire_thread_lock,
     _canonical,
     _ensure_current_process,
     _fork_transition,
@@ -70,6 +71,8 @@
 
 # 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
@@ -262,6 +265,13 @@ def _reset_after_fork_in_child(self) -> None:  # pragma: forked child
         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
@@ -643,11 +653,17 @@ def close(self) -> None:
             self._release(force=True, close=True)
 
     def _acquire_level_of(
-        self, holder: int, mode: Literal["read", "write"], timeout: float, *, blocking: bool
+        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.
-        return self._acquire(mode, timeout, blocking=blocking, holder=holder)
+        # 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():
@@ -763,7 +779,13 @@ def _raise_if_unusable(self) -> None:
             raise sqlite3.ProgrammingError(msg)
 
     def _acquire(
-        self, mode: Literal["read", "write"], timeout: float, *, blocking: bool, holder: int | None = None
+        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()
@@ -781,17 +803,27 @@ def _acquire(
             )
             try:
                 return self._attempt_acquire(
-                    mode, remaining if final else _MAX_TRANSITION_WAIT, blocking=blocking, holder=holder
+                    mode,
+                    remaining if final else _MAX_TRANSITION_WAIT,
+                    blocking=blocking,
+                    holder=holder,
+                    cancel_check=cancel_check,
                 )
             except Timeout:
-                if final:
+                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
+        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()
@@ -807,11 +839,15 @@ def _attempt_acquire(
                 self._acquisition_thread_ids.add(thread_id)
             try:
                 start_time = time.perf_counter()
-                self._acquire_transaction_lock(blocking=blocking, timeout=timeout)
+                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_pid, start_time)
+                        mode,
+                        timeout,
+                        holder,
+                        blocking=blocking,
+                        operation=_Operation(operation_pid, start_time, cancel_check),
                     )
                 finally:
                     self._transaction_lock.release()
@@ -826,9 +862,9 @@ def _do_acquire_inner(
         holder: int,
         *,
         blocking: bool,
-        operation: tuple[int, float],
+        operation: _Operation,
     ) -> AcquireReturnProxy:
-        operation_pid, start_time = operation
+        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():
@@ -891,34 +927,53 @@ def _configure_and_begin(
         timeout: float,
         *,
         blocking: bool,
-        operation: tuple[int, float],
+        operation: _Operation,
     ) -> 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()
+            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
-            #
-            # 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;"
+            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;"
-            connection.executescript(statements).close()
+            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
@@ -1015,15 +1070,16 @@ def _validate_reentrant(self, mode: Literal["read", "write"], cur: int) -> Acqui
         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:
-            # Lock.acquire raises OverflowError past TIMEOUT_MAX, which is 49.7 days on Windows.
-            acquired = self._transaction_lock.acquire(blocking=True, timeout=min(timeout, threading.TIMEOUT_MAX))
-        if not acquired:
+    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:
diff --git a/src/filelock/_soft_rw/_async.py b/src/filelock/_soft_rw/_async.py
index 02675fc6..f751d6b9 100644
--- a/src/filelock/_soft_rw/_async.py
+++ b/src/filelock/_soft_rw/_async.py
@@ -2,11 +2,12 @@
 
 from __future__ import annotations
 
-__lazy_modules__ = ["asyncio", "filelock._async", "functools"]
+__lazy_modules__ = ["asyncio", "filelock._async", "filelock._error", "functools", "threading"]
 
 import asyncio
 import functools
 import os
+import threading
 from contextlib import asynccontextmanager
 from typing import TYPE_CHECKING, Final, Literal, ParamSpec, TypeVar
 
@@ -20,6 +21,7 @@
     _task_owners_for,
     _wait_until_done,
 )
+from filelock._error import Timeout
 from filelock._soft_rw._sync import SoftReadWriteLock, _lost_hold_error
 
 if TYPE_CHECKING:
@@ -315,17 +317,22 @@ def _leave_backend(self) -> None:
         # Only the tasks' level: a sync user of the same singleton keeps its own.
         self._lock._release_level_of(_TASKS_THREAD_ID)  # ruff:ignore[private-member-access]  # the wrapper owns this backend hold
 
-    async def _run_acquire(self, acquire: Callable[[], AcquireReturnProxy]) -> None:
-        # run_in_executor cannot recall work the pool already started, so canceling the caller does not stop the sync
-        # acquire: it still commits its claim, sets the hold, and starts the heartbeat, which keeps the record fresh
-        # forever so no peer on any host can evict it as stale. Wait the submitted call out and hand the claim back,
-        # the way AsyncReadWriteLock does.
-        acquire_future = self._submit(acquire)
+    async def _run_acquire(self, acquire: Callable[..., AcquireReturnProxy]) -> None:
+        # run_in_executor cannot recall work the pool already started. A sync acquire that won the lock before it saw
+        # the cancellation has published its claim and started the heartbeat, which keeps the record fresh so no peer
+        # evicts it. Wait the call out and hand the claim back, as AsyncReadWriteLock does.
+        abandoned = threading.Event()
+        acquire_future = self._submit(acquire, cancel_check=abandoned.is_set)
         try:
             await _wait_until_done(acquire_future)
         except asyncio.CancelledError as cancellation:
+            # Stops a backend wait on another process's hold, which would otherwise outlast the caller by as long.
+            abandoned.set()
             try:
                 await _drain_future(acquire_future)
+            except Timeout:
+                # Nothing was taken, so the caller's cancellation is the whole outcome.
+                raise cancellation from None
             except BaseException as error:  # ruff:ignore[blind-except]  # reported with the cancellation below
                 _raise_cancelled_error(cancellation, error)
             try:
diff --git a/src/filelock/_soft_rw/_sync.py b/src/filelock/_soft_rw/_sync.py
index 48d35831..caa74298 100644
--- a/src/filelock/_soft_rw/_sync.py
+++ b/src/filelock/_soft_rw/_sync.py
@@ -22,6 +22,7 @@
 
 from filelock._api import (
     AcquireReturnProxy,
+    _acquire_thread_lock,
     _canonical,
     _check_timeout_max,
     _ensure_current_process,
@@ -420,10 +421,12 @@ def release(self, *, force: bool = False) -> None:
             return
         self._release(threading.get_ident(), force=force)
 
-    def _acquire_level_of(self, holder: int, mode: Mode, timeout: float, *, blocking: bool) -> AcquireReturnProxy:
+    def _acquire_level_of(
+        self, holder: int, mode: Mode, 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.
-        return self._acquire(mode, timeout, blocking=blocking, holder=holder)
+        # 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, holder: int) -> None:
         self._release(holder, force=False)
@@ -487,6 +490,7 @@ def _acquire(
         *,
         blocking: bool | None,
         holder: int | None = None,
+        cancel_check: Callable[[], bool] | None = None,
     ) -> AcquireReturnProxy:
         if self._creator_pid != os.getpid():  # pragma: forked child
             msg = f"SoftReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance"
@@ -502,18 +506,17 @@ def _acquire(
             if self._hold is not None:
                 return self._validate_reentrant(mode, holder)
 
-        start = time.perf_counter()
-        if not blocking:
-            acquired = self._locks.transaction.acquire(blocking=False)
-        elif timeout == -1:
-            acquired = self._locks.transaction.acquire(blocking=True)
-        else:
-            # Lock.acquire raises OverflowError past TIMEOUT_MAX, which is 49.7 days on Windows.
-            acquired = self._locks.transaction.acquire(blocking=True, timeout=min(timeout, threading.TIMEOUT_MAX))
-        if not acquired:
+        deadline = None if timeout == -1 else time.perf_counter() + timeout
+        if not _acquire_thread_lock(
+            self._locks.transaction,
+            blocking=blocking,
+            timeout=timeout,
+            poll_interval=self.poll_interval,
+            cancel_check=cancel_check,
+        ):
             raise Timeout(self.lock_file) from None
         try:
-            return self._do_acquire_inner(mode, timeout, start, holder, blocking=blocking)
+            return self._do_acquire_inner(mode, deadline, holder, blocking=blocking, cancel_check=cancel_check)
         finally:
             self._locks.transaction.release()
 
@@ -523,16 +526,15 @@ def _closed_error(self) -> RuntimeError:
     def _do_acquire_inner(
         self,
         mode: Mode,
-        effective_timeout: float,
-        start: float,
+        deadline: float | None,
         holder: int,
         *,
         blocking: bool,
+        cancel_check: Callable[[], bool] | None,
     ) -> AcquireReturnProxy:
         with self._locks.internal:
             if self._hold is not None:
                 return self._validate_reentrant(mode, holder)
-        deadline = None if effective_timeout == -1 else start + effective_timeout
         participant = Participant(
             self._files,
             self._root,
@@ -544,7 +546,7 @@ def _do_acquire_inner(
         )
         participant.publish()
         try:
-            self._wait_for(participant.advance, deadline=deadline, blocking=blocking)
+            self._wait_for(participant.advance, deadline=deadline, blocking=blocking, cancel_check=cancel_check)
         except BaseException:
             # A contender that gives up must not stay named in the snapshot: a writer left there would block every
             # reader until a peer waits out the stale threshold. Its own token is all leave() touches, so it cannot
@@ -622,15 +624,17 @@ def _wait_for(
         *,
         deadline: float | None,
         blocking: bool,
+        cancel_check: Callable[[], bool] | None,
     ) -> None:
         while not predicate():
             # The flag is read without the lock here only to stop waiting early; publishing the hold re-checks it.
             if self._closed:
                 raise self._closed_error()
             now = time.perf_counter()
-            if not blocking:
+            if not blocking or (deadline is not None and now >= deadline):
                 raise Timeout(self.lock_file)
-            if deadline is not None and now >= deadline:
+            # An abandoned acquire gives up like a timed-out one, so the caller's rollback withdraws its claim.
+            if cancel_check is not None and cancel_check():
                 raise Timeout(self.lock_file)
             sleep_for = self.poll_interval
             if deadline is not None:
diff --git a/src/filelock/asyncio.py b/src/filelock/asyncio.py
index 92738894..8c370373 100644
--- a/src/filelock/asyncio.py
+++ b/src/filelock/asyncio.py
@@ -22,7 +22,7 @@
 import sys
 import time
 from concurrent import futures
-from dataclasses import dataclass, fields
+from dataclasses import dataclass, field, fields
 from inspect import iscoroutinefunction
 from threading import RLock
 from typing import TYPE_CHECKING, Final, NoReturn, TypeVar, cast
@@ -231,7 +231,6 @@ def __init__(  # ruff:ignore[too-many-arguments]  # public constructor: one para
         self._fallback_to_soft = fallback_to_soft
         self._preserve_lock_file = preserve_lock_file  # already validated by the metaclass
         self._on_acquired = on_acquired  # already validated by the metaclass
-        self._transition_gate: Final[_AsyncTransitionGate] = _AsyncTransitionGate()
         # BaseFileLock.__init__ never runs here, and the finalizer releases thread-local holds through these.
         self._transition_lock = RLock()
         self._thread_holds = {}
@@ -366,7 +365,7 @@ async def _acquire_with_admission(
         poll_interval: float,
         start_time: float,
     ) -> AsyncAcquireReturnProxy:
-        async with self._transition_gate.hold_for_acquire(
+        async with self._async_context.transition_gate.hold_for_acquire(
             blocking=blocking,
             cancel_check=cancel_check,
             deadline=None if timeout < 0 else start_time + timeout,
@@ -525,7 +524,7 @@ async def release(self, force: bool = False) -> None:  # ty: ignore[invalid-meth
         """
         if self._creator_pid != os.getpid() or not self.is_locked:
             return
-        async with self._transition_gate.hold():
+        async with self._async_context.transition_gate.hold():
             await self._release_serialized(force=force)
 
     async def _release_serialized(self, *, force: bool) -> None:
@@ -813,6 +812,11 @@ def _release_without_loop(self) -> None:
         self._release_thread_holds(exited_only=False)
 
 
+def _new_transition_gate() -> _AsyncTransitionGate:
+    # Called per context rather than passed as the factory, so the lazy import resolves on first use, not at import.
+    return _AsyncTransitionGate()
+
+
 @dataclass
 class AsyncFileLockContext(FileLockContext):
     """A dataclass which holds the context for a ``BaseAsyncFileLock`` object."""
@@ -826,6 +830,10 @@ class AsyncFileLockContext(FileLockContext):
     #: The loop
     loop: asyncio.AbstractEventLoop | None = None
 
+    #: Serializes the tasks that transition this context. An acquire keeps it while it polls, so a thread-local lock
+    #: needs one per thread. A thread waiting for the lock must not keep the thread that holds it from releasing.
+    transition_gate: _AsyncTransitionGate = field(default_factory=_new_transition_gate)
+
 
 class AsyncThreadLocalFileContext(AsyncFileLockContext, ThreadLocalFileContext):
     """A thread local version of the ``FileLockContext`` class."""
diff --git a/tests/soft_rw/test_soft_rw_async.py b/tests/soft_rw/test_soft_rw_async.py
index 34add7f9..1591a026 100644
--- a/tests/soft_rw/test_soft_rw_async.py
+++ b/tests/soft_rw/test_soft_rw_async.py
@@ -395,14 +395,15 @@ async def test_async_acquire_cancellation_hands_the_claim_back(
 @pytest.mark.asyncio
 @XFAIL_WITHOUT_COROUTINE_CANCELLATION
 async def test_async_acquire_cancellation_surfaces_a_failed_acquire(tmp_path: Path, mocker: MockerFixture) -> None:
-    acquire_error = Timeout(str(tmp_path / "foo.lock"))
+    # A Timeout here means the acquire gave up on the cancellation, so a real failure has to be another error.
+    acquire_error = OSError("acquire failed")
     gate = _Gate(mocker, "_acquire_level_of", fail_with=acquire_error)
     lock = _make(tmp_path)
     task = asyncio.create_task(lock.acquire_write(timeout=5))
     await gate.started.wait()
     task.cancel("cancel acquire")
     gate.resume()
-    with pytest.raises(Timeout) as info:
+    with pytest.raises(OSError, match="acquire failed") as info:
         await task
 
     assert info.value is acquire_error
diff --git a/tests/test_async_filelock_transition_admission.py b/tests/test_async_filelock_transition_admission.py
index 47cabdfb..f7ae1135 100644
--- a/tests/test_async_filelock_transition_admission.py
+++ b/tests/test_async_filelock_transition_admission.py
@@ -7,7 +7,7 @@
 
 import pytest
 
-from filelock import AsyncFileLock, AsyncSoftFileLock, Timeout
+from filelock import AsyncFileLock, AsyncSoftFileLock, BaseAsyncFileLock, Timeout
 from filelock._async import _AsyncTransitionGate
 from tests.async_filelock_cancellation_helpers import (
     assert_cancellation_message,
@@ -187,5 +187,33 @@ async def acquire() -> None:  # pragma: win32 no cover
     return asyncio.create_task(acquire()), started
 
 
+@pytest.mark.parametrize(
+    "lock_type", [pytest.param(AsyncFileLock, id="flock"), pytest.param(AsyncSoftFileLock, id="soft")]
+)
+@pytest.mark.asyncio
+async def test_thread_local_lock_releases_while_another_thread_waits_for_it(
+    tmp_path: Path, lock_type: type[BaseAsyncFileLock]
+) -> None:
+    lock = lock_type(tmp_path / "a", thread_local=True, run_in_executor=False)
+    await lock.acquire()
+    polling = threading.Event()
+
+    def note_poll() -> bool:
+        polling.set()
+        return False
+
+    async def acquire_and_release() -> None:
+        await lock.acquire(timeout=5, cancel_check=note_poll, poll_interval=0.01)
+        await lock.release()
+
+    waiter = asyncio.create_task(asyncio.to_thread(asyncio.run, acquire_and_release()))
+    assert await asyncio.to_thread(polling.wait, 5)
+
+    await asyncio.wait_for(lock.release(), 5)
+
+    await waiter
+    assert not lock.is_locked
+
+
 class _Interrupt(BaseException):
     """Stands in for an exception a signal handler raises, without pytest treating it as a KeyboardInterrupt."""
diff --git a/tests/test_async_read_write_cancel_wait.py b/tests/test_async_read_write_cancel_wait.py
new file mode 100644
index 00000000..64a737cc
--- /dev/null
+++ b/tests/test_async_read_write_cancel_wait.py
@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+import asyncio
+import threading
+import time
+from typing import TYPE_CHECKING, Final, Literal
+
+import pytest
+
+pytest.importorskip("sqlite3")
+
+from collections.abc import Callable
+
+from filelock import AsyncReadWriteLock, ReadWriteLock, Timeout
+from filelock._soft_rw import AsyncSoftReadWriteLock, SoftReadWriteLock
+from tests.capability_marks import XFAIL_WITHOUT_COROUTINE_CANCELLATION
+
+if TYPE_CHECKING:
+    from collections.abc import Generator
+    from pathlib import Path
+
+    from pytest_mock import MockerFixture
+
+# The other process's hold ends this long after the wait starts, so a wait that ignores its deadline ends then.
+_HOLD: Final[float] = 3
+# The caller's own deadline, far below _HOLD.
+_DEADLINE: Final[float] = 0.2
+
+pytestmark = XFAIL_WITHOUT_COROUTINE_CANCELLATION
+
+
+@pytest.fixture(autouse=True)
+def _clear_singleton_caches() -> Generator[None]:
+    caches: Final = (ReadWriteLock._instances, SoftReadWriteLock._instances)
+    for cache in caches:
+        cache.clear()
+    yield
+    # One loop for both caches, so a platform where only one kind of lock runs still closes through it.
+    for lock in filter(None, [ref() for cache in caches for ref in list(cache.valuerefs())]):
+        lock.close()
+    for cache in caches:
+        cache.clear()
+
+
+def _sqlite(path: Path, *, is_singleton: bool) -> tuple[ReadWriteLock, AsyncReadWriteLock]:
+    return ReadWriteLock(path / "lock.db", is_singleton=is_singleton), AsyncReadWriteLock(path / "lock.db")
+
+
+def _soft(  # pragma: needs hard-link
+    path: Path, *, is_singleton: bool
+) -> tuple[SoftReadWriteLock, AsyncSoftReadWriteLock]:
+    return (
+        SoftReadWriteLock(
+            path / "lock", is_singleton=is_singleton, heartbeat_interval=0.1, stale_threshold=0.5, poll_interval=0.02
+        ),
+        AsyncSoftReadWriteLock(path / "lock", heartbeat_interval=0.1, stale_threshold=0.5, poll_interval=0.02),
+    )
+
+
+_Make = Callable[..., tuple[ReadWriteLock | SoftReadWriteLock, AsyncReadWriteLock | AsyncSoftReadWriteLock]]
+_FAMILIES: Final = pytest.mark.parametrize(
+    "make", [pytest.param(_sqlite, id="sqlite"), pytest.param(_soft, id="soft", marks=pytest.mark.requires_hard_links)]
+)
+_MODES: Final = pytest.mark.parametrize("mode", [pytest.param("read", id="read"), pytest.param("write", id="write")])
+
+
+async def _acquire_within_deadline(
+    lock: AsyncReadWriteLock | AsyncSoftReadWriteLock, mode: Literal["read", "write"], timeout: float
+) -> float:
+    start = time.perf_counter()
+    with pytest.raises(TimeoutError) as info:
+        async with asyncio.timeout(_DEADLINE):
+            await (lock.acquire_read if mode == "read" else lock.acquire_write)(timeout)
+    assert not isinstance(info.value, Timeout)
+    return time.perf_counter() - start
+
+
+@_FAMILIES
+@_MODES
+@pytest.mark.parametrize(
+    "timeout", [pytest.param(-1, id="no-lock-timeout"), pytest.param(_HOLD * 2, id="lock-timeout")]
+)
+@pytest.mark.asyncio
+async def test_async_acquire_behind_another_process_honors_the_callers_deadline(
+    tmp_path: Path, make: _Make, mode: Literal["read", "write"], timeout: float
+) -> None:
+    holder, lock = make(tmp_path, is_singleton=False)
+    holder.acquire_write()
+    ending = asyncio.get_running_loop().call_later(_HOLD, holder.release)
+
+    elapsed = await _acquire_within_deadline(lock, mode, timeout)
+
+    assert elapsed < _HOLD
+    ending.cancel()
+    holder.release()
+    await (lock.acquire_read if mode == "read" else lock.acquire_write)(_HOLD * 2)
+    await lock.release()
+    await lock.close()
+
+
+@_FAMILIES
+@pytest.mark.asyncio
+async def test_async_acquire_behind_a_sync_acquire_of_the_same_instance_honors_the_callers_deadline(
+    tmp_path: Path, make: _Make
+) -> None:
+    holder, lock = make(tmp_path, is_singleton=False)
+    shared, _ = make(tmp_path, is_singleton=True)
+    holder.acquire_write()
+
+    # A thread waiting in the shared instance's acquire keeps that instance busy for the task too.
+    def write_once() -> None:
+        with shared.write_lock(timeout=_HOLD * 2):
+            pass
+
+    waiting = threading.Thread(target=write_once)
+    waiting.start()
+    ending = asyncio.get_running_loop().call_later(_HOLD, holder.release)
+
+    elapsed = await _acquire_within_deadline(lock, "write", -1)
+
+    assert elapsed < _HOLD
+    ending.cancel()
+    holder.release()
+    await asyncio.to_thread(waiting.join, _HOLD * 2)
+    await lock.acquire_write(timeout=_HOLD)
+    await lock.release()
+    await lock.close()
+
+
+@_FAMILIES
+@pytest.mark.asyncio
+async def test_async_acquire_behind_a_sync_acquire_of_the_same_instance_times_out_at_its_own_deadline(
+    tmp_path: Path, make: _Make, mocker: MockerFixture
+) -> None:
+    # With one thread alive the wait is not sliced, so the thread lock is polled across several intervals.
+    mocker.patch.object(threading, "active_count", return_value=1)
+    holder, lock = make(tmp_path, is_singleton=False)
+    shared, _ = make(tmp_path, is_singleton=True)
+    holder.acquire_write()
+
+    def write_once() -> None:
+        with shared.write_lock(timeout=_HOLD * 2):
+            pass
+
+    waiting = threading.Thread(target=write_once)
+    waiting.start()
+    ending = asyncio.get_running_loop().call_later(_HOLD, holder.release)
+    start = time.perf_counter()
+
+    with pytest.raises(Timeout):
+        await lock.acquire_write(_DEADLINE)
+
+    assert time.perf_counter() - start < _HOLD
+    ending.cancel()
+    holder.release()
+    await asyncio.to_thread(waiting.join, _HOLD * 2)
+    await lock.close()
diff --git a/tests/test_async_read_write_cancellation.py b/tests/test_async_read_write_cancellation.py
index c0e17aca..f2f6a314 100644
--- a/tests/test_async_read_write_cancellation.py
+++ b/tests/test_async_read_write_cancellation.py
@@ -362,6 +362,58 @@ async def fail_in_context() -> None:
     assert_read_write_lock_state(lock_file, probe_mode, available=True)
 
 
+@pytest.mark.asyncio
+async def test_read_retried_after_its_select_waited_starts_a_fresh_transaction(
+    lock_file: str, mocker: MockerFixture
+) -> None:
+    busy_once = True
+    seen_in_transaction: list[bool] = []
+
+    def executescript(real_connection: sqlite3.Connection, statement: str) -> sqlite3.Cursor:
+        nonlocal busy_once
+        if "SELECT name" in statement:
+            if busy_once:
+                # A writer can take the lock between the deferred BEGIN and the SELECT that needs the shared lock.
+                busy_once = False
+                real_connection.execute("BEGIN TRANSACTION")
+                msg = "database is locked"
+                raise sqlite3.OperationalError(msg)
+            seen_in_transaction.append(real_connection.in_transaction)
+        return real_connection.executescript(statement)
+
+    _patch_async_connection(mocker, executescript=executescript)
+    # With other threads alive the acquire waits in short slices that never reach the in-place retry under test.
+    mocker.patch.object(threading, "active_count", return_value=1)
+    lock = AsyncReadWriteLock(lock_file, is_singleton=False)
+
+    async with lock.read_lock():
+        assert_read_write_lock_state(lock_file, "write", available=False)
+    await lock.close()
+    assert seen_in_transaction == [False]
+
+
+@pytest.mark.asyncio
+async def test_write_retried_after_its_begin_waited_is_acquired(lock_file: str, mocker: MockerFixture) -> None:
+    busy_once = True
+
+    def executescript(real_connection: sqlite3.Connection, statement: str) -> sqlite3.Cursor:
+        nonlocal busy_once
+        if "BEGIN EXCLUSIVE" in statement and busy_once:
+            busy_once = False
+            msg = "database is locked"
+            raise sqlite3.OperationalError(msg)
+        return real_connection.executescript(statement)
+
+    _patch_async_connection(mocker, executescript=executescript)
+    mocker.patch.object(threading, "active_count", return_value=1)
+    lock = AsyncReadWriteLock(lock_file, is_singleton=False)
+
+    async with lock.write_lock():
+        assert_read_write_lock_state(lock_file, "read", available=False)
+    await lock.close()
+    assert not busy_once
+
+
 def _patch_async_rollback(
     mocker: MockerFixture,
     loop: asyncio.AbstractEventLoop,
@@ -511,7 +563,7 @@ def connect(
             timeout=timeout,
         )
         connection_count += 1
-        if connection_count != 2:
+        if connection_count < 2:
             return real_connection
         configuration: dict[str, Callable[[str], sqlite3.Cursor] | Callable[[], None]] = {}
         if executescript is not None: