fix(redis): stop lock waiters pinning executor threads (LAB-3596)

docs/features/distributed-locking.md

@@ -256,7 +256,10 @@ The decorator wrapper calls it with `timeout=30.0` (lock self-expiry) and
 ```
 1. Try to SET lock key (NX - only if not exists)
 2. If SET succeeds → lock acquired, yield True
-3. If SET fails → lock held, wait up to blocking_timeout
+3. If SET fails → lock held, retry every 0.1 s for up to blocking_timeout.
+   Each retry is one non-blocking SET NX; the wait between retries is an
+   asyncio.sleep on the event loop, so a waiter never holds an executor thread
+   while waiting between attempts
 4. On context exit: DEL lock key (only if still holder)
    Lock auto-expires via Redis TTL if holder crashes
 ```

src/cachekit/backends/redis/provider.py

@@ -352,32 +352,45 @@ async def acquire_lock(
             BackendError: If Redis operation fails
 
         Note:
-            Uses asyncio.to_thread() to run sync Redis lock operations without blocking event loop.
-            Sets thread_local=False to avoid thread-local token issues with thread pool.
+            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.
+            Sets thread_local=False because attempts and release may run on different
+            executor threads.
         """
         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
         # (the wrapper no longer pollutes the cache_key passed in).
         scoped_key = f"{self._scoped_key(key)}:lock"
-        lock = None
         try:
             from redis.lock import Lock
 
-            # Create Redis lock with tenant-scoped key
-            # CRITICAL: thread_local=False allows lock to work across thread pool
             lock = Lock(
                 self._client,
                 name=scoped_key,
                 timeout=timeout,
-                blocking_timeout=blocking_timeout if blocking_timeout is not None else 0,
-                thread_local=False,  # Disable thread-local storage for async/thread pool compatibility
+                thread_local=False,  # attempts and release may land on different executor threads
             )
 
-            # Run sync lock.acquire() in thread pool to avoid blocking event loop
-            acquired = await asyncio.to_thread(lock.acquire, blocking=blocking_timeout is not None)
+            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
+            while True:
+                acquired = await asyncio.to_thread(lock.acquire, blocking=False, token=token)
+                # 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:
+                    break
+                await asyncio.sleep(lock.sleep)
             try:
                 yield acquired
             finally:

tests/unit/backends/test_redis_backend.py

@@ -7,15 +7,27 @@
 Regression coverage for #154: the shared pools must use decode_responses=False so
 binary payloads (LZ4 / Arrow IPC / AES-256-GCM ciphertext) are never UTF-8 decoded,
 and RedisBackend.get() must return those raw bytes (or None) without coercion.
+
+Regression coverage for the distributed-lock executor stall: ``acquire_lock`` must
+not hold an executor thread while a waiter polls (see
+``TestRedisLockWaitersDoNotPinExecutorThreads``).
 """
 
 from __future__ import annotations
 
+import asyncio
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
 from unittest.mock import Mock, patch
 
 import pytest
+from redis.commands.core import Script
+from redis.connection import Encoder
+from redis.lock import Lock
 
 from cachekit.backends.redis import RedisBackend
+from cachekit.backends.redis.provider import PerRequestRedisBackend
 
 
 @pytest.mark.unit
@@ -250,3 +262,116 @@ def test_get_returns_none_for_non_bytes_response(self):
         # bytes|None narrowing guard must hold defensively (no str coercion).
         backend = self._backend_returning("unexpected-str")
         assert backend.get("k") is None
+
+
+class _FakeRedis:
+    """Just enough of ``redis.Redis`` for ``redis.lock.Lock``: SET NX PX plus the release script.
+
+    Guarded by a mutex because ``acquire_lock`` runs each attempt on an executor thread.
+    """
+
+    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
+
+    def get_encoder(self) -> Encoder:
+        return Encoder("utf-8", "strict", False)
+
+    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:
+        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
+
+    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."""
+        with self._mutex:
+            if self._store.get(name) != token:
+                return 0
+            del self._store[name]
+            return 1
+
+
+@pytest.mark.unit
+class TestRedisLockWaitersDoNotPinExecutorThreads:
+    """A lock waiter must not hold an executor thread while it waits.
+
+    ``acquire_lock`` used to run redis-py's *blocking* ``Lock.acquire`` inside
+    ``asyncio.to_thread``. With more concurrent misses on one key than the default
+    executor has threads (``min(32, cpu_count + 4)``, 8 when ``cpu_count`` is 4), every
+    thread sat in a polling loop, the holder's own ``get``/``set``/``release`` (also
+    ``to_thread`` calls) queued behind them, every waiter hit ``blocking_timeout`` and
+    recomputed — a stampede from the feature that exists to prevent one. This pins the
+    executor at 2 threads and runs 4 contenders: red on the blocking implementation
+    (two waiters time out), green when the wait happens on the event loop.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _fresh_release_script(self, monkeypatch):
+        # Lock caches its Script objects class-wide. An earlier test may have registered
+        # lua_release against a MagicMock client, whose "release" never deletes our key;
+        # reset it so register_scripts() binds it to this test's fake.
+        monkeypatch.setattr(Lock, "lua_release", None)
+
+    async def test_all_contenders_acquire_when_executor_is_smaller_than_contention(self):
+        asyncio.get_running_loop().set_default_executor(ThreadPoolExecutor(max_workers=2))
+        backend = PerRequestRedisBackend(_FakeRedis(), tenant_id="t")
+
+        async def contend() -> bool:
+            async with backend.acquire_lock("k", timeout=30.0, blocking_timeout=2.0) as acquired:
+                await asyncio.sleep(0.05)  # the holder's compute
+                return acquired
+
+        results = await asyncio.gather(*(contend() for _ in range(4)))
+        assert results == [True] * 4, f"waiters starved the executor and timed out: {results}"
+
+    async def test_waiter_gives_up_with_false_when_lock_is_held_past_its_window(self):
+        fake = _FakeRedis()
+        backend = PerRequestRedisBackend(fake, tenant_id="t")
+        holder_acquired = asyncio.Event()
+        holder_released = asyncio.Event()
+        blocking_timeout = 0.45
+
+        async def hold() -> None:
+            async with backend.acquire_lock("k", timeout=30.0, blocking_timeout=None) as acquired:
+                assert acquired is True
+                holder_acquired.set()
+                await asyncio.sleep(0.8)  # longer than the waiter's window
+            holder_released.set()
+
+        async def wait() -> tuple[bool, bool, float]:
+            await holder_acquired.wait()
+            started = time.monotonic()
+            async with backend.acquire_lock("k", timeout=30.0, blocking_timeout=blocking_timeout) as acquired:
+                return acquired, holder_released.is_set(), started
+
+        holder = asyncio.create_task(hold())
+        acquired, holder_had_released, started = await wait()
+        await holder
+
+        assert acquired is False
+        assert holder_had_released is False, "waiter must give up on its own deadline, not wait for the release"
+        waiter_attempts = [t - started for t in fake.nx_attempts if t >= started]
+        assert len(waiter_attempts) >= 2, f"a blocking waiter must retry before giving up: {waiter_attempts}"
+        # Contract: no attempt lands past the deadline. Scheduling jitter only ever delays an
+        # attempt, so the tolerance can hide a slightly late legitimate attempt but never an
+        # extra one — that would land a full lock.sleep (0.1 s) later.
+        assert max(waiter_attempts) <= blocking_timeout + 0.03, f"attempt past the deadline: {waiter_attempts}"
+
+    async def test_non_blocking_acquire_makes_exactly_one_attempt(self):
+        fake = _FakeRedis()
+        backend = PerRequestRedisBackend(fake, tenant_id="t")
+
+        async with backend.acquire_lock("k", timeout=30.0, blocking_timeout=None) as held:
+            assert held is True
+            async with backend.acquire_lock("k", timeout=30.0, blocking_timeout=None) as contended:
+                assert contended is False
+
+        assert len(fake.nx_attempts) == 2, "blocking_timeout=None must be a single SET NX per acquire_lock"

tests/unit/test_wrapper_lock_bare_key.py

@@ -276,7 +276,7 @@ def __init__(self, _client: Any, *, name: str, **_kwargs: Any) -> None:
                 """Record the lock name (the on-wire key) used to construct the lock."""
                 captured_lock_names.append(name)
 
-            def acquire(self, blocking: bool = True) -> bool:
+            def acquire(self, blocking: bool = True, token: Any = None) -> bool:
                 """Pretend acquisition always succeeds (no real Redis round-trip)."""
                 return True