diff --git a/docs/features/distributed-locking.md b/docs/features/distributed-locking.md index 85bbf811..bec2efa3 100644 --- a/docs/features/distributed-locking.md +++ b/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 ``` diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py index 901f98e2..707b738e 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 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: ``: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..69ca320b 100644 --- a/tests/unit/backends/test_redis_backend.py +++ b/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" 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