キャッシュのメタデータ（関数名、入力ID、バージョン、有効期限等）をデータベースに永続化できること。

抽象インターフェースにより、DBやストレージなどのバックエンド実装を差し替え可能であること。

データベース操作が指定されたタイムアウト時間内に完了しない場合、呼び出し元を無期限にブロックせずにエラーを返し、システムの可用性を維持すること。特にSQLiteの書き込みスレッドがスタックした場合でも、フェイルファストによって制御を戻せること。

関数の実行結果をキャッシュし、同一入力に対して再実行せずにキャッシュから結果を返せること。同期関数・非同期関数の両方をサポートすること。

関数の引数から安定かつ一意なキャッシュキーを生成できること。引数の型・構造に応じた正規化を行い、辞書のキー順序やコレクション型の違いを正しく区別できること。

大きなキャッシュデータを外部ストレージ（ローカルファイルシステムやS3等）に保存できること。

DB直接保存とBlob保存を自動判定できること。

関数の戻り値を安全にシリアライズ・デシリアライズできること。ユーザー定義型のカスタムエンコーダ・デコーダを登録可能であること。

キャッシュの保存処理をバックグラウンドで非同期に実行し、関数の応答レイテンシに影響を与えないモードを提供すること。

キャッシュの有効期限を関数名パターンに基づくルールで設定できること。個別の関数に対して保持期間を指定でき、期限切れのキャッシュを自動的に無効化できること。

関数の実行頻度をトークンバケット方式で制限し、外部APIの呼び出しレートを制御できること。同期・非同期の両方に対応すること。

関数の実行前、キャッシュヒット時、キャッシュミス時にカスタムコールバック（フック）を実行できること。スレッドセーフなフック実装を提供すること。

期限切れキャッシュの削除、孤立Blobファイルのガベージコレクション、時間ベースの一括削除（prune）等のメンテナンス操作ができること。

コマンドラインインターフェースからキャッシュの一覧表示、詳細確認、統計表示、削除、GC等の管理操作ができること。

全コンポーネント（DB、ストレージ、シリアライザ、リミッター等）が Protocol/ABC に基づく依存注入により差し替え可能であること。ファクトリ関数がデフォルトの組み立てを提供すること。

同一キャッシュキーに対する並行リクエストを直列化し、1つのリクエストのみが関数を実行し、他のリクエストはその結果を共有することで重複実行を防止できること。

本ドキュメントは、Python関数キャッシュおよびレート制限ライブラリ『beautyspot』の要件を定義する。

データパイプラインや機械学習の実験、外部API呼び出し等において、同一入力での不要な再実行を防ぐことで、実行時間の短縮とリソースの節約を実現する。開発者が最小限のコード変更で強力なキャッシュ・レート制限機能を導入できることを目指す。

本プロジェクト（Beautyspot）で使用される主要な用語の定義です。

本章ではシステムが提供すべき具体的な機能要件について定義する。

• Beautyspot: 本キャッシュライブラリの名称。

• Spot: 本ライブラリにおけるキャッシュ・レート制限のコアエンジンの名称。

• キャッシュキー (Cache Key): 関数の入力引数等から一意に生成される識別子。

• キャッシュヒット: 保存された結果を再利用し、元の関数を実行しないこと。

• キャッシュミス: 結果が未保存または無効のため、元の関数を実行すること。

• メタデータDB (Metadata DB): キャッシュの有効期限、サイズ、関数名などの付帯情報を保存するデータストア。

• Blobストレージ (Blob Storage): キャッシュの実際のペイロード（戻り値のシリアライズ結果）を保存するストレージ。

• シリアライザ (Serializer): 関数の戻り値をストレージに保存可能なバイト列等に変換、および復元するコンポーネント。

• バックグラウンドIO (Background IO): メインスレッド（またはタスク）の実行をブロックせず、非同期にキャッシュ保存などを行う仕組み。

• レート制限 (Rate Limiting): 外部API呼び出し等の頻度を制御する仕組み。トークンバケット方式を採用。

• Thundering Herd: キャッシュミス時に、多数の並行リクエストが同時に同一関数を実行しようとして過負荷になる現象。

• フック (Hook): キャッシュのライフサイクル（ヒット、ミス、保存前後等）に介入できるコールバック機構。

• 依存注入 (Dependency Injection / DI): 各コンポーネント（DB、ストレージ等）を外部から差し替え可能にする設計パターン。

仕様書の機能グループ (Groups)

仕様書アイテムに付与される groups 属性の意味は以下の通りです。

コンポーネント構成

classDiagram
  class TaskDBCore {
    <<Protocol>>
    +init_schema() void
    +get(cache_key) TaskRecord
    +save(cache_key, ...) void
    +delete(cache_key) bool
  }
  class Maintenable {
    <<Protocol>>
    +delete_expired() int
    +prune(older_than) int
    +delete_all() int
  }
  class TaskDBMaintenable {
    <<Protocol>>
  }
  TaskDBCore <|-- TaskDBMaintenable
  Maintenable <|-- TaskDBMaintenable

  class SQLiteTaskDB {
    -db_path: Path
    -_write_queue: Queue
    +init_schema() void
    +get(cache_key) TaskRecord
    +save(cache_key, ...) void
  }
  TaskDBMaintenable <|-- SQLiteTaskDB

コンポーネント責務

データフロー

sequenceDiagram
  participant Spot as core.Spot
  participant DB as SQLiteTaskDB
  participant Q as WriteQueue
  participant Thread as WriterThread
  participant SQLite as SQLiteDB

  Spot->>DB: get(cache_key)
  DB->>SQLite: SELECT
  SQLite-->>DB: TaskRecord
  DB-->>Spot: TaskRecord

  Spot->>DB: save(metadata)
  DB->>Q: put(WriteTask)
  DB-->>Spot: void (Non-blocking)

  Thread->>Q: get()
  Thread->>SQLite: INSERT / UPDATE

技術選定

非機能要件方針

• 性能: 読み込みはメインスレッドで高速に処理（WALモード）、書き込みは専用スレッドによる非同期化。
• スレッド安全性: SQLiteへの書き込みは1つのスレッドに直列化し、データベースロックを回避する。
• 拡張性: すべての操作はProtocol (TaskDBCore, Maintenable) で定義され、利用側（core.Spotや_CacheManager）は実装に依存しない。

コンポーネント構成

graph TD
  A[bs.Spot Factory] -->|DI/Composition| B[core.Spot]
  B --> C[TaskDBBase]
  B --> D[SerializerProtocol]
  B --> E[BlobStorageBase]
  B --> F[StoragePolicyProtocol]
  B --> G[LimiterProtocol]
  B --> H[LifecyclePolicy]

コンポーネント責務

データフロー

sequenceDiagram
  participant User as ユーザーコード
  participant Spot as core.Spot
  participant DB as TaskDB
  participant Ser as Serializer
  participant Blob as BlobStorage

  User->>Spot: fn(args)
  Spot->>DB: find_by_key(cache_key)
  alt キャッシュヒット
    DB-->>Spot: cached_data
    Spot->>Ser: unpack(data)
  else キャッシュミス
    Spot->>Spot: fn(args) 実行
    Spot->>Ser: pack(result)
    Spot->>DB: insert(metadata)
    opt Blob保存
      Spot->>Blob: save(key, data)
    end
  end
  Spot-->>User: result

技術選定

非機能要件方針

• 透過性: ユーザー関数の入出力に影響を与えないこと。
• 拡張性: 全コンポーネントはProtocol/ABCで抽象化し、DIで差し替え可能とすること。

コンポーネント構成

graph TD
  A[core.Spot] -->|キャッシュキー生成要求| B[KeyGenPolicy]
  B -->|バインド| C[bound_keygen]
  C -->|引数別戦略| D[Strategy]
  C -->|正規化| E[canonicalize]
  C -->|ハッシュ計算| F[KeyGen.hash_items]
  E --> F

コンポーネント責務

データフロー

sequenceDiagram
  participant Spot as core.Spot
  participant KP as KeyGenPolicy
  participant Sig as inspect.signature
  participant Can as canonicalize
  participant KG as KeyGen

  Spot->>KP: bind(func)
  KP->>Sig: signature(func)
  Sig-->>KP: bound_keygen 生成
  KP-->>Spot: bound_keygen
  Spot->>Spot: bound_keygen(*args, **kwargs)
  loop 各引数
    alt Strategy.IGNORE
      Spot->>Spot: スキップ
    else Strategy.FILE_CONTENT
      Spot->>KG: from_file_content(path)
    else Strategy.PATH_STAT
      Spot->>KG: from_path_stat(path)
    else Strategy.DEFAULT
      Spot->>Can: canonicalize(val)
      Can-->>Spot: 正規化済みデータ
    end
  end
  Spot->>KG: hash_items(items_to_hash)
  KG-->>Spot: SHA-256 ハッシュ文字列 (キャッシュキー)

技術選定

非機能要件方針

• 安定性: プロセスの再起動やPythonのバージョン（辞書の順序等）に依存せず、同一入力に対して常に同一のハッシュを生成すること。
• 性能: 複雑なオブジェクト構造でも高速に正規化とハッシュ計算を行えること。ファイル内容のハッシュ化はチャンク読み込みでメモリ効率を確保する。
• 拡張性: ユーザーが独自の KeyGenPolicy を定義可能にし、柔軟なキャッシュ戦略をサポートする。

コンポーネント構成

graph TD
  A[core.Spot] -->|判定| B[StoragePolicyProtocol];
  A -->|保存/読込| C[BlobStorageBase];
  D[LocalStorage] --> C;
  E[S3Storage] --> C;
  A -->|メタデータ| F[TaskDBBase];

コンポーネント責務

データフロー

sequenceDiagram
  participant Spot as core.Spot
  participant Policy as StoragePolicy
  participant DB as TaskDB
  participant Blob as BlobStorage

  Spot->>Policy: should_save_as_blob(data)
  alt Policy: True (Blob保存)
    Spot->>Blob: save(key, data)
    Blob-->>Spot: location (path/URI)
    Spot->>DB: insert(metadata, storage_type=FILE, blob_key=location)
  else Policy: False (Inline保存)
    Spot->>DB: insert(metadata, storage_type=DIRECT_BLOB, result_data=data)
  end

技術選定

非機能要件方針

• 安全性: LocalStorage では is_relative_to によるパストラバーサル防止を徹底

• 信頼性: 保存時は fsync を実行し、OSレベルでのデータ到達を確認

• 効率性: S3 保存時は upload_fileobj を使用し、5GB超のマルチパートアップロードに自動対応

コンポーネント構成

graph TD
  A[MsgpackSerializer] -->|Registry| B[TypeRegistryProtocol]
  A -->|LRU Cache| C[Thread-Local Storage]
  A -->|Core| D[msgpack-python]

コンポーネント責務

データフロー

sequenceDiagram
  participant User as ユーザーコード
  participant Ser as MsgpackSerializer
  participant Reg as TypeRegistry
  participant MP as msgpack

  User->>Ser: dumps(obj)
  Ser->>Ser: 型チェック & MRO解決
  opt カスタム型
    Ser->>Reg: エンコーダ取得
    Reg-->>Ser: encoder_fn
    Ser->>MP: ExtType(code, encoder_fn(obj))
  end
  MP-->>Ser: binary
  Ser-->>User: bytes

技術選定

非機能要件方針

• 性能: MRO（継承関係）解決結果をLRUキャッシュし、深層木探索のコストを最小化
• 柔軟性: ユーザー定義のエンコーダがプリミティブを返せば、再帰的にシリアライズ可能
• 堅牢性: デコード時の ExtCode 未登録エラーを検知し、互換性問題を明示的に報告

コンポーネント構成

graph TD
  A[core.Spot] -->|非同期タスク投入| B[_BackgroundLoop]
  B -->|Daemon Thread| C[asyncio.AbstractEventLoop]
  C -->|I/O| D[TaskDB / BlobStorage]
  A -->|ライフサイクル制御| E[atexit / ContextManager]

コンポーネント責務

データフロー

sequenceDiagram
  participant Main as メインスレッド
  participant BGLoop as _BackgroundLoop
  participant Store as ストレージ

  Main->>BGLoop: submit(save_coro)
  Note over Main: 関数は即座に結果を返す (save_sync=False)
  BGLoop->>Store: 書き込み実行 (I/O)
  alt 成功
      Store-->>BGLoop: OK
  else 失敗
      BGLoop->>Main: on_background_error(error)
  end

技術選定

非機能要件方針

• 透過性: 保存失敗時もユーザーコードの例外にはせず、コールバック経由で通知

• 一貫性: flush() により、クリティカルな処理の直後での保存完了を保証

• 安全性: シャットダウンシーケンス中に新規タスクを受け付けない排他制御を導入

コンポーネント構成

graph TD
  A[core.Spot] -->|マッチング要求| B[LifecyclePolicy]
  B -->|ルールリスト| C[Rule]
  C -->|パターン照合| D[fnmatch]
  B -->|パース| E[Retention]

コンポーネント責務

データフロー

sequenceDiagram
  participant Spot as core.Spot
  participant Policy as LifecyclePolicy
  participant DB as TaskDB

  Spot->>Policy: resolve_with_fallback("module.func")
  Policy->>Policy: ルール順次照合 (fnmatch)
  Policy-->>Spot: expires_at (timestamp)
  Spot->>DB: save(..., expires_at)
  Note over DB: get() 時に現在時刻と比較判定

技術選定

非機能要件方針

• 優先順位: 完全修飾名（モジュール名込み）が短縮名より優先される二段構えのマッチング
• 堅牢性: ポリシーにマッチしない場合は「無期限」とするフェイルセーフな設計
• 拡張性: Retention.FOREVER 等の定数により、グローバルポリシーの個別上書きに対応

コンポーネント構成

graph TD
  A["@spot.consume"] -->|宣言的適用| B[LimiterProtocol]
  C[core.Spot] -->|DI| B
  D[TokenBucket] --> B
  D -->|アルゴリズム| E[GCRA]
  D -->|時刻同期| F[monotonic clock]

コンポーネント責務

データフロー

sequenceDiagram
  participant User as ユーザーコード
  participant Dec as @spot.consume
  participant Bucket as TokenBucket

  User->>Dec: call func()
  Dec->>Bucket: consume(cost)
  Bucket->>Bucket: 次回実行許可時刻の計算 (GCRA)
  alt 許可
      Bucket-->>Dec: OK
      Dec->>User: execute func()
  else 拒否 (Over rate)
      Bucket-->>Dec: raise ValueError/RateLimitError
  end

技術選定

非機能要件方針

• 正確性: time.monotonic() を使用し、システム時刻の変更（NTP同期等）の影響を受けない

• バースト制御: max_burst 設定により、アイドル後の過度なバーストを防止

• スレッド安全: threading.Lock により、マルチスレッド環境での二重消費を防止

コンポーネント構成

graph TD
  A[core.Spot] -->|イベント通知| B[HookBase]
  C[ThreadSafeHookBase]　--> B
  B -->|コンテキスト| D[Context Objects]
  E[PreExecuteContext] --> D
  F[CacheHitContext] --> D
  G[CacheMissContext] --> D

コンポーネント責務

データフロー

sequenceDiagram
  participant Spot as core.Spot
  participant Hook as HookBase
  participant Fn as Target Function

  Spot->>Hook: pre_execute(ctx)
  alt キャッシュヒット
      Spot->>Hook: on_cache_hit(ctx)
  else キャッシュミス
      Spot->>Fn: execute()
      Fn-->>Spot: result
      Spot->>Hook: on_cache_miss(ctx)
  end

技術選定

非機能要件方針

• 透過性: フックの実行失敗（例外）は主処理の結果に影響を与えない

• 柔軟性: 関数ごとに異なるフックリストを hooks=[...] で指定可能

• パフォーマンス: フック未登録時のオーバーヘッドを最小化するガード条件を実装

コンポーネント構成

graph TD
  A[MaintenanceService] -->|操作集約| B[TaskDBBase]
  A -->|操作集約| C[BlobStorageBase]
  A -->|自動実行| D[Probabilistic Eviction]

コンポーネント責務

データフロー (GCプロセス)

sequenceDiagram
  participant Service as MaintenanceService
  participant DB as TaskDB
  participant Storage as BlobStorage

  Service->>DB: get_blob_refs()
  DB-->>Service: 有効な参照リスト (Whitelist)
  Service->>Storage: list_keys()
  Storage-->>Service: 全ファイルリスト
  Service->>Service: 差分抽出 (孤立ファイルの特定)
  Service->>Storage: delete(orphan_keys)
  Note over Service: 猶予期間 (grace_period) を考慮して削除

技術選定

非機能要件方針

• 安全性: DBレコードの削除 → Blobの削除の順序を徹底し、参照エラーを防止
• 可用性: メンテナンス中も通常のキャッシュ読み取りを妨げないロック粒度
• 観測性: 削除されたタスク数や解放された容量を統計として返却

コンポーネント構成

graph TD
  A[beautyspot CLI] -->|Command| B[Typer App]
  B -->|Business Logic| C[MaintenanceService]
  B -->|Output| D[Rich Console]
  B -->|Dashboard| E[Dashboard App]

コンポーネント責務

データフロー

sequenceDiagram
  participant User as ユーザー
  participant CLI as CLI App
  participant Service as MaintenanceService
  participant Output as Rich Console

  User->>CLI: beautyspot gc --name my-project
  CLI->>Service: clean_garbage()
  Service-->>CLI: result (removed count, etc.)
  CLI->>Output: print summary table
  Output-->>User: formatted output

技術選定

非機能要件方針

• UX: 破壊的な操作（clear等）には対話的な確認プロンプトを表示
• 互換性: 標準出力のパイプ連携を妨げないよう、色の自動無効化に対応
• 利便性: 特定のプロジェクトだけでなく、全DBを横断的にスキャンする機能を提供

コンポーネント構成

graph TD
  A[bs.Spot Factory] -->|構築| B[core.Spot]
  A -->|DI| C[TaskDBBase]
  A -->|DI| D[BlobStorageBase]
  A -->|DI| E[SerializerProtocol]
  A -->|DI| F[StoragePolicyProtocol]
  A -->|DI| G[LimiterProtocol]

コンポーネント責務

技術選定

非機能要件方針

• 結合度: core.Spot は具体的なクラス名（SQLite等）を知らず、Protocolのみに依存する
• テスト容易性: メモリ上のDBや仮装ストレージを注入することで、高速な単体テストが可能
• 拡張性: ユーザーが独自のシリアライザやストレージを自作して注入可能

コンポーネント構成

graph TD
  A[core.Spot] -->|管理要求| B[CacheManager]
  B -->|In-flight追跡| C[_inflight Dict]
  C -->|同期待機| D[threading.Event]
  C -->|非同期待機| E[asyncio.Future]

コンポーネント責務

コンポーネント責務インターフェースCacheManagerキャッシュキーごとの実行状態管理と実行のシリアライズget_or_create_inflight()_inflight実行中のタスクを保持し、後続リクエストをイベントで待機させるExecutionStatecore.Spot待機結果の受け取りとエラーの伝播cached_run()

データフロー

sequenceDiagram
  participant T1 as Thread 1 (First)
  participant T2 as Thread 2 (Second)
  participant CM as CacheManager
  participant Fn as Target Function

  T1->>CM: get_or_create(key)
  Note over CM: 新規作成 (Executor)
  T2->>CM: get_or_create(key)
  Note over CM: 既存あり (Waiter)

  T1->>Fn: execute()
  Fn-->>T1: result
  T1->>CM: set_result(key, result)
  Note over CM: Eventをセット
  CM-->>T2: notify result
  T1-->>T1: return result
  T2-->>T2: return result

技術選定

技術領域選定理由待機機構Event / FutureOS/ランタイムレベルの待機を使用し、CPU負荷（ビジーループ）を回避スコープキャッシュキー単位異なる関数の実行は妨げず、同一入力のみを直列化安全策タイムアウト & リトライ実行者がハングした場合に、待機側がデッドロックしないよう保護

非機能要件方針

• 一貫性: 実行者が成功すれば全員に結果を、失敗すれば全員に同じ例外を伝播

• 効率性: メモリリークを防ぐため、結果配布後に _inflight エントリを即座に削除

• 信頼性: 強参照 (StrongReference) で実行中のタスクを保持し、GCによる消失を防止

本ドキュメントは、『beautyspot』のアーキテクチャ設計を定義する。

システムを構成する主要コンポーネントとその責務、およびコンポーネント間の相互作用を定義する。

インターフェース

class SQLiteTaskDB(TaskDBBase):
    def __init__(self, db_path: str | Path, timeout: float = 30.0): ...
    def save(self, task: TaskRecord) -> None: ...
    def get(self, cache_key: str, *, include_expired: bool = False) -> TaskRecord | None: ...

振る舞い

ライタースレッド方式

• 書き込み操作は全て _WriteTask キューに投入され、専用のバックグラウンドスレッドで直列に実行される。これにより SQLite の database is locked エラーを回避する。

読み取り並行性

• 読み取りはスレッドローカルな接続を使用し、PRAGMA query_only = ON 状態で並行して実行される。

スキーマ管理

• 起動時に tasks テーブルの存在を確認し、必要に応じて自動的に ALTER TABLE によるカラム追加（マイグレーション）を行う。

タイムアウトとフェイルファスト

• 全ての書き込み操作は timeout 秒以内に完了する必要がある。
• PENDING（未着手）状態: timeout を超えた場合、タスクはキューから破棄（キャンセル）され、呼び出し元に TimeoutError を返す。
• RUNNING（実行中）状態: timeout を超えた場合、sqlite3.Connection.interrupt() を用いて実行中のクエリを中断し、呼び出し元に TimeoutError を返す。これにより、ライタースレッドの恒久的なハングを防ぎ、システムの可用性を維持する。

パラメータ詳細

エラーハンドリング

• タイムアウト: TimeoutError を送出する。
• 接続失敗: sqlite3.Error をキャッチし、適切にラップして再送出する。
• シャットダウン時: ライタースレッドへの新規タスク投入を停止し、保留中のタスクを完了させてから接続を閉じる。

エッジケース

• 破損DB: DBファイルが壊れている場合は、初期化時に検知しエラーとする。
• インメモリDB: :memory: 指定時は、プロセスの生存期間中のみキャッシュが保持される。

インターフェース

class CacheManager:
    def herd_sync(
        self, key: str, serializer: Optional[SerializerProtocol] = None
    ) -> Generator[HerdWaitResult, None, None]: ...

    async def herd_async(
        self,
        key: str,
        serializer: Optional[SerializerProtocol],
        loop: asyncio.AbstractEventLoop,
        executor: Any,
    ) -> AsyncGenerator[HerdWaitResult, None]: ...

振る舞い

同時実行制御

1. キャッシュキーをキーとした _inflight 辞書をチェックする
2. 存在しない場合: HerdWaitResult を新規作成し、自分が「実行者 (Executor)」となる
3. 存在する場合: 既存の HerdWaitResult を返し、自分は「待機者 (Waiter)」となる
4. 完了通知: 実行者が処理を終えたら、HerdWaitResult の Event (または Future) をセットし、結果を共有する
5. 自動クリーンアップ: with または async with ブロックを抜ける際、実行者は自動的に notify_and_cleanup_inflight を呼び出し、インフライト状態を解除する。これにより例外発生時も状態が残留しないことが保証される。

強参照によるインフライト状態の保持

目的

実行中のタスク状態（イベント・結果・Future）が GC によって消失しないよう、 _inflight 辞書を通じて強参照で保持する。WeakRef は使用しない。

データ構造

_inflight 辞書はキャッシュキーをキーとし、以下の3要素タプルを値として保持する:

_inflight: dict[str, tuple[threading.Event, list[asyncio.Future], list]]

ライフサイクル

1. 作成: herd_sync / herd_async コンテキスト開始時に _inflight にキーが存在しない場合、 _inflight_lock を保持した状態で新規タプルを挿入する
2. 保持: 実行者が処理を完了するまで、_inflight 辞書がタプルへの唯一の管理参照を保持する。 待機者はロック取得時にタプル要素への参照を取得するが、_inflight エントリが正規の所有者である
3. 解放: コンテキスト終了（__exit__ / __aexit__）時に notify_and_cleanup_inflight で以下の順序で解放する: a. _inflight_lock を取得し、エントリの同一性を event の identity (is) で確認する b. 辞書からエントリを削除する（del _inflight[cache_key]） c. ロックを解放した後、event.set() で同期待機者に通知する d. futures リスト内の各 asyncio.Future に結果またはエラーを伝播する

不変条件

• _inflight 辞書への挿入・削除は常に _inflight_lock の保護下で行う
• 1つのキャッシュキーに対して _inflight 内には高々1つのエントリしか存在しない
• エントリの削除は、挿入時の event と同一オブジェクト（is 比較）を持つスレッドのみが行える
• result_box への書き込みは実行者スレッドのみが行い、待機者は読み取り専用とする
• futures リストへの追加は _inflight_lock の保護下でのみ行う

GC安全性

_inflight 辞書は CacheManager インスタンスの属性であり、CacheManager が 生存している限り、全てのインフライトエントリは GC の対象にならない。 CacheManager 自体は Spot インスタンスが所有するため、with spot: ブロック内での 安全性が保証される。

パラメータ詳細

エラーハンドリング

• 実行者失敗: 実行者が例外を投げた場合、待機者全員にも同じ例外が伝播される。
• タイムアウト: タイムアウト時は待機を解除し、自らが実行者となって再試行（またはエラー）を試みる。

エッジケース

• 非同期混在: 同一キーに対して同期実行と非同期実行が混在する場合も、スレッドセーフなロックにより一つの実行に絞り込まれる。

インターフェース

@spot.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,
)
def my_func(x, y): ...

振る舞い

基本フロー

1. デコレータが対象関数をラップする
2. 呼び出し時にキャッシュキーを生成する
3. DBからキャッシュを検索する
4. ヒット時: デシリアライズして結果を返す（関数は実行しない）
5. ミス時: 関数を実行し、結果をシリアライズしてDB/Blobに保存する

sync/async 自動判定

• デコレーション時に inspect.iscoroutinefunction(fn) で判定する
• sync関数 → _execute_sync() に委譲する同期ラッパーを返す
• async関数 → _execute_async() に委譲する非同期ラッパーを返す

パラメータ詳細

エラーハンドリング

• ジェネレータ関数（isgeneratorfunction または isasyncgenfunction）を渡した場合 → デコレーション時に ConfigurationError を送出する
• cached_run 利用時に引数として関数が一つも指定されなかった場合 → ValidationError を送出する

エッジケース

• ラップされた関数の __name__, __doc__ は functools.wraps によって保持される
• @mark (括弧なし) と @mark() (括弧あり) の両方の記法をサポートする

インターフェース

@contextmanager
def cached_run(
    self,
    *funcs: Any,
    **kwargs: Any
) -> Iterator[Any]: ...

振る舞い

基本フロー

1. Spot インスタンスからコンテキストマネージャとして呼び出される。
2. 引数 funcs に渡された関数が0個の場合はエラーを送出する。
3. 各関数に対して、Spot.mark(**kwargs) で生成したキャッシュデコレータを適用し、ラップされた関数を生成する。
4. ラップされた関数を yield してコンテキストブロック内のユーザーコードに提供する。
5. ユーザーコード内での実行は、通常の @spot.mark 適用済み関数と同様にキャッシュ機構を経由する。
6. コンテキストブロック終了時のクリーンアップ処理は特に行わない（一時的なラッパーはスコープを抜けると破棄される）。

パラメータ詳細

戻り値（Yields）の仕様

• 1つの関数を渡した場合: ラップされた単一の callable が返る。
• 複数の関数を渡した場合: ラップされた callable のタプルが返る。

エラーハンドリング

• 引数エラー: funcs が空（0個）の場合、beautyspot.exceptions.ValidationError を送出し、コンテキストに入らない。

エッジケース

• 非同期関数やジェネレータ関数を渡した場合の挙動は、基底となる mark デコレータの制約に完全に依存する。
• 適用されるキャッシュキーは元の関数に基づくため、同名・同シグネチャの別関数が混同されないように KeyGen のスコープ規則に従う。

概要

@spot.mark() デコレータは同期関数と非同期（async def）関数の両方をサポートし、関数定義に応じて適切な実行・保存・キャッシュ検索フローを透過的に切り替える。

振る舞い

sync/async の自動判定

• デコレーション時に inspect.iscoroutinefunction(func) を用いて判定する。
• ランタイム時の判定オーバーヘッドを削減するため、ラッパー構築時に分岐を行う。

実行フロー

• 同期関数の場合:
• _execute_sync() に委譲する。
• キャッシュ保存（DBやBlobへのI/O）は、デフォルトでバックグラウンドのスレッドプール (ThreadPoolExecutor) にタスクを投げて非同期化される（save_sync=False の場合）。
• 非同期関数の場合:
• _execute_async() に委譲する。
• キャッシュ検索や保存処理などのI/Oバウンドな操作は、asyncio イベントループ経由でノンブロッキングに処理され、保存自体も非同期でキューに投入される。

エラーハンドリング

• ジェネレータ関数の拒否: inspect.isgeneratorfunction または inspect.isasyncgenfunction で真となるジェネレータ・非同期ジェネレータが渡された場合、サポート対象外として即座に ConfigurationError を送出する。
• 保存タスク中のバックグラウンドエラーは、同期・非同期ともに on_background_error コールバックが設定されていればそこに処理が委譲される。

エッジケース

• デコレーション後に手動で関数の性質（コルーチンかどうか）を変更するような動的な書き換えには対応しない（デコレーション時の静的な判定結果に依存する）。

インターフェース

class KeyGen:
    @staticmethod
    def _default(args: tuple, kwargs: dict) -> str: ...

    @staticmethod
    def hash_items(items: list) -> str: ...

振る舞い

基本フロー (_default)

1. 正規化: args と kwargs を canonicalize() により再帰的に正規化する
2. パック: 正規化された構造 [args, kwargs] を MessagePack でバイナリにシリアライズする
3. ハッシュ: バイナリデータに対して SHA-256 ハッシュを計算し、16進数文字列を返す

リストハッシュ (hash_items)

• 任意個の正規化済みアイテムを含むリストを受け取り、一括でパックしてハッシュ化する。KeyGenPolicy で使用される。

パラメータ詳細

エラーハンドリング

• 再帰制限超過: RecursionError 発生時は、安全のため str((args, kwargs)) のハッシュをフォールバックとして使用する
• パック失敗: MessagePack シリアライズに失敗した場合は、警告をログ出力した上で str() ベースのハッシュを返す

エッジケース

• 空の引数: 引数なしの場合も [(), {}] の構造として一定のハッシュ値を生成する
• 非決定的な repr: フォールバック時の str() はオブジェクトによっては実行ごとに変わる可能性があるため、明示的な正規化が推奨される

インターフェース

@singledispatch
def canonicalize(obj: Any) -> Any: ...

振る舞い

singledispatch を使用し、型に応じた最適な正規化形式へ再帰的に変換する。

主要な型別の処理

• 基本型 (int, float, str, bytes): そのまま返す
• bool: ("__bool__", value) のタプルに変換（1 と True の衝突を防止）
• dict: キーでソートし、[[k1, v1], [k2, v2], ...] のリストに変換
• list / tuple / set / frozenset / deque: それぞれ専用の型タグ（例: "__list__") を付与し、要素を再帰的に正規化。集合型はソートを行う
• Enum: ("__enum__", module, qualname, value) に変換
• カスタムオブジェクト: __dict__ または __slots__ から属性を取得し、型名と共に正規化

パラメータ詳細

エラーハンドリング

• 未対応型: 警告をログ出力し、str(obj) にフォールバックする
• 循環参照: RecursionError をキャッチし、その階層で str() による正規化に切り替える

エッジケース

• 異なる型の比較: _safe_sort_key により、Python 3 で比較不可能な型同士（例: int と str）が辞書のキーに含まれていても安定したソート順を保証する

インターフェース

class Strategy(Enum):
    DEFAULT = auto()      # 再帰的正規化
    IGNORE = auto()       # キー計算から除外
    FILE_CONTENT = auto() # ファイルの内容をハッシュ
    PATH_STAT = auto()    # パス+サイズ+更新時刻をハッシュ

class KeyGenPolicy:
    def bind(self, func: Callable) -> Callable[..., str]: ...

振る舞い

1. バインド: bind(func) 時に inspect.signature を解析し、引数名とデフォルト値を把握する
2. 適用: 呼び出し時に引数を実値にバインドし、デフォルト値を補填する
3. 戦略実行: 引数名ごとに指定された Strategy を適用し、正規化された値のリストを作成する
4. 集約: 全ての値を KeyGen.hash_items() で一つのハッシュにまとめる

パラメータ詳細

エラーハンドリング

• ファイル不在: FILE_CONTENT / PATH_STAT 指定時にファイルがない場合、"MISSING_{path}" という固定文字列をハッシュ対象とする
• アクセスエラー: ファイル読み込み失敗時は "ERROR_{path}" を返す

エッジケース

• シグネチャ不一致: 引数が関数のシグネチャと合わない場合は TypeError を送出する（実行前バリデーション）
• デフォルト値の変更: 関数のデフォルト値が変わると、同じ引数でもハッシュ値が変わる（意図的な挙動）

インターフェース

@runtime_checkable
class TaskDBCore(Protocol):
    def get(self, cache_key: str, *, include_expired: bool = False) -> TaskRecord | None: ...
    def save(self, record: TaskRecord) -> None: ...

class TaskDBBase(ABC):
    # TaskDBCore を満たしつつ、メンテナンスのデフォルト実装を提供

振る舞い

階層構造

1. TaskDBCore: キャッシュの実行に必要な最小限のメソッド定義
2. Maintenable: GCや統計取得に必要な管理用メソッドの定義
3. TaskDBBase: 共通のバリデーションや例外処理を実装する基底クラス

パラメータ詳細 (主要メソッド)

エラーハンドリング

• プロトコルに未実装のメソッドを呼び出した場合、実行時に NotImplementedError を送出する代わりに、runtime_checkable による事前チェックを推奨する。

エッジケース

• 複合実装: 複数のプロトコルを一つのクラスで実装する場合、TaskDBMaintenable 構成プロトコルとして扱う。

インターフェース

class LocalStorage(BlobStorageMaintenable):
    def __init__(self, base_dir: str | Path): ...
    def save(self, key: str, data: bytes) -> str: ...
    def load(self, location: str) -> bytes: ...

振る舞い

アトミック書き込み (save)

1. 指定ディレクトリ内に tempfile.mkstemp で一時ファイルを作成
2. データを書き込み、flush() および os.fsync() でディスク到達を保証
3. os.replace で最終的なパスへリネーム（アトミックな置換）

パストラバーサル対策 (load / delete)

• 解決後の絶対パスが base_dir の配下にあることを is_relative_to で厳密にチェックし、範囲外へのアクセスを遮断する。

パラメータ詳細

エラーハンドリング

• ファイル不在: load 時にファイルがない場合 CacheCorruptedError を送出
• 権限エラー: PermissionError 発生時は、GCでの回収に委ねるため警告ログのみ出力

エッジケース

• ディスクフル: 書き込み中に容量不足になった場合、一時ファイルを削除して OSError を送出
• Windowsパス: パス区切り文字として / と \ の両方を検証対象とする

インターフェース

class S3Storage(BlobStorageMaintenable):
    def __init__(self, s3_uri: str, s3_opts: dict | None = None): ...

振る舞い

URIベースの管理

• s3://bucket/prefix/key.bin 形式のURIをロケーション識別子として使用する。

大容量対応 (save)

• boto3.upload_fileobj を使用。5GBを超えるデータに対しては、自動的にマルチパートアップロードに切り替えられ、PutObjectの制限を回避する。

高速なメタデータ取得 (get_mtime)

• head_object を使用し、データ本体をダウンロードせずに最終更新時刻（LastModified）を取得する。

パラメータ詳細

エラーハンドリング

• ライブラリ不在: boto3 がインポートできない環境でインスタンス化を試みると ImportError を送出
• ネットワークエラー: botocore.exceptions.ClientError をキャッチし、状況に応じて CacheCorruptedError に変換

エッジケース

• リージョン不一致: S3クライアントの設定とバケットのリージョンが異なる場合の動作は Boto3 のデフォルト設定に依存
• 削除の冪等性: S3 は存在しないオブジェクトの削除要求に対してエラーを返さないため、常に成功として扱う

インターフェース

class StoragePolicyProtocol(Protocol):
    def should_save_as_blob(self, data: bytes) -> bool: ...

振る舞い

判定ロジック

• ThresholdStoragePolicy: len(data) が threshold を超えた場合に True を返す。
• WarningOnlyPolicy: 常に False を返すが、閾値を超えた場合にログ出力のみ行う（互換モード）。
• AlwaysBlobPolicy: 常に True を返す。

パラメータ詳細

エラーハンドリング

• ポリシー内の判定で例外が発生した場合は、安全のため False (DB保存) とみなし、エラーをログに記録する。

エッジケース

• ゼロバイト: 空のデータに対する挙動は各ポリシーの閾値設定に依存する（通常はDB保存）。
• 閾値ちょうどのサイズ: len(data) > threshold のため、閾値と等しい場合はDB保存となる。

インターフェース

class MsgpackSerializer(SerializerProtocol, TypeRegistryProtocol):
    def dumps(self, obj: Any) -> bytes: ...
    def loads(self, data: bytes) -> Any: ...

振る舞い

シリアライズ (dumps)

1. オブジェクトの型を type(obj) で取得
2. 登録済み型またはそのサブクラス（MRO順）を探索
3. ヒットした場合、カスタムエンコーダを実行し ExtType としてパック
4. ヒットしない場合、MessagePack のデフォルト処理（または SerializationError）

デシリアライズ (loads)

1. MessagePack ストリームをパース
2. ExtType を発見した場合、コードに対応するカスタムデコーダを呼び出す

パラメータ詳細

エラーハンドリング

• 未登録型: シリアライズ時に対応するエンコーダが見つからない場合、SerializationError を送出
• デコード失敗: データ破損やコード不一致時は SerializationError を送出

エッジケース

• 動的型: 名前が同じでも ID が異なる動的生成型は、LRUキャッシュによりメモリリークを防ぎつつ効率的に処理される
• スレッド安全性: registry_generation カウンタを使用して、スレッドローカルキャッシュの整合性を維持