From a982b76a93fb4d0583b91d9f51bb313b2c983ff6 Mon Sep 17 00:00:00 2001 From: Yassin Kortam Date: Thu, 12 Nov 2026 10:32:44 -0800 Subject: [PATCH] fix(proxy): fetch blog posts without blocking the event loop (#51369) Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/litellm_core_utils/get_blog_posts.py | 51 ++++- .../public_endpoints/public_endpoints.py | 5 +- .../test_blog_posts_endpoint.py | 183 ++++++++++++++++-- 3 files changed, 218 insertions(+), 21 deletions(-) diff --git a/litellm/litellm_core_utils/get_blog_posts.py b/litellm/litellm_core_utils/get_blog_posts.py index c3b4114d5b9..1201a73d6ce 100644 --- a/litellm/litellm_core_utils/get_blog_posts.py +++ b/litellm/litellm_core_utils/get_blog_posts.py @@ -8,9 +8,11 @@ Disable remote fetching entirely: export LITELLM_LOCAL_BLOG_POSTS=True """ +import asyncio import os import time import xml.etree.ElementTree as ET +from collections.abc import Awaitable, Callable from email.utils import parsedate_to_datetime from importlib.resources import files from typing import Final @@ -78,6 +80,7 @@ class GetBlogPosts: _cached_posts: tuple[BlogPost, ...] | None = None _last_fetch_time: float = 0.0 + _refresh_lock: tuple[asyncio.AbstractEventLoop, asyncio.Lock] | None = None @staticmethod def load_local_blog_posts() -> tuple[BlogPost, ...]: @@ -98,7 +101,7 @@ class GetBlogPosts: return response.text @staticmethod - def parse_rss_to_posts(xml_text: str, max_posts: int = BLOG_POSTS_MAX) -> tuple[BlogPost, ...]: + def parse_rss_to_posts(xml_text: str | bytes, max_posts: int = BLOG_POSTS_MAX) -> tuple[BlogPost, ...]: """ Parse RSS XML into blog posts. @@ -147,14 +150,52 @@ class GetBlogPosts: url, str(e), ) - return cls.load_local_blog_posts() + return cls._cache_posts(cls.load_local_blog_posts()) if not cls.validate_blog_posts(posts): + return cls._cache_posts(cls.load_local_blog_posts()) + + return cls._cache_posts(posts) + + @classmethod + def _cache_posts(cls, posts: tuple[BlogPost, ...]) -> tuple[BlogPost, ...]: + cls._cached_posts = posts + cls._last_fetch_time = time.time() + return posts + + @classmethod + async def aget_blog_posts(cls, url: str, fetch: Callable[[str], Awaitable[bytes]]) -> tuple[BlogPost, ...]: + if is_offline_mode() or os.getenv("LITELLM_LOCAL_BLOG_POSTS", "").lower() == "true": return cls.load_local_blog_posts() - cls._cached_posts = posts - cls._last_fetch_time = now - return posts + cached: Final = cls._cached_posts + if cached is not None and (time.time() - cls._last_fetch_time) < BLOG_POSTS_TTL_SECONDS: + return cached + + loop: Final = asyncio.get_running_loop() + state: Final = cls._refresh_lock + lock: Final = state[1] if state is not None and state[0] is loop else asyncio.Lock() + cls._refresh_lock = (loop, lock) + async with lock: + refreshed: Final = cls._cached_posts + if refreshed is not None and (time.time() - cls._last_fetch_time) < BLOG_POSTS_TTL_SECONDS: + return refreshed + return await cls._refresh_blog_posts(url, fetch) + + @classmethod + async def _refresh_blog_posts(cls, url: str, fetch: Callable[[str], Awaitable[bytes]]) -> tuple[BlogPost, ...]: + try: + xml_text: Final = await fetch(url) + posts: Final = cls.parse_rss_to_posts(xml_text) + except (httpx.HTTPError, ET.ParseError, ValueError) as e: + verbose_logger.warning( + "LiteLLM: Failed to fetch blog posts from %s: %s. Falling back to local backup.", url, str(e) + ) + return cls._cache_posts(cls.load_local_blog_posts()) + + if not cls.validate_blog_posts(posts): + return cls._cache_posts(cls.load_local_blog_posts()) + return cls._cache_posts(posts) def get_blog_posts(url: str) -> tuple[BlogPost, ...]: diff --git a/litellm/proxy/public_endpoints/public_endpoints.py b/litellm/proxy/public_endpoints/public_endpoints.py index 45abd3c4d5f..48898afd8fa 100644 --- a/litellm/proxy/public_endpoints/public_endpoints.py +++ b/litellm/proxy/public_endpoints/public_endpoints.py @@ -17,7 +17,6 @@ from litellm._logging import verbose_logger from litellm.litellm_core_utils.get_blog_posts import ( BlogPostsResponse, GetBlogPosts, - get_blog_posts, ) from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap from litellm.litellm_core_utils.offline_mode import is_offline_mode @@ -561,7 +560,9 @@ async def get_litellm_blog_posts() -> BlogPostsResponse: Falls back to the bundled local backup on any failure. """ try: - return BlogPostsResponse(posts=get_blog_posts(url=litellm.blog_posts_url)) + return BlogPostsResponse( + posts=await GetBlogPosts.aget_blog_posts(url=litellm.blog_posts_url, fetch=_fetch_remote_bytes) + ) except Exception as e: verbose_logger.warning("LiteLLM: get_litellm_blog_posts endpoint fallback triggered: %s", str(e)) return BlogPostsResponse(posts=GetBlogPosts.load_local_blog_posts()) diff --git a/tests/unit/proxy/public_endpoints/test_blog_posts_endpoint.py b/tests/unit/proxy/public_endpoints/test_blog_posts_endpoint.py index 5d460617dda..85391bb521b 100644 --- a/tests/unit/proxy/public_endpoints/test_blog_posts_endpoint.py +++ b/tests/unit/proxy/public_endpoints/test_blog_posts_endpoint.py @@ -1,11 +1,16 @@ """Tests for the /public/litellm_blog_posts endpoint.""" -from unittest.mock import patch +import asyncio +import time +from typing import Final +from unittest.mock import AsyncMock, patch +import httpx import pytest from fastapi.testclient import TestClient -from litellm.litellm_core_utils.get_blog_posts import BlogPost, GetBlogPosts +from litellm.litellm_core_utils.get_blog_posts import BLOG_POSTS_TTL_SECONDS, BlogPost, GetBlogPosts +from litellm.proxy.public_endpoints.public_endpoints import get_litellm_blog_posts SAMPLE_POSTS = ( BlogPost( @@ -27,15 +32,17 @@ def client(): app = FastAPI() app.include_router(router) - return TestClient(app) + GetBlogPosts._cached_posts = None + try: + yield TestClient(app) + finally: + GetBlogPosts._cached_posts = None def test_get_blog_posts_returns_response_shape(client): - with patch( - "litellm.proxy.public_endpoints.public_endpoints.get_blog_posts", - return_value=SAMPLE_POSTS, - ): - response = client.get("/public/litellm_blog_posts") + GetBlogPosts._cached_posts = SAMPLE_POSTS + GetBlogPosts._last_fetch_time = time.time() + response = client.get("/public/litellm_blog_posts") assert response.status_code == 200 assert response.json() == { @@ -58,11 +65,9 @@ def test_get_blog_posts_returns_every_fetched_post(client): for i in range(10) ) - with patch( - "litellm.proxy.public_endpoints.public_endpoints.get_blog_posts", - return_value=many_posts, - ): - response = client.get("/public/litellm_blog_posts") + GetBlogPosts._cached_posts = many_posts + GetBlogPosts._last_fetch_time = time.time() + response = client.get("/public/litellm_blog_posts") assert response.status_code == 200 assert [p["title"] for p in response.json()["posts"]] == [f"Post {i}" for i in range(10)] @@ -71,7 +76,7 @@ def test_get_blog_posts_returns_every_fetched_post(client): def test_get_blog_posts_returns_local_backup_on_failure(client): """Endpoint returns the local backup when the fetcher raises.""" with patch( - "litellm.proxy.public_endpoints.public_endpoints.get_blog_posts", + "litellm.proxy.public_endpoints.public_endpoints._fetch_remote_bytes", side_effect=Exception("fetch failed"), ): response = client.get("/public/litellm_blog_posts") @@ -79,3 +84,153 @@ def test_get_blog_posts_returns_local_backup_on_failure(client): assert response.status_code == 200 backup = [post.model_dump(mode="json") for post in GetBlogPosts.load_local_blog_posts()] assert response.json()["posts"] == backup + + +@pytest.mark.asyncio +async def test_blog_fetch_allows_concurrent_coroutine_progress() -> None: + progress: Final = asyncio.Event() + + async def fetch(*args: object, **kwargs: object) -> httpx.Response: + await asyncio.sleep(0) + await progress.wait() + return httpx.Response(503, request=httpx.Request("GET", "https://example.test/rss")) + + async def other_request() -> None: + await asyncio.sleep(0) + progress.set() + + with patch("litellm.llms.custom_httpx.http_handler.get_async_httpx_client") as client_factory: + client_factory.return_value.get = AsyncMock(side_effect=fetch) + with patch("litellm.litellm_core_utils.get_blog_posts.httpx.get", side_effect=httpx.ConnectTimeout("timeout")): + GetBlogPosts._cached_posts = None + try: + task: Final = asyncio.create_task(other_request()) + response: Final = await get_litellm_blog_posts() + assert task.done() + assert response.posts == GetBlogPosts.load_local_blog_posts() + finally: + await task + GetBlogPosts._cached_posts = None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", [httpx.ConnectTimeout("timeout"), "not xml", ""]) +async def test_blog_fetch_failure_is_cached(failure: Exception | str) -> None: + with patch("litellm.llms.custom_httpx.http_handler.get_async_httpx_client") as client_factory: + client_factory.return_value.get = AsyncMock( + side_effect=failure if isinstance(failure, Exception) else None, + return_value=httpx.Response( + 200, + text=failure if isinstance(failure, str) else "", + request=httpx.Request("GET", "https://example.test"), + ), + ) + with patch( + "litellm.litellm_core_utils.get_blog_posts.httpx.get", side_effect=httpx.ConnectTimeout("timeout") + ) as sync: + GetBlogPosts._cached_posts = None + try: + first: Final = await get_litellm_blog_posts() + second: Final = await get_litellm_blog_posts() + assert first.posts == second.posts == GetBlogPosts.load_local_blog_posts() + assert client_factory.return_value.get.call_count + sync.call_count == 1 + with patch( + "litellm.litellm_core_utils.get_blog_posts.time.time", + return_value=GetBlogPosts._last_fetch_time + BLOG_POSTS_TTL_SECONDS, + ): + expired: Final = await get_litellm_blog_posts() + assert expired.posts == first.posts + assert client_factory.return_value.get.call_count == 2 + finally: + GetBlogPosts._cached_posts = None + + +@pytest.mark.asyncio +async def test_blog_fetch_success_is_cached() -> None: + rss: Final = ( + "Async posthttps://example.test/post" + ) + with patch("litellm.llms.custom_httpx.http_handler.get_async_httpx_client") as client_factory: + client_factory.return_value.get = AsyncMock( + return_value=httpx.Response(200, text=rss, request=httpx.Request("GET", "https://example.test/rss")) + ) + GetBlogPosts._cached_posts = None + try: + first: Final = await get_litellm_blog_posts() + second: Final = await get_litellm_blog_posts() + assert ( + first.posts + == second.posts + == (BlogPost(title="Async post", description="", date="", url="https://example.test/post"),) + ) + client_factory.return_value.get.assert_awaited_once() + finally: + GetBlogPosts._cached_posts = None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("expired", [False, True]) +@pytest.mark.parametrize("fails", [False, True]) +async def test_concurrent_blog_refresh_is_shared(expired: bool, fails: bool) -> None: + started: Final = asyncio.Event() + release: Final = asyncio.Event() + rss: Final = ( + b"Sharedhttps://example.test/post" + ) + + async def fetch(url: str) -> bytes: + started.set() + await release.wait() + if fails: + raise httpx.ConnectTimeout("timeout") + return rss + + fetcher: Final = AsyncMock(side_effect=fetch) + GetBlogPosts._cached_posts = SAMPLE_POSTS if expired else None + GetBlogPosts._last_fetch_time = 0.0 + tasks: Final = tuple( + asyncio.create_task(GetBlogPosts.aget_blog_posts("https://example.test/rss", fetcher)) for _ in range(32) + ) + try: + await started.wait() + await asyncio.sleep(0) + release.set() + results: Final = await asyncio.gather(*tasks) + expected: Final = GetBlogPosts.load_local_blog_posts() if fails else GetBlogPosts.parse_rss_to_posts(rss) + assert all(posts == expected for posts in results) + assert fetcher.await_count == 1 + finally: + release.set() + await asyncio.gather(*tasks, return_exceptions=True) + GetBlogPosts._cached_posts = None + + +@pytest.mark.asyncio +async def test_cancelled_blog_refresh_does_not_wedge_waiters() -> None: + started: Final = asyncio.Event() + release: Final = asyncio.Event() + + async def fetch(url: str) -> bytes: + started.set() + await release.wait() + return ( + b"Recoveredhttps://example.test/post" + ) + + GetBlogPosts._cached_posts = None + first: Final = asyncio.create_task(GetBlogPosts.aget_blog_posts("https://example.test/rss", fetch)) + await started.wait() + second: Final = asyncio.create_task(GetBlogPosts.aget_blog_posts("https://example.test/rss", fetch)) + try: + await asyncio.sleep(0) + first.cancel() + with pytest.raises(asyncio.CancelledError): + await first + release.set() + posts: Final = await asyncio.wait_for(second, timeout=1) + assert posts == (BlogPost(title="Recovered", description="", date="", url="https://example.test/post"),) + finally: + first.cancel() + release.set() + await asyncio.gather(first, second, return_exceptions=True) + GetBlogPosts._cached_posts = None