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)