Use CAS deletion in get_and_delete to prevent race conditions

CHANGELOG.md

@@ -153,12 +153,23 @@ Note that 0.3.3 was never released; 0.3.4 follows 0.3.2.
   like a miss, `set()` dropped the write, `increment()` returned 0 and
   `delete_if_equals()` raised `TypeError`. A failed server is now taken out of
   rotation at once and tried again after one second, instead of after 60.
-  ([#197](https://github.com/allen0099/FastAPI-CacheX/issues/197))- **`SessionConfig` warns when `cookie_same_site="none"` is set without
+  ([#197](https://github.com/allen0099/FastAPI-CacheX/issues/197))
+
+- **`SessionConfig` warns when `cookie_same_site="none"` is set without
   `cookie_https_only=True`.** Browsers reject a `SameSite=None` cookie that is
   not `Secure`, so the session cookie was silently never stored. The
   combination is still accepted.
   ([#167](https://github.com/allen0099/FastAPI-CacheX/issues/167))
 
+- **Memcached `get_and_delete()` uses CAS deletion to avoid deleting concurrent writes.**
+  The get-then-delete sequence allowed a concurrent writer to update the key
+  between the two calls, causing `get_and_delete()` to delete the new value
+  while returning the old one. It now issues `gets` and a `cas` write with
+  `exptime=-1`. If the key is updated before `cas` runs, the operation retries
+  to retrieve and remove the current value, matching Redis `GETDEL`, and raises
+  `CacheXError` if retries run out.
+  ([#175](https://github.com/allen0099/FastAPI-CacheX/issues/175))
+
 
 ## [0.3.7] - 2026-09-25
 

CLAUDE.md

@@ -86,7 +86,7 @@ Backend keys are namespaced automatically (default prefix: `fastapi_cachex:`).
 
 Four non-abstract atomic primitives live on the base class with non-atomic fallbacks, and every built-in backend overrides them (see `docs/BACKENDS.md` "Atomic backend primitives"):
 - `increment(key, delta=1, ttl=None) -> int`: fixed-window counter; `ttl` applies only when the counter is created. Redis runs a registered Lua script, Memcached uses `ADD` + `INCR`/`DECR`, memory works under its lock. A counter reads back through `get()` as a `CacheEntry` with `COUNTER_FINGERPRINT` (`types.py`).
-- `get_and_delete(key) -> CacheEntry | None`: one-shot retrieval (Redis `GETDEL`, Memcached get + `delete(noreply=False)` winner check). `StateManager.consume_state`, `delete_state`, `CacheManager.delete` and `invalidate()` use it. `delete()` keeps returning `None` for 0.3.x compatibility.
+- `get_and_delete(key) -> CacheEntry | None`: one-shot retrieval (Redis `GETDEL`, Memcached `gets` + `cas(..., exptime=-1)` with retry). `StateManager.consume_state`, `delete_state`, `CacheManager.delete` and `invalidate()` use it. `delete()` keeps returning `None` for 0.3.x compatibility.
 - `set_if_absent(key, value, ttl=None) -> bool`: claim-if-free for locks/slots. Redis `SET NX EX`, Memcached `ADD`, memory under its lock.
 - `delete_if_equals(key, expected) -> bool`: release only while the key still holds `expected` (compared as decoded `CacheEntry`). Redis compares in Python then deletes via a Lua script that re-checks the raw bytes; Memcached uses `GETS` + `CAS` with exptime `-1` (immediate expiry), since classic `DELETE` has no CAS.
 

docs/BACKENDS.md

@@ -196,8 +196,9 @@ if await backend.set_if_absent(f"stream:{user_id}", owner, ttl=300):
   response whose body is a number. A counter written with
   `set(key, counter_entry(n))` can be incremented on every backend.
 - `get_and_delete(key) -> CacheEntry | None` — Memory pops under its lock, Redis
-  uses `GETDEL` (server 6.2+) and Memcached returns the value only when its own
-  `DELETE` won. `StateManager.consume_state`, `StateManager.delete_state`,
+  uses `GETDEL` (server 6.2+) and Memcached uses `GETS` + a `CAS` write with
+  `exptime=-1` (retrying if another writer replaced the value in between).
+  `StateManager.consume_state`, `StateManager.delete_state`,
   `CacheManager.delete` and `invalidate()` are built on it.
 - `set_if_absent(key, value, ttl=None) -> bool` — stores `value` only when
   `key` does not exist (an expired key counts as absent) and reports whether it

docs/STATE.md

@@ -159,8 +159,8 @@ CacheXError
   Memcached. With `MemoryBackend` a state only exists in the process that created it, so an
   authorization callback that lands on a different worker fails.
 - **The one-time guarantee comes from the backend's atomic operation.** `get_and_delete()` is
-  `GETDEL` on Redis (requires Redis server 6.2 or newer), a get followed by
-  `delete(noreply=False)` where only the caller whose delete succeeded wins on Memcached,
+  `GETDEL` on Redis (requires Redis server 6.2 or newer), `gets` followed by
+  `cas(..., exptime=-1)` on Memcached (retrying if another writer replaced the value in between),
   and a `pop` under the lock on the memory backend.
   A custom backend that implements only the abstract methods falls back to
   `BaseCacheBackend`'s non-atomic version, so a concurrent replay could succeed on both

fastapi_cachex/backends/memcached.py

@@ -33,6 +33,9 @@
 # again. Until then every call to it raises.
 _DEAD_TIMEOUT = 1
 
+# Maximum retries for CAS operations when another writer replaces the value.
+_CAS_MAX_RETRIES = 16
+
 
 def _expiry(ttl: int | None) -> int:
     """Convert a TTL in seconds to the exptime Memcached expects.
@@ -164,23 +167,36 @@ async def set(self, key: str, value: CacheEntry, ttl: int | None = None) -> None
     async def get_and_delete(self, key: str) -> CacheEntry | None:
         """Atomically retrieve and remove a cached entry (see base class).
 
-        Memcached has no combined primitive, but DELETE is atomic: the value is
-        returned only when this call is the one that removed it, so exactly one
-        concurrent caller wins.
+        Uses ``gets`` and ``cas`` with negative ``exptime`` to ensure the entry
+        is deleted only if it was not modified after reading. If another writer
+        replaces the value between ``gets`` and ``cas``, the operation retries
+        up to a bounded number of times to return and remove the current value.
+
+        Raises:
+            CacheXError: If concurrent writes cause the CAS deletion to exceed
+                the maximum retry attempts.
         """
         prefixed_key = self._make_key(key)
-        raw = await asyncio.to_thread(self.client.get, prefixed_key)
-        if raw is None:
-            logger.debug("Memcached GET_AND_DELETE MISS; key=%s", key)
-            return None
-        deleted = await asyncio.to_thread(
-            self.client.delete, prefixed_key, noreply=False
-        )
-        if not deleted:
-            logger.debug("Memcached GET_AND_DELETE LOST RACE; key=%s", key)
-            return None
-        logger.debug("Memcached GET_AND_DELETE HIT; key=%s", key)
-        return decode_entry(raw)
+        for _ in range(_CAS_MAX_RETRIES):
+            raw, cas_token = await asyncio.to_thread(self.client.gets, prefixed_key)
+            if raw is None:
+                logger.debug("Memcached GET_AND_DELETE MISS; key=%s", key)
+                return None
+            cas_result = await asyncio.to_thread(
+                self.client.cas, prefixed_key, b"", cas_token, -1, noreply=False
+            )
+            if cas_result is True:
+                logger.debug("Memcached GET_AND_DELETE HIT; key=%s", key)
+                return decode_entry(raw)
+            if cas_result is None:
+                logger.debug(
+                    "Memcached GET_AND_DELETE LOST RACE (DELETED); key=%s", key
+                )
+                return None
+            logger.debug("Memcached GET_AND_DELETE RETRY; key=%s", key)
+
+        msg = f"Memcached get_and_delete exceeded {_CAS_MAX_RETRIES} attempts on key {key!r}"
+        raise CacheXError(msg)
 
     async def set_if_absent(
         self, key: str, value: CacheEntry, ttl: int | None = None

i18n/zh-TW/docs/BACKENDS.md

@@ -143,7 +143,7 @@ if await backend.set_if_absent(f"stream:{user_id}", owner, ttl=300):
 ```
 
 - `increment(key, delta=1, ttl=None) -> int`：記憶體後端在鎖內執行讀取—修改—寫入，Redis 執行 Lua 腳本（`EXISTS` + `INCRBY` + `EXPIRE`），Memcached 則使用 `ADD` + `INCR`/`DECR`（Memcached 的計數器最低停在 0）。計數器可透過 `get()` 讀到，形式為 fingerprint 為 `COUNTER_FINGERPRINT`、內容為十進位數值的 `CacheEntry`，因此 `delete`／`clear*` 與監控路由都會把它當成一般項目處理。對存放快取回應的鍵執行 increment 會拋出 `CacheXError`。
-- `get_and_delete(key) -> CacheEntry | None`：記憶體後端在鎖內 pop，Redis 使用 `GETDEL`（伺服器 6.2 以上），Memcached 則只在自己的 `DELETE` 勝出時才回傳該值。`StateManager.consume_state`、`StateManager.delete_state`、`CacheManager.delete` 與 `invalidate()` 都建立在它之上。
+- `get_and_delete(key) -> CacheEntry | None`：記憶體後端在鎖內 pop，Redis 使用 `GETDEL`（伺服器 6.2 以上），Memcached 使用 `GETS` + `exptime=-1` 的 `CAS` 寫入（若中間有其他寫入者替換了值則會重試）。`StateManager.consume_state`、`StateManager.delete_state`、`CacheManager.delete` 與 `invalidate()` 都建立在它之上。
 - `set_if_absent(key, value, ttl=None) -> bool`：只在 `key` 不存在時儲存 `value`（已過期的鍵視為不存在），並回報是否有寫入。記憶體後端在鎖內檢查，Redis 使用 `SET NX EX`，Memcached 使用 `ADD`。
 - `delete_if_equals(key, expected) -> bool`：只在 `key` 仍存放 `expected` 時才移除它，因此項目已過期的持有者無法釋放已被他人取得的鎖。請在你儲存的項目中放入唯一的權杖，並以同一個項目釋放。記憶體後端在鎖內比較，Redis 透過 Lua 腳本刪除，並在腳本中重新檢查先前比較過的值，Memcached 則使用 `GETS` + 一個讓項目立即過期的 `CAS` 寫入（傳統協定的 `DELETE` 不接受 CAS 權杖）。
 

i18n/zh-TW/docs/STATE.md

@@ -128,6 +128,6 @@ CacheXError
 ## 注意事項 {#notes}
 
 - **後端必須在多個行程之間共用。** 多 worker 部署請使用 Redis 或 Memcached。使用 `MemoryBackend` 時，state 只存在於建立它的行程中，因此落到其他 worker 的授權回呼會失敗。
-- **一次性保證來自後端的原子操作。** `get_and_delete()` 在 Redis 上是 `GETDEL`（需要 Redis 伺服器 6.2 或更新版本）；在 Memcached 上是先 get 再 `delete(noreply=False)`，只有 delete 成功的呼叫者勝出；在記憶體後端上則是在鎖內 `pop`。只實作抽象方法的自訂後端會退回使用 `BaseCacheBackend` 的非原子性版本，因此並行的重送可能兩邊都成功。這種情況請覆寫 `get_and_delete()`。
+- **一次性保證來自後端的原子操作。** `get_and_delete()` 在 Redis 上是 `GETDEL`（需要 Redis 伺服器 6.2 或更新版本）；在 Memcached 上是 `gets` 後接 `cas(..., exptime=-1)`（若中間有其他寫入者替換了值則會重試）；在記憶體後端上則是在鎖內 `pop`。只實作抽象方法的自訂後端會退回使用 `BaseCacheBackend` 的非原子性版本，因此並行的重送可能兩邊都成功。這種情況請覆寫 `get_and_delete()`。
 - 不要在 state 中存放敏感資料。`metadata` 會以明文 JSON 存放在快取後端中。
 - **日誌中絕不會出現 state 本身。** 來自 `fastapi_cachex.state.manager` 的日誌以 `state_ref` 識別 state，也就是其 SHA-256 的前 12 個十六進位字元；你可以從已知的 state 計算出它來比對。未知或已過期的 state 以 INFO 等級記錄，因為那是常見的用戶端輸入；格式錯誤的儲存資料則以 WARNING 等級記錄一次。

tests/backends/test_memcached.py

@@ -7,6 +7,8 @@
 import pytest_asyncio
 
 from fastapi_cachex.backends import MemcachedBackend
+from fastapi_cachex.backends.codec import encode_entry
+from fastapi_cachex.backends.memcached import _CAS_MAX_RETRIES
 from fastapi_cachex.backends.memcached import _DEAD_TIMEOUT
 from fastapi_cachex.exceptions import CacheXError
 from fastapi_cachex.lock import CacheLock
@@ -490,15 +492,103 @@ async def test_memcached_get_and_delete_returns_then_removes(
 
 @requires_memcached
 @pytest.mark.asyncio
-async def test_memcached_get_and_delete_loses_the_race_when_delete_fails(
+async def test_memcached_get_and_delete_loses_the_race_when_cas_reports_deletion(
     memcached_backend: MemcachedBackend, monkeypatch
 ) -> None:
     await memcached_backend.set("once", CacheEntry(fingerprint="e", content=b"x"), 60)
-    monkeypatch.setattr(memcached_backend.client, "delete", lambda *a, **kw: False)
+    monkeypatch.setattr(memcached_backend.client, "cas", lambda *a, **kw: None)
 
     assert await memcached_backend.get_and_delete("once") is None
 
 
+@pytest.mark.asyncio
+async def test_memcached_get_and_delete_retries_on_concurrent_write() -> None:
+    """When another writer updates the key before CAS, retry gets + cas."""
+    backend = stubbed_backend()
+    entry1 = CacheEntry(fingerprint="e1", content=b"first")
+    entry2 = CacheEntry(fingerprint="e2", content=b"second")
+
+    # First gets returns entry1 and token1; second gets returns entry2 and token2
+    backend.client.gets.side_effect = [
+        (encode_entry(entry1), b"101"),
+        (encode_entry(entry2), b"102"),
+    ]
+    # First cas fails (writer changed key); second cas succeeds
+    backend.client.cas.side_effect = [False, True]
+
+    result = await backend.get_and_delete("key")
+
+    assert result == entry2
+    assert backend.client.gets.call_count == 2
+    assert backend.client.cas.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_memcached_get_and_delete_returns_none_when_key_deleted_before_cas() -> (
+    None
+):
+    """When the key is deleted/expired between gets and cas, cas returns None."""
+    backend = stubbed_backend()
+    entry = CacheEntry(fingerprint="e", content=b"v")
+    backend.client.gets.return_value = (encode_entry(entry), b"101")
+    backend.client.cas.return_value = None
+
+    result = await backend.get_and_delete("key")
+
+    assert result is None
+    assert backend.client.gets.call_count == 1
+    assert backend.client.cas.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_memcached_get_and_delete_exhausts_retries_when_writes_continue() -> None:
+    """When writers keep replacing the value, stop after max retries and raise."""
+    backend = stubbed_backend()
+    entry = CacheEntry(fingerprint="e", content=b"v")
+    backend.client.gets.return_value = (encode_entry(entry), b"101")
+    backend.client.cas.return_value = False
+
+    with pytest.raises(
+        CacheXError,
+        match=f"Memcached get_and_delete exceeded {_CAS_MAX_RETRIES} attempts on key 'key'",
+    ):
+        await backend.get_and_delete("key")
+
+    assert backend.client.gets.call_count == _CAS_MAX_RETRIES
+    assert backend.client.cas.call_count == _CAS_MAX_RETRIES
+
+
+@requires_memcached
+@pytest.mark.asyncio
+async def test_memcached_get_and_delete_live_concurrent_write_regression(
+    memcached_backend: MemcachedBackend, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Simulate a concurrent write between gets and cas against live Memcached."""
+    initial = CacheEntry(fingerprint="e1", content=b"initial")
+    updated = CacheEntry(fingerprint="e2", content=b"updated")
+    await memcached_backend.set("race_key", initial, 60)
+
+    original_gets = memcached_backend.client.gets
+    overwrite_done = False
+
+    def gets_with_concurrent_write(key, *args, **kwargs):
+        nonlocal overwrite_done
+        res = original_gets(key, *args, **kwargs)
+        if not overwrite_done:
+            overwrite_done = True
+            # Simulate a concurrent writer replacing the value before cas runs
+            memcached_backend.client.set(key, encode_entry(updated), 60)
+        return res
+
+    monkeypatch.setattr(memcached_backend.client, "gets", gets_with_concurrent_write)
+
+    # get_and_delete should detect CAS mismatch on initial, retry, and return updated
+    result = await memcached_backend.get_and_delete("race_key")
+    assert result == updated
+    # Key should now be deleted from memcached
+    assert await memcached_backend.get("race_key") is None
+
+
 @requires_memcached
 @pytest.mark.asyncio
 async def test_memcached_get_and_delete_has_exactly_one_winner(