feat(session)!: make ordinary session saves conditional

changelog.d/128.added.md

@@ -0,0 +1,4 @@
+**`set_if_equals(key, expected, value, ttl=None)` on every backend.** It stores
+`value` only while `key` still holds `expected`, atomically on the memory,
+Redis and Memcached backends. `BaseCacheBackend` provides a non-atomic
+fallback for custom backends.

changelog.d/128.changed.2.md

@@ -0,0 +1,8 @@
+**Rotating a session ID refuses a session that ended meanwhile.**
+`regenerate_session_id()` removes the old record atomically and raises
+`SessionNotFoundError` or `SessionInvalidError`, storing nothing under a new
+ID, if another request deleted, invalidated or rotated the session since it
+was read. `rotate_session_id()` answers `401` in that case and sends no token,
+and `login()` starts a new session for the user without the ended session's
+data. See the
+[migration guide](https://fastapi-cachex.readthedocs.io/en/stable/MIGRATING_0_4/#session-writes).

changelog.d/128.changed.md

@@ -0,0 +1,9 @@
+**Session saves are conditional.** The middleware's save of `request.session`
+changes, sliding renewal in `get_session()` and `update_session()` store a
+session only while the backend still holds what the request last read or
+wrote, so a request cannot bring back a session that another request deleted,
+invalidated or rotated. A dropped save is logged and the response is sent
+without a session token, unless the request already stored a renewal for a
+session that is still valid. When two requests change the same session at once,
+the first save wins. `update_session()` returns `True` or `False`. See the
+[migration guide](https://fastapi-cachex.readthedocs.io/en/stable/MIGRATING_0_4/#session-writes).

docs/BACKENDS.md

@@ -325,16 +325,23 @@ if await backend.set_if_absent(f"stream:{user_id}", owner, ttl=300):
   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
+- `set_if_equals(key, expected, value, ttl=None) -> bool` — stores `value` only
+  while `key` still holds `expected`: a compare-and-set that fails if anything
+  changed, deleted or expired the key since the caller read it. Sessions save
+  through it (see [Session writes](MIGRATING_0_4.md#session-writes)). Memory
+  compares under its lock, Redis compares in Python then runs a Lua script
+  (`GET` compare + `SET`, with `EX` when `ttl` is set), and Memcached uses
+  `GETS` + a `CAS` write of the new value.
+
+All six 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.
 
 Complete runnable example: [`examples/rate_limit.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/rate_limit.py).
 
 ## TTL values
 
-Every `ttl` argument (`set`, `set_if_absent`, `increment`, and the `CacheManager`
+Every `ttl` argument (`set`, `set_if_absent`, `set_if_equals`, `increment`, and the `CacheManager`
 and `StateManager` methods and defaults built on them) is either `None`, meaning
 the entry never expires, or an `int` number of seconds from 1 up to `MAX_TTL`
 (2**31 - 1, about 68 years). The checks run before any backend I/O:

docs/MIGRATING_0_4.md

@@ -185,9 +185,13 @@ SessionConfig(
 
 0.4.0 writes sessions conditionally ([#128](https://github.com/allen0099/FastAPI-CacheX/issues/128)): a request that loaded a session before another request deleted, invalidated or rotated it can no longer bring the record back when it saves. No code change is needed.
 
-- Ordinary saves become conditional: the middleware's save of `request.session` changes, sliding renewal and `update_session()`. Each succeeds only while the stored record still equals what this request last read or wrote. Deleting, invalidating, expiring and rotating stay unconditional, so a security action always wins.
-- A rejected save is dropped and logged. The response is still sent, without a session token.
+- Ordinary saves become conditional: the middleware's save of `request.session` changes, sliding renewal and `update_session()`. Each succeeds only while the stored record still equals what this request last read or wrote. Deleting, invalidating and expiring stay unconditional, so a security action always wins.
+- A rejected save is dropped and logged. The response is still sent, without a session token. The exception is a renewal that this request already stored: while the session is still valid (the save lost only to another save), the renewed token is sent, so a JWT client does not keep a token that expires before the record.
+- Rotating the ID (`regenerate_session_id()`) removes the old record atomically and goes ahead only if that record was still valid, so a copy read before another request deleted, invalidated or rotated the session cannot come back under a new ID. Otherwise it raises `SessionNotFoundError` or `SessionInvalidError` and stores nothing under a new ID (the old record is removed either way), and of two concurrent rotations of one session only the first succeeds. A change saved by another request in between does not stop the rotation. `rotate_session_id()` answers `401` in that case and sends no token, and `login()` starts a new session for the user instead, without the data of the ended one. A `Session` built by hand, never read from or written to the backend, is rotated as before.
 - **Side effect:** when two requests change the same session at the same time (two tabs adding to a cart), the first save wins and the second is dropped. Today the last save wins, so one of the two changes is already lost; 0.4.0 changes which one. Merging such changes is tracked in [#376](https://github.com/allen0099/FastAPI-CacheX/issues/376).
+- `update_session()` now returns `True` when the session was stored and `False` when the save was dropped. A `Session` built by hand, never read from the backend, is stored only if no record exists under its ID yet.
+- When `get_session()` renews a session (sliding expiry) and the renewal loses a race with another save, it reads the session again and retries once. If it loses again, the request keeps the session it read and no renewed token is sent.
+- `Session` equality ignores which backend entry each copy last read or wrote, so a loaded session still equals one built with the same fields.
 - The backend gains `set_if_equals(key, expected, value, ttl=None)`, next to `delete_if_equals` and `expire_if_equals`. The base class provides a non-atomic fallback, so a custom backend keeps working unchanged; override it to make the save atomic.
 
 ### Token sources {#token-source-priority}

docs/SESSION.md

@@ -121,8 +121,12 @@ settings from `REDIS_HOST`, `REDIS_PORT`, `REDIS_DB` and `REDIS_PASSWORD`.
 ```
 <!-- fmt:on -->
 
-Changes made to a `Session` object inside a handler (flash messages, `session.data`,
-`session.user`) are only persisted when you call `session_manager.update_session(session)`.
+Changes made to a `Session` object inside a handler (flash messages, `session.data`) are only
+persisted when you call `session_manager.update_session(session)`. The save is conditional: it
+stores the session only while the backend still holds what this object last read or wrote, and
+returns `False` otherwise. A session that another request deleted, invalidated or rotated in the
+meantime is not brought back, and of two requests that change the same session at the same time,
+the first save wins (see [Session writes](MIGRATING_0_4.md#session-writes)).
 
 `delete_user_sessions()` and `clear_expired_sessions()` enumerate every key in the backend via
 `get_all_keys()` and load each session under `backend_key_prefix`, so their cost grows with the
@@ -181,7 +185,10 @@ async def me(session=Depends(require_user_session)):
   (`SessionManager.create_anonymous_session()`, with IP/User-Agent bindings applied as
   configured) and sends its token back through the request's transport.
 - Modifying it on a loaded session saves the new contents to the backend via `update_session()`,
-  replacing `Session.data` with the dict's contents.
+  replacing `Session.data` with the dict's contents. If another request deleted, invalidated or
+  rotated the session, or saved it first, while this one ran, the save is dropped and logged. The
+  response sends no session token, except a renewed one that this request already stored for a
+  session that is still valid.
 - Clearing it (`request.session.clear()`) on a loaded session logs out: the backend session is
   deleted even if its data was already empty, and a cookie client also receives a `Set-Cookie`
   that expires the cookie. Keys written after `clear()` in the same request go into a new
@@ -564,12 +571,16 @@ new ID, keeping its data, user, `created_at` and expiry. Either middleware sees
 and sends a token for it through the transport the request used: `Set-Cookie` for a
 cookie, the response header for a header token. After that the old token no longer
 resolves to a session. For a new visitor there is no session to rotate, so it returns
-`False`.
+`False`. If another request deleted, invalidated or rotated the session while this one ran,
+nothing is saved or sent and it answers `401`; `login()` starts a new session for the user instead (see
+[Session writes](MIGRATING_0_4.md#session-writes)).
 
 A handler that already holds the request's session object can call
-`await manager.regenerate_session_id(session)` directly, with the same effect. Get it from
-`get_optional_session` and skip the call when it is `None`; `SessionDep` answers `401` to a
-visitor who has no session yet.
+`await manager.regenerate_session_id(session)` directly. Get it from `get_optional_session` and
+skip the call when it is `None`; `SessionDep` answers `401` to a visitor who has no session yet.
+Unlike `rotate_session_id()`, a direct call raises `SessionNotFoundError` or `SessionInvalidError`
+when another request ended the session meanwhile, so catch `SessionError` and answer as for a
+missing session.
 
 Outside a middleware, load the session with the same bindings the middleware would pass, and hand
 the returned token to the client yourself:

fastapi_cachex/backends/base.py

@@ -255,6 +255,39 @@ async def expire_if_equals(self, key: str, expected: CacheEntry, ttl: int) -> bo
         await self.set(key, expected, ttl=ttl)
         return True
 
+    async def set_if_equals(
+        self,
+        key: str,
+        expected: CacheEntry,
+        value: CacheEntry,
+        ttl: int | None = None,
+    ) -> bool:
+        """Store ``value`` only while ``key`` still holds ``expected``.
+
+        A compare-and-set: a caller that read ``expected`` earlier overwrites
+        it only if nothing changed, deleted or expired the key since. Sessions
+        save through it, so a request that loaded a session cannot bring it
+        back after another request deleted or invalidated it.
+
+        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 overwrite
+            expected: The entry the caller last read or wrote (compared with ``==``)
+            value: Entry to store in its place
+            ttl: Time to live in seconds (``None`` = never expires)
+
+        Returns:
+            Whether ``value`` was stored
+        """
+        validate_ttl(ttl)
+        if await self.get(key) != expected:
+            return False
+        await self.set(key, value, 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

@@ -331,6 +331,50 @@ def _expire_if_equals(
         )
         return bool(updated)
 
+    async def set_if_equals(
+        self,
+        key: str,
+        expected: CacheEntry,
+        value: CacheEntry,
+        ttl: int | None = None,
+    ) -> bool:
+        """Atomically store ``value`` while ``key`` holds ``expected`` (see base class).
+
+        GETS reads the value to compare along with its CAS token, and the CAS
+        write succeeds only if nothing wrote or deleted the key since.
+        """
+        validate_ttl(ttl)
+        # Converted up front so a ttl Memcached cannot store fails before I/O.
+        exptime = _expiry(ttl)
+        return await asyncio.to_thread(
+            self._set_if_equals, key, expected, value, ttl, exptime
+        )
+
+    def _set_if_equals(
+        self,
+        key: str,
+        expected: CacheEntry,
+        value: CacheEntry,
+        ttl: int | None,
+        exptime: int,
+    ) -> bool:
+        """Run ``set_if_equals``'s GETS, compare and CAS in one worker thread."""
+        prefixed_key = self._make_key(key)
+        raw, cas_token = self.client.gets(prefixed_key)
+        if raw is None or decode_entry(raw) != expected:
+            logger.debug("Memcached SET_IF_EQUALS MISMATCH; key=%s", key)
+            return False
+        stored = self.client.cas(
+            prefixed_key, encode_entry(value), cas_token, exptime, noreply=False
+        )
+        logger.debug(
+            "Memcached SET_IF_EQUALS %s; key=%s ttl=%s",
+            "HIT" if stored else "LOST RACE",
+            key,
+            ttl,
+        )
+        return bool(stored)
+
     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

@@ -240,6 +240,29 @@ async def expire_if_equals(self, key: str, expected: CacheEntry, ttl: int) -> bo
             logger.debug("Memory cache EXPIRE_IF_EQUALS HIT; key=%s ttl=%s", key, ttl)
             return True
 
+    async def set_if_equals(
+        self,
+        key: str,
+        expected: CacheEntry,
+        value: CacheEntry,
+        ttl: int | None = None,
+    ) -> bool:
+        """Atomically store ``value`` while ``key`` holds ``expected`` (see base class)."""
+        validate_ttl(ttl)
+        async with self.lock:
+            now = time.time()
+            item = self.cache.get(key)
+            if item is None or not _is_live(item, now):
+                logger.debug("Memory cache SET_IF_EQUALS MISS; key=%s", key)
+                return False
+            if item.value != expected:
+                logger.debug("Memory cache SET_IF_EQUALS MISMATCH; key=%s", key)
+                return False
+            expiry = now + ttl if ttl is not None else None
+            self.cache[key] = CacheItem(value=value, expiry=expiry)
+            logger.debug("Memory cache SET_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

@@ -62,6 +62,21 @@
 return 0
 """
 
+# SET that only fires while the key still holds the exact bytes the caller read,
+# so a value replaced or deleted in the meantime stays as it is. KEYS[1] = key,
+# ARGV[1] = bytes read, ARGV[2] = bytes to store, ARGV[3] = ttl (0 = none).
+_SET_IF_EQUALS_SCRIPT = """
+if redis.call('GET', KEYS[1]) ~= ARGV[1] then
+    return 0
+end
+if tonumber(ARGV[3]) > 0 then
+    redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[3])
+else
+    redis.call('SET', KEYS[1], ARGV[2])
+end
+return 1
+"""
+
 
 # Constructor keywords that 0.4.0 removed. They would otherwise fall through
 # **kwargs to redis-py and change how keys and replies are encoded.
@@ -174,6 +189,7 @@ def __init__(
         self._expire_if_equals_script = self.client.register_script(
             _EXPIRE_IF_EQUALS_SCRIPT
         )
+        self._set_if_equals_script = self.client.register_script(_SET_IF_EQUALS_SCRIPT)
 
     @staticmethod
     def load_from_config(config: RedisConfig) -> "AsyncRedisCacheBackend":
@@ -374,6 +390,36 @@ async def expire_if_equals(self, key: str, expected: CacheEntry, ttl: int) -> bo
         )
         return bool(updated)
 
+    async def set_if_equals(
+        self,
+        key: str,
+        expected: CacheEntry,
+        value: CacheEntry,
+        ttl: int | None = None,
+    ) -> bool:
+        """Atomically store ``value`` while ``key`` holds ``expected`` (see base class).
+
+        The stored value is decoded and compared here, then a Lua script
+        writes the key only if it still holds the bytes that were compared,
+        so a value written or deleted 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 SET_IF_EQUALS MISMATCH; key=%s", key)
+            return False
+        stored = await self._set_if_equals_script(
+            keys=[prefixed_key], args=[raw, encode_entry(value), ttl or 0]
+        )
+        logger.debug(
+            "Redis SET_IF_EQUALS %s; key=%s ttl=%s",
+            "HIT" if stored else "LOST RACE",
+            key,
+            ttl,
+        )
+        return bool(stored)
+
     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/session/dependencies.py

@@ -14,6 +14,7 @@
 
 from fastapi_cachex.exceptions import ProxyNotSetError
 
+from .exceptions import SessionError
 from .middleware import _SESSION_READ_KEY
 from .middleware import _log_in
 from .middleware import _RequestSession
@@ -249,14 +250,26 @@ async def sudo(request: Request, session: AuthenticatedSession):
         True if a loaded session was given a new ID, False if none was loaded
 
     Raises:
-        HTTPException: 500 if no session middleware has registered a
-            SessionManager yet
+        HTTPException: 401 if another request logged the session out (deleted,
+            invalidated or rotated it) since this one loaded it; 500 if no
+            session middleware has registered a SessionManager yet
     """
     manager = get_session_manager(request)
     session: Session | None = getattr(request.state, "__fastapi_cachex_session", None)
     if session is None:
         return False
-    await manager.regenerate_session_id(session)
+    try:
+        await manager.regenerate_session_id(session)
+    except SessionError as e:
+        # Another request logged the session out while this one ran (#128).
+        request_session = request.scope.get("session")
+        if isinstance(request_session, _RequestSession):
+            request_session.ended = True
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Authentication required",
+            headers={"WWW-Authenticate": "Bearer"},
+        ) from e
     return True
 
 
@@ -283,6 +296,10 @@ async def login(
     - With no session loaded (a new visitor, or a token that did not
       resolve) a new session is created with the user, bound to the client
       IP and User-Agent as configured.
+    - A loaded session that another request deleted, invalidated or rotated
+      while this one ran is not rotated back to life (#128). A new session is
+      created as above, and nothing written to ``request.session`` before
+      the call is carried over.
 
     The middleware then saves the session, including anything written to
     ``request.session`` after the call (and, unless the loaded session was a