From d34cd911e044a5295e4f9b9f8ab8e80dc96dc3ed Mon Sep 17 00:00:00 2001 From: Mark S Date: Mon, 14 Sep 2026 15:08:21 +1000 Subject: [PATCH 1/2] fix(redis): stop lock waiters pinning executor threads (LAB-3596) PerRequestRedisBackend.acquire_lock ran redis-py's blocking Lock.acquire inside asyncio.to_thread, so every waiter held an executor thread for up to blocking_timeout. The default executor has min(32, cpu_count + 4) threads - 8 on a 4-vCPU GitHub-hosted runner - so ten concurrent misses on one cold key pinned all of them; the holder's own get/set/release (also to_thread calls) queued behind the waiters, every waiter timed out at 5 s and recomputed. test_concurrent_access in tests/integration/test_redis_integration.py failed on 4 of 5 matrix versions with 9 unique results and 8 lock-timeout warnings. Each acquisition attempt is now one non-blocking SET NX round-trip via to_thread, paced by asyncio.sleep on the event loop, with the same give-up rule as redis-py's own Lock.acquire. No executor thread is held across a wait. Regression test (runs on pull requests): tests/unit/backends/test_redis_backend.py::TestRedisLockWaitersDoNotPinExecutorThreads pins the default executor at 2 threads and runs 4 contenders - red on the old implementation, green on this one. --- docs/features/distributed-locking.md | 4 +- src/cachekit/backends/redis/provider.py | 31 +++++++--- tests/unit/backends/test_redis_backend.py | 75 +++++++++++++++++++++++ tests/unit/test_wrapper_lock_bare_key.py | 2 +- 4 files changed, 101 insertions(+), 11 deletions(-) diff --git a/docs/features/distributed-locking.md b/docs/features/distributed-locking.md index 85bbf811..8c96d353 100644 --- a/docs/features/distributed-locking.md +++ b/docs/features/distributed-locking.md @@ -256,7 +256,9 @@ 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 4. On context exit: DEL lock key (only if still holder) Lock auto-expires via Redis TTL if holder crashes ``` diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py index 901f98e2..0fa7301e 100644 --- a/src/cachekit/backends/redis/provider.py +++ b/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 on a 4-vCPU host), 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: ``: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: diff --git a/tests/unit/backends/test_redis_backend.py b/tests/unit/backends/test_redis_backend.py index 6c41f060..8a9df24b 100644 --- a/tests/unit/backends/test_redis_backend.py +++ b/tests/unit/backends/test_redis_backend.py @@ -7,15 +7,26 @@ 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 +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 +261,67 @@ 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() + + 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 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 on a 4-vCPU CI host), 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. + """ + + async def test_all_contenders_acquire_when_executor_is_smaller_than_contention(self, monkeypatch): + # Lock caches its Script objects on the class. An earlier test may have registered + # them against a MagicMock client, whose "release" would never delete our key. + for attr in ("lua_release", "lua_extend", "lua_reacquire"): + monkeypatch.setattr(Lock, attr, None) + + 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}" diff --git a/tests/unit/test_wrapper_lock_bare_key.py b/tests/unit/test_wrapper_lock_bare_key.py index 85c34471..68cf1349 100644 --- a/tests/unit/test_wrapper_lock_bare_key.py +++ b/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 From bfbb45977d9edda13f22445a4beda57a654308b8 Mon Sep 17 00:00:00 2001 From: Mark S Date: Mon, 14 Sep 2026 15:35:00 +1000 Subject: [PATCH 2/2] test(redis): lock give-up path coverage and wording from review (LAB-3596) - assert a waiter yields False on its own deadline while the holder still holds, retries before giving up, and never attempts past blocking_timeout - assert blocking_timeout=None is a single SET NX - reset only Lock.lua_release between tests; extend/reacquire are unused here - docs: a waiter holds no executor thread while waiting between attempts; each SET NX round-trip still uses one briefly - neutral wording for the executor-size example in both docstrings --- docs/features/distributed-locking.md | 1 + src/cachekit/backends/redis/provider.py | 4 +- tests/unit/backends/test_redis_backend.py | 62 ++++++++++++++++++++--- 3 files changed, 59 insertions(+), 8 deletions(-) diff --git a/docs/features/distributed-locking.md b/docs/features/distributed-locking.md index 8c96d353..bec2efa3 100644 --- a/docs/features/distributed-locking.md +++ b/docs/features/distributed-locking.md @@ -259,6 +259,7 @@ The decorator wrapper calls it with `timeout=30.0` (lock self-expiry) and 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 ``` diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py index 0fa7301e..707b738e 100644 --- a/src/cachekit/backends/redis/provider.py +++ b/src/cachekit/backends/redis/provider.py @@ -357,8 +357,8 @@ async def acquire_lock( 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 on a 4-vCPU host), so once concurrent misses on one key reach that size the - holder's own ``get``/``set``/``release`` — also ``to_thread`` calls — queue behind + 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. diff --git a/tests/unit/backends/test_redis_backend.py b/tests/unit/backends/test_redis_backend.py index 8a9df24b..69ca320b 100644 --- a/tests/unit/backends/test_redis_backend.py +++ b/tests/unit/backends/test_redis_backend.py @@ -17,6 +17,7 @@ import asyncio import threading +import time from concurrent.futures import ThreadPoolExecutor from unittest.mock import Mock, patch @@ -272,6 +273,7 @@ class _FakeRedis: 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) @@ -281,6 +283,8 @@ def register_script(self, script: str) -> 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 @@ -301,7 +305,7 @@ class TestRedisLockWaitersDoNotPinExecutorThreads: ``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 on a 4-vCPU CI host), every + 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 @@ -309,12 +313,14 @@ class TestRedisLockWaitersDoNotPinExecutorThreads: (two waiters time out), green when the wait happens on the event loop. """ - async def test_all_contenders_acquire_when_executor_is_smaller_than_contention(self, monkeypatch): - # Lock caches its Script objects on the class. An earlier test may have registered - # them against a MagicMock client, whose "release" would never delete our key. - for attr in ("lua_release", "lua_extend", "lua_reacquire"): - monkeypatch.setattr(Lock, attr, None) + @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") @@ -325,3 +331,47 @@ async def contend() -> bool: 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"