Coverage for src / beautyspot / cache.py: 88%
185 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-03-10 23:54 +0900
1# src/beautyspot/cache.py
3import hashlib
4import logging
5import threading
6import time
7import asyncio
8from datetime import datetime, timezone
9from typing import Any, Callable, Optional, NamedTuple
11from beautyspot.db import TaskDBMaintenable
12from beautyspot.storage import BlobStorageMaintenable, StoragePolicyProtocol
13from beautyspot.serializer import SerializerProtocol
14from beautyspot.lifecycle import (
15 LifecyclePolicy,
16 RetentionSpec,
17 parse_retention,
18 _ForeverSentinel,
19 _FOREVER,
20)
21from beautyspot.cachekey import KeyGen
22from beautyspot.exceptions import CacheCorruptedError
23from beautyspot.content_types import ContentType
25logger = logging.getLogger(__name__)
26logger.addHandler(logging.NullHandler())
28# --- キャッシュミスを表す番兵オブジェクト ---
29CACHE_MISS = object()
32class HerdWaitResult(NamedTuple):
33 """Thundering Herd 待機フェーズの結果。"""
35 is_executor: bool # True: 自分が実行者になった
36 result: Any # is_executor=False のときの結果 or 例外
37 event: threading.Event | None # is_executor=True のときのイベント
38 result_box: list # is_executor=True のときの共有リスト
39 is_error: bool # result が例外の場合 True
42class CacheManager:
43 """
44 キャッシュの読み書き、キー生成、および並行実行制御（Thundering Herd対策）を
45 担当するコンポーネント。
46 """
48 HERD_POLL: float = 5.0
49 HERD_TIMEOUT: float = 300.0
50 HERD_MAX_RETRIES: int = 3
52 def __init__(
53 self,
54 db: TaskDBMaintenable,
55 storage: BlobStorageMaintenable,
56 serializer: SerializerProtocol,
57 storage_policy: StoragePolicyProtocol,
58 lifecycle_policy: Optional[LifecyclePolicy] = None,
59 ):
60 self.db = db
61 self.storage = storage
62 self.serializer = serializer
63 self.storage_policy = storage_policy
65 if lifecycle_policy is not None:
66 self.lifecycle_policy = lifecycle_policy
67 else:
68 self.lifecycle_policy = LifecyclePolicy.default()
70 # サンダリングハード対策: 同一キーの並行実行を直列化する
71 # tuple: (threading.Event, list[asyncio.Future], list[result])
72 self._inflight: dict[
73 str, tuple[threading.Event, list[asyncio.Future], list]
74 ] = {}
75 self._inflight_lock = threading.Lock()
77 def make_cache_key(
78 self,
79 func_identifier: str,
80 args: tuple,
81 kwargs: dict,
82 resolved_key_fn: Optional[Callable],
83 version: str | None,
84 ) -> tuple[str, str]:
85 """キャッシュキーと入力IDを生成する。"""
86 iid = (
87 resolved_key_fn(*args, **kwargs)
88 if resolved_key_fn
89 else KeyGen._default(args, kwargs)
90 )
92 key_source = f"{func_identifier}:{iid}"
93 if version:
94 key_source += f":{version}"
96 ck = hashlib.sha256(key_source.encode()).hexdigest()
97 return iid, ck
99 def calculate_expires_at(
100 self,
101 func_identifier: str,
102 func_name: str,
103 local_retention: RetentionSpec,
104 ) -> Optional[datetime]:
105 """有効期限を計算する。"""
106 if local_retention is _FOREVER:
107 return None
109 if isinstance(local_retention, _ForeverSentinel): 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true
110 raise RuntimeError(
111 "Internal Error: _ForeverSentinel reached calculate_expires_at."
112 )
114 retention = parse_retention(local_retention)
116 if retention is None:
117 retention = self.lifecycle_policy.resolve_with_fallback(
118 func_identifier, func_name
119 )
121 if retention is None: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
122 return None
124 return datetime.now(timezone.utc) + retention
126 def get(
127 self, cache_key: str, serializer: Optional[SerializerProtocol] = None
128 ) -> Any:
129 """同期的にキャッシュから値を取得する。"""
130 use_serializer = serializer or self.serializer
131 entry = self.db.get(cache_key)
133 if not entry:
134 return CACHE_MISS
136 r_type = entry["result_type"]
137 r_val = entry["result_value"]
138 r_data = entry.get("result_data")
140 try:
141 if r_type == "DIRECT_BLOB":
142 if r_data is None:
143 return CACHE_MISS
144 return use_serializer.loads(r_data)
146 elif r_type == "FILE":
147 if r_val is None:
148 logger.warning(
149 f"Data corruption: 'FILE' record has no path for key `{cache_key}`"
150 )
151 return CACHE_MISS
152 data_bytes = self.storage.load(r_val)
153 return use_serializer.loads(data_bytes)
155 else:
156 logger.warning(
157 f"Unknown result_type '{r_type}' for cache_key `{cache_key}`"
158 )
159 return CACHE_MISS
161 except CacheCorruptedError as e:
162 logger.debug(f"Cache corrupted for {cache_key}: {e}")
163 return CACHE_MISS
164 except Exception as e:
165 logger.error(
166 f"Failed to deserialize cache for `{cache_key}`: {e}", exc_info=True
167 )
168 return CACHE_MISS
170 def set(
171 self,
172 cache_key: str,
173 func_name: str,
174 func_identifier: str,
175 input_id: str,
176 version: str | None,
177 result: Any,
178 content_type: str | ContentType | None,
179 save_blob: bool | None,
180 expires_at: Optional[datetime] = None,
181 serializer: Optional[SerializerProtocol] = None,
182 ) -> None:
183 """同期的にキャッシュへ値を保存する。"""
184 use_serializer = serializer or self.serializer
185 data_bytes = use_serializer.dumps(result)
187 should_use_blob = save_blob
188 if should_use_blob is None:
189 should_use_blob = self.storage_policy.should_save_as_blob(data_bytes)
191 if should_use_blob:
192 r_val = self.storage.save(cache_key, data_bytes)
193 try:
194 self.db.save(
195 cache_key=cache_key,
196 func_name=func_name,
197 func_identifier=func_identifier,
198 input_id=input_id,
199 version=version,
200 result_type="FILE",
201 content_type=content_type,
202 result_value=r_val,
203 result_data=None,
204 expires_at=expires_at,
205 )
206 except Exception:
207 try:
208 self.storage.delete(r_val)
209 except Exception as rollback_err:
210 logger.warning(f"Failed to rollback blob '{r_val}': {rollback_err}")
211 raise
212 else:
213 self.db.save(
214 cache_key=cache_key,
215 func_name=func_name,
216 func_identifier=func_identifier,
217 input_id=input_id,
218 version=version,
219 result_type="DIRECT_BLOB",
220 content_type=content_type,
221 result_value=None,
222 result_data=data_bytes,
223 expires_at=expires_at,
224 )
226 # --- Thundering Herd Protection ---
228 def wait_herd_sync(
229 self, cache_key: str, serializer: Optional[SerializerProtocol] = None
230 ) -> HerdWaitResult:
231 """同期パスでの Thundering Herd 待機。"""
232 retries = 0
233 while True:
234 with self._inflight_lock:
235 if cache_key not in self._inflight:
236 event = threading.Event()
237 result_box: list = []
238 self._inflight[cache_key] = (event, [], result_box)
239 return HerdWaitResult(True, None, event, result_box, False)
241 wait_event, _, wait_box = self._inflight[cache_key]
243 deadline = time.monotonic() + self.HERD_TIMEOUT
244 while not wait_event.wait(timeout=self.HERD_POLL):
245 if time.monotonic() >= deadline:
246 retries += 1
247 if retries > self.HERD_MAX_RETRIES: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
248 raise TimeoutError(f"Herd wait timeout for {cache_key} exceeded max retries ({self.HERD_MAX_RETRIES})")
249 logger.warning(f"Herd wait timeout for {cache_key} (retry {retries}/{self.HERD_MAX_RETRIES})")
250 break
252 if wait_box:
253 success, val = wait_box[0]
254 return HerdWaitResult(False, val, None, [], not success)
256 # 万が一の結果漏れに備えて再チェック
257 cached = self.get(cache_key, serializer)
258 if cached is not CACHE_MISS: 258 ↛ 233line 258 didn't jump to line 233 because the condition on line 258 was always true
259 return HerdWaitResult(False, cached, None, [], False)
261 async def wait_herd_async(
262 self,
263 cache_key: str,
264 serializer: Optional[SerializerProtocol],
265 loop: asyncio.AbstractEventLoop,
266 executor: Any,
267 ) -> HerdWaitResult:
268 """非同期パスでの Thundering Herd 待機。"""
269 retries = 0
270 while True:
271 fut = None
272 with self._inflight_lock:
273 if cache_key not in self._inflight:
274 event = threading.Event()
275 result_box: list = []
276 self._inflight[cache_key] = (event, [], result_box)
277 return HerdWaitResult(True, None, event, result_box, False)
279 wait_event, futs, wait_box = self._inflight[cache_key]
280 if not wait_box:
281 fut = loop.create_future()
282 futs.append(fut)
284 signal = await self._await_herd_signal_async(
285 fut, wait_event, wait_box, cache_key, loop, executor
286 )
287 if signal is None: 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true
288 retries += 1
289 if retries > self.HERD_MAX_RETRIES:
290 raise TimeoutError(f"Herd wait timeout for {cache_key} exceeded max retries ({self.HERD_MAX_RETRIES})")
291 logger.warning(f"Herd wait timeout for {cache_key} (retry {retries}/{self.HERD_MAX_RETRIES})")
292 continue
294 success, val = signal
295 return HerdWaitResult(False, val, None, [], not success)
297 async def _await_herd_signal_async(
298 self,
299 fut: Optional[asyncio.Future],
300 wait_event: threading.Event,
301 wait_box: list,
302 cache_key: str,
303 loop: asyncio.AbstractEventLoop,
304 executor: Any,
305 ) -> Optional[tuple[bool, Any]]:
306 if fut is not None:
307 try:
308 val = await asyncio.wait_for(
309 asyncio.shield(fut), timeout=self.HERD_TIMEOUT
310 )
311 return (True, val)
312 except asyncio.TimeoutError:
313 return None
314 except Exception as e:
315 return (False, e)
317 if wait_box: 317 ↛ 320line 317 didn't jump to line 320 because the condition on line 317 was always true
318 return wait_box[0]
320 deadline = time.monotonic() + self.HERD_TIMEOUT
321 while not await loop.run_in_executor(executor, wait_event.wait, self.HERD_POLL):
322 if time.monotonic() >= deadline:
323 return None
325 return wait_box[0] if wait_box else None
327 def notify_and_cleanup_inflight(
328 self,
329 cache_key: str,
330 event: Optional[threading.Event],
331 result_box: list,
332 ) -> None:
333 """待機中のスレッド/タスクに通知し、管理情報を削除する。"""
334 futs_to_notify: list = []
335 with self._inflight_lock:
336 val = self._inflight.get(cache_key)
337 if val is not None and val[0] is event:
338 _, futs_to_notify, _ = val
339 del self._inflight[cache_key]
341 if event is not None: 341 ↛ exitline 341 didn't return from function 'notify_and_cleanup_inflight' because the condition on line 341 was always true
342 event.set()
343 if result_box and futs_to_notify:
344 success, res_val = result_box[0]
345 for fut in futs_to_notify:
346 if not fut.done(): 346 ↛ 345line 346 didn't jump to line 345 because the condition on line 346 was always true
347 self._notify_future(fut, success, res_val)
349 def _notify_future(self, fut: asyncio.Future, success: bool, val: Any) -> None:
350 def _set():
351 if not fut.done(): 351 ↛ exitline 351 didn't return from function '_set' because the condition on line 351 was always true
352 if success:
353 fut.set_result(val)
354 elif isinstance(val, BaseException):
355 fut.set_exception(val)
356 else:
357 fut.set_exception(RuntimeError(f"Non-Exception error: {repr(val)}"))
359 fut.get_loop().call_soon_threadsafe(_set)