Spot (Core)

Spot クラスは beautyspot のメインエントリポイントです。関数のマーキング（登録）、キャッシュのルックアップ、および実行結果の永続化をオーケストレートします。

beautyspot.core

BeautySpot コアモジュール。

タスク管理、シリアライズ、キャッシュとストレージを含むリソース管理を行うメインクラス群を提供します。

Spot

タスク管理、シリアライズ、キャッシュとストレージを含むリソース管理を行うメインクラス。

依存オブジェクト（CacheManagerやLimiterProtocolなど）を注入して初期化されます。 通常は直接インスタンス化せず、bs.Spot(...) ファクトリ関数を通じて使用することが推奨されます。

Parameters:

Name	Type	Description	Default
name	str	Spotインスタンスの名前。	required
cache	CacheManager	キャッシュマネージャーのインスタンス。	required
limiter	LimiterProtocol	レートリミッターのインスタンス。	required
save_sync	bool	キャッシュ保存のデフォルト同期動作。デフォルトはTrue。	True
eviction_rate	float	キャッシュの自動破棄を実行する確率（0.0〜1.0）。デフォルトは0.0。	0.0
drain_timeout	float	バックグラウンドタスク完了待機のタイムアウト（秒）。デフォルトは5.0。	5.0
drain_poll_interval	float	バックグラウンドタスク待機時のポーリング間隔（秒）。デフォルトは0.5。	0.5
on_background_error	Optional[Callable[[Exception, SaveErrorContext], None]]	バックグラウンド保存時のエラーハンドラ。	None

Source code in src/beautyspot/core.py

class Spot:
    """タスク管理、シリアライズ、キャッシュとストレージを含むリソース管理を行うメインクラス。

    依存オブジェクト（CacheManagerやLimiterProtocolなど）を注入して初期化されます。
    通常は直接インスタンス化せず、`bs.Spot(...)` ファクトリ関数を通じて使用することが推奨されます。

    Args:
        name (str): Spotインスタンスの名前。
        cache (CacheManager): キャッシュマネージャーのインスタンス。
        limiter (LimiterProtocol): レートリミッターのインスタンス。
        save_sync (bool, optional): キャッシュ保存のデフォルト同期動作。デフォルトはTrue。
        eviction_rate (float, optional): キャッシュの自動破棄を実行する確率（0.0〜1.0）。デフォルトは0.0。
        drain_timeout (float, optional): バックグラウンドタスク完了待機のタイムアウト（秒）。デフォルトは5.0。
        drain_poll_interval (float, optional): バックグラウンドタスク待機時のポーリング間隔（秒）。デフォルトは0.5。
        on_background_error (Optional[Callable[[Exception, SaveErrorContext], None]], optional): バックグラウンド保存時のエラーハンドラ。
    """

    def __init__(
        self,
        name: str,
        cache: CacheManager,
        limiter: LimiterProtocol,
        save_sync: bool = True,
        eviction_rate: float = 0.0,
        drain_timeout: float = 5.0,
        drain_poll_interval: float = 0.5,
        on_background_error: Optional[
            Callable[[Exception | BaseException, SaveErrorContext], None]
        ] = None,
    ) -> None:
        self.name = name
        if not (0.0 <= eviction_rate <= 1.0):
            raise ValueError("eviction_rate must be between 0.0 and 1.0")
        self.eviction_rate = eviction_rate
        if drain_timeout <= 0:
            raise ValueError("drain_timeout must be positive")
        if drain_poll_interval <= 0:
            raise ValueError("drain_poll_interval must be positive")
        self._drain_timeout = drain_timeout
        self._drain_poll_interval = drain_poll_interval

        # --- コンポーネントの保持 ---
        self.cache = cache
        self.limiter = limiter

        # --- オプション設定の適用 ---
        self._save_sync = save_sync
        self.on_background_error = on_background_error

        # --- DBの初期化 ---
        self.cache.db.init_schema()

        # --- バックグラウンド IO 管理 ---
        self._bg_loop: _BackgroundLoop | None = None
        self._executor: Executor | None = None
        self._finalizer: weakref.finalize | None = None

        self._bg_init_lock = threading.Lock()
        self._shutdown_called = False
        self._owns_db = False

        self._active_futures: set = set()
        self._futures_lock = threading.Lock()

        self._maintenance_service: MaintenanceService | None = None
        self._maintenance_lock = threading.Lock()
        self._eviction_guard_lock = threading.Lock()
        self._eviction_running = False
        self._last_eviction_time = 0.0

    def __enter__(self) -> "Spot":
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self._drain_futures()

    def _track_future(self, future: Any):
        if future is None:
            return
        with self._futures_lock:
            self._active_futures.add(future)

        def _on_done(f):
            with self._futures_lock:
                self._active_futures.discard(f)

        future.add_done_callback(_on_done)

    @property
    def maintenance(self) -> MaintenanceService:
        svc = self._maintenance_service
        if svc is None:
            with self._maintenance_lock:
                if self._maintenance_service is None:
                    self._maintenance_service = MaintenanceService(
                        db=self.cache.db,
                        storage=self.cache.storage,
                        serializer=self.cache.serializer,
                    )
                svc = self._maintenance_service
        assert svc is not None
        return svc

    def _ensure_bg_resources(self) -> tuple[_BackgroundLoop, Executor]:
        bg, ex = self._bg_loop, self._executor
        if bg is not None and ex is not None:
            return bg, ex

        with self._bg_init_lock:
            if self._shutdown_called:
                raise RuntimeError(
                    "Cannot submit background tasks after shutdown() has been called."
                )
            if self._bg_loop is None or self._executor is None:
                if self._bg_loop is None:
                    self._bg_loop = _BackgroundLoop(drain_timeout=self._drain_timeout)
                if self._executor is None:
                    self._executor = ThreadPoolExecutor()
                if self._finalizer is None:
                    self._finalizer = weakref.finalize(
                        self,
                        Spot._shutdown_resources,
                        self._bg_loop,
                        self._executor,
                        self.cache.db,
                        self._owns_db,
                    )
                    self._finalizer.atexit = False
            return self._bg_loop, self._executor

    @staticmethod
    def _shutdown_resources(
        bg_loop: _BackgroundLoop,
        executor: Executor,
        db: TaskDBCore,
        owns_db: bool,
    ) -> None:
        bg_loop.stop(save_sync=False)
        executor.shutdown(wait=False, cancel_futures=True)
        if owns_db and isinstance(db, Shutdownable):
            db.shutdown(wait=False)

    def shutdown(self, save_sync: bool = True):
        """Spotインスタンスをシャットダウンし、バックグラウンドリソースを解放する。

        Args:
            save_sync (bool, optional): 同期的に未完了の保存タスクを待機するかどうか。Trueの場合は完了を待つ。デフォルトはTrue。
        """
        with self._bg_init_lock:
            self._shutdown_called = True
        if self._finalizer is not None and self._finalizer.alive:
            self._finalizer.detach()
        if save_sync:
            self._drain_futures()

        if self._bg_loop is not None:
            self._bg_loop.stop(save_sync=save_sync)

        if self._executor is not None:
            self._executor.shutdown(wait=save_sync, cancel_futures=not save_sync)

    def flush(self, timeout: Optional[float] = None) -> None:
        """バックグラウンドで実行中のすべての保存タスクとDBの書き込みの完了を待機する。

        Args:
            timeout (Optional[float], optional): 待機する最大時間（秒）。指定しない場合は初期化時の `drain_timeout` が使用される。
        """
        timeout_val = timeout if timeout is not None else self._drain_timeout
        deadline = time.monotonic() + timeout_val

        while True:
            with self._futures_lock:
                snapshot = list(self._active_futures)
            if not snapshot:
                break
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break
            wait_timeout = min(self._drain_poll_interval, remaining)
            wait(snapshot, timeout=wait_timeout)

        db_remaining = deadline - time.monotonic()
        if db_remaining > 0 and isinstance(self.cache.db, Flushable):
            self.cache.db.flush(timeout=db_remaining)

    def _drain_futures(self) -> None:
        self.flush()

    def _get_func_identifier(self, func: Callable) -> str:
        module = getattr(func, "__module__", None) or func.__class__.__module__
        qualname = getattr(func, "__qualname__", None) or func.__class__.__qualname__
        return f"{module}.{qualname}"

    def _trigger_auto_eviction(self) -> None:
        if self.eviction_rate <= 0.0:
            return
        if random.random() >= self.eviction_rate:
            return

        with self._eviction_guard_lock:
            if self._eviction_running:
                return
            now = time.monotonic()
            if now - self._last_eviction_time < 60.0:
                return
            self._eviction_running = True

        logger.debug(f"Triggering auto-eviction (rate: {self.eviction_rate})")

        with self._futures_lock:
            pending_futures = list(self._active_futures)

        def _run_clean_safe():
            try:
                self.maintenance.clean_garbage(orphan_grace_seconds=60.0)
            except Exception as e:
                logger.error(f"Auto-eviction failed: {e}", exc_info=True)

        def _clear_eviction_flag():
            with self._eviction_guard_lock:
                self._last_eviction_time = time.monotonic()
                self._eviction_running = False

        try:
            bg_loop, executor = self._ensure_bg_resources()

            async def _run_clean_coro():
                loop = asyncio.get_running_loop()
                if pending_futures:
                    await asyncio.wait(
                        [asyncio.wrap_future(f) for f in pending_futures],
                        timeout=self._drain_timeout,
                    )
                await loop.run_in_executor(executor, _run_clean_safe)

            future = bg_loop.submit(_run_clean_coro())
            if future:
                self._track_future(future)
                future.add_done_callback(lambda f: _clear_eviction_flag())
            else:
                _clear_eviction_flag()
        except Exception:
            _clear_eviction_flag()

    def _resolve_key_fn(
        self,
        func: Callable,
        keygen: Optional[Union[Callable, KeyGenPolicy]] = None,
        input_key_fn: Optional[Union[Callable, KeyGenPolicy]] = None,
    ) -> Optional[Callable]:
        if keygen is not None and input_key_fn is not None:
            raise IncompatibleProviderError("Cannot specify both 'keygen' and 'input_key_fn'.")
        if input_key_fn is not None:
            warnings.warn("`input_key_fn` is deprecated, use `keygen` instead.", DeprecationWarning, stacklevel=3)
        target = keygen or input_key_fn
        if isinstance(target, KeyGenPolicy):
            return target.bind(func)
        return target

    def register(
        self,
        code: int,
        encoder: Callable[[T], Any],
        decoder: Optional[Callable[[Any], T]] = None,
        decoder_factory: Optional[Callable[[Type[T]], Callable[[Any], T]]] = None,
    ) -> Callable[[Type[T]], Type[T]]:
        """カスタム型をシリアライザに登録するためのデコレータ。

        `decoder` または `decoder_factory` のいずれかを必ず提供する必要があります。

        Args:
            code (int): カスタム型の一意な識別コード。
            encoder (Callable[[T], Any]): カスタム型オブジェクトからシリアライズ可能な形式（辞書など）に変換する関数。
            decoder (Optional[Callable[[Any], T]], optional): デシリアライズ時にデータをカスタム型オブジェクトに復元する関数。
            decoder_factory (Optional[Callable[[Type[T]], Callable[[Any], T]]], optional): 型に基づいてデコーダ関数を生成するファクトリ関数。

        Returns:
            Callable[[Type[T]], Type[T]]: クラスデコレータ。

        Raises:
            IncompatibleProviderError: `decoder` と `decoder_factory` の両方が未指定の場合に発生。
        """
        if decoder is None and decoder_factory is None:
            raise IncompatibleProviderError(
                "Must provide either `decoder` or `decoder_factory`."
            )

        def decorator(cls: Type) -> Type:
            actual_decoder = decoder
            if decoder_factory:
                actual_decoder = decoder_factory(cls)

            if actual_decoder is None:
                raise ValueError("Decoder resolution failed.")

            self.register_type(cls, code, encoder, actual_decoder)
            return cls

        return decorator

    def register_type(
        self,
        type_class: Type[T],
        code: int,
        encoder: Callable[[T], Any],
        decoder: Callable[[Any], T],
    ):
        """カスタム型を直接シリアライザに登録する。

        Args:
            type_class (Type[T]): 登録するカスタム型のクラス。
            code (int): カスタム型の一意な識別コード。
            encoder (Callable[[T], Any]): エンコーダ関数。
            decoder (Callable[[Any], T]): デコーダ関数。

        Raises:
            NotImplementedError: 現在のシリアライザが型登録をサポートしていない場合。
        """
        if isinstance(self.cache.serializer, TypeRegistryProtocol):
            self.cache.serializer.register(type_class, code, encoder, decoder)
        else:
            raise NotImplementedError(
                "Current serializer does not support type registration."
            )

    @staticmethod
    def _dispatch_hooks(
        hooks: Optional[Sequence[HookBase]], method_name: str, context: Any
    ) -> None:
        if not hooks:
            return
        for hook in hooks:
            try:
                getattr(hook, method_name)(context)
            except Exception as e:
                logger.error(
                    f"Error in hook '{type(hook).__name__}.{method_name}': {e}",
                    exc_info=True,
                )

    async def _dispatch_hooks_async(
        self,
        hooks: Optional[Sequence[HookBase]],
        method_name: str,
        context: Any,
        loop: asyncio.AbstractEventLoop,
        executor: Executor,
    ) -> None:
        if not hooks:
            return
        await loop.run_in_executor(
            executor, self._dispatch_hooks, hooks, method_name, context
        )

    # --- Core Logic ---

    def _resolve_settings(
        self,
        save_blob: bool | None,
        version: str | None,
        content_type: str | ContentType | None,
        save_sync: bool | None,
    ) -> tuple[bool | None, str | None, str | None, bool]:
        return (
            save_blob,
            version,
            content_type,
            (save_sync if save_sync is not None else self._save_sync),
        )

    def _prepare_execution(
        self,
        func: Callable,
        args: tuple,
        kwargs: dict,
        save_blob: bool | None,
        effective_key_fn: Optional[Callable],
        version: str | None,
        content_type: Optional[str | ContentType],
        save_sync: bool | None,
        hooks: Optional[Sequence[HookBase]],
    ) -> _ExecutionContext:
        s_blob, s_ver, s_ct, s_save_sync = self._resolve_settings(
            save_blob, version, content_type, save_sync
        )
        func_identifier = self._get_func_identifier(func)
        iid, ck = self.cache.make_cache_key(
            func_identifier, args, kwargs, effective_key_fn, s_ver
        )
        return _ExecutionContext(
            s_blob,
            s_ver,
            s_ct,
            s_save_sync,
            func_identifier,
            iid,
            ck,
            dict(kwargs) if hooks else kwargs,
        )

    def _build_cache_hit_context(
        self,
        func_name: str,
        input_id: str,
        cache_key: str,
        args: tuple,
        hook_kwargs: dict,
        result: Any,
        version: str | None,
    ) -> CacheHitContext:
        return CacheHitContext(
            func_name=func_name,
            input_id=str(input_id),
            cache_key=cache_key,
            args=args,
            kwargs=hook_kwargs,
            result=result,
            version=version,
        )

    def _build_save_kwargs(
        self,
        cache_key: str,
        func: Callable,
        func_identifier: str,
        input_id: str,
        version: str | None,
        result: Any,
        content_type: str | None,
        save_blob: bool | None,
        serializer: Optional[SerializerProtocol],
        retention: RetentionSpec,
    ) -> dict:
        expires_at = self.cache.calculate_expires_at(
            func_identifier, func.__name__, retention
        )
        return {
            "cache_key": cache_key,
            "func_name": func.__name__,
            "func_identifier": func_identifier,
            "input_id": str(input_id),
            "version": version,
            "result": result,
            "content_type": content_type,
            "save_blob": save_blob,
            "serializer": serializer,
            "expires_at": expires_at,
        }

    def _persist_result_sync(self, save_sync: bool, save_kwargs: dict) -> None:
        if save_sync:
            try:
                self.cache.set(**save_kwargs)
            except Exception as e:
                self._handle_save_error(e, save_kwargs)
                raise
        else:
            try:
                self._submit_background_save(**save_kwargs)
            except Exception as e:
                self._handle_save_error(e, save_kwargs)
                if self.on_background_error is None:
                    raise

    async def _persist_result_async(self, save_sync: bool, save_kwargs: dict) -> None:
        if save_sync:
            try:
                bg_loop, exec_pool = self._ensure_bg_resources()
                coro = self._save_result_async(
                    executor=exec_pool, safe=False, **save_kwargs
                )
                future = bg_loop.submit(coro)
                if future is None:
                    self._notify_save_discarded(save_kwargs)
                    raise RuntimeError(
                        f"Cache save for '{save_kwargs.get('func_name')}' "
                        "was discarded because the background loop is shutting down."
                    )
                else:
                    await asyncio.wrap_future(future)
            except Exception as e:
                self._handle_save_error(e, save_kwargs)
                raise
        else:
            try:
                self._submit_background_save(**save_kwargs)
            except Exception as e:
                self._handle_save_error(e, save_kwargs)
                if self.on_background_error is None:
                    raise

    def _execute_sync(
        self,
        func: Callable,
        args: tuple,
        kwargs: dict,
        save_blob: bool | None,
        effective_key_fn: Optional[Callable],
        version: str | None,
        content_type: Optional[str | ContentType],
        serializer: Optional[SerializerProtocol],
        retention: RetentionSpec,
        save_sync: bool | None,
        hooks: Optional[Sequence[HookBase]] = None,
    ) -> Any:
        ctx = self._prepare_execution(
            func,
            args,
            kwargs,
            save_blob,
            effective_key_fn,
            version,
            content_type,
            save_sync,
            hooks,
        )
        self._dispatch_hooks(
            hooks,
            "pre_execute",
            PreExecuteContext(
                func.__name__, str(ctx.input_id), ctx.cache_key, args, ctx.hook_kwargs
            ),
        )
        cached = self.cache.get(ctx.cache_key, serializer)
        if cached is not CACHE_MISS:
            self._dispatch_hooks(
                hooks,
                "on_cache_hit",
                self._build_cache_hit_context(
                    func.__name__,
                    ctx.input_id,
                    ctx.cache_key,
                    args,
                    ctx.hook_kwargs,
                    cached,
                    ctx.version,
                ),
            )
            return cached
        herd = self.cache.wait_herd_sync(ctx.cache_key, serializer)
        if not herd.is_executor:
            if herd.is_error:
                raise herd.result
            self._dispatch_hooks(
                hooks,
                "on_cache_hit",
                self._build_cache_hit_context(
                    func.__name__,
                    ctx.input_id,
                    ctx.cache_key,
                    args,
                    ctx.hook_kwargs,
                    herd.result,
                    ctx.version,
                ),
            )
            return herd.result
        try:
            res = func(*args, **kwargs)
            self._dispatch_hooks(
                hooks,
                "on_cache_miss",
                CacheMissContext(
                    func.__name__,
                    str(ctx.input_id),
                    ctx.cache_key,
                    args,
                    ctx.hook_kwargs,
                    res,
                    ctx.version,
                ),
            )
            herd.result_box.append((True, res))

            # 実行成功後、同期モード(save_sync=True)の場合はキャッシュ保存エラーを伝播させる
            try:
                save_kwargs = self._build_save_kwargs(
                    ctx.cache_key,
                    func,
                    ctx.func_identifier,
                    ctx.input_id,
                    ctx.version,
                    res,
                    ctx.content_type,
                    ctx.save_blob,
                    serializer,
                    retention,
                )
                self._persist_result_sync(ctx.save_sync, save_kwargs)
            except Exception as e:
                if ctx.save_sync:
                    raise
                logger.error(f"Failed to submit background cache save, but execution succeeded: {e}")

            return res

        except BaseException as e:
            if not herd.result_box:
                herd.result_box.append((False, e))
            raise
        finally:
            self.cache.notify_and_cleanup_inflight(
                ctx.cache_key, herd.event, herd.result_box
            )
            self._trigger_auto_eviction()

    async def _execute_async(
        self,
        func: Callable,
        args: tuple,
        kwargs: dict,
        save_blob: bool | None,
        effective_key_fn: Optional[Callable],
        version: str | None,
        content_type: Optional[str | ContentType],
        serializer: Optional[SerializerProtocol],
        retention: RetentionSpec,
        save_sync: bool | None,
        hooks: Optional[Sequence[HookBase]] = None,
    ) -> Any:
        ctx = self._prepare_execution(
            func,
            args,
            kwargs,
            save_blob,
            effective_key_fn,
            version,
            content_type,
            save_sync,
            hooks,
        )
        loop = asyncio.get_running_loop()
        _, executor = self._ensure_bg_resources()
        await self._dispatch_hooks_async(
            hooks,
            "pre_execute",
            PreExecuteContext(
                func.__name__, str(ctx.input_id), ctx.cache_key, args, ctx.hook_kwargs
            ),
            loop,
            executor,
        )
        cached = await loop.run_in_executor(
            executor, self.cache.get, ctx.cache_key, serializer
        )
        if cached is not CACHE_MISS:
            await self._dispatch_hooks_async(
                hooks,
                "on_cache_hit",
                self._build_cache_hit_context(
                    func.__name__,
                    ctx.input_id,
                    ctx.cache_key,
                    args,
                    ctx.hook_kwargs,
                    cached,
                    ctx.version,
                ),
                loop,
                executor,
            )
            return cached
        herd = await self.cache.wait_herd_async(
            ctx.cache_key, serializer, loop, executor
        )
        if not herd.is_executor:
            if herd.is_error:
                raise herd.result
            await self._dispatch_hooks_async(
                hooks,
                "on_cache_hit",
                self._build_cache_hit_context(
                    func.__name__,
                    ctx.input_id,
                    ctx.cache_key,
                    args,
                    ctx.hook_kwargs,
                    herd.result,
                    ctx.version,
                ),
                loop,
                executor,
            )
            return herd.result
        try:
            res = await func(*args, **kwargs)
            await self._dispatch_hooks_async(
                hooks,
                "on_cache_miss",
                CacheMissContext(
                    func.__name__,
                    str(ctx.input_id),
                    ctx.cache_key,
                    args,
                    ctx.hook_kwargs,
                    res,
                    ctx.version,
                ),
                loop,
                executor,
            )
            herd.result_box.append((True, res))

            # 実行成功後、同期モード(save_sync=True)の場合はキャッシュ保存エラーを伝播させる
            try:
                save_kwargs = self._build_save_kwargs(
                    ctx.cache_key,
                    func,
                    ctx.func_identifier,
                    ctx.input_id,
                    ctx.version,
                    res,
                    ctx.content_type,
                    ctx.save_blob,
                    serializer,
                    retention,
                )
                await self._persist_result_async(ctx.save_sync, save_kwargs)
            except Exception as e:
                if ctx.save_sync:
                    raise
                logger.error(f"Failed to persist cache asynchronously, but execution succeeded: {e}")

            return res

        except BaseException as e:
            if not herd.result_box:
                herd.result_box.append((False, e))
            raise
        finally:
            self.cache.notify_and_cleanup_inflight(
                ctx.cache_key, herd.event, herd.result_box
            )
            self._trigger_auto_eviction()

    def _handle_save_error(self, err: BaseException | Exception, save_kwargs: dict) -> None:
        logger.error(
            f"Cache save failed for '{save_kwargs.get('func_name')}': {err}",
            exc_info=True,
        )
        if self.on_background_error:
            try:
                import sys

                res = save_kwargs.get("result")
                self.on_background_error(
                    err,
                    SaveErrorContext(
                        func_name=save_kwargs.get("func_name", "unknown"),
                        cache_key=save_kwargs.get("cache_key", ""),
                        input_id=save_kwargs.get("input_id", ""),
                        version=save_kwargs.get("version"),
                        content_type=save_kwargs.get("content_type"),
                        save_blob=save_kwargs.get("save_blob"),
                        expires_at=save_kwargs.get("expires_at"),
                        result_type=type(res).__name__,
                        result_size=sys.getsizeof(res) if res is not None else None,
                    ),
                )
            except Exception:
                logger.error(
                    "Error occurred within the 'on_background_error' callback",
                    exc_info=True,
                )

    def _notify_save_discarded(self, save_kwargs: dict) -> None:
        msg = f"Background save for '{save_kwargs.get('func_name')}' discarded during shutdown."
        logger.warning(msg)
        warnings.warn(msg, ResourceWarning)
        self._handle_save_error(RuntimeError(msg), save_kwargs)

    def _submit_background_save(self, **save_kwargs) -> None:
        bg_loop, executor = self._ensure_bg_resources()
        coro = self._save_result_async(executor=executor, **save_kwargs)
        future = bg_loop.submit(coro)
        if future:
            self._track_future(future)
        else:
            self._notify_save_discarded(save_kwargs)

    async def _save_result_async(
        self, /, executor: Executor, safe: bool = True, **kwargs
    ) -> None:
        loop = asyncio.get_running_loop()
        target = (lambda **kw: self._save_result_safe(**kw)) if safe else self.cache.set
        try:
            await loop.run_in_executor(executor, functools.partial(target, **kwargs))
        except (asyncio.CancelledError, RuntimeError) as e:
            # Executor might be forcibly shut down during program exit or Spot.shutdown(save_sync=False)
            msg = f"Background save for '{kwargs.get('func_name')}' cancelled during shutdown."
            logger.warning(msg)
            self._handle_save_error(e, kwargs)
            if not safe:
                raise

    def _save_result_safe(self, **kwargs):
        try:
            self.cache.set(**kwargs)
        except Exception as e:
            self._handle_save_error(e, kwargs)

    def consume(self, cost: Union[int, Callable] = 1):
        """関数実行前にレートリミッターのトークンを消費するデコレータ。

        Args:
            cost (Union[int, Callable], optional): 消費するコスト。整数、または実行時の引数を受け取りコストを計算する関数を指定可能。デフォルトは1。

        Returns:
            Callable: デコレートされた関数。
        """
        def decorator(func):
            is_async = inspect.iscoroutinefunction(func)

            @functools.wraps(func)
            def sync_wrapper(*args, **kwargs):
                self.limiter.consume(cost(*args, **kwargs) if callable(cost) else cost)
                return func(*args, **kwargs)

            @functools.wraps(func)
            async def async_wrapper(*args, **kwargs):
                await self.limiter.consume_async(
                    cost(*args, **kwargs) if callable(cost) else cost
                )
                return await func(*args, **kwargs)

            return async_wrapper if is_async else sync_wrapper

        return decorator

    @overload
    def mark(self, _func: Callable[P, R]) -> Callable[P, R]: ...

    @overload
    def mark(
        self,
        *,
        save_blob: Optional[bool] = None,
        keygen: Optional[Union[Callable, KeyGenPolicy]] = None,
        input_key_fn: Optional[Union[Callable, KeyGenPolicy]] = None,
        version: str | None = None,
        content_type: Optional[str | ContentType] = None,
        serializer: Optional[SerializerProtocol] = None,
        save_sync: Optional[bool] = None,
        retention: RetentionSpec = None,
        hooks: Optional[Sequence[HookBase]] = None,
    ) -> Callable[[Callable[P, R]], Callable[P, R]]: ...

    def mark(self, _func: Optional[Callable] = None, **kwargs) -> Any:
        """関数を修飾し、実行結果のキャッシュ機能とメタデータ管理を追加するデコレータ。

        関数の実行結果は、引数とその他の設定から計算されたキャッシュキーに基づいて保存・再利用されます。

        Args:
            _func (Optional[Callable], optional): デコレート対象の関数。
            save_blob (Optional[bool], optional): 大きな戻り値をBlobストレージに保存するかどうか。
            keygen (Optional[Union[Callable, KeyGenPolicy]], optional): キャッシュキーの生成ロジックを指定する。
            input_key_fn (Optional[Union[Callable, KeyGenPolicy]], optional): 非推奨。`keygen` を使用すること。
            version (Optional[str], optional): 関数のキャッシュバージョン。ロジック変更時にインクリメントすることでキャッシュを無効化できる。
            content_type (Optional[Union[str, ContentType]], optional): 戻り値のMIMEタイプ。
            serializer (Optional[SerializerProtocol], optional): この関数に適用するカスタムシリアライザ。
            save_sync (Optional[bool], optional): 保存処理を同期的に行うかどうか。
            retention (RetentionSpec, optional): キャッシュの保持ポリシー。
            hooks (Optional[Sequence[HookBase]], optional): 実行前後やキャッシュヒット時に発火するフックのリスト。

        Returns:
            Any: デコレートされた関数、またはデコレータ関数。
        """
        def decorator(func):
            if inspect.isgeneratorfunction(func) or inspect.isasyncgenfunction(func):
                raise ConfigurationError(f"Generators not supported: {func.__name__}")
            key_fn = self._resolve_key_fn(
                func, kwargs.get("keygen"), kwargs.get("input_key_fn")
            )
            is_async = inspect.iscoroutinefunction(func)

            @functools.wraps(func)
            def sync_wrapper(*args, **kw):
                return self._execute_sync(
                    func,
                    args,
                    kw,
                    kwargs.get("save_blob"),
                    key_fn,
                    kwargs.get("version"),
                    kwargs.get("content_type"),
                    kwargs.get("serializer"),
                    kwargs.get("retention"),
                    kwargs.get("save_sync"),
                    kwargs.get("hooks"),
                )

            @functools.wraps(func)
            async def async_wrapper(*args, **kw):
                return await self._execute_async(
                    func,
                    args,
                    kw,
                    kwargs.get("save_blob"),
                    key_fn,
                    kwargs.get("version"),
                    kwargs.get("content_type"),
                    kwargs.get("serializer"),
                    kwargs.get("retention"),
                    kwargs.get("save_sync"),
                    kwargs.get("hooks"),
                )

            return async_wrapper if is_async else sync_wrapper

        return decorator(_func) if _func else decorator

    @overload
    def cached_run(self, __func: Callable[P, R], **kwargs: Any) -> ContextManager[Callable[P, R]]: ...

    @overload
    def cached_run(self, *funcs: *Ts, **kwargs: Any) -> ContextManager[tuple[*Ts]]: ...

    @contextmanager
    def cached_run(self, *funcs: Any, **kwargs) -> Iterator[Any]:
        """コンテキストマネージャ内で一時的に関数を `mark` し、キャッシュ機能を適用する。

        デコレータを直接付与できない外部ライブラリの関数などをキャッシュする際に使用します。

        Args:
            *funcs (Any): キャッシュ対象にする関数（複数可）。
            **kwargs: `mark` デコレータに渡すオプションパラメータ。

        Yields:
            Callable | tuple[Callable, ...]: キャッシュ機能が付与された関数。複数の場合はタプルで返る。

        Raises:
            ValidationError: 関数が1つも指定されなかった場合。
        """
        if not funcs:
            raise ValidationError(
                "At least one function must be provided to cached_run."
            )
        wrappers = [self.mark(**kwargs)(f) for f in funcs]
        yield wrappers[0] if len(wrappers) == 1 else tuple(wrappers)

cached_run(*funcs, **kwargs)

cached_run(__func: Callable[P, R], **kwargs: Any) -> ContextManager[Callable[P, R]]

cached_run(*funcs: *Ts, **kwargs: Any) -> ContextManager[tuple[*Ts,]]

コンテキストマネージャ内で一時的に関数を mark し、キャッシュ機能を適用する。

デコレータを直接付与できない外部ライブラリの関数などをキャッシュする際に使用します。

Parameters:

Name	Type	Description	Default
*funcs	Any	キャッシュ対象にする関数（複数可）。	()
**kwargs		mark デコレータに渡すオプションパラメータ。	{}

Yields:

Type	Description
Any	Callable | tuple[Callable, ...]: キャッシュ機能が付与された関数。複数の場合はタプルで返る。

Raises:

Type	Description
ValidationError	関数が1つも指定されなかった場合。

Source code in src/beautyspot/core.py

@contextmanager
def cached_run(self, *funcs: Any, **kwargs) -> Iterator[Any]:
    """コンテキストマネージャ内で一時的に関数を `mark` し、キャッシュ機能を適用する。

    デコレータを直接付与できない外部ライブラリの関数などをキャッシュする際に使用します。

    Args:
        *funcs (Any): キャッシュ対象にする関数（複数可）。
        **kwargs: `mark` デコレータに渡すオプションパラメータ。

    Yields:
        Callable | tuple[Callable, ...]: キャッシュ機能が付与された関数。複数の場合はタプルで返る。

    Raises:
        ValidationError: 関数が1つも指定されなかった場合。
    """
    if not funcs:
        raise ValidationError(
            "At least one function must be provided to cached_run."
        )
    wrappers = [self.mark(**kwargs)(f) for f in funcs]
    yield wrappers[0] if len(wrappers) == 1 else tuple(wrappers)

consume(cost=1)

関数実行前にレートリミッターのトークンを消費するデコレータ。

Parameters:

Name	Type	Description	Default
cost	Union[int, Callable]	消費するコスト。整数、または実行時の引数を受け取りコストを計算する関数を指定可能。デフォルトは1。	1

Returns:

Name	Type	Description
Callable		デコレートされた関数。

Source code in src/beautyspot/core.py

def consume(self, cost: Union[int, Callable] = 1):
    """関数実行前にレートリミッターのトークンを消費するデコレータ。

    Args:
        cost (Union[int, Callable], optional): 消費するコスト。整数、または実行時の引数を受け取りコストを計算する関数を指定可能。デフォルトは1。

    Returns:
        Callable: デコレートされた関数。
    """
    def decorator(func):
        is_async = inspect.iscoroutinefunction(func)

        @functools.wraps(func)
        def sync_wrapper(*args, **kwargs):
            self.limiter.consume(cost(*args, **kwargs) if callable(cost) else cost)
            return func(*args, **kwargs)

        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs):
            await self.limiter.consume_async(
                cost(*args, **kwargs) if callable(cost) else cost
            )
            return await func(*args, **kwargs)

        return async_wrapper if is_async else sync_wrapper

    return decorator

flush(timeout=None)

バックグラウンドで実行中のすべての保存タスクとDBの書き込みの完了を待機する。

Parameters:

Name	Type	Description	Default
timeout	Optional[float]	待機する最大時間（秒）。指定しない場合は初期化時の drain_timeout が使用される。	None

Source code in src/beautyspot/core.py

def flush(self, timeout: Optional[float] = None) -> None:
    """バックグラウンドで実行中のすべての保存タスクとDBの書き込みの完了を待機する。

    Args:
        timeout (Optional[float], optional): 待機する最大時間（秒）。指定しない場合は初期化時の `drain_timeout` が使用される。
    """
    timeout_val = timeout if timeout is not None else self._drain_timeout
    deadline = time.monotonic() + timeout_val

    while True:
        with self._futures_lock:
            snapshot = list(self._active_futures)
        if not snapshot:
            break
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            break
        wait_timeout = min(self._drain_poll_interval, remaining)
        wait(snapshot, timeout=wait_timeout)

    db_remaining = deadline - time.monotonic()
    if db_remaining > 0 and isinstance(self.cache.db, Flushable):
        self.cache.db.flush(timeout=db_remaining)

mark(_func=None, **kwargs)

mark(_func: Callable[P, R]) -> Callable[P, R]

mark(*, save_blob: Optional[bool] = None, keygen: Optional[Union[Callable, KeyGenPolicy]] = None, input_key_fn: Optional[Union[Callable, KeyGenPolicy]] = None, version: str | None = None, content_type: Optional[str | ContentType] = None, serializer: Optional[SerializerProtocol] = None, save_sync: Optional[bool] = None, retention: RetentionSpec = None, hooks: Optional[Sequence[HookBase]] = None) -> Callable[[Callable[P, R]], Callable[P, R]]

関数を修飾し、実行結果のキャッシュ機能とメタデータ管理を追加するデコレータ。

関数の実行結果は、引数とその他の設定から計算されたキャッシュキーに基づいて保存・再利用されます。

Parameters:

Name	Type	Description	Default
_func	Optional[Callable]	デコレート対象の関数。	None
save_blob	Optional[bool]	大きな戻り値をBlobストレージに保存するかどうか。	required
keygen	Optional[Union[Callable, KeyGenPolicy]]	キャッシュキーの生成ロジックを指定する。	required
input_key_fn	Optional[Union[Callable, KeyGenPolicy]]	非推奨。keygen を使用すること。	required
version	Optional[str]	関数のキャッシュバージョン。ロジック変更時にインクリメントすることでキャッシュを無効化できる。	required
content_type	Optional[Union[str, ContentType]]	戻り値のMIMEタイプ。	required
serializer	Optional[SerializerProtocol]	この関数に適用するカスタムシリアライザ。	required
save_sync	Optional[bool]	保存処理を同期的に行うかどうか。	required
retention	RetentionSpec	キャッシュの保持ポリシー。	required
hooks	Optional[Sequence[HookBase]]	実行前後やキャッシュヒット時に発火するフックのリスト。	required

Returns:

Name	Type	Description
Any	Any	デコレートされた関数、またはデコレータ関数。

Source code in src/beautyspot/core.py

def mark(self, _func: Optional[Callable] = None, **kwargs) -> Any:
    """関数を修飾し、実行結果のキャッシュ機能とメタデータ管理を追加するデコレータ。

    関数の実行結果は、引数とその他の設定から計算されたキャッシュキーに基づいて保存・再利用されます。

    Args:
        _func (Optional[Callable], optional): デコレート対象の関数。
        save_blob (Optional[bool], optional): 大きな戻り値をBlobストレージに保存するかどうか。
        keygen (Optional[Union[Callable, KeyGenPolicy]], optional): キャッシュキーの生成ロジックを指定する。
        input_key_fn (Optional[Union[Callable, KeyGenPolicy]], optional): 非推奨。`keygen` を使用すること。
        version (Optional[str], optional): 関数のキャッシュバージョン。ロジック変更時にインクリメントすることでキャッシュを無効化できる。
        content_type (Optional[Union[str, ContentType]], optional): 戻り値のMIMEタイプ。
        serializer (Optional[SerializerProtocol], optional): この関数に適用するカスタムシリアライザ。
        save_sync (Optional[bool], optional): 保存処理を同期的に行うかどうか。
        retention (RetentionSpec, optional): キャッシュの保持ポリシー。
        hooks (Optional[Sequence[HookBase]], optional): 実行前後やキャッシュヒット時に発火するフックのリスト。

    Returns:
        Any: デコレートされた関数、またはデコレータ関数。
    """
    def decorator(func):
        if inspect.isgeneratorfunction(func) or inspect.isasyncgenfunction(func):
            raise ConfigurationError(f"Generators not supported: {func.__name__}")
        key_fn = self._resolve_key_fn(
            func, kwargs.get("keygen"), kwargs.get("input_key_fn")
        )
        is_async = inspect.iscoroutinefunction(func)

        @functools.wraps(func)
        def sync_wrapper(*args, **kw):
            return self._execute_sync(
                func,
                args,
                kw,
                kwargs.get("save_blob"),
                key_fn,
                kwargs.get("version"),
                kwargs.get("content_type"),
                kwargs.get("serializer"),
                kwargs.get("retention"),
                kwargs.get("save_sync"),
                kwargs.get("hooks"),
            )

        @functools.wraps(func)
        async def async_wrapper(*args, **kw):
            return await self._execute_async(
                func,
                args,
                kw,
                kwargs.get("save_blob"),
                key_fn,
                kwargs.get("version"),
                kwargs.get("content_type"),
                kwargs.get("serializer"),
                kwargs.get("retention"),
                kwargs.get("save_sync"),
                kwargs.get("hooks"),
            )

        return async_wrapper if is_async else sync_wrapper

    return decorator(_func) if _func else decorator

register(code, encoder, decoder=None, decoder_factory=None)

カスタム型をシリアライザに登録するためのデコレータ。

decoder または decoder_factory のいずれかを必ず提供する必要があります。

Parameters:

Name	Type	Description	Default
code	int	カスタム型の一意な識別コード。	required
encoder	Callable[[T], Any]	カスタム型オブジェクトからシリアライズ可能な形式（辞書など）に変換する関数。	required
decoder	Optional[Callable[[Any], T]]	デシリアライズ時にデータをカスタム型オブジェクトに復元する関数。	None
decoder_factory	Optional[Callable[[Type[T]], Callable[[Any], T]]]	型に基づいてデコーダ関数を生成するファクトリ関数。	None

Returns:

Type	Description
Callable[[Type[T]], Type[T]]	Callable[[Type[T]], Type[T]]: クラスデコレータ。

Raises:

Type	Description
IncompatibleProviderError	decoder と decoder_factory の両方が未指定の場合に発生。

Source code in src/beautyspot/core.py

def register(
    self,
    code: int,
    encoder: Callable[[T], Any],
    decoder: Optional[Callable[[Any], T]] = None,
    decoder_factory: Optional[Callable[[Type[T]], Callable[[Any], T]]] = None,
) -> Callable[[Type[T]], Type[T]]:
    """カスタム型をシリアライザに登録するためのデコレータ。

    `decoder` または `decoder_factory` のいずれかを必ず提供する必要があります。

    Args:
        code (int): カスタム型の一意な識別コード。
        encoder (Callable[[T], Any]): カスタム型オブジェクトからシリアライズ可能な形式（辞書など）に変換する関数。
        decoder (Optional[Callable[[Any], T]], optional): デシリアライズ時にデータをカスタム型オブジェクトに復元する関数。
        decoder_factory (Optional[Callable[[Type[T]], Callable[[Any], T]]], optional): 型に基づいてデコーダ関数を生成するファクトリ関数。

    Returns:
        Callable[[Type[T]], Type[T]]: クラスデコレータ。

    Raises:
        IncompatibleProviderError: `decoder` と `decoder_factory` の両方が未指定の場合に発生。
    """
    if decoder is None and decoder_factory is None:
        raise IncompatibleProviderError(
            "Must provide either `decoder` or `decoder_factory`."
        )

    def decorator(cls: Type) -> Type:
        actual_decoder = decoder
        if decoder_factory:
            actual_decoder = decoder_factory(cls)

        if actual_decoder is None:
            raise ValueError("Decoder resolution failed.")

        self.register_type(cls, code, encoder, actual_decoder)
        return cls

    return decorator

register_type(type_class, code, encoder, decoder)

カスタム型を直接シリアライザに登録する。

Parameters:

Name	Type	Description	Default
type_class	Type[T]	登録するカスタム型のクラス。	required
code	int	カスタム型の一意な識別コード。	required
encoder	Callable[[T], Any]	エンコーダ関数。	required
decoder	Callable[[Any], T]	デコーダ関数。	required

Raises:

Type	Description
NotImplementedError	現在のシリアライザが型登録をサポートしていない場合。

Source code in src/beautyspot/core.py

def register_type(
    self,
    type_class: Type[T],
    code: int,
    encoder: Callable[[T], Any],
    decoder: Callable[[Any], T],
):
    """カスタム型を直接シリアライザに登録する。

    Args:
        type_class (Type[T]): 登録するカスタム型のクラス。
        code (int): カスタム型の一意な識別コード。
        encoder (Callable[[T], Any]): エンコーダ関数。
        decoder (Callable[[Any], T]): デコーダ関数。

    Raises:
        NotImplementedError: 現在のシリアライザが型登録をサポートしていない場合。
    """
    if isinstance(self.cache.serializer, TypeRegistryProtocol):
        self.cache.serializer.register(type_class, code, encoder, decoder)
    else:
        raise NotImplementedError(
            "Current serializer does not support type registration."
        )

shutdown(save_sync=True)

Spotインスタンスをシャットダウンし、バックグラウンドリソースを解放する。

Parameters:

Name	Type	Description	Default
save_sync	bool	同期的に未完了の保存タスクを待機するかどうか。Trueの場合は完了を待つ。デフォルトはTrue。	True

Source code in src/beautyspot/core.py

def shutdown(self, save_sync: bool = True):
    """Spotインスタンスをシャットダウンし、バックグラウンドリソースを解放する。

    Args:
        save_sync (bool, optional): 同期的に未完了の保存タスクを待機するかどうか。Trueの場合は完了を待つ。デフォルトはTrue。
    """
    with self._bg_init_lock:
        self._shutdown_called = True
    if self._finalizer is not None and self._finalizer.alive:
        self._finalizer.detach()
    if save_sync:
        self._drain_futures()

    if self._bg_loop is not None:
        self._bg_loop.stop(save_sync=save_sync)

    if self._executor is not None:
        self._executor.shutdown(wait=save_sync, cancel_futures=not save_sync)

主なコンセプト

1. タスクの登録と実行

@spot.mark デコレータを使用、または spot.cached_run() で関数を登録することで、その関数は自動的にキャッシュ対応となります。入力引数に基づいて一意のキャッシュキーが生成され、同じ引数での呼び出しはストレージから結果を復元します。

2. 非ブロッキング保存 (Non-blocking Persistence)

v2.0 では、Spot 初期化時に default_wait=False を設定するか、@mark(wait=False) を指定することで、保存処理をバックグラウンドスレッドで実行できます。これにより、計算終了直後に制御がユーザーに戻り、I/O 待ちによる遅延が解消されます。

3. 同期ポイントとしての Flush

with spot: ブロック（コンテキストマネージャ）を使用すると、そのブロックを抜ける際に、実行中のすべてのバックグラウンド保存タスクの完了を待機（Flush）します。 これは、プログラムの終了前やバッチ処理の区切りでデータの整合性を保証するために重要です。

4. ライフサイクルフック (Lifecycle Hooks)

v2.0から、関数の実行パイプラインに介入できるクラスベースのフックシステム (HookBase) が導入されました。@spot.mark(hooks=[...]) を指定することで、関数の実行前 (pre_execute)、キャッシュヒット時 (on_cache_hit)、キャッシュミス時 (on_cache_miss) にカスタムロジックを差し込めます。これにより、レイテンシ計測やAPIコスト計算などを容易に実装できます。

使用例

基本的なデコレータの使用

@spot.mark(version="1.0", save_blob=True)
def heavy_task(data):
    # 重い処理
    return result

局所的なキャッシュ実行 (cached_run)

デコレートせずに関数を一時的にキャッシュ化したい場合に使用します。with ブロックを使うと変数スコープを明確にできますが、返されたラッパーはブロック外でも有効です。

with spot.cached_run(my_func, version="v1") as task:
    result = task(arg)

# task はブロック外でも @spot.mark(version="v1") 相当のラッパーとして有効

バックグラウンド保存の制御

# 保存を待たずに即座に値を返す
@spot.mark(wait=False)
def async_save_task(x):
    return x * 10

with spot:
    async_save_task(1)
    async_save_task(2)
# ここを抜ける時に 1 と 2 の保存完了が保証される

メトリクス収集フックの使用

from beautyspot.hooks import HookBase

class MetricsHook(HookBase):
    def on_cache_miss(self, context):
        print(f"[{context.func_name}] 実関数が実行されました。")

    def on_cache_hit(self, context):
        print(f"[{context.func_name}] キャッシュが利用されました！")

@spot.mark(hooks=[MetricsHook()])
def fetch_data(query: str):
    return "Result"

関連コンポーネント

• CacheManager: キャッシュのキー生成、読み書き、Thundering Herd対策などの内部ロジックを担当するコンポーネントです。

注意事項

• シャットダウン: spot.shutdown() を呼ぶと Executor が停止し、それ以降バックグラウンド保存は利用できなくなります。
• スレッドセーフ: 内部で ThreadPoolExecutor を使用しているため、注入される DB や Storage はスレッドセーフである必要があります。

動作フローチャート

キャッシュヒットした場合（同期パス）

キャッシュにデータが存在する場合、バックグラウンドスレッドは関与せず、メインスレッド内で直接データの復元が行われます。

sequenceDiagram
    box rgb(255, 40, 40, 0.1) メインスレッド
        participant User as ユーザーコード
        participant UserFunc as markされた関数
        participant Spot as Spotインスタンス
        participant Cache as CacheManager
        participant Storage as DB / Storage
    end

    User->>UserFunc: 関数呼び出し (引数: args, kwargs)
    UserFunc->>Spot: キャッシュルックアップ要求

    Spot->>Cache: make_cache_key() (引数からハッシュキーを生成)
    Cache-->>Spot: cache_key, input_id

    Spot->>Spot: pre_execute フック発火

    Spot->>Cache: get(cache_key) (キャッシュの取得要求)
    Cache->>Storage: データの読み込み
    Storage-->>Cache: シリアライズされたデータ
    Cache->>Cache: デシリアライズ (復元)
    Cache-->>Spot: 復元されたオブジェクト

    Spot->>Spot: on_cache_hit フック発火

    Spot-->>UserFunc: キャッシュされた結果
    Note right of UserFunc: 実関数は実行されない
    UserFunc-->>User: 関数の実行結果

キャッシュヒットしなかった場合（バックグラウンド保存）

sequenceDiagram
    participant OS as OS / atexit

    box rgb(255, 40, 40, 0.1) メインスレッド
        participant User as ユーザーコード<br/>(User Code)
        participant UserFunc as markされた関数<br/>(User Function)
        participant Spot as Spotインスタンス
    end

    participant BGLoop as BGLoopスレッド<br/>(_BackgroundLoop)
    participant Executor as ワーカースレッド<br/>(ThreadPoolExecutor)

    User->>Spot: Spot.__init__() (インスタンス初期化、リソースは未作成)
    Note right of User: バックグラウンドリソースは遅延初期化される

    User->>UserFunc: 関数呼び出し (デコレートされたラッパー経由)
    UserFunc->>Spot: Spot._execute_sync() (キャッシュルックアップ＆実行要求)

    Note right of Spot: キャッシュミス発生と仮定

    Spot->>Spot: Spot._persist_result_sync(save_sync=False) (結果の非同期保存処理を開始)
    Spot->>Spot: Spot._ensure_bg_resources() (バックグラウンド用スレッドの確保/生成)

    opt リソース未作成の場合（初回のみ）
        Spot->>BGLoop: _BackgroundLoop.__init__() (非同期IOを管理するイベントループスレッドを生成)
        activate BGLoop
        Note right of BGLoop: 内部で新しくイベントループを作成し<br/>asyncio.run_forever()を実行
        Spot->>Executor: ThreadPoolExecutor.__init__() (実際のIO処理を行うスレッドプールを生成)
        activate Executor
        Spot->>Spot: weakref.finalize(Spot._shutdown_resources) (プロセス強制終了時のリソース解放フックを登録)
    end

    Spot->>BGLoop: _BackgroundLoop.submit() (保存用コルーチンをイベントループに投入)
    Note right of Spot: _active_tasksカウンタを加算しインフライトタスクとして追跡
    BGLoop-->>Spot: asyncio.Future (イベントループ内でスケジュールされたタスクの参照)
    Spot->>Spot: Spot._track_future() (タスク完了をFlush等で待機できるようリストに追加)

    Spot-->>UserFunc: 戻り値 (保存完了を待たずに即座に返す)
    UserFunc-->>User: 関数の実行結果

    Note over BGLoop, Executor: --- バックグラウンド保存処理 ---

    BGLoop->>BGLoop: _BackgroundLoop._task_wrapper() (コルーチン実行開始・終了をフックしタスク状態を管理)
    BGLoop->>Executor: loop.run_in_executor() (同期的なDB/ストレージ保存関数をワーカースレッドに委譲)

    activate Executor
    Executor->>Executor: Spot._save_result_safe() (ファイルシステムやDBに対する実際のIO書き込みを実行)

    alt 保存成功
        Executor-->>BGLoop: 結果を返す (IO処理の正常完了を通知)
    else 例外発生
        Executor->>Spot: Spot._handle_save_error() (エラーのロギングとユーザーフックの実行)
        Note right of Executor: 例外は握り潰さずロギング＆<br/>on_background_errorフックを実行
        Executor-->>BGLoop: 処理完了を通知 (ワーカースレッド内の例外は上位イベントループに伝搬させない)
    end
    deactivate Executor

    BGLoop->>BGLoop: _BackgroundLoop._active_tasks を減算 (追跡カウンタを減らし、0になればシャットダウン可能か判定)

    Note over User, Executor: --- シャットダウン / 終了処理 ---

    alt ユーザーコードが明示的に終了する場合
        User->>Spot: Spot.shutdown(save_sync=True) (Spotの明示的な終了要求、未完了タスクを待機)
        Spot->>Spot: Spot._drain_futures() (登録済みFutureの完了をタイムアウト付きで同期待機)
        Spot->>BGLoop: _BackgroundLoop.stop(save_sync=True) (新規タスク受付を停止し、ループの終了を要求)

        BGLoop->>BGLoop: _BackgroundLoop._is_shutting_down = True (内部フラグを立てて新規タスクを拒否)
        Note right of User: Mainスレッドは最大 drain_timeout 秒間、<br/>BGLoopスレッドの完了を thread.join() で待機する

        alt 実行中のタスクが残っている場合
            BGLoop->>BGLoop: _BackgroundLoop._task_wrapper() のfinally内で<br/>loop.stop() を発行 (最後のタスク完了時にループを停止)
        else アクティブなタスクが0の場合
            BGLoop->>BGLoop: 即座に loop.stop() を発行 (待つタスクがないため即座にループを停止)
        end

        BGLoop-->>Spot: イベントループ停止 ＆ スレッド終了 (BGLoopスレッドの完全な停止)
        deactivate BGLoop

        Spot->>Executor: ThreadPoolExecutor.shutdown(wait=True) (ワーカースレッドプールへ完了待機と停止を指示)
        Executor-->>Spot: スレッドプール終了 (全ワーカーの停止)
        deactivate Executor

        Spot-->>User: シャットダウン完了 (全てのリソースが安全に解放された状態)

    else プロセス終了時 (atexit / weakref.finalize による自動終了)
        OS->>Spot: Spot._shutdown_resources() (終了フックからの強制クリーンアップ呼び出し)
        Note right of Spot: atexitまたはガベージコレクションによって発火
        Spot->>BGLoop: _BackgroundLoop.stop(save_sync=False) (同期待機せずに即座の停止を要求)

        BGLoop->>BGLoop: _BackgroundLoop._is_shutting_down = True (内部フラグを立てて新規タスクを拒否)
        BGLoop->>BGLoop: 即座に loop.stop() を発行 (実行中のタスクがあっても強制停止をスケジュール)

        Spot->>Executor: ThreadPoolExecutor.shutdown(wait=False, cancel_futures=True) (スレッドプールの即時停止と未実行タスクのキャンセル)

        Note right of OS: タイムアウトを待たずに即座にプロセスを終了し、OSがリソースを回収する
    end