fix(proxy): invalidate cached project object on project update and delete

enterprise/litellm_enterprise/proxy/management_endpoints/project_endpoints.py

@@ -19,6 +19,7 @@
 from litellm._logging import verbose_proxy_logger
 from litellm._uuid import uuid
 from litellm.proxy._types import *
+from litellm.proxy.auth.auth_checks import delete_cached_project_object
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 from litellm.proxy.management_endpoints.common_utils import _set_object_metadata_field
 from litellm.proxy.management_helpers.utils import (
@@ -514,6 +515,7 @@ async def update_project(
         litellm_proxy_admin_name,
         premium_user,
         prisma_client,
+        user_api_key_cache,
     )
 
     try:
@@ -672,6 +674,11 @@ async def update_project(
             include={"litellm_budget_table": True, "object_permission": True},
         )
 
+        await delete_cached_project_object(
+            project_id=data.project_id,
+            user_api_key_cache=user_api_key_cache,
+        )
+
         return updated_project
     except Exception as e:
         verbose_proxy_logger.exception(
@@ -710,7 +717,7 @@ async def delete_project(
     }'
     ```
     """
-    from litellm.proxy.proxy_server import premium_user, prisma_client
+    from litellm.proxy.proxy_server import premium_user, prisma_client, user_api_key_cache
 
     try:
         if not premium_user:
@@ -773,6 +780,11 @@ async def delete_project(
                 prisma_models.LiteLLM_ProjectTable | None
             ) = await prisma_client.db.litellm_projecttable.delete(where={"project_id": project_id})
 
+            await delete_cached_project_object(
+                project_id=project_id,
+                user_api_key_cache=user_api_key_cache,
+            )
+
             deleted_projects.append(deleted_project)
 
         return deleted_projects

litellm/proxy/auth/auth_checks.py

@@ -4262,6 +4262,10 @@ async def _project_soft_budget_check(
             )
 
 
+def _project_cache_key(project_id: str) -> str:
+    return f"project_id:{project_id}"
+
+
 async def get_project_object(
     project_id: str,
     prisma_client: PrismaClient | None,
@@ -4279,7 +4283,7 @@ async def get_project_object(
         return None
 
     # Check cache first
-    cache_key: Final = f"project_id:{project_id}"
+    cache_key: Final = _project_cache_key(project_id)
     deserialized_project: Final = await user_api_key_cache.async_get_cache(
         key=cache_key,
         model_type=LiteLLM_ProjectTableCachedObj,
@@ -4310,6 +4314,32 @@ async def get_project_object(
     return project_obj
 
 
+async def delete_cached_project_object(
+    project_id: str,
+    user_api_key_cache: UserApiKeyCache,
+) -> None:
+    """
+    Every endpoint that mutates litellm_projecttable must call this: get_project_object
+    serves auth cache-first with no freshness check, so without invalidation a stale
+    project (e.g. a pre-update empty model allowlist) keeps being enforced until the
+    TTL expires (LIT-3803). Best-effort on both steps: the DB write has already
+    committed, so a cache backend error must not fail the endpoint; the stale entry
+    then expires via TTL.
+    """
+    from litellm.proxy.common_utils.auth_cache_invalidation_pubsub import publish_auth_cache_invalidation
+
+    cache_key: Final = _project_cache_key(project_id)
+    try:
+        await user_api_key_cache.async_delete_cache(key=cache_key)
+    except Exception as e:  # noqa: BLE001  # best-effort eviction: any cache backend error must not fail the mutation
+        verbose_proxy_logger.warning(
+            "Failed to evict cached project entry %s; a stale project may be served until its TTL expires: %s",
+            cache_key,
+            e,
+        )
+    await publish_auth_cache_invalidation(cache_key=cache_key)
+
+
 async def _organization_max_budget_check(
     valid_token: UserAPIKeyAuth | None,
     team_object: LiteLLM_TeamTable | None,

litellm/proxy/common_utils/auth_cache_invalidation_pubsub.py

@@ -0,0 +1,153 @@
+import asyncio
+import json
+from dataclasses import asdict, dataclass
+from typing import TYPE_CHECKING, Final
+
+from litellm._logging import verbose_proxy_logger
+from litellm.proxy.common_utils.config_sync_pubsub import (
+    _ConfigSyncPubSub,
+    _pubsub_capable_client,
+    coordination_redis_cache,
+)
+
+if TYPE_CHECKING:
+    from litellm.caching.redis_cache import RedisCache
+    from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
+
+AUTH_CACHE_INVALIDATION_CHANNEL: Final = "litellm_proxy.auth_cache_invalidation"
+_POLL_TIMEOUT_SECONDS: Final = 1.0
+_BACKOFF_INITIAL_SECONDS: Final = 5.0
+_BACKOFF_MAX_SECONDS: Final = 60.0
+
+
+def auth_cache_invalidation_channel(redis_cache: "RedisCache") -> str:
+    if redis_cache.namespace is None:
+        return AUTH_CACHE_INVALIDATION_CHANNEL
+    return f"{redis_cache.namespace}:{AUTH_CACHE_INVALIDATION_CHANNEL}"
+
+
+@dataclass(frozen=True, slots=True)
+class _CacheInvalidationMessage:
+    cache_key: str
+
+
+def _cache_invalidation_message_json(cache_key: str) -> str:
+    return json.dumps(asdict(_CacheInvalidationMessage(cache_key=cache_key)))
+
+
+def _cache_key_from_message_data(data: object) -> str | None:
+    if isinstance(data, bytes):
+        data = data.decode("utf-8", errors="replace")
+    if not isinstance(data, str):
+        return None
+    try:
+        parsed: Final = json.loads(data)
+    except json.JSONDecodeError:
+        return None
+    if not isinstance(parsed, dict):
+        return None
+    cache_key: Final = parsed.get("cache_key")
+    return cache_key if isinstance(cache_key, str) else None
+
+
+async def publish_auth_cache_invalidation(cache_key: str) -> None:
+    """
+    Best-effort broadcast so every worker drops its local in-memory copy of a
+    mutated management object; without this, only the handling worker and Redis
+    are evicted and other workers keep serving the stale object until its TTL.
+    """
+    redis_cache: Final = coordination_redis_cache()
+    if redis_cache is None:
+        return
+    try:
+        client: Final = _pubsub_capable_client(redis_cache)
+        if client is None:
+            verbose_proxy_logger.debug(
+                "auth cache invalidation publish for %s skipped: cluster redis client has no pub/sub support",
+                cache_key,
+            )
+            return
+        await client.publish(auth_cache_invalidation_channel(redis_cache), _cache_invalidation_message_json(cache_key))
+    except Exception as e:  # noqa: BLE001  # best-effort publish; mutations must never fail on redis errors
+        verbose_proxy_logger.warning("auth cache invalidation publish for %s failed: %s", cache_key, e)
+
+
+class AuthCacheInvalidationSubscriber:
+    __slots__ = ("_redis_cache", "_task", "_user_api_key_cache")
+
+    def __init__(
+        self,
+        redis_cache: "RedisCache",
+        user_api_key_cache: "UserApiKeyCache",
+    ) -> None:
+        self._redis_cache = redis_cache
+        self._user_api_key_cache = user_api_key_cache
+        self._task: asyncio.Task[None] | None = None
+
+    def start(self) -> None:
+        if self._task is not None:
+            return
+        self._task = asyncio.create_task(self._run())
+
+    async def stop(self) -> None:
+        task: Final = self._task
+        if task is None:
+            return
+        self._task = None
+        _ = task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    async def _run(self) -> None:
+        backoff_seconds = _BACKOFF_INITIAL_SECONDS  # rebind-ok: exponential backoff accumulator across reconnects
+        while True:
+            try:
+                client = _pubsub_capable_client(self._redis_cache)  # rebind-ok: re-resolved on every reconnect
+                if client is None:
+                    verbose_proxy_logger.warning(
+                        "auth cache invalidation subscriber disabled: cluster redis client has no pub/sub support; "
+                        "cross-worker eviction falls back to the local cache TTL"
+                    )
+                    return
+                pubsub = client.pubsub()  # rebind-ok: fresh pubsub per reconnect
+                try:
+                    await pubsub.subscribe(auth_cache_invalidation_channel(self._redis_cache))
+                    backoff_seconds = _BACKOFF_INITIAL_SECONDS  # rebind-ok: reset after successful subscribe
+                    await self._consume(pubsub)
+                finally:
+                    await self._close_pubsub(pubsub)
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:  # noqa: BLE001  # any redis failure falls through to backoff and reconnect
+                verbose_proxy_logger.warning(
+                    "auth cache invalidation subscriber redis error: %s; reconnecting in %.0fs",
+                    e,
+                    backoff_seconds,
+                )
+                await asyncio.sleep(backoff_seconds)
+                backoff_seconds = min(backoff_seconds * 2, _BACKOFF_MAX_SECONDS)  # rebind-ok: backoff accumulator
+
+    async def _consume(self, pubsub: _ConfigSyncPubSub) -> None:
+        while True:
+            message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=_POLL_TIMEOUT_SECONDS)
+            if message is None:
+                continue
+            self._apply_message(message)
+
+    def _apply_message(self, message: object) -> None:
+        data: Final = message.get("data") if isinstance(message, dict) else None
+        cache_key: Final = _cache_key_from_message_data(data)
+        if cache_key is None:
+            return
+        in_memory_cache: Final = self._user_api_key_cache.in_memory_cache
+        if in_memory_cache is not None:
+            in_memory_cache.delete_cache(cache_key)
+
+    @staticmethod
+    async def _close_pubsub(pubsub: _ConfigSyncPubSub) -> None:
+        try:
+            await pubsub.aclose()
+        except Exception as e:  # noqa: BLE001  # best-effort close of a possibly-broken connection
+            verbose_proxy_logger.debug("auth cache invalidation pubsub close failed: %s", e)

litellm/proxy/proxy_server.py

@@ -297,6 +297,9 @@ def generate_feedback_box():
     _should_return_raw_model_name,
     create_response,
 )
+from litellm.proxy.common_utils.auth_cache_invalidation_pubsub import (
+    AuthCacheInvalidationSubscriber,
+)
 from litellm.proxy.common_utils.callback_utils import initialize_callbacks_on_proxy
 from litellm.proxy.common_utils.config_sync_pubsub import ConfigSyncSubscriber
 from litellm.proxy.common_utils.debug_utils import init_verbose_loggers
@@ -1189,6 +1192,8 @@ async def _run_pw_migration():
 
     await proxy_config.stop_config_sync_subscriber()
 
+    await proxy_config.stop_auth_cache_invalidation_subscriber()
+
     await proxy_shutdown_event()
 
 
@@ -3900,6 +3905,7 @@ def __init__(self) -> None:
         self._last_hashicorp_vault_config: dict[str, Any] | None = None
         self.worker_registry: list[WorkerRegistryEntry] = []
         self.config_sync_subscriber: ConfigSyncSubscriber | None = None
+        self.auth_cache_invalidation_subscriber: AuthCacheInvalidationSubscriber | None = None
         from litellm.litellm_core_utils.get_model_cost_map import (
             get_model_cost_map_loaded_at,
         )
@@ -6370,6 +6376,30 @@ async def stop_config_sync_subscriber(self) -> None:
         except Exception as e:
             verbose_proxy_logger.error("Error stopping config sync subscriber: %s", e)
 
+    def start_auth_cache_invalidation_subscriber(
+        self,
+        redis_cache: RedisCache | None,
+        user_api_key_cache: UserApiKeyCache,
+    ) -> None:
+        if redis_cache is None or self.auth_cache_invalidation_subscriber is not None:
+            return
+        subscriber: Final = AuthCacheInvalidationSubscriber(
+            redis_cache=redis_cache,
+            user_api_key_cache=user_api_key_cache,
+        )
+        self.auth_cache_invalidation_subscriber = subscriber
+        subscriber.start()
+
+    async def stop_auth_cache_invalidation_subscriber(self) -> None:
+        subscriber: Final = self.auth_cache_invalidation_subscriber
+        if subscriber is None:
+            return
+        self.auth_cache_invalidation_subscriber = None
+        try:
+            await subscriber.stop()
+        except Exception as e:  # noqa: BLE001  # best-effort: a failing stop must not break proxy shutdown
+            verbose_proxy_logger.error("Error stopping auth cache invalidation subscriber: %s", e)
+
     async def _init_non_llm_objects_in_db(self, prisma_client: PrismaClient):
         """
         Use this to read non-llm objects from the db and initialize them
@@ -8296,6 +8326,11 @@ async def initialize_scheduled_background_jobs(
             misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME,
         )
 
+        proxy_config.start_auth_cache_invalidation_subscriber(
+            redis_cache=redis_usage_cache,
+            user_api_key_cache=user_api_key_cache,
+        )
+
         if store_model_in_db is True:
             ### GET STORED CREDENTIALS ###
             scheduler.add_job(