From d56c0581da0d8baaecdc161aa22d4f6b15609598 Mon Sep 17 00:00:00 2001 From: allen0099 Date: Mon, 5 Oct 2026 08:24:34 +0000 Subject: [PATCH] feat(cache): run the handler once for concurrent misses with coalesce=True `@cache(coalesce=True)` coalesces concurrent misses of one key within a process. The first GET to miss renders as usual; requests that miss the same key meanwhile wait for it, then read the backend again and are served the stored entry. If nothing was stored, each waiting request renders itself, all at once. HEAD waits for a running GET but never leads, and a request whose backend read fails does not wait. Off by default. coalesce must be a bool, needs a positive ttl and is rejected with private or no_cache; no_store ignores it with the usual warning. Closes #252 --- changelog.d/252.added.md | 8 + docs/APP_CACHE.md | 3 +- docs/CACHE_FLOW.md | 13 +- docs/HTTP_CACHING.md | 34 +++ fastapi_cachex/_coalesce.py | 70 ++++++ fastapi_cachex/cache.py | 92 +++++++- i18n/zh-TW/docs/APP_CACHE.md | 2 +- i18n/zh-TW/docs/CACHE_FLOW.md | 7 +- i18n/zh-TW/docs/HTTP_CACHING.md | 17 ++ tests/conftest.py | 4 + tests/test_cache_coalesce.py | 384 ++++++++++++++++++++++++++++++++ 11 files changed, 615 insertions(+), 19 deletions(-) create mode 100644 changelog.d/252.added.md create mode 100644 fastapi_cachex/_coalesce.py create mode 100644 tests/test_cache_coalesce.py diff --git a/changelog.d/252.added.md b/changelog.d/252.added.md new file mode 100644 index 0000000..eb23d11 --- /dev/null +++ b/changelog.d/252.added.md @@ -0,0 +1,8 @@ +**`@cache(coalesce=True)` runs the handler once for concurrent misses.** +Requests that miss a key while another request in the same process is +rendering it wait for that request, then are served the stored entry, so a +cold or just-expired key no longer sends every concurrent request to the +handler. When the first response is not stored (a cookie, `private`, an error +status), the waiting requests run the handler themselves, all at once. Off by +default; it needs a positive `ttl` and is rejected with `private` or +`no_cache`. Each worker process still runs the handler once per cold key. diff --git a/docs/APP_CACHE.md b/docs/APP_CACHE.md index a54488f..fb361d4 100644 --- a/docs/APP_CACHE.md +++ b/docs/APP_CACHE.md @@ -99,7 +99,8 @@ Complete runnable example: [`examples/app_cache.py`](https://github.com/allen009 When the factory is expensive (a slow database query, a rate-limited upstream API) and the key is hot, cache expiry turns into simultaneous recomputations. Distributed stampede protection built on `CacheLock` is on by default; you can -tune or turn it off per call or manager-wide: +tune or turn it off per call or manager-wide (for `@cache` routes, see +[Concurrent misses](HTTP_CACHING.md#concurrent-misses)): ```python # Per-call protection: diff --git a/docs/CACHE_FLOW.md b/docs/CACHE_FLOW.md index 6e3089c..d7176b5 100644 --- a/docs/CACHE_FLOW.md +++ b/docs/CACHE_FLOW.md @@ -9,9 +9,10 @@ keys), [`_vary.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fast [`_cache_control.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_cache_control.py) (`Cache-Control`), [`_stored_response.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_stored_response.py) -(storing and replaying a response, ETags and 304s) and +(storing and replaying a response, ETags and 304s), [`_rendering.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_rendering.py) (running the -handler and adding its dependencies' headers). +handler and adding its dependencies' headers) and +[`_coalesce.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_coalesce.py) (`coalesce=`). ## Overall flow @@ -167,8 +168,9 @@ Arguments are validated when the decorator is applied, and a `CacheXError` is raised if `public` and `private` are both set, if only one of `stale` / `stale_ttl` is given, if `ttl` is not an `int`, is negative or is larger than `MAX_TTL`, if `vary` is not a list of header field names, if `sort_query` is -not a `bool` or is passed with a custom `key_builder`, or if `key_builder` is -an `async` callable. +not a `bool` or is passed with a custom `key_builder`, if `key_builder` is +an `async` callable, or if `coalesce` is not a `bool`, or is set without a +positive `ttl` or together with `private` or `no_cache`. The header value is built once per decorated route: @@ -259,6 +261,9 @@ if bypass or (credential and not cache_authorized): # HEAD: key_builder sees the request with method GET cache_key = key_builder(request) + vary_components(request) # built only here entry = await backend.get(cache_key) # expired entries are already skipped here +if coalesce and entry is None and (leader := running_miss(cache_key)): + await leader # GET only leads; HEAD only waits + entry = await backend.get(cache_key) # None: render below, without waiting if client_etag and no_cache: fresh = await render() # no-cache: always re-render first diff --git a/docs/HTTP_CACHING.md b/docs/HTTP_CACHING.md index ec2c0fc..46bce70 100644 --- a/docs/HTTP_CACHING.md +++ b/docs/HTTP_CACHING.md @@ -244,6 +244,40 @@ empty body. The server drops the body of every HEAD response. A route that already accepted HEAD ran the handler on every HEAD request before 0.4.2 and added none of these headers; it now skips the handler on a hit. +### Concurrent misses + +By default every request that misses runs the handler: 20 concurrent requests +to a cold key, or arriving just after its entry expired, run it 20 times. +`coalesce=True` runs it once per key in each process: + +```python +@app.get("/report") +@cache(ttl=60, coalesce=True) +async def report() -> dict[str, int]: ... +``` + +The first request to miss renders as usual. Requests that miss the same key +while it runs wait for it to answer, then read the backend again and are served +the stored entry, with an `ETag`, `Age` and 304 handling like any hit (#252). +When the first response was not stored (it set a cookie, was `private` or +`no-store`, had an error status or a status a dependency set, was streamed, or +the handler raised), each waiting request runs the handler itself, all at once +rather than one after another. + +- Only requests in one process are coalesced, without a lock in the backend: + each worker still runs the handler once per cold key. For one run across + workers, cache the expensive part with `get_or_set()`, whose + [stampede protection](APP_CACHE.md#stampede-protection) uses a distributed + lock. +- A waiting request waits as long as the first one's handler takes; there is + no separate timeout. A request whose backend read fails does not wait. +- A HEAD request waits for a running GET of the same key but never makes + others wait, since its response is not stored. +- It needs a positive `ttl`. With `private=True` or `no_cache=True`, which + never serve a stored entry, the decorator raises `CacheXError`; with + `no_store=True` it is ignored with the usual warning. Requests with + credentials that bypass the backend are not coalesced. + ### Requests with credentials A single-page app that sends `Authorization` on every request, or a site where diff --git a/fastapi_cachex/_coalesce.py b/fastapi_cachex/_coalesce.py new file mode 100644 index 0000000..9af4768 --- /dev/null +++ b/fastapi_cachex/_coalesce.py @@ -0,0 +1,70 @@ +"""In-process coalescing of concurrent misses for `@cache(coalesce=True)` (#252). + +The first request to miss a key leads: it runs the handler as usual. Requests +that miss the same key while it runs follow: they wait until the leader has +answered, then read the backend again. A follower that still finds no entry +(the leader's response was not stored, or the handler failed) renders its own +response without waiting again, so followers never queue behind each other. + +Only requests in this process are coalesced. The registry is keyed by event +loop and backend as well as the cache key, so two apps or loops never wait on +each other. +""" + +import asyncio + +# One future per key being rendered by a leader; done once it has answered. +_IN_FLIGHT: dict[tuple[int, int, str], asyncio.Future[None]] = {} + + +class _Flight: + """One request's part in coalescing: leader, follower or neither. + + The wrapper creates one per request and calls ``finish()`` once the + request has been answered, however it ended, so a leader always releases + its followers and leaves nothing behind in the registry. + """ + + def __init__(self) -> None: + self._key: tuple[int, int, str] | None = None + self._future: asyncio.Future[None] | None = None + + def join( + self, backend: object, cache_key: str, *, may_lead: bool + ) -> asyncio.Future[None] | None: + """Return the leader's future to wait on, or ``None`` to render. + + With ``may_lead`` and no leader for the key, this request becomes the + leader. Without it (HEAD, whose response is never stored), it renders + without blocking anyone. + """ + loop = asyncio.get_running_loop() + key = (id(loop), id(backend), cache_key) + running = _IN_FLIGHT.get(key) + if running is not None or not may_lead: + return running + self._key = key + self._future = loop.create_future() + _IN_FLIGHT[key] = self._future + return None + + def finish(self) -> None: + """Release the followers, if this request led.""" + if self._key is None or self._future is None: + return + # Only the leader registers or removes its key, and followers wait + # through a shield, so the future is still pending. `pop` because a + # test teardown may have cleared the registry already. + _IN_FLIGHT.pop(self._key, None) + self._future.set_result(None) + self._key = None + self._future = None + + +async def _wait_for(leader: asyncio.Future[None]) -> None: + """Wait until the leader has answered. + + Shielded: a follower that is cancelled stops waiting without cancelling + the leader's future for the other followers. + """ + await asyncio.shield(leader) diff --git a/fastapi_cachex/cache.py b/fastapi_cachex/cache.py index 3846d27..ac85ed5 100644 --- a/fastapi_cachex/cache.py +++ b/fastapi_cachex/cache.py @@ -27,6 +27,8 @@ from ._cache_control import _cache_control_for from ._cache_control import _unshareable_reason from ._cache_control import _with_cache_control +from ._coalesce import _Flight +from ._coalesce import _wait_for from ._key_builders import _append_key_components from ._key_builders import _build_key from ._key_builders import _resolve_key_builder @@ -50,6 +52,7 @@ from ._vary import _validate_vary from ._vary import _vary_components from .backends.base import MAX_TTL +from .backends.base import BaseCacheBackend from .directives import DirectiveType from .exceptions import BackendNotFoundError from .exceptions import CacheXError @@ -57,6 +60,7 @@ from .headers import add_vary from .proxy import BackendProxy from .proxy import get_backend_or_fallback +from .types import CacheEntry from .types import CacheKeyBuilder from .types import log_ref @@ -111,6 +115,22 @@ def _log_backend_failure( logger.debug("Cache backend %s; key_ref=%s key=%s", what, key_ref, cache_key) +async def _read_entry( + backend: BaseCacheBackend, cache_key: str, request: Request, *, fail_open: bool +) -> tuple[CacheEntry | None, bool]: + """Read the entry under ``cache_key`` and whether the read worked. + + A failed read is a miss if failing open. + """ + try: + return await backend.get(cache_key), True + except Exception as e: + if not fail_open: + raise + _log_backend_failure("read failed; serving uncached", request, cache_key, e) + return None, False + + # Where `FastAPICacheXSessionMiddleware` puts the session it loaded; # `get_session` reads it. _SESSION_STATE_KEY = "__fastapi_cachex_session" @@ -357,6 +377,7 @@ def cache( cache_authorized: bool = False, vary: Sequence[str] | None = None, sort_query: bool | None = None, + coalesce: bool = False, ) -> Callable[[HandlerCallable], AsyncResponseCallable]: """Cache decorator for FastAPI route handlers. @@ -470,6 +491,19 @@ def cache( is (``build_cache_key()`` sorts unless told otherwise), and passing ``sort_query`` with one is rejected. Pass the same value to ``invalidate()``. + coalesce: Run the handler once for concurrent misses of one key in + this process. The first request to miss renders as usual; requests + that miss the same key meanwhile wait for it to answer, then read + the backend again and are served the stored entry. If the first + response was not stored (it set a cookie, was ``private``, had an + error status, was streamed, or the handler raised), each of them + runs the handler itself, at once. A request whose backend read + fails does not wait. Each worker process still runs + the handler once per cold key, and a waiting request waits as + long as the first one's handler takes. A HEAD request waits for a + running GET but never makes others wait. Requires a positive + ``ttl`` and is rejected with ``private`` or ``no_cache``, which + never serve a stored entry. Returns: Decorator function that wraps route handlers with caching logic @@ -481,7 +515,9 @@ def cache( negative or is larger than ``MAX_TTL``, or if ``vary`` is not a list of header field names (a single string is rejected), if ``sort_query`` is not a ``bool`` or is passed with - ``key_builder``, or if ``key_builder`` is an ``async`` callable. + ``key_builder``, if ``key_builder`` is an ``async`` callable, or + if ``coalesce`` is not a ``bool`` or is set without a positive + ``ttl`` or with ``private`` or ``no_cache``. At request time, if ``key_builder`` returns anything but a ``str``. @@ -519,6 +555,18 @@ def decorator(func: HandlerCallable) -> AsyncResponseCallable: if ttl is not None and ttl > MAX_TTL: msg = f"ttl must be at most {MAX_TTL} seconds" raise CacheXError(msg) + if not isinstance(coalesce, bool): + # Unreachable for type checkers; guards untyped callers. + msg = f"coalesce must be a bool, got {type(coalesce).__name__}" # type: ignore[unreachable] + raise CacheXError(msg) + if coalesce and not no_store and (private or no_cache or not ttl): + # Such a route never serves a stored entry, so there would be + # nothing to wait for. + msg = ( + "coalesce needs a positive ttl and cannot be combined with " + "private or no_cache, which never serve a stored entry" + ) + raise CacheXError(msg) vary_names = _validate_vary(vary) builder = _resolve_key_builder(key_builder, sort_query) if any(name.lower() == "cookie" for name in vary_names): @@ -539,6 +587,7 @@ def decorator(func: HandlerCallable) -> AsyncResponseCallable: ("private", private), ("immutable", immutable), ("must_revalidate", must_revalidate), + ("coalesce", coalesce), ) if value is not None and value is not False ] @@ -635,6 +684,7 @@ def decorator(func: HandlerCallable) -> AsyncResponseCallable: ) async def respond( + flight: _Flight, sub_response: Response | None, dependency_lines: Sequence[tuple[bytes, bytes]], dependency_status: int | None, @@ -768,13 +818,23 @@ async def respond( _vary_components(req, vary_names), ) - try: - cached_data = await cache_backend.get(cache_key) - except Exception as e: - if not fail_open: - raise - _log_backend_failure("read failed; serving uncached", req, cache_key, e) - cached_data = None + cached_data, read_ok = await _read_entry( + cache_backend, cache_key, req, fail_open=fail_open + ) + # A failed read does not wait: the backend is likely down, so the + # leader could not store its response either. A miss read just + # before a leader stores may still lead a second render; that + # window is one backend round trip. + if coalesce and read_ok and cached_data is None: + leader = flight.join( + cache_backend, cache_key, may_lead=req.method == "GET" + ) + if leader is not None: + logger.debug("Waiting for a running miss; key=%s", cache_key) + await _wait_for(leader) + cached_data, _ = await _read_entry( + cache_backend, cache_key, req, fail_open=fail_open + ) current_response: Response | None = None current_body: bytes | None = None @@ -964,9 +1024,19 @@ async def serve(*args: Any, **kwargs: Any) -> Response: None if sub_response is None else sub_response.status_code ) req: Request | None = kwargs.get(request_name) - response = await respond( - sub_response, dependency_lines, dependency_status, *args, **kwargs - ) + flight = _Flight() + try: + response = await respond( + flight, + sub_response, + dependency_lines, + dependency_status, + *args, + **kwargs, + ) + finally: + # Releases the requests waiting on this one (#252). + flight.finish() return _with_dependency_headers( response, sub_response, diff --git a/i18n/zh-TW/docs/APP_CACHE.md b/i18n/zh-TW/docs/APP_CACHE.md index b9b19de..f0d3235 100644 --- a/i18n/zh-TW/docs/APP_CACHE.md +++ b/i18n/zh-TW/docs/APP_CACHE.md @@ -57,7 +57,7 @@ await manager.clear_pattern("user:*") # 比對 "myapp:user:*" ## Cache stampede 保護 {#stampede-protection} -當 `factory` 的運算成本很高（例如慢速資料庫查詢、受速率限制的外部 API）且該鍵又是熱門鍵時，快取過期會導致多個請求同時重新計算。以 `CacheLock` 實作的分散式 cache stampede 保護預設開啟；你可以在單次呼叫或 manager 全域調整或關閉它： +當 `factory` 的運算成本很高（例如慢速資料庫查詢、受速率限制的外部 API）且該鍵又是熱門鍵時，快取過期會導致多個請求同時重新計算。以 `CacheLock` 實作的分散式 cache stampede 保護預設開啟；你可以在單次呼叫或 manager 全域調整或關閉它（`@cache` 路由請見[同時發生的未命中](HTTP_CACHING.md#concurrent-misses)）： ```python # 單次呼叫保護： diff --git a/i18n/zh-TW/docs/CACHE_FLOW.md b/i18n/zh-TW/docs/CACHE_FLOW.md index 068ef04..e1f31e1 100644 --- a/i18n/zh-TW/docs/CACHE_FLOW.md +++ b/i18n/zh-TW/docs/CACHE_FLOW.md @@ -1,6 +1,6 @@ # FastAPI-CacheX 快取流程 {#fastapi-cachex-cache-flow} -本文件詳細說明 FastAPI-CacheX 如何將快取邏輯套用到 HTTP 請求上。裝飾器位於 [`fastapi_cachex/cache.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/cache.py)，它呼叫的各部分則位於同一目錄的私有模組：[`_key_builders.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_key_builders.py)（快取鍵）、[`_vary.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_vary.py)（`vary=`）、[`_cache_control.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_cache_control.py)（`Cache-Control`）、[`_stored_response.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_stored_response.py)（儲存與重播回應、ETag 與 304），以及 [`_rendering.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_rendering.py)（執行 handler 並加上依賴項的標頭）。 +本文件詳細說明 FastAPI-CacheX 如何將快取邏輯套用到 HTTP 請求上。裝飾器位於 [`fastapi_cachex/cache.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/cache.py)，它呼叫的各部分則位於同一目錄的私有模組：[`_key_builders.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_key_builders.py)（快取鍵）、[`_vary.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_vary.py)（`vary=`）、[`_cache_control.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_cache_control.py)（`Cache-Control`）、[`_stored_response.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_stored_response.py)（儲存與重播回應、ETag 與 304）、[`_rendering.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_rendering.py)（執行 handler 並加上依賴項的標頭），以及 [`_coalesce.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/fastapi_cachex/_coalesce.py)（`coalesce=`）。 ## 整體流程 {#overall-flow} @@ -122,7 +122,7 @@ cache_key = "|".join( @cache(ttl=60, stale="error", stale_ttl=300) # stale-if-error=300 ``` -參數會在套用裝飾器時驗證；若同時設定 `public` 與 `private`、只提供 `stale`／`stale_ttl` 其中之一、`ttl` 不是 `int`、為負數或大於 `MAX_TTL`、`vary` 不是由標頭欄位名稱組成的 list、`sort_query` 不是 `bool` 或與自訂的 `key_builder` 一起傳入，或 `key_builder` 是 `async` 可呼叫物件，會拋出 `CacheXError`。 +參數會在套用裝飾器時驗證；若同時設定 `public` 與 `private`、只提供 `stale`／`stale_ttl` 其中之一、`ttl` 不是 `int`、為負數或大於 `MAX_TTL`、`vary` 不是由標頭欄位名稱組成的 list、`sort_query` 不是 `bool` 或與自訂的 `key_builder` 一起傳入，`key_builder` 是 `async` 可呼叫物件，或 `coalesce` 不是 `bool`、未搭配正的 `ttl` 或與 `private`、`no_cache` 一起設定，會拋出 `CacheXError`。 標頭值在每個被裝飾的路由上只建立一次： @@ -190,6 +190,9 @@ if bypass or (credential and not cache_authorized): # HEAD：key_builder 拿到的請求方法是 GET cache_key = key_builder(request) + vary_components(request) # 只在這裡建立 entry = await backend.get(cache_key) # 過期的項目已在此略過 +if coalesce and entry is None and (leader := running_miss(cache_key)): + await leader # 只有 GET 會帶頭；HEAD 只會等待 + entry = await backend.get(cache_key) # None：在下方自行產生，不再等待 if client_etag and no_cache: fresh = await render() # no-cache：一律先重新產生 diff --git a/i18n/zh-TW/docs/HTTP_CACHING.md b/i18n/zh-TW/docs/HTTP_CACHING.md index 5902ae2..c13955e 100644 --- a/i18n/zh-TW/docs/HTTP_CACHING.md +++ b/i18n/zh-TW/docs/HTTP_CACHING.md @@ -120,6 +120,23 @@ HEAD 請求會讀取 GET 以同一個快取鍵儲存的項目，這是 RFC 9110 0.4.2 之前，已接受 HEAD 的路由每次收到 HEAD 請求都會執行 handler，也不會加上這些標頭；現在命中時不會執行 handler。 +### 同時發生的未命中 {#concurrent-misses} + +預設情況下，每個未命中的請求都會執行 handler：20 個同時送到冷鍵的請求，或在項目剛過期時抵達的請求，會執行 20 次。`coalesce=True` 讓每個行程對每個快取鍵只執行一次： + +```python +@app.get("/report") +@cache(ttl=60, coalesce=True) +async def report() -> dict[str, int]: ... +``` + +第一個未命中的請求照常產生回應。在它執行期間未命中同一個快取鍵的請求，會等它回應後再讀取一次後端，並以儲存的項目回應，`ETag`、`Age` 與 304 的處理與一般命中相同（#252）。若第一個回應沒有儲存（它設定了 cookie、是 `private` 或 `no-store`、狀態碼是錯誤或由依賴項設定、是串流回應，或 handler 拋出例外），每個等待中的請求會自己執行 handler，而且是同時執行，不會一個接一個排隊。 + +- 只會合併同一個行程內的請求，不會在後端加鎖：每個 worker 對每個冷鍵仍會執行一次 handler。若要讓所有 worker 合計只執行一次，請以 `get_or_set()` 快取成本高的部分，它的 [cache stampede 保護](APP_CACHE.md#stampede-protection)使用分散式鎖。 +- 等待中的請求會等到第一個請求的 handler 結束，沒有另外的逾時。讀取後端失敗的請求不會等待。 +- HEAD 請求會等待同一個快取鍵正在執行的 GET，但永遠不會讓其他請求等它，因為它的回應不會儲存。 +- 它需要正的 `ttl`。搭配永遠不會以儲存的項目回應的 `private=True` 或 `no_cache=True` 時，裝飾器會拋出 `CacheXError`；搭配 `no_store=True` 時則會被忽略，並發出一般的警告。帶有憑證而繞過後端的請求不會被合併。 + ### 帶有憑證的請求 {#requests-with-credentials} 每個請求都送出 `Authorization` 的單頁應用程式，或每位訪客都有 Session 的網站，在只加上 `@cache` 的路由上完全不會命中快取：每個請求都會繞過後端（見上文）。請依 handler 回傳的內容選擇： diff --git a/tests/conftest.py b/tests/conftest.py index 42a3b05..1b67250 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -13,6 +13,7 @@ import pytest_asyncio from fastapi_cachex import manager +from fastapi_cachex._coalesce import _IN_FLIGHT from fastapi_cachex.backends import MemcachedBackend from fastapi_cachex.backends import memory from fastapi_cachex.backends.base import BaseCacheBackend @@ -106,6 +107,9 @@ async def setup_default_backend() -> AsyncGenerator[None, None]: BackendProxy.set(backend) yield await backend.aclose() + # A leader that never finished (a test that failed midway) must not leave + # a future of this loop for a later test to wait on. + _IN_FLIGHT.clear() @pytest.fixture(autouse=True) diff --git a/tests/test_cache_coalesce.py b/tests/test_cache_coalesce.py new file mode 100644 index 0000000..a889213 --- /dev/null +++ b/tests/test_cache_coalesce.py @@ -0,0 +1,384 @@ +"""``@cache(coalesce=True)`` runs the handler once for concurrent misses (#252).""" + +import asyncio +import importlib +from collections.abc import AsyncIterator +from collections.abc import Callable +from typing import Any + +import pytest +from fastapi import FastAPI +from fastapi import Request +from fastapi import Response + +from fastapi_cachex._coalesce import _IN_FLIGHT +from fastapi_cachex.cache import cache +from fastapi_cachex.exceptions import CacheXError +from fastapi_cachex.proxy import BackendProxy + +try: # Starlette's TestClient moved to httpx2; the `lowest` env still has httpx. + import httpx2 as httpx # type: ignore[import-not-found, unused-ignore] +except ImportError: # pragma: no cover - depends on the installed starlette + import httpx # type: ignore[no-redef, import-not-found, unused-ignore] + +# `fastapi_cachex.cache` is the decorator once the package is imported. +cache_module = importlib.import_module("fastapi_cachex.cache") + +KEY = "http:v2|GET|testserver|/items|" +FOLLOWERS = 4 + + +async def _settle(*requests: "asyncio.Future[httpx.Response]") -> list[httpx.Response]: + """Wait for ``requests``; fail instead of hanging on one never released.""" + return list(await asyncio.wait_for(asyncio.gather(*requests), timeout=5)) + + +class Waiters: + """Counts requests that started waiting for a leader. + + Also the test's progress signal: the handler and the counter call + ``notify()`` on every change, and ``until()`` waits for a condition on + them, failing instead of hanging when it never holds. + """ + + def __init__(self) -> None: + self.count = 0 + self._changed = asyncio.Event() + + def notify(self) -> None: + self._changed.set() + + async def until(self, condition: Callable[[], bool]) -> None: + async def wait() -> None: + while not condition(): + self._changed.clear() + await self._changed.wait() + + await asyncio.wait_for(wait(), timeout=5) + + +class Handler: + """A handler that blocks until released, counting runs and overlap.""" + + def __init__(self, waiters: Waiters) -> None: + self.waiters = waiters + self.release = asyncio.Event() + self.calls: list[str] = [] + self.running = 0 + self.max_running = 0 + + async def run(self, method: str) -> None: + self.calls.append(method) + self.running += 1 + self.max_running = max(self.max_running, self.running) + self.waiters.notify() + try: + # Read at call time, so a test can swap in a new event for the + # requests that come later. + await self.release.wait() + finally: + self.running -= 1 + self.waiters.notify() + + +@pytest.fixture +def waiters(monkeypatch: pytest.MonkeyPatch) -> Waiters: + counter = Waiters() + wait_for = cache_module._wait_for + + async def counting(leader: "asyncio.Future[None]") -> None: + counter.count += 1 + counter.notify() + await wait_for(leader) + + monkeypatch.setattr(cache_module, "_wait_for", counting) + return counter + + +@pytest.fixture +async def client_for() -> AsyncIterator[Callable[[FastAPI], httpx.AsyncClient]]: + clients: list[httpx.AsyncClient] = [] + + def make(app: FastAPI) -> httpx.AsyncClient: + client = httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://testserver" + ) + clients.append(client) + return client + + yield make + for client in clients: + await client.aclose() + + +def _app(handler: Handler, **cache_kwargs: Any) -> FastAPI: + app = FastAPI() + options: dict[str, Any] = {"ttl": 60, "coalesce": True, **cache_kwargs} + + @app.api_route("/items", methods=["GET", "HEAD"]) + @cache(**options) + async def items(request: Request, response: Response) -> dict[str, int]: + await handler.run(request.method) + if request.headers.get("x-cookie"): + response.set_cookie("seen", "1") + if request.headers.get("x-fail"): + msg = "handler failed" + raise RuntimeError(msg) + return {"n": 1} + + return app + + +async def _leader_then_followers( + client: httpx.AsyncClient, + handler: Handler, + waiters: Waiters, + *, + leader_headers: dict[str, str] | None = None, + follower_method: str = "GET", + follower_headers: dict[str, str] | None = None, +) -> tuple["asyncio.Task[httpx.Response]", list[httpx.Response]]: + """Start a leader, let FOLLOWERS requests queue behind it, then release.""" + leader = asyncio.ensure_future(client.get("/items", headers=leader_headers)) + await waiters.until(lambda: bool(handler.calls)) + followers = [ + asyncio.ensure_future( + client.request(follower_method, "/items", headers=follower_headers) + ) + for _ in range(FOLLOWERS) + ] + await waiters.until(lambda: waiters.count == FOLLOWERS) + handler.release.set() + return leader, await _settle(*followers) + + +async def test_concurrent_misses_run_the_handler_once( + client_for: Callable[[FastAPI], httpx.AsyncClient], waiters: Waiters +) -> None: + handler = Handler(waiters) + client = client_for(_app(handler)) + + leader, followers = await _leader_then_followers(client, handler, waiters) + + first = await leader + assert handler.calls == ["GET"] + assert {r.status_code for r in followers} == {200} + assert {r.headers["etag"] for r in followers} == {first.headers["etag"]} + assert all("age" in r.headers for r in followers) + assert await BackendProxy.get().get_all_keys() == [KEY] + assert _IN_FLIGHT == {} + + +async def test_followers_render_at_once_when_nothing_was_stored( + client_for: Callable[[FastAPI], httpx.AsyncClient], waiters: Waiters +) -> None: + handler = Handler(waiters) + client = client_for(_app(handler)) + cookie = {"X-Cookie": "1"} + leader = asyncio.ensure_future(client.get("/items", headers=cookie)) + await waiters.until(lambda: bool(handler.calls)) + followers = [ + asyncio.ensure_future(client.get("/items", headers=cookie)) + for _ in range(FOLLOWERS) + ] + await waiters.until(lambda: waiters.count == FOLLOWERS) + + first_release, handler.release = handler.release, asyncio.Event() + first_release.set() + + assert "set-cookie" in (await leader).headers + # Nothing was stored, so every follower runs the handler, all at once + # rather than one after another. + await waiters.until(lambda: handler.running == FOLLOWERS) + handler.release.set() + assert {r.status_code for r in await _settle(*followers)} == {200} + assert len(handler.calls) == 1 + FOLLOWERS + + +async def test_followers_render_when_the_leader_fails( + client_for: Callable[[FastAPI], httpx.AsyncClient], waiters: Waiters +) -> None: + handler = Handler(waiters) + client = client_for(_app(handler)) + leader = asyncio.ensure_future(client.get("/items", headers={"X-Fail": "1"})) + await waiters.until(lambda: bool(handler.calls)) + followers = [asyncio.ensure_future(client.get("/items")) for _ in range(FOLLOWERS)] + await waiters.until(lambda: waiters.count == FOLLOWERS) + + handler.release.set() + + with pytest.raises(RuntimeError, match="handler failed"): + await leader + assert {r.status_code for r in await _settle(*followers)} == {200} + # Not 1 + FOLLOWERS: the first follower to render may store its response + # before the others read again. + assert len(handler.calls) >= 2 + assert _IN_FLIGHT == {} + + +async def test_a_cancelled_leader_releases_its_followers( + client_for: Callable[[FastAPI], httpx.AsyncClient], waiters: Waiters +) -> None: + handler = Handler(waiters) + client = client_for(_app(handler)) + leader = asyncio.ensure_future(client.get("/items")) + await waiters.until(lambda: bool(handler.calls)) + follower = asyncio.ensure_future(client.get("/items")) + await waiters.until(lambda: waiters.count == 1) + + leader.cancel() + await waiters.until(lambda: len(handler.calls) == 2) + handler.release.set() + + assert [r.status_code for r in await _settle(follower)] == [200] + assert _IN_FLIGHT == {} + + +async def test_a_cancelled_follower_leaves_the_others_waiting( + client_for: Callable[[FastAPI], httpx.AsyncClient], waiters: Waiters +) -> None: + handler = Handler(waiters) + client = client_for(_app(handler)) + leader = asyncio.ensure_future(client.get("/items")) + await waiters.until(lambda: bool(handler.calls)) + followers = [asyncio.ensure_future(client.get("/items")) for _ in range(FOLLOWERS)] + await waiters.until(lambda: waiters.count == FOLLOWERS) + + followers[0].cancel() + await asyncio.sleep(0) + handler.release.set() + + assert [r.status_code for r in await _settle(leader, *followers[1:])] == [200] * ( + FOLLOWERS + ) + assert handler.calls == ["GET"] + + +async def test_a_follower_gets_a_304_for_the_stored_etag( + client_for: Callable[[FastAPI], httpx.AsyncClient], waiters: Waiters +) -> None: + handler = Handler(waiters) + handler.release.set() + client = client_for(_app(handler)) + etag = (await client.get("/items")).headers["etag"] + await BackendProxy.get().clear() + handler.release.clear() + handler.calls.clear() + + leader, followers = await _leader_then_followers( + client, handler, waiters, follower_headers={"If-None-Match": etag} + ) + + await leader + assert handler.calls == ["GET"] + assert {r.status_code for r in followers} == {304} + + +async def test_head_waits_for_a_running_get( + client_for: Callable[[FastAPI], httpx.AsyncClient], waiters: Waiters +) -> None: + handler = Handler(waiters) + client = client_for(_app(handler)) + + leader, followers = await _leader_then_followers( + client, handler, waiters, follower_method="HEAD" + ) + + await leader + assert handler.calls == ["GET"] + assert {r.status_code for r in followers} == {200} + + +async def test_head_never_makes_others_wait( + client_for: Callable[[FastAPI], httpx.AsyncClient], waiters: Waiters +) -> None: + handler = Handler(waiters) + client = client_for(_app(handler)) + heads = [asyncio.ensure_future(client.head("/items")) for _ in range(3)] + + await waiters.until(lambda: handler.running == len(heads)) + handler.release.set() + + assert {r.status_code for r in await _settle(*heads)} == {200} + assert waiters.count == 0 + assert _IN_FLIGHT == {} + + +async def test_different_keys_do_not_wait_on_each_other( + client_for: Callable[[FastAPI], httpx.AsyncClient], waiters: Waiters +) -> None: + handler = Handler(waiters) + client = client_for(_app(handler)) + requests = [ + asyncio.ensure_future(client.get("/items", params={"page": page})) + for page in range(3) + ] + + await waiters.until(lambda: handler.running == len(requests)) + handler.release.set() + + assert {r.status_code for r in await _settle(*requests)} == {200} + assert waiters.count == 0 + + +async def test_a_failed_backend_read_does_not_wait( + client_for: Callable[[FastAPI], httpx.AsyncClient], + waiters: Waiters, + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def unreachable(key: str) -> None: + msg = "backend down" + raise ConnectionError(msg) + + monkeypatch.setattr(BackendProxy.get(), "get", unreachable) + handler = Handler(waiters) + client = client_for(_app(handler)) + requests = [asyncio.ensure_future(client.get("/items")) for _ in range(3)] + + await waiters.until(lambda: handler.running == len(requests)) + handler.release.set() + + assert {r.status_code for r in await _settle(*requests)} == {200} + assert waiters.count == 0 + assert _IN_FLIGHT == {} + + +async def test_without_coalesce_every_miss_runs_the_handler( + client_for: Callable[[FastAPI], httpx.AsyncClient], waiters: Waiters +) -> None: + handler = Handler(waiters) + client = client_for(_app(handler, coalesce=False)) + requests = [asyncio.ensure_future(client.get("/items")) for _ in range(3)] + + await waiters.until(lambda: handler.running == len(requests)) + handler.release.set() + + assert {r.status_code for r in await _settle(*requests)} == {200} + assert waiters.count == 0 + + +@pytest.mark.parametrize( + "kwargs", + [ + pytest.param({"ttl": None}, id="no-ttl"), + pytest.param({"ttl": 0}, id="ttl-0"), + pytest.param({"ttl": 60, "private": True}, id="private"), + pytest.param({"ttl": 60, "no_cache": True}, id="no-cache"), + ], +) +def test_coalesce_is_rejected_where_nothing_stored_is_served( + kwargs: dict[str, Any], +) -> None: + with pytest.raises(CacheXError, match="coalesce needs a positive ttl"): + cache(coalesce=True, **kwargs)(lambda: None) + + +@pytest.mark.parametrize("value", [1, "yes", None]) +def test_coalesce_must_be_a_bool(value: object) -> None: + with pytest.raises(CacheXError, match="coalesce must be a bool"): + cache(ttl=60, coalesce=value)(lambda: None) # type: ignore[arg-type] + + +def test_no_store_warns_that_it_overrides_coalesce() -> None: + with pytest.warns(UserWarning, match="coalesce"): + cache(no_store=True, coalesce=True)(lambda: None)