From fee35e94cf999461f9f8435fbf0c1b3747238827 Mon Sep 17 00:00:00 2001
From: Mark S <ray@insighttimer.com>
Date: Tue, 15 Sep 2026 03:54:05 +1000
Subject: [PATCH 1/5] fix(redis): release a lock won after acquire_lock
 cancellation (LAB-3606)

asyncio.to_thread cannot be interrupted once the executor thread starts the
SET NX round-trip, so cancelling the task awaiting acquire_lock only stops the
coroutine from seeing the result -- not the thread from winning the lock. The
try/finally release block was never reached in that case, orphaning the key
for its full TTL. Run each attempt as its own task and await it through
asyncio.shield; on cancellation, wait for the attempt's real result and
release before re-raising.
---
 docs/features/distributed-locking.md      |  5 ++
 src/cachekit/backends/redis/provider.py   | 27 +++++++---
 tests/unit/backends/test_redis_backend.py | 61 +++++++++++++++++++++--
 3 files changed, 84 insertions(+), 9 deletions(-)

diff --git a/docs/features/distributed-locking.md b/docs/features/distributed-locking.md
index bec2efa3..b8c07080 100644
--- a/docs/features/distributed-locking.md
+++ b/docs/features/distributed-locking.md
@@ -162,6 +162,11 @@ Three behavioural edges to design around:
 # the lock itself self-expires after 30 s (lock_timeout) as the safety net.
 ```
 
+Cancelling the task awaiting `acquire_lock` mid-attempt does not trigger this
+degradation: the in-flight `SET NX` is always awaited to completion, and a lock
+it goes on to win is released before the cancellation propagates, so a
+cancelled waiter never orphans a held lock.
+
 ### TTL Shorter Than Compute Time
 ```python
 @cache(ttl=1)  # 1 second TTL
diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py
index 707b738e..d3fb8dc0 100644
--- a/src/cachekit/backends/redis/provider.py
+++ b/src/cachekit/backends/redis/provider.py
@@ -384,8 +384,27 @@ async def acquire_lock(
             loop = asyncio.get_running_loop()
             deadline = None if blocking_timeout is None else loop.time() + blocking_timeout
             token = uuid.uuid4().hex  # one token for the whole acquisition, however many attempts
+
+            async def _release() -> None:
+                try:
+                    await asyncio.to_thread(lock.release)
+                except Exception as e:
+                    # Lock may have expired - log but don't fail
+                    logger.debug("Error releasing Redis lock (may have expired): %s", e)
+
             while True:
-                acquired = await asyncio.to_thread(lock.acquire, blocking=False, token=token)
+                # asyncio.to_thread cannot be interrupted once the executor thread starts the
+                # SET NX round-trip, so a cancellation of the awaiting task doesn't stop it from
+                # winning the lock — only from seeing that it did. Run the attempt as its own task
+                # and await it shielded: on cancellation, wait for the attempt's real result and
+                # release before re-raising, instead of orphaning a won lock for its full TTL.
+                attempt = asyncio.ensure_future(asyncio.to_thread(lock.acquire, blocking=False, token=token))
+                try:
+                    acquired = await asyncio.shield(attempt)
+                except asyncio.CancelledError:
+                    if await attempt:
+                        await _release()
+                    raise
                 # Same give-up rule as redis-py's Lock.acquire: stop once the next attempt
                 # would land past the deadline. blocking_timeout=None means a single attempt.
                 if acquired or deadline is None or loop.time() + lock.sleep > deadline:
@@ -396,11 +415,7 @@ async def acquire_lock(
             finally:
                 # Release lock if acquired (also run in thread pool)
                 if acquired:
-                    try:
-                        await asyncio.to_thread(lock.release)
-                    except Exception as e:
-                        # Lock may have expired - log but don't fail
-                        logger.debug("Error releasing Redis lock (may have expired): %s", e)
+                    await _release()
         except Exception as exc:
             raise classify_redis_error(exc, operation="acquire_lock", key=key) from exc
 
diff --git a/tests/unit/backends/test_redis_backend.py b/tests/unit/backends/test_redis_backend.py
index 69ca320b..4f19b4f6 100644
--- a/tests/unit/backends/test_redis_backend.py
+++ b/tests/unit/backends/test_redis_backend.py
@@ -274,6 +274,14 @@ def __init__(self) -> None:
         self._store: dict[str, bytes] = {}
         self._mutex = threading.Lock()
         self.nx_attempts: list[float] = []  # monotonic time of every SET NX, i.e. every acquire attempt
+        # Test hooks for the cancellation-mid-attempt race: when set, an NX SET call
+        # signals nx_entered (so the test knows the executor thread is inside the call),
+        # blocks on block_nx until the test releases it, then signals nx_done once the
+        # store write has actually landed — independent of whatever asyncio did with the
+        # coroutine that was awaiting it.
+        self.nx_entered: threading.Event | None = None
+        self.block_nx: threading.Event | None = None
+        self.nx_done: threading.Event | None = None
 
     def get_encoder(self) -> Encoder:
         return Encoder("utf-8", "strict", False)
@@ -282,13 +290,21 @@ def register_script(self, script: str) -> Script:
         return Script(self, script)
 
     def set(self, name: str, value: bytes, nx: bool = False, px: int | None = None) -> bool | None:
+        if nx and self.nx_entered is not None:
+            self.nx_entered.set()
+        if nx and self.block_nx is not None:
+            self.block_nx.wait()
         with self._mutex:
             if nx:
                 self.nx_attempts.append(time.monotonic())
             if nx and name in self._store:
-                return None
-            self._store[name] = value
-            return True
+                result = None
+            else:
+                self._store[name] = value
+                result = True
+        if nx and self.nx_done is not None:
+            self.nx_done.set()
+        return result
 
     def evalsha(self, _sha: str, _numkeys: int, name: str, token: bytes) -> int:
         """The only script ``Lock`` runs here is LUA_RELEASE: delete iff the token still matches."""
@@ -375,3 +391,42 @@ async def test_non_blocking_acquire_makes_exactly_one_attempt(self):
                 assert contended is False
 
         assert len(fake.nx_attempts) == 2, "blocking_timeout=None must be a single SET NX per acquire_lock"
+
+    async def test_cancellation_mid_attempt_releases_a_lock_it_goes_on_to_win(self):
+        """Cancelling the awaiter while the SET NX round-trip is in flight must not orphan the key.
+
+        ``asyncio.to_thread`` can't be interrupted once the executor thread starts the
+        round-trip, so cancellation only stops the awaiting coroutine from seeing the
+        result — not the thread from winning the lock. Red on the pre-fix code (the
+        `try`/`finally` release block is never reached because the cancellation
+        propagates straight out of the `while True` loop); green with the shield.
+        """
+        fake = _FakeRedis()
+        fake.nx_entered = threading.Event()
+        fake.block_nx = threading.Event()
+        fake.nx_done = threading.Event()
+        backend = PerRequestRedisBackend(fake, tenant_id="t")
+
+        async def acquire() -> None:
+            async with backend.acquire_lock("k", timeout=30.0, blocking_timeout=None):
+                pass
+
+        task = asyncio.create_task(acquire())
+        deadline = time.monotonic() + 2.0
+        while not fake.nx_entered.is_set():
+            assert time.monotonic() < deadline, "executor thread never entered the SET NX call"
+            await asyncio.sleep(0.01)
+
+        task.cancel()
+        fake.block_nx.set()  # let the executor thread finish the SET NX (it wins the lock)
+
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+        # The executor thread runs independently of the cancelled coroutine, so wait for
+        # its write to actually land before checking the store — otherwise the assertion
+        # below races the background thread instead of testing the fix.
+        assert fake.nx_done.wait(2.0), "executor thread never finished the SET NX"
+
+        lock_name = backend._scoped_key("k") + ":lock"
+        assert lock_name not in fake._store, "lock won after cancellation must still be released"

From 81f4561b6edd054ff12907b2ab81d153b6c663dd Mon Sep 17 00:00:00 2001
From: Mark S <ray@insighttimer.com>
Date: Tue, 15 Sep 2026 04:09:04 +1000
Subject: [PATCH 2/5] fix(redis): never let a failed attempt mask acquire_lock
 cancellation (LAB-3606)

Panel review on #293: if the shielded SET NX attempt raises (e.g. a Redis
ConnectionError) instead of returning, `await attempt` inside the
CancelledError handler re-raised that exception, which escaped to the outer
classify_redis_error handler instead of the CancelledError -- silently
breaking cancellation propagation. Swallow the attempt's own failure and
always re-raise the cancellation. Also reword the docs claim from "never
orphans" to "single cancellation never orphans" -- the release itself is
cancellable, so a second cancellation during it re-orphans the key within the
same 30s TTL ceiling (accepted, not fixed).
---
 docs/features/distributed-locking.md    | 4 +++-
 src/cachekit/backends/redis/provider.py | 8 +++++++-
 2 files changed, 10 insertions(+), 2 deletions(-)

diff --git a/docs/features/distributed-locking.md b/docs/features/distributed-locking.md
index b8c07080..8234f5b2 100644
--- a/docs/features/distributed-locking.md
+++ b/docs/features/distributed-locking.md
@@ -165,7 +165,9 @@ Three behavioural edges to design around:
 Cancelling the task awaiting `acquire_lock` mid-attempt does not trigger this
 degradation: the in-flight `SET NX` is always awaited to completion, and a lock
 it goes on to win is released before the cancellation propagates, so a
-cancelled waiter never orphans a held lock.
+**single** cancellation never orphans a held lock. A *second* cancellation
+landing during that release is not shielded and re-orphans the key — bounded
+by the same 30 s TTL as the crash case above.
 
 ### TTL Shorter Than Compute Time
 ```python
diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py
index d3fb8dc0..7e5c4769 100644
--- a/src/cachekit/backends/redis/provider.py
+++ b/src/cachekit/backends/redis/provider.py
@@ -402,7 +402,13 @@ async def _release() -> None:
                 try:
                     acquired = await asyncio.shield(attempt)
                 except asyncio.CancelledError:
-                    if await attempt:
+                    # Recover the shielded attempt's real result; never let its own failure
+                    # (e.g. a Redis ConnectionError) mask the cancellation — always re-raise.
+                    try:
+                        won = await attempt
+                    except Exception:
+                        won = False
+                    if won:
                         await _release()
                     raise
                 # Same give-up rule as redis-py's Lock.acquire: stop once the next attempt

From fe5974d4e428c8c411b125ad17fd9e30ff3c5dfc Mon Sep 17 00:00:00 2001
From: Mark S <ray@insighttimer.com>
Date: Tue, 15 Sep 2026 07:21:56 +1000
Subject: [PATCH 3/5] fix(redis): drain repeated cancellation through
 acquire_lock attempt and release (LAB-3606)

The first cut awaited the shielded attempt bare once its shield was
cancelled, so a second task.cancel() landed on the attempt task itself:
its result was lost and the release skipped, orphaning a won lock for
its TTL. The release had the same hole from the other side: with the
executor saturated, to_thread(lock.release) sits in the pool queue, and
a cancellation there drops the queued work item so the release never
runs at all.

Both round-trips now go through _await_uninterrupted, which waits on the
future with asyncio.wait (never cancels or unwraps its input), absorbs
every cancellation, and re-raises the last one only after the future has
really finished. The acquisition handler reads the attempt's real
outcome: a Redis failure is logged at WARNING with its traceback instead
of being swallowed (it cannot have won), a win is released, a loss is
left alone. The release helper catches redis.RedisError only, so a
non-Redis failure surfaces as a BackendError instead of a debug line.

Regression tests cover a second cancellation during attempt drain and
during a queued release (1-thread executor), a failing attempt under
cancellation, a losing attempt under cancellation, and Redis failing the
release. Docs drop the "second cancellation re-orphans" caveat; the one
remaining gap is Redis itself failing the release, bounded by the TTL.
---
 docs/features/distributed-locking.md      |  11 +-
 src/cachekit/backends/redis/provider.py   |  69 ++++++++---
 tests/unit/backends/test_redis_backend.py | 142 ++++++++++++++++++++--
 3 files changed, 188 insertions(+), 34 deletions(-)

diff --git a/docs/features/distributed-locking.md b/docs/features/distributed-locking.md
index 8234f5b2..2a947bf4 100644
--- a/docs/features/distributed-locking.md
+++ b/docs/features/distributed-locking.md
@@ -163,11 +163,12 @@ Three behavioural edges to design around:
 ```
 
 Cancelling the task awaiting `acquire_lock` mid-attempt does not trigger this
-degradation: the in-flight `SET NX` is always awaited to completion, and a lock
-it goes on to win is released before the cancellation propagates, so a
-**single** cancellation never orphans a held lock. A *second* cancellation
-landing during that release is not shielded and re-orphans the key — bounded
-by the same 30 s TTL as the crash case above.
+degradation: the in-flight `SET NX` is awaited to completion — through repeated
+cancellations too — and a lock it goes on to win is released before the
+cancellation propagates. The release round-trip is drained the same way, so a
+cancel landing while it is still queued for an executor thread cannot drop it.
+What remains is Redis itself failing the release, which leaves the key until
+the same 30 s TTL as the crash case above.
 
 ### TTL Shorter Than Compute Time
 ```python
diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py
index 7e5c4769..deedce25 100644
--- a/src/cachekit/backends/redis/provider.py
+++ b/src/cachekit/backends/redis/provider.py
@@ -11,11 +11,13 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
+import uuid
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from contextvars import ContextVar
-from typing import Any, Optional
+from typing import Any, Optional, TypeVar
 from urllib.parse import quote as url_encode
 
 import redis
@@ -29,6 +31,29 @@
 # Module-level ContextVar for async-safe tenant isolation
 tenant_context: ContextVar[Optional[str]] = ContextVar("tenant_context", default=None)
 
+T = TypeVar("T")
+
+
+async def _await_uninterrupted(fut: asyncio.Future[T]) -> T:
+    """Await ``fut`` to completion even if the current task is cancelled meanwhile.
+
+    ``asyncio.to_thread`` work is uninterruptible once an executor thread picks it up, and a
+    still-queued work item is dropped if its future is cancelled first — so a cancelled awaiter
+    either loses the outcome of a round-trip that still completes, or loses the round-trip
+    itself. ``asyncio.wait`` never cancels its inputs and never unwraps their result, so keep
+    waiting on ``fut`` until it is really done, absorbing every cancellation, then re-raise the
+    last one: callers read ``fut`` for the real outcome before letting it propagate.
+    """
+    cancelled: Optional[asyncio.CancelledError] = None
+    while not fut.done():
+        try:
+            await asyncio.wait({fut})
+        except asyncio.CancelledError as exc:
+            cancelled = exc
+    if cancelled is not None:
+        raise cancelled
+    return fut.result()
+
 
 class PerRequestRedisBackend:
     """Per-request Redis backend wrapper with tenant isolation.
@@ -362,10 +387,10 @@ async def acquire_lock(
             the waiters, every waiter times out, and all of them recompute.
             Sets thread_local=False because attempts and release may run on different
             executor threads.
+            Cancellation is drained, not raced: an in-flight attempt or release round-trip
+            always runs to completion, a lock the attempt wins is released, and only then is
+            the ``CancelledError`` re-raised.
         """
-        import asyncio
-        import uuid
-
         # Derive the on-wire Redis lock name from the bare cache key: ``<scoped_key>:lock``.
         # Keeping this suffix on the wire preserves compatibility with existing Redis
         # deployments — the lock identity didn't change, only the protocol boundary
@@ -385,30 +410,34 @@ async def acquire_lock(
             deadline = None if blocking_timeout is None else loop.time() + blocking_timeout
             token = uuid.uuid4().hex  # one token for the whole acquisition, however many attempts
 
-            async def _release() -> None:
+            def _release_sync() -> None:
                 try:
-                    await asyncio.to_thread(lock.release)
-                except Exception as e:
+                    lock.release()
+                except redis.RedisError as e:
                     # Lock may have expired - log but don't fail
                     logger.debug("Error releasing Redis lock (may have expired): %s", e)
 
+            async def _release() -> None:
+                # Uninterrupted: with the executor saturated the release sits in the pool's queue,
+                # and a cancellation landing then would drop it and orphan the key for its TTL.
+                await _await_uninterrupted(asyncio.ensure_future(asyncio.to_thread(_release_sync)))
+
             while True:
-                # asyncio.to_thread cannot be interrupted once the executor thread starts the
-                # SET NX round-trip, so a cancellation of the awaiting task doesn't stop it from
-                # winning the lock — only from seeing that it did. Run the attempt as its own task
-                # and await it shielded: on cancellation, wait for the attempt's real result and
-                # release before re-raising, instead of orphaning a won lock for its full TTL.
+                # A cancellation of the awaiting task doesn't stop the executor thread's SET NX
+                # from winning the lock — only the coroutine from seeing that it did. Await the
+                # attempt uninterrupted: on cancellation, recover its real result and release a
+                # won lock before re-raising, instead of orphaning it for its full TTL.
                 attempt = asyncio.ensure_future(asyncio.to_thread(lock.acquire, blocking=False, token=token))
                 try:
-                    acquired = await asyncio.shield(attempt)
+                    acquired = await _await_uninterrupted(attempt)
                 except asyncio.CancelledError:
-                    # Recover the shielded attempt's real result; never let its own failure
-                    # (e.g. a Redis ConnectionError) mask the cancellation — always re-raise.
-                    try:
-                        won = await attempt
-                    except Exception:
-                        won = False
-                    if won:
+                    # The attempt has finished. One that failed (e.g. a Redis ConnectionError) cannot
+                    # have won; log it rather than let it mask the cancellation.
+                    if (err := attempt.exception()) is not None:
+                        logger.warning(
+                            "Redis lock attempt for %s failed while acquire_lock was being cancelled", key, exc_info=err
+                        )
+                    elif attempt.result():
                         await _release()
                     raise
                 # Same give-up rule as redis-py's Lock.acquire: stop once the next attempt
diff --git a/tests/unit/backends/test_redis_backend.py b/tests/unit/backends/test_redis_backend.py
index 4f19b4f6..535c7c8f 100644
--- a/tests/unit/backends/test_redis_backend.py
+++ b/tests/unit/backends/test_redis_backend.py
@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
@@ -24,6 +25,7 @@
 import pytest
 from redis.commands.core import Script
 from redis.connection import Encoder
+from redis.exceptions import ConnectionError as RedisConnectionError
 from redis.lock import Lock
 
 from cachekit.backends.redis import RedisBackend
@@ -282,6 +284,7 @@ def __init__(self) -> None:
         self.nx_entered: threading.Event | None = None
         self.block_nx: threading.Event | None = None
         self.nx_done: threading.Event | None = None
+        self.nx_error: Exception | None = None  # raised by the NX SET once unblocked, in place of a result
 
     def get_encoder(self) -> Encoder:
         return Encoder("utf-8", "strict", False)
@@ -294,6 +297,8 @@ def set(self, name: str, value: bytes, nx: bool = False, px: int | None = None)
             self.nx_entered.set()
         if nx and self.block_nx is not None:
             self.block_nx.wait()
+        if nx and self.nx_error is not None:
+            raise self.nx_error
         with self._mutex:
             if nx:
                 self.nx_attempts.append(time.monotonic())
@@ -315,6 +320,19 @@ def evalsha(self, _sha: str, _numkeys: int, name: str, token: bytes) -> int:
             return 1
 
 
+async def _entered(event: threading.Event, what: str) -> None:
+    """Poll a thread-side event from the loop; ``to_thread(event.wait)`` would take the very executor thread under test."""
+    deadline = time.monotonic() + 2.0
+    while not event.is_set():
+        assert time.monotonic() < deadline, what
+        await asyncio.sleep(0.01)
+
+
+async def _acquire_once(backend: PerRequestRedisBackend) -> None:
+    async with backend.acquire_lock("k", timeout=30.0, blocking_timeout=None):
+        pass
+
+
 @pytest.mark.unit
 class TestRedisLockWaitersDoNotPinExecutorThreads:
     """A lock waiter must not hold an executor thread while it waits.
@@ -407,15 +425,8 @@ async def test_cancellation_mid_attempt_releases_a_lock_it_goes_on_to_win(self):
         fake.nx_done = threading.Event()
         backend = PerRequestRedisBackend(fake, tenant_id="t")
 
-        async def acquire() -> None:
-            async with backend.acquire_lock("k", timeout=30.0, blocking_timeout=None):
-                pass
-
-        task = asyncio.create_task(acquire())
-        deadline = time.monotonic() + 2.0
-        while not fake.nx_entered.is_set():
-            assert time.monotonic() < deadline, "executor thread never entered the SET NX call"
-            await asyncio.sleep(0.01)
+        task = asyncio.create_task(_acquire_once(backend))
+        await _entered(fake.nx_entered, "executor thread never entered the SET NX call")
 
         task.cancel()
         fake.block_nx.set()  # let the executor thread finish the SET NX (it wins the lock)
@@ -430,3 +441,116 @@ async def acquire() -> None:
 
         lock_name = backend._scoped_key("k") + ":lock"
         assert lock_name not in fake._store, "lock won after cancellation must still be released"
+
+    async def test_second_cancellation_while_draining_the_attempt_still_releases_the_lock(self):
+        """A cancel landing while the first one waits out the in-flight SET NX must not orphan the key.
+
+        Awaiting the attempt bare once its shield is cancelled hands the *next* ``task.cancel()``
+        straight to the attempt task: its result is lost and the release skipped. Red on that
+        code, green once the attempt is awaited uninterrupted.
+        """
+        fake = _FakeRedis()
+        fake.nx_entered = threading.Event()
+        fake.block_nx = threading.Event()
+        fake.nx_done = threading.Event()
+        backend = PerRequestRedisBackend(fake, tenant_id="t")
+
+        task = asyncio.create_task(_acquire_once(backend))
+        await _entered(fake.nx_entered, "executor thread never entered the SET NX call")
+
+        task.cancel()
+        await asyncio.sleep(0)  # first cancellation lands; acquire_lock is now waiting out the attempt
+        task.cancel()
+        fake.block_nx.set()  # the executor thread finishes the SET NX and wins the lock
+
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert fake.nx_done.wait(2.0), "executor thread never finished the SET NX"
+        assert backend._scoped_key("k") + ":lock" not in fake._store, "lock won under repeated cancellation must be released"
+
+    async def test_second_cancellation_while_the_release_is_queued_still_releases_the_lock(self):
+        """A cancel landing while ``lock.release`` still waits for an executor thread must not orphan the key.
+
+        With every executor thread busy — the saturation this class exists for — the release
+        sits in the pool's queue, and a bare ``await to_thread(lock.release)`` lets the next
+        ``task.cancel()`` cancel that queued work item, so the release never runs. Red on that
+        code, green once the release is awaited uninterrupted.
+        """
+        fake = _FakeRedis()
+        backend = PerRequestRedisBackend(fake, tenant_id="t")
+        pool = ThreadPoolExecutor(max_workers=1)
+        asyncio.get_running_loop().set_default_executor(pool)
+        holding = asyncio.Event()
+
+        async def hold() -> None:
+            async with backend.acquire_lock("k", timeout=30.0, blocking_timeout=None) as acquired:
+                assert acquired
+                holding.set()
+                await asyncio.Event().wait()  # hold the lock until cancelled
+
+        task = asyncio.create_task(hold())
+        await holding.wait()
+
+        busy = threading.Event()
+        pool.submit(busy.wait)  # the only executor thread is now taken; the release will queue behind it
+        task.cancel()
+        await asyncio.sleep(0)  # first cancellation lands; the release is queued for the pool
+        task.cancel()
+        busy.set()
+
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        pool.shutdown(wait=True)  # whatever survived in the queue has run by now
+        assert backend._scoped_key("k") + ":lock" not in fake._store, "lock must be released despite repeated cancellation"
+
+    async def test_attempt_failing_during_cancellation_is_logged_not_raised(self, caplog):
+        """A Redis error from the in-flight attempt must not replace the ``CancelledError``; it is logged instead."""
+        fake = _FakeRedis()
+        fake.nx_entered = threading.Event()
+        fake.block_nx = threading.Event()
+        fake.nx_error = RedisConnectionError("redis went away")
+        backend = PerRequestRedisBackend(fake, tenant_id="t")
+
+        task = asyncio.create_task(_acquire_once(backend))
+        await _entered(fake.nx_entered, "executor thread never entered the SET NX call")
+        task.cancel()
+        fake.block_nx.set()  # the executor thread now fails the SET NX
+
+        with pytest.raises(asyncio.CancelledError), caplog.at_level(logging.WARNING, logger="cachekit.backends.redis.provider"):
+            await task
+        assert any(
+            r.levelno == logging.WARNING and r.exc_info and isinstance(r.exc_info[1], RedisConnectionError)
+            for r in caplog.records
+        ), "a failed attempt swallowed by cancellation must be logged with its traceback"
+
+    async def test_cancellation_mid_attempt_that_loses_leaves_the_holders_lock_alone(self, caplog):
+        """A cancelled attempt that loses to an existing holder has nothing to release: no release call, nothing logged."""
+        fake = _FakeRedis()
+        fake.nx_entered = threading.Event()
+        fake.block_nx = threading.Event()
+        backend = PerRequestRedisBackend(fake, tenant_id="t")
+        lock_name = backend._scoped_key("k") + ":lock"
+        fake._store[lock_name] = b"someone-else"
+
+        task = asyncio.create_task(_acquire_once(backend))
+        await _entered(fake.nx_entered, "executor thread never entered the SET NX call")
+        task.cancel()
+        fake.block_nx.set()  # the executor thread finishes the SET NX and loses
+
+        with pytest.raises(asyncio.CancelledError), caplog.at_level(logging.DEBUG, logger="cachekit.backends.redis.provider"):
+            await task
+        assert fake._store[lock_name] == b"someone-else"
+        assert not caplog.records, "a lost attempt must not try to release (a bare release would log at DEBUG)"
+
+    async def test_release_failing_in_redis_is_logged_and_leaves_the_key_to_its_ttl(self, caplog, monkeypatch):
+        """Redis failing the release is the one gap left: the caller sees no error, the key lives until its TTL."""
+        fake = _FakeRedis()
+        monkeypatch.setattr(fake, "evalsha", Mock(side_effect=RedisConnectionError("redis went away")))
+        backend = PerRequestRedisBackend(fake, tenant_id="t")
+
+        with caplog.at_level(logging.DEBUG, logger="cachekit.backends.redis.provider"):
+            async with backend.acquire_lock("k", timeout=30.0, blocking_timeout=None) as acquired:
+                assert acquired
+
+        assert backend._scoped_key("k") + ":lock" in fake._store, "a failed release leaves the key for its TTL"
+        assert any("releasing" in r.getMessage() and r.levelno == logging.DEBUG for r in caplog.records)

From 48f76a46fb96b78fc8dbb459e41e7cf57b937a0a Mon Sep 17 00:00:00 2001
From: Mark S <ray@insighttimer.com>
Date: Tue, 15 Sep 2026 07:40:53 +1000
Subject: [PATCH 4/5] fix(redis): run lock round-trips as plain executor
 futures; warn on a failed release (LAB-3606)

Expert-panel findings on the drain:

- ensure_future(to_thread(...)) made each round-trip a Task, so any
  all_tasks() sweep (asyncio.run teardown, graceful shutdown) cancelled it
  under _await_uninterrupted and the win was lost again. Plain
  loop.run_in_executor futures are invisible to that sweep. Regression test
  mimics the sweep.
- A release failing for anything but LockNotOwnedError orphans the key
  until its TTL; it was logged at DEBUG with no key or traceback while the
  attempt failure got WARNING. Now WARNING with exc_info; the benign
  expired/taken-over case stays at DEBUG. Test parametrised over both.
- Keys in log lines go through redact_cache_key (issue #163), local import
  because cache_handler imports the backends package.
- Docs paragraph scoped to RedisBackend: CachekitIOBackend.acquire_lock
  has the same gap (LAB-3648). Stale "shield" wording removed from test
  docstrings; duplicated WHY comments trimmed to one pointer each.
---
 docs/features/distributed-locking.md      | 13 +++---
 src/cachekit/backends/redis/provider.py   | 33 ++++++++-----
 tests/unit/backends/test_redis_backend.py | 56 +++++++++++++++++------
 3 files changed, 71 insertions(+), 31 deletions(-)

diff --git a/docs/features/distributed-locking.md b/docs/features/distributed-locking.md
index 2a947bf4..1c0b1344 100644
--- a/docs/features/distributed-locking.md
+++ b/docs/features/distributed-locking.md
@@ -162,13 +162,12 @@ Three behavioural edges to design around:
 # the lock itself self-expires after 30 s (lock_timeout) as the safety net.
 ```
 
-Cancelling the task awaiting `acquire_lock` mid-attempt does not trigger this
-degradation: the in-flight `SET NX` is awaited to completion — through repeated
-cancellations too — and a lock it goes on to win is released before the
-cancellation propagates. The release round-trip is drained the same way, so a
-cancel landing while it is still queued for an executor thread cannot drop it.
-What remains is Redis itself failing the release, which leaves the key until
-the same 30 s TTL as the crash case above.
+On `RedisBackend`, cancelling the task mid-`acquire_lock` does not orphan the
+lock: the in-flight `SET NX` and the release both run to completion — however
+many cancellations land — before the `CancelledError` propagates. Only Redis
+failing the release leaves the key, until the same 30 s TTL as the crash case
+above. `CachekitIOBackend` does not yet drain cancellation this way: a cancel
+mid-request can leave a server-granted lock held until its server-side timeout.
 
 ### TTL Shorter Than Compute Time
 ```python
diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py
index deedce25..9ef1592c 100644
--- a/src/cachekit/backends/redis/provider.py
+++ b/src/cachekit/backends/redis/provider.py
@@ -12,6 +12,7 @@
 from __future__ import annotations
 
 import asyncio
+import functools
 import logging
 import uuid
 from collections.abc import AsyncIterator
@@ -21,6 +22,7 @@
 from urllib.parse import quote as url_encode
 
 import redis
+from redis.exceptions import LockNotOwnedError
 
 from cachekit.backends.base import BaseBackend
 from cachekit.backends.errors import BackendError
@@ -43,6 +45,9 @@ async def _await_uninterrupted(fut: asyncio.Future[T]) -> T:
     itself. ``asyncio.wait`` never cancels its inputs and never unwraps their result, so keep
     waiting on ``fut`` until it is really done, absorbing every cancellation, then re-raise the
     last one: callers read ``fut`` for the real outcome before letting it propagate.
+
+    Pass a plain future (``loop.run_in_executor``), never a Task: ``all_tasks()`` sweeps such as
+    ``asyncio.run`` teardown cancel Tasks out from under the drain, and the outcome is lost again.
     """
     cancelled: Optional[asyncio.CancelledError] = None
     while not fut.done():
@@ -396,6 +401,8 @@ async def acquire_lock(
         # deployments — the lock identity didn't change, only the protocol boundary
         # (the wrapper no longer pollutes the cache_key passed in).
         scoped_key = f"{self._scoped_key(key)}:lock"
+        from cachekit.cache_handler import redact_cache_key  # local: cache_handler imports the backends package
+
         try:
             from redis.lock import Lock
 
@@ -411,23 +418,25 @@ async def acquire_lock(
             token = uuid.uuid4().hex  # one token for the whole acquisition, however many attempts
 
             def _release_sync() -> None:
+                # Catch inside the executor callable, not around _release(): once a cancellation has
+                # landed, _await_uninterrupted re-raises it and an error left on the future would only
+                # surface as asyncio's "exception was never retrieved" at GC.
                 try:
                     lock.release()
+                except LockNotOwnedError as e:
+                    logger.debug("Redis lock already expired or taken over before release: %s", e)  # nothing to orphan
                 except redis.RedisError as e:
-                    # Lock may have expired - log but don't fail
-                    logger.debug("Error releasing Redis lock (may have expired): %s", e)
+                    logger.warning(
+                        "Redis lock release for %s failed; the key lives until its TTL", redact_cache_key(key), exc_info=e
+                    )
 
             async def _release() -> None:
-                # Uninterrupted: with the executor saturated the release sits in the pool's queue,
-                # and a cancellation landing then would drop it and orphan the key for its TTL.
-                await _await_uninterrupted(asyncio.ensure_future(asyncio.to_thread(_release_sync)))
+                # Drained: a cancel landing while this still queues for a thread must not drop the release.
+                await _await_uninterrupted(loop.run_in_executor(None, _release_sync))
 
             while True:
-                # A cancellation of the awaiting task doesn't stop the executor thread's SET NX
-                # from winning the lock — only the coroutine from seeing that it did. Await the
-                # attempt uninterrupted: on cancellation, recover its real result and release a
-                # won lock before re-raising, instead of orphaning it for its full TTL.
-                attempt = asyncio.ensure_future(asyncio.to_thread(lock.acquire, blocking=False, token=token))
+                # Drained: a cancel cannot stop the thread's SET NX from winning, only hide that it did.
+                attempt = loop.run_in_executor(None, functools.partial(lock.acquire, blocking=False, token=token))
                 try:
                     acquired = await _await_uninterrupted(attempt)
                 except asyncio.CancelledError:
@@ -435,7 +444,9 @@ async def _release() -> None:
                     # have won; log it rather than let it mask the cancellation.
                     if (err := attempt.exception()) is not None:
                         logger.warning(
-                            "Redis lock attempt for %s failed while acquire_lock was being cancelled", key, exc_info=err
+                            "Redis lock attempt for %s failed while acquire_lock was being cancelled",
+                            redact_cache_key(key),
+                            exc_info=err,
                         )
                     elif attempt.result():
                         await _release()
diff --git a/tests/unit/backends/test_redis_backend.py b/tests/unit/backends/test_redis_backend.py
index 535c7c8f..3f999e5e 100644
--- a/tests/unit/backends/test_redis_backend.py
+++ b/tests/unit/backends/test_redis_backend.py
@@ -26,6 +26,7 @@
 from redis.commands.core import Script
 from redis.connection import Encoder
 from redis.exceptions import ConnectionError as RedisConnectionError
+from redis.exceptions import LockNotOwnedError
 from redis.lock import Lock
 
 from cachekit.backends.redis import RedisBackend
@@ -417,7 +418,7 @@ async def test_cancellation_mid_attempt_releases_a_lock_it_goes_on_to_win(self):
         round-trip, so cancellation only stops the awaiting coroutine from seeing the
         result — not the thread from winning the lock. Red on the pre-fix code (the
         `try`/`finally` release block is never reached because the cancellation
-        propagates straight out of the `while True` loop); green with the shield.
+        propagates straight out of the `while True` loop); green once the attempt is awaited uninterrupted.
         """
         fake = _FakeRedis()
         fake.nx_entered = threading.Event()
@@ -445,9 +446,8 @@ async def test_cancellation_mid_attempt_releases_a_lock_it_goes_on_to_win(self):
     async def test_second_cancellation_while_draining_the_attempt_still_releases_the_lock(self):
         """A cancel landing while the first one waits out the in-flight SET NX must not orphan the key.
 
-        Awaiting the attempt bare once its shield is cancelled hands the *next* ``task.cancel()``
-        straight to the attempt task: its result is lost and the release skipped. Red on that
-        code, green once the attempt is awaited uninterrupted.
+        A plain ``asyncio.shield`` hands the *next* ``task.cancel()`` straight to the attempt
+        itself: its result is lost and the release skipped.
         """
         fake = _FakeRedis()
         fake.nx_entered = threading.Event()
@@ -468,13 +468,37 @@ async def test_second_cancellation_while_draining_the_attempt_still_releases_the
         assert fake.nx_done.wait(2.0), "executor thread never finished the SET NX"
         assert backend._scoped_key("k") + ":lock" not in fake._store, "lock won under repeated cancellation must be released"
 
+    async def test_all_tasks_sweep_mid_attempt_still_releases_the_lock(self):
+        """``asyncio.run()`` teardown cancels everything in ``all_tasks()``: a round-trip run as a Task dies under the drain.
+
+        A plain executor future is invisible to that sweep, so the win is still read and released.
+        """
+        fake = _FakeRedis()
+        fake.nx_entered = threading.Event()
+        fake.block_nx = threading.Event()
+        fake.nx_done = threading.Event()
+        backend = PerRequestRedisBackend(fake, tenant_id="t")
+
+        task = asyncio.create_task(_acquire_once(backend))
+        await _entered(fake.nx_entered, "executor thread never entered the SET NX call")
+
+        me = asyncio.current_task()
+        for t in asyncio.all_tasks():  # what asyncio.run()'s _cancel_all_tasks does
+            if t is not me:
+                t.cancel()
+        fake.block_nx.set()  # the executor thread finishes the SET NX and wins the lock
+
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert fake.nx_done.wait(2.0), "executor thread never finished the SET NX"
+        assert backend._scoped_key("k") + ":lock" not in fake._store, "lock won during a shutdown sweep must be released"
+
     async def test_second_cancellation_while_the_release_is_queued_still_releases_the_lock(self):
         """A cancel landing while ``lock.release`` still waits for an executor thread must not orphan the key.
 
         With every executor thread busy — the saturation this class exists for — the release
         sits in the pool's queue, and a bare ``await to_thread(lock.release)`` lets the next
-        ``task.cancel()`` cancel that queued work item, so the release never runs. Red on that
-        code, green once the release is awaited uninterrupted.
+        ``task.cancel()`` cancel that queued work item, so the release never runs.
         """
         fake = _FakeRedis()
         backend = PerRequestRedisBackend(fake, tenant_id="t")
@@ -489,7 +513,7 @@ async def hold() -> None:
                 await asyncio.Event().wait()  # hold the lock until cancelled
 
         task = asyncio.create_task(hold())
-        await holding.wait()
+        await asyncio.wait_for(holding.wait(), 2.0)
 
         busy = threading.Event()
         pool.submit(busy.wait)  # the only executor thread is now taken; the release will queue behind it
@@ -539,13 +563,19 @@ async def test_cancellation_mid_attempt_that_loses_leaves_the_holders_lock_alone
 
         with pytest.raises(asyncio.CancelledError), caplog.at_level(logging.DEBUG, logger="cachekit.backends.redis.provider"):
             await task
-        assert fake._store[lock_name] == b"someone-else"
-        assert not caplog.records, "a lost attempt must not try to release (a bare release would log at DEBUG)"
-
-    async def test_release_failing_in_redis_is_logged_and_leaves_the_key_to_its_ttl(self, caplog, monkeypatch):
+        assert not caplog.records, "a lost attempt must not try to release (a release without a token logs)"
+
+    @pytest.mark.parametrize(
+        ("error", "level"),
+        [
+            (RedisConnectionError("redis went away"), logging.WARNING),  # key orphaned until its TTL: worth a warning
+            (LockNotOwnedError("expired"), logging.DEBUG),  # already gone or taken over: nothing to orphan
+        ],
+    )
+    async def test_release_failing_in_redis_is_logged_not_raised(self, caplog, monkeypatch, error, level):
         """Redis failing the release is the one gap left: the caller sees no error, the key lives until its TTL."""
         fake = _FakeRedis()
-        monkeypatch.setattr(fake, "evalsha", Mock(side_effect=RedisConnectionError("redis went away")))
+        monkeypatch.setattr(fake, "evalsha", Mock(side_effect=error))
         backend = PerRequestRedisBackend(fake, tenant_id="t")
 
         with caplog.at_level(logging.DEBUG, logger="cachekit.backends.redis.provider"):
@@ -553,4 +583,4 @@ async def test_release_failing_in_redis_is_logged_and_leaves_the_key_to_its_ttl(
                 assert acquired
 
         assert backend._scoped_key("k") + ":lock" in fake._store, "a failed release leaves the key for its TTL"
-        assert any("releasing" in r.getMessage() and r.levelno == logging.DEBUG for r in caplog.records)
+        assert [r.levelno for r in caplog.records if "release" in r.getMessage()] == [level]

From 09f7873d465bfba132fc356239a65029023daa7a Mon Sep 17 00:00:00 2001
From: Mark S <ray@insighttimer.com>
Date: Tue, 15 Sep 2026 08:06:18 +1000
Subject: [PATCH 5/5] =?UTF-8?q?docs(redis):=20correct=20acquire=5Flock=20n?=
 =?UTF-8?q?ote=20=E2=80=94=20executor=20drain,=20not=20to=5Fthread=20(LAB-?=
 =?UTF-8?q?3606)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/cachekit/backends/redis/provider.py | 16 +++++++++-------
 1 file changed, 9 insertions(+), 7 deletions(-)

diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py
index 9ef1592c..f97f316b 100644
--- a/src/cachekit/backends/redis/provider.py
+++ b/src/cachekit/backends/redis/provider.py
@@ -383,13 +383,15 @@ async def acquire_lock(
 
         Note:
             Each acquisition attempt is one non-blocking ``SET NX`` round-trip run via
-            ``asyncio.to_thread()``; the wait between attempts is an ``asyncio.sleep`` on
-            the event loop, never a sleep inside an executor thread. A blocking
-            ``Lock.acquire`` run via ``to_thread`` would pin one executor thread per waiter for
-            up to ``blocking_timeout``. The default executor has only ``min(32, cpu_count + 4)``
-            threads (8 when ``cpu_count`` is 4), so once concurrent misses on one key reach that size
-            the holder's own ``get``/``set``/``release`` — also ``to_thread`` calls — queue behind
-            the waiters, every waiter times out, and all of them recompute.
+            ``loop.run_in_executor()`` and drained through ``_await_uninterrupted`` (so a
+            cancellation cannot drop a round-trip that still completes); the wait between
+            attempts is an ``asyncio.sleep`` on the event loop, never a sleep inside an
+            executor thread. A blocking ``Lock.acquire`` run in the executor would pin one
+            executor thread per waiter for up to ``blocking_timeout``. The default executor
+            has only ``min(32, cpu_count + 4)`` threads (8 when ``cpu_count`` is 4), so once
+            concurrent misses on one key reach that size the holder's own
+            ``get``/``set``/``release`` — also executor calls — queue behind the waiters,
+            every waiter times out, and all of them recompute.
             Sets thread_local=False because attempts and release may run on different
             executor threads.
             Cancellation is drained, not raced: an in-flight attempt or release round-trip