feat: add CacheLock and the expire_if_equals backend primitive

CHANGELOG.md

@@ -17,6 +17,11 @@ Note that 0.3.3 was never released; 0.3.4 follows 0.3.2.
 
 ## [Unreleased]
 
+### Added
+
+- **`CacheLock`, a distributed lock built on backend primitives.** Usable as an async context manager or with direct `acquire`/`release`/`extend`/`locked` calls, raising `LockTimeoutError` on timeout. ([#64](https://github.com/allen0099/FastAPI-CacheX/issues/64))
+- **`expire_if_equals()` backend primitive for owner-checked TTL renewal.** Added to `BaseCacheBackend`, `MemoryBackend`, `AsyncRedisCacheBackend`, and `MemcachedBackend`. ([#64](https://github.com/allen0099/FastAPI-CacheX/issues/64))
+
 ### Changed
 
 - **GitHub release notes list one line per change.** Each changelog entry now

README.md

@@ -88,6 +88,7 @@ async def report(cache: AppCache):
 - [Backends](https://fastapi-cachex.readthedocs.io/en/latest/BACKENDS/) — choosing and configuring a backend, atomic primitives
 - [Session management](https://fastapi-cachex.readthedocs.io/en/latest/SESSION/) and [JWT claims](https://fastapi-cachex.readthedocs.io/en/latest/JWT_CLAIMS/)
 - [OAuth state](https://fastapi-cachex.readthedocs.io/en/latest/STATE/) — one-shot OAuth/CSRF state tokens
+- [Distributed lock](https://fastapi-cachex.readthedocs.io/en/latest/LOCK/) — `CacheLock` for multi-process mutual exclusion
 - [API reference](https://fastapi-cachex.readthedocs.io/en/latest/api/http-caching/)
 - [Development guide](https://fastapi-cachex.readthedocs.io/en/latest/DEVELOPMENT/) and [contributing](https://fastapi-cachex.readthedocs.io/en/latest/CONTRIBUTING/)
 - [Changelog](https://github.com/allen0099/FastAPI-CacheX/blob/master/CHANGELOG.md) · [Known limitations and planned work](https://github.com/allen0099/FastAPI-CacheX/issues)

docs/BACKENDS.md

@@ -161,8 +161,14 @@ if await backend.set_if_absent(f"stream:{user_id}", owner, ttl=300):
   through a Lua script that re-checks the value it compared, and Memcached uses
   `GETS` + a `CAS` write that expires the entry immediately (the classic
   protocol's `DELETE` takes no CAS token).
-
-All four have a non-atomic fallback on `BaseCacheBackend`, so a third-party backend
+- `expire_if_equals(key, expected, ttl) -> bool` — updates the TTL on `key` to `ttl`
+  seconds only while it still holds `expected`, so a long-running lock holder can
+  renew its lease without risking overwriting someone else's lock if it expired.
+  Memory updates under its lock, Redis compares in Python then executes a Lua
+  script (`GET` compare + `EXPIRE`), and Memcached uses `GETS` + `CAS` writing the
+  same bytes with the new exptime (`TOUCH` takes no CAS token).
+
+All five have a non-atomic fallback on `BaseCacheBackend`, so a third-party backend
 that only implements the abstract methods keeps working; override them to get
 real atomicity.
 

docs/LOCK.md

@@ -0,0 +1,40 @@
+# Distributed Lock
+
+`CacheLock` provides a distributed lock helper built on backend atomic primitives (`set_if_absent`, `delete_if_equals`, `expire_if_equals`). It guarantees mutual exclusion across multiple processes or containers sharing the same cache backend.
+
+```python
+from fastapi import HTTPException
+from fastapi_cachex import CacheLock, LockTimeoutError
+
+# Usage as an async context manager:
+async with CacheLock(f"report:{report_id}", ttl=30):
+    ...  # only one process holds the lock at any time
+
+# Explicit acquire and release calls:
+lock = CacheLock(f"stream:{user_id}", ttl=60)
+if not await lock.acquire(blocking=False):
+    raise HTTPException(409, detail="Lock already held")
+try:
+    ...
+    await lock.extend(60)  # long-running tasks renew before expiry
+finally:
+    await lock.release()
+```
+
+## Behavior
+
+- **Safety & Token Ownership**: Each `CacheLock` instance generates a unique token (`secrets.token_hex(16)`) stored inside a `CacheEntry`. Releases (`release()`) and extensions (`extend()`) use owner-checked backend primitives (`delete_if_equals` and `expire_if_equals`), so a holder whose lock expired cannot release or renew a lock claimed by someone else.
+- **Blocking & Non-blocking Modes**:
+  - Non-blocking (`acquire(blocking=False)`): Performs a single atomic `set_if_absent` and immediately returns `True` if acquired or `False` if held.
+  - Blocking (`acquire(blocking=True, timeout=None, poll_interval=0.1)`): Retries at `poll_interval` seconds until acquired or until `timeout` seconds elapse. The default `timeout=None` makes a blocking `acquire()` (and `async with CacheLock(...)`) wait indefinitely until the lock becomes free. If a finite `timeout` is reached, `acquire()` returns `False`.
+- **Context Manager Timeouts**: Entering a context manager (`async with CacheLock(...)`) invokes `acquire()`. If acquisition fails or times out, it raises `LockTimeoutError`.
+- **TTL Expiration**: If a task takes longer than its `ttl` and fails to renew, the lock entry expires in the backend and becomes free. Another process or container can then acquire the lock while the original code is still running. Subsequent calls to `extend()` or `release()` by the original holder will safely return `False` without throwing an error. Always choose a `ttl` longer than the expected work, or call `extend()` periodically during long-running operations.
+- **TTL Renewal (`extend`)**: `extend(ttl)` updates the key's TTL only while the lock is still owned by this holder instance, preventing race conditions on expired locks.
+- **One Instance per Acquisition Rule**: A single `CacheLock` instance tracks its active ownership state. Re-entering or sharing a single `CacheLock` instance across concurrent tasks raises a `RuntimeError`. Instantiate a new `CacheLock` instance for each acquisition.
+- **Namespace**: Lock keys live under their own `lock:` prefix by default (e.g. `lock:report:123`), separate from `cache:` and `oauth_state:`.
+
+> [!NOTE]
+> `CacheLock` works on all built-in backends (`MemoryBackend`, `AsyncRedisCacheBackend`, `MemcachedBackend`).
+> Redis executes Lua scripts for atomic operations, Memcached uses `ADD` and `CAS`, and the memory backend operates under its internal lock.
+
+The full class signature and options are documented in the [API reference](api/lock.md).

docs/api/lock.md

@@ -0,0 +1,5 @@
+# Distributed lock
+
+Distributed locking using backend atomic primitives.
+
+::: fastapi_cachex.lock.CacheLock

fastapi_cachex/__init__.py

@@ -11,6 +11,8 @@
 from .dependencies import CacheBackend as CacheBackend
 from .dependencies import get_app_cache as get_app_cache
 from .dependencies import get_cache_backend as get_cache_backend
+from .exceptions import LockTimeoutError as LockTimeoutError
+from .lock import CacheLock as CacheLock
 from .manager import CacheManager as CacheManager
 from .manager_proxy import CacheManagerProxy as CacheManagerProxy
 from .proxy import BackendProxy as BackendProxy
@@ -69,10 +71,12 @@ def _read_version() -> str:
     "BackendProxy",
     "CacheBackend",
     "CacheKeyBuilder",
+    "CacheLock",
     "CacheManager",
     "CacheManagerProxy",
     "FastAPICacheXSessionMiddleware",
     "InvalidStateError",
+    "LockTimeoutError",
     "Session",
     "SessionConfig",
     "SessionError",

fastapi_cachex/backends/base.py

@@ -159,6 +159,32 @@ async def delete_if_equals(self, key: str, expected: CacheEntry) -> bool:
         await self.delete(key)
         return True
 
+    async def expire_if_equals(self, key: str, expected: CacheEntry, ttl: int) -> bool:
+        """Update expiry on ``key`` to ``ttl`` seconds only while it still holds ``expected``.
+
+        Re-setting a lock's TTL with a plain ``set`` is unsafe: if the holder's
+        entry expired and someone else claimed the key in the meantime, a plain
+        ``set`` overwrites the new holder's entry. Comparing against the value
+        the caller stored makes the renewal a no-op in that case.
+
+        The base implementation is a best-effort, NON-atomic get-compare-set
+        fallback for third-party backends; the built-in backends override it
+        with an atomic implementation.
+
+        Args:
+            key: Cache key to update expiry for
+            expected: The entry the caller stored (compared with ``==``)
+            ttl: Time to live in seconds
+
+        Returns:
+            Whether the expiry was updated
+        """
+        validate_ttl(ttl)
+        if await self.get(key) != expected:
+            return False
+        await self.set(key, expected, ttl=ttl)
+        return True
+
     async def increment(self, key: str, delta: int = 1, ttl: int | None = None) -> int:
         """Atomically add ``delta`` to the integer counter stored at ``key``.
 

fastapi_cachex/backends/memcached.py

@@ -215,6 +215,29 @@ async def delete_if_equals(self, key: str, expected: CacheEntry) -> bool:
         )
         return bool(deleted)
 
+    async def expire_if_equals(self, key: str, expected: CacheEntry, ttl: int) -> bool:
+        """Atomically update expiry on ``key`` while it holds ``expected`` (see base class).
+
+        TOUCH in the Memcached protocol takes no CAS token, so the renewal is a
+        CAS write with the same bytes read by GETS and the new exptime.
+        """
+        validate_ttl(ttl)
+        prefixed_key = self._make_key(key)
+        raw, cas_token = await asyncio.to_thread(self.client.gets, prefixed_key)
+        if raw is None or decode_entry(raw) != expected:
+            logger.debug("Memcached EXPIRE_IF_EQUALS MISMATCH; key=%s", key)
+            return False
+        updated = await asyncio.to_thread(
+            self.client.cas, prefixed_key, raw, cas_token, _expiry(ttl), noreply=False
+        )
+        logger.debug(
+            "Memcached EXPIRE_IF_EQUALS %s; key=%s ttl=%s",
+            "HIT" if updated else "LOST RACE",
+            key,
+            ttl,
+        )
+        return bool(updated)
+
     def _add_delta(self, prefixed_key: str, delta: int) -> int | None:
         """Apply ``delta`` with INCR/DECR; ``None`` when the key does not exist."""
         if delta < 0:

fastapi_cachex/backends/memory.py

@@ -191,6 +191,21 @@ async def delete_if_equals(self, key: str, expected: CacheEntry) -> bool:
             logger.debug("Memory cache DELETE_IF_EQUALS HIT; key=%s", key)
             return True
 
+    async def expire_if_equals(self, key: str, expected: CacheEntry, ttl: int) -> bool:
+        """Atomically update expiry on ``key`` while it holds ``expected`` (see base class)."""
+        validate_ttl(ttl)
+        async with self.lock:
+            item = self.cache.get(key)
+            if item is None or not _is_live(item, time.time()):
+                logger.debug("Memory cache EXPIRE_IF_EQUALS MISS; key=%s", key)
+                return False
+            if item.value != expected:
+                logger.debug("Memory cache EXPIRE_IF_EQUALS MISMATCH; key=%s", key)
+                return False
+            item.expiry = time.time() + ttl
+            logger.debug("Memory cache EXPIRE_IF_EQUALS HIT; key=%s ttl=%s", key, ttl)
+            return True
+
     async def increment(self, key: str, delta: int = 1, ttl: int | None = None) -> int:
         """Atomically add ``delta`` to the counter at ``key`` (see base class).
 

fastapi_cachex/backends/redis.py

@@ -60,6 +60,15 @@ def _escape_glob(text: str) -> str:
 return 0
 """
 
+# EXPIRE that only fires while the key still holds the exact bytes the caller read,
+# so a value replaced in the meantime survives. KEYS[1] = key, ARGV[1] = bytes, ARGV[2] = ttl.
+_EXPIRE_IF_EQUALS_SCRIPT = """
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+    return redis.call('EXPIRE', KEYS[1], ARGV[2])
+end
+return 0
+"""
+
 
 class AsyncRedisCacheBackend(BaseCacheBackend):
     """Async Redis cache backend implementation.
@@ -135,6 +144,9 @@ def __init__(
         self._delete_if_equals_script = self.client.register_script(
             _DELETE_IF_EQUALS_SCRIPT
         )
+        self._expire_if_equals_script = self.client.register_script(
+            _EXPIRE_IF_EQUALS_SCRIPT
+        )
 
     @staticmethod
     def load_from_config(config: RedisConfig) -> "AsyncRedisCacheBackend":
@@ -259,6 +271,30 @@ async def delete_if_equals(self, key: str, expected: CacheEntry) -> bool:
         )
         return bool(deleted)
 
+    async def expire_if_equals(self, key: str, expected: CacheEntry, ttl: int) -> bool:
+        """Atomically update expiry on ``key`` while it holds ``expected`` (see base class).
+
+        The stored value is decoded and compared here, then a Lua script
+        updates the expiry on the key only if it still holds the bytes that
+        were compared, so a value written in between is never overwritten.
+        """
+        validate_ttl(ttl)
+        prefixed_key = self._make_key(key)
+        raw = await self.client.get(prefixed_key)
+        if raw is None or decode_entry(raw) != expected:
+            logger.debug("Redis EXPIRE_IF_EQUALS MISMATCH; key=%s", key)
+            return False
+        updated = await self._expire_if_equals_script(
+            keys=[prefixed_key], args=[raw, ttl]
+        )
+        logger.debug(
+            "Redis EXPIRE_IF_EQUALS %s; key=%s ttl=%s",
+            "HIT" if updated else "LOST RACE",
+            key,
+            ttl,
+        )
+        return bool(updated)
+
     async def increment(self, key: str, delta: int = 1, ttl: int | None = None) -> int:
         """Atomically add ``delta`` to the counter at ``key`` (see base class).
 

fastapi_cachex/exceptions.py

@@ -15,3 +15,7 @@ class BackendNotFoundError(CacheXError):
 
 class RequestNotFoundError(CacheXError):
     """Exception raised when a request is not found."""
+
+
+class LockTimeoutError(CacheXError):
+    """Exception raised when acquiring a lock times out."""