music-assistant-server

1"""Cached-payload mixin for providers whose recommendations come from one bulk payload."""
2
3from __future__ import annotations
4
5import asyncio
6import dataclasses
7import logging
8import time
9from typing import TYPE_CHECKING, TypeVar, cast
10
11from music_assistant_models.media_items import RecommendationFolder
12from music_assistant_models.unique_list import UniqueList
13
14if TYPE_CHECKING:
15    from collections.abc import Coroutine
16    from typing import Any
17
18    from music_assistant_models.media_items import BrowseFolder, ItemMapping, MediaItemType
19
20    from music_assistant.helpers.json import SerializableType
21
22    # type the mixin against the Provider base it composes with, so self.mass,
23    # self.logger, instance_id and the unload() chain resolve without redeclaring
24    # any (final) Provider members; at runtime the mixin stays a plain object base
25    from music_assistant.models.provider import Provider as _MixinBase
26else:
27    _MixinBase = object
28
29_T = TypeVar("_T")
30
31# Key of the persistent payload cache entry (stored with provider=instance_id).
32# Kept identical to the key the previous @use_cache-based implementation produced,
33# so payloads persisted before the redesign still warm cold instances.
34_PAYLOAD_CACHE_KEY = "_cached_recommendation_payload"
35
36
37class RecommendationPayloadMixin(_MixinBase):
38    """
39    Mixin serving recommendation rows and items from a single (cached) bulk payload.
40
41    The provider implements _fetch_recommendation_payload() with its existing bulk
42    backend fetch+parse; the mixin derives both the fast rows call and the per-row
43    items call from that payload.
44
45    Caching works in two layers:
46    - In memory: the last fetched payload is kept on the instance and served directly
47      while it is younger than recommendation_payload_ttl (no cache-db round trip).
48      A stale in-memory payload is served immediately while a single background
49      refresh replaces it (stale-while-revalidate).
50    - Persistent: every successful fetch is stored in mass.cache (persistent=True), so
51      a cold instance (e.g. after a restart) warms from the cache db with at most one
52      read; a stale persisted payload is likewise served while one refresh runs.
53
54    Concurrent callers share one in-flight fetch (single-flight); the shared fetch is
55    isolated from caller cancellation, so a timed-out caller cannot cancel it for the
56    other waiters and its result still lands in memory and cache.
57
58    All background work runs in tasks created via mass.create_task, so it is cancelled
59    on server stop; the mixin's unload() additionally cancels any in-flight fetch or
60    refresh when the provider itself is unloaded. For that override to be reachable,
61    list the mixin before the provider base class, e.g.
62    ``class MyProvider(RecommendationPayloadMixin, MusicProvider)``. A cancelled fetch
63    does not poison later calls: the next call simply starts a fresh one.
64    """
65
66    recommendation_payload_ttl: int = 3600
67    """Seconds a fetched payload is served as fresh. Subclasses may override."""
68
69    _recommendation_payload_task: asyncio.Task[list[RecommendationFolder]] | None = None
70    _recommendation_refresh_task: asyncio.Task[list[RecommendationFolder]] | None = None
71    _recommendation_payload_memory: list[RecommendationFolder] | None = None
72    _recommendation_payload_timestamp: float = 0.0
73    _recommendation_payload_cache_checked: bool = False
74
75    async def unload(self, is_removed: bool = False) -> None:
76        """
77        Handle unload/close of the provider.
78
79        Cancels any in-flight payload fetch/refresh task and awaits its completion,
80        so no payload work keeps running while the provider tears down, before
81        continuing the regular provider unload chain.
82
83        :param is_removed: True when the provider is removed from the configuration.
84        """
85        tasks = [
86            task
87            for task in (self._recommendation_payload_task, self._recommendation_refresh_task)
88            if task is not None and not task.done()
89        ]
90        for task in tasks:
91            task.cancel()
92        if tasks:
93            await asyncio.gather(*tasks, return_exceptions=True)
94        self._recommendation_payload_task = None
95        self._recommendation_refresh_task = None
96        await super().unload(is_removed)
97
98    async def _recommendation_rows_from_payload(self) -> list[RecommendationFolder]:
99        """Return all payload folders as rows, without items."""
100        # replace() keeps every payload field (image, is_playable, media_type, type, ...)
101        # instead of enumerating them, so new folder fields survive the copy automatically
102        return [
103            dataclasses.replace(folder, items=UniqueList())
104            for folder in await self._recommendation_payload()
105        ]
106
107    async def _recommendation_items_from_payload(
108        self, item_id: str
109    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
110        """
111        Return the items of the payload folder matching the given item_id.
112
113        :param item_id: The item_id of the recommendation folder (empty result if unknown).
114        """
115        for folder in await self._recommendation_payload():
116            if folder.item_id == item_id:
117                return folder.items
118        return UniqueList()
119
120    async def _recommendation_payload(self) -> list[RecommendationFolder]:
121        """Return the full recommendations payload, cached and deduplicated."""
122        if (payload := self._recommendation_payload_memory) is not None:
123            if (
124                time.monotonic() - self._recommendation_payload_timestamp
125                < self.recommendation_payload_ttl
126            ):
127                return payload
128            # stale: serve immediately while a (deduplicated) background refresh runs
129            self._schedule_recommendation_refresh()
130            return payload
131        # cold instance: single-flight load from the persistent cache or the backend
132        task = self._recommendation_payload_task
133        if task is None or task.done():
134            task = self._start_recommendation_task(
135                self._cold_load_recommendation_payload(),
136                task_id=f"recommendation_payload_fetch.{self.instance_id}",
137            )
138            self._recommendation_payload_task = task
139        # wait for the shared load instead of awaiting it directly: a timed-out caller must
140        # not cancel it out from under the other waiters (and the load must still complete
141        # to warm memory + cache). asyncio.shield achieves the same, but as of Python 3.14 a
142        # cancelled caller makes it report the load's exception through
143        # loop.call_exception_handler, even when another caller already handled it.
144        if not task.done():
145            await asyncio.wait((task,))
146        return task.result()
147
148    async def _refresh_recommendation_payload(self) -> list[RecommendationFolder]:
149        """
150        Force-fetch a fresh payload and store it in memory and the payload cache.
151
152        Unlike _recommendation_payload, this never serves cached data: use it when the
153        cached payload is known to be outdated (e.g. after detecting rotated backend ids).
154        Subsequent _recommendation_payload calls serve the refreshed payload.
155        """
156        # wait for the shared refresh rather than awaiting it: see _recommendation_payload
157        task = self._schedule_recommendation_refresh()
158        if not task.done():
159            await asyncio.wait((task,))
160        return task.result()
161
162    def _schedule_recommendation_refresh(self) -> asyncio.Task[list[RecommendationFolder]]:
163        """Return the in-flight refresh task, starting one if none is running."""
164        task = self._recommendation_refresh_task
165        if task is None or task.done():
166            task = self._start_recommendation_task(
167                self._fetch_and_store_recommendation_payload(),
168                task_id=f"recommendation_payload_refresh.{self.instance_id}",
169            )
170            self._recommendation_refresh_task = task
171        return task
172
173    async def _cold_load_recommendation_payload(self) -> list[RecommendationFolder]:
174        """Load the payload on a cold instance: persisted cache entry or backend fetch."""
175        if not self._recommendation_payload_cache_checked:
176            data, is_fresh, found = await self.mass.cache.get_with_freshness(
177                _PAYLOAD_CACHE_KEY,
178                provider=self.instance_id,
179                base_class=RecommendationFolder,
180                include_expired=True,
181            )
182            # set after the await: a cancelled read is retried on the next cold call
183            self._recommendation_payload_cache_checked = True
184            if found and data is not None:
185                payload = cast("list[RecommendationFolder]", data)
186                self._recommendation_payload_memory = payload
187                if is_fresh:
188                    self._recommendation_payload_timestamp = time.monotonic()
189                else:
190                    # expired entry: serve it now, refresh it in the background
191                    self._recommendation_payload_timestamp = (
192                        time.monotonic() - self.recommendation_payload_ttl
193                    )
194                    self._schedule_recommendation_refresh()
195                return payload
196        return await self._fetch_and_store_recommendation_payload()
197
198    async def _fetch_and_store_recommendation_payload(self) -> list[RecommendationFolder]:
199        """Fetch the payload from the backend and store it in memory + persistent cache."""
200        payload = await self._fetch_recommendation_payload()
201        # memory first: warm calls are served from here without touching the cache db,
202        # and a caller landing before the background store completes still gets a hit
203        self._recommendation_payload_memory = payload
204        self._recommendation_payload_timestamp = time.monotonic()
205        self._start_recommendation_task(
206            self.mass.cache.set(
207                key=_PAYLOAD_CACHE_KEY,
208                data=cast("SerializableType", payload),
209                expiration=self.recommendation_payload_ttl,
210                provider=self.instance_id,
211                persistent=True,
212                allow_expired_cache=True,
213            )
214        )
215        return payload
216
217    def _start_recommendation_task(
218        self, coro: Coroutine[Any, Any, _T], task_id: str | None = None
219    ) -> asyncio.Task[_T]:
220        """Create a tracked background task that logs failures through the provider logger."""
221        task = self.mass.create_task(coro, task_id=task_id)
222        # always retrieve the task's exception: without a waiter (background refresh,
223        # cancelled sole caller) it would otherwise surface as a raw loop-handler error
224        task.add_done_callback(self._log_recommendation_task_failure)
225        return task
226
227    def _log_recommendation_task_failure(self, task: asyncio.Task[Any]) -> None:
228        """Log (and thereby retrieve) the exception of a finished payload task, if any."""
229        if task.cancelled():
230            return
231        if (err := task.exception()) is not None:
232            self.logger.warning(
233                "Recommendation payload task failed: %s",
234                str(err),
235                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
236            )
237
238    async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
239        """Fetch and parse the full recommendations payload (folders WITH items)."""
240        # the provider using this mixin supplies its bulk backend fetch here
241        raise NotImplementedError
242