music-assistant-server

1"""Helper utilities for the cache controller."""
2
3from __future__ import annotations
4
5import asyncio
6import functools
7from collections.abc import Awaitable, Callable, Coroutine
8from copy import deepcopy
9from dataclasses import dataclass
10from typing import (
11    TYPE_CHECKING,
12    Any,
13    Concatenate,
14    ParamSpec,
15    Protocol,
16    TypeVar,
17    cast,
18    get_type_hints,
19)
20
21from music_assistant.controllers.cache.constants import (
22    BYPASS_CACHE,
23    DEFAULT_CACHE_EXPIRATION,
24    LOGGER,
25)
26from music_assistant.helpers.api import parse_value
27from music_assistant.helpers.json import SerializableType
28
29if TYPE_CHECKING:
30    from music_assistant import MusicAssistant
31
32
33class _Cacheable(Protocol):
34    """Protocol for objects that can use the @use_cache decorator."""
35
36    @property
37    def domain(self) -> str: ...
38
39    @property
40    def mass(self) -> MusicAssistant: ...
41
42
43ProviderT = TypeVar("ProviderT", bound="_Cacheable")
44P = ParamSpec("P")
45R = TypeVar("R")
46
47
48def use_cache(
49    expiration: int = DEFAULT_CACHE_EXPIRATION,
50    category: int = 0,
51    persistent: bool = False,
52    cache_checksum: str | None = None,
53    allow_bypass: bool | None = None,
54    base_class: Any = None,
55    allow_expired_cache: bool = False,
56    cache_none: bool = True,
57    single_flight: bool = True,
58) -> Callable[
59    [Callable[Concatenate[ProviderT, P], Awaitable[R]]],
60    Callable[Concatenate[ProviderT, P], Coroutine[Any, Any, R]],
61]:
62    """
63    Return decorator that can be used to cache a method's result.
64
65    :param expiration: Time in seconds the cache entry should be valid.
66    :param category: Category to group cache objects.
67    :param persistent: If True, the entry survives cache clears.
68    :param cache_checksum: Optional checksum to store with the cache object.
69    :param allow_bypass: Whether to respect the BYPASS_CACHE context variable.
70    :param base_class: If provided, reconstruct cached data using base_class.from_dict().
71        Handles both single dicts and lists of dicts automatically.
72        If not provided, falls back to type-annotation based reconstruction.
73    :param allow_expired_cache: If True, enable stale-while-revalidate. When the cached
74        entry has expired, return it immediately and trigger a background refresh that
75        re-runs the wrapped function and updates the cache. Expired entries also
76        survive the cache auto-cleanup task so they remain available as fallback data.
77    :param cache_none: Whether a None result is cached and served like any other value
78        (a negative hit, e.g. "no lyrics exist for this track"). Set to False for
79        methods where None signals a (transient) failure, so the call is retried on
80        the next invocation instead of serving a cached None.
81    :param single_flight: Whether concurrent callers that miss the cache on the same key
82        share one execution of the wrapped function instead of each running it. Callers of
83        a shared fetch each get their own copy of the result, or share the one object when
84        it cannot be copied. Set to False when sharing a fetch would be wrong, e.g. because
85        the call advances a cursor on the provider side or sends a one-shot event, so
86        concurrent callers must each run their own fetch instead of sharing one. This does
87        not apply while a stale entry is served under allow_expired_cache: that refresh is
88        always a single background call, whichever way single_flight is set.
89    """
90    if allow_bypass is None:
91        allow_bypass = not persistent
92
93    def _decorator(
94        func: Callable[Concatenate[ProviderT, P], Awaitable[R]],
95    ) -> Callable[Concatenate[ProviderT, P], Coroutine[Any, Any, R]]:
96        def _reconstruct(cachedata: Any) -> R:
97            if base_class is not None:
98                return cast("R", cachedata)
99            # fallback: reconstruct using the (memoized) return-type annotation
100            return cast("R", parse_value(func.__name__, cachedata, _resolve_return_hint(func)))
101
102        @functools.wraps(func)
103        async def wrapper(self: ProviderT, *args: P.args, **kwargs: P.kwargs) -> R:
104            cache = self.mass.cache
105            provider_id = getattr(self, "instance_id", self.domain)
106
107            # create a cache key dynamically based on the (remaining) args/kwargs
108            cache_key_parts = [func.__name__, *args]
109            for key in sorted(kwargs.keys()):
110                cache_key_parts.append(f"{key}{kwargs[key]}")
111            cache_key = ".".join(map(str, cache_key_parts))
112
113            # single lookup that returns the entry, its freshness and whether it was found;
114            # found distinguishes a stored None (a negative hit) from a cache miss. expired
115            # entries are only read (and deserialized) when this call serves stale data
116            cachedata, is_fresh, found = await cache.get_with_freshness(
117                cache_key,
118                provider=provider_id,
119                checksum=cache_checksum,
120                category=category,
121                allow_bypass=allow_bypass,
122                base_class=base_class,
123                include_expired=allow_expired_cache,
124            )
125            # a found value counts as a hit, except a None that this call opted out of
126            # caching (cache_none=False) which is treated as a miss and re-fetched
127            cache_hit = found and (cache_none or cachedata is not None)
128
129            if cache_hit and is_fresh:
130                return _reconstruct(cachedata)
131
132            def _store_task(result: R) -> Any:
133                return cache.set(
134                    key=cache_key,
135                    data=cast("SerializableType", result),
136                    expiration=expiration,
137                    provider=provider_id,
138                    category=category,
139                    checksum=cache_checksum,
140                    persistent=persistent,
141                    allow_expired_cache=allow_expired_cache,
142                )
143
144            if cache_hit and allow_expired_cache:
145                # serve stale data and refresh in the background;
146                # task_id deduplicates concurrent refreshes for the same entry
147                async def _background_refresh() -> None:
148                    try:
149                        result = await func(self, *args, **kwargs)
150                        if cache_none or result is not None:
151                            await _store_task(result)
152                    except Exception:
153                        LOGGER.exception(
154                            "Background cache refresh failed for %s/%s",
155                            provider_id,
156                            cache_key,
157                        )
158
159                self.mass.create_task(
160                    _background_refresh(),
161                    task_id=f"cache_refresh.{provider_id}.{cache_key}",
162                )
163                return _reconstruct(cachedata)
164
165            # cache miss (or expired entry without stale-while-revalidate):
166            # fetch synchronously, store in background
167            async def _fetch_and_store() -> R:
168                result = await func(self, *args, **kwargs)
169                if cache_none or result is not None:
170                    self.mass.create_task(_store_task(result))
171                return result
172
173            # a caller that bypasses this method's cache asked for the backend, so it
174            # fetches alone rather than joining or publishing a flight
175            if not single_flight or (allow_bypass and BYPASS_CACHE.get()):
176                return await _fetch_and_store()
177
178            async def _flight() -> _FlightOutcome[R]:
179                # the outcome is returned rather than raised, to keep a routine failure quiet:
180                # a MediaNotFoundError out of a cached lookup is an ordinary result that
181                # callers deal with themselves, while a raising task would both draw a warning
182                # from create_task and, once every caller has gone, have asyncio report it
183                # against the shielded future
184                outcome: _FlightOutcome[R] = _FlightOutcome()
185                try:
186                    outcome.result = await _fetch_and_store()
187                except Exception as err:
188                    outcome.error = err
189                return outcome
190
191            # task_id folds concurrent callers for this key onto one execution
192            flight = self.mass.create_task(
193                _flight(), task_id=f"cache_flight.{provider_id}.{cache_key}"
194            )
195            if flight is asyncio.current_task():
196                # a body that calls back into itself for the same key is handed the very
197                # fetch it is running in, and awaiting that would wait on itself forever
198                return await _fetch_and_store()
199            # the shield keeps the shared task out of every caller's cancellation scope: a
200            # caller giving up cancels neither the fetch nor the callers still waiting
201            outcome = await asyncio.shield(flight)
202            if outcome.error is not None:
203                raise outcome.error
204            # the fetched objects stay behind with the flight, which keeps them pristine for
205            # the cache write; callers get a clone each because they do mutate results in
206            # place (per-user podcast resume state, for one)
207            try:
208                return deepcopy(cast("R", outcome.result))
209            except Exception as err:
210                LOGGER.warning(
211                    "Cannot copy the shared result for %s/%s, callers share one object: %s",
212                    provider_id,
213                    cache_key,
214                    err,
215                )
216                return cast("R", outcome.result)
217
218        return wrapper
219
220    return _decorator
221
222
223@dataclass(slots=True)
224class _FlightOutcome[ResultT]:
225    """Outcome of one shared fetch, handed to every caller awaiting that fetch."""
226
227    result: ResultT | None = None
228    error: Exception | None = None
229
230
231@functools.cache
232def _resolve_return_hint(func: Callable[..., Any]) -> Any:
233    """
234    Return the resolved return-type annotation of func, memoized per function.
235
236    A function's return annotation is invariant for the process lifetime, so it is resolved
237    once and cached instead of re-running get_type_hints() — which re-evaluates the PEP-563
238    string annotations — on every cache hit. Resolution stays lazy (it happens on the first
239    hit, not at decoration time), so forward-reference handling is unchanged.
240
241    :param func: The decorated function whose return-type annotation to resolve.
242    """
243    return get_type_hints(func)["return"]
244