SPEC Specification

インターフェース

@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 を送出する（実行前バリデーション）
• デフォルト値の変更: 関数のデフォルト値が変わると、同じ引数でもハッシュ値が変わる（意図的な挙動）

インターフェース

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

振る舞い

ライタースレッド方式

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

読み取り並行性

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

スキーマ管理

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

パラメータ詳細

エラーハンドリング

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

エッジケース

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

インターフェース

@runtime_checkable
class TaskDBCore(Protocol):
    def get(self, input_key: str) -> 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 カウンタを使用して、スレッドローカルキャッシュの整合性を維持

インターフェース

def register(
    code: int,
    encoder: Callable[[Any], Any],
    decoder: Callable[[Any], Any],
) -> Callable: ... # デコレータ形式

def register_type(
    type_class: Type,
    code: int,
    encoder: Callable[[Any], Any],
    decoder: Callable[[Any], Any],
) -> None: ...

振る舞い

1. 重複チェック: 同一コードまたは同一クラスが既に登録されていないか確認
2. CoW更新: 現在のレジストリ辞書をコピーし、新しいエントリを追加した後に参照を差し替える
3. 世代更新: _registry_generation をインクリメントし、全スレッドのLRUキャッシュを無効化対象とする

パラメータ詳細

エラーハンドリング

• コード重複: ValueError を送出し、レジストリの破損を防ぐ
• 範囲外コード: 128以上のコードを指定した場合は ValueError

エッジケース

• 継承: 基底クラスを登録すると、サブクラスも自動的にそのエンコーダを使用する（オーバーライド可能）

インターフェース

class _BackgroundLoop:
    def submit(self, coro: Coroutine) -> None: ...
    def stop(self) -> None: ...
    def is_running(self) -> bool: ...

振る舞い

1. 起動: 初回の submit 時にデーモンスレッドを作成し、asyncio.run() を開始する
2. 投入: run_coroutine_threadsafe を使用して、メインスレッドからコルーチンをイベントループへ投入
3. 待機: 投入されたタスクの完了は待たず、制御を即座に呼び出し元へ返す

パラメータ詳細

エラーハンドリング

• 投入失敗: ループ停止中にタスクが投入された場合、エラーをログ出力し無視する
• タスク内エラー: コルーチン内で発生した例外は、on_background_error コールバックで処理される

エッジケース

• 多重起動: すでにスレッドが実行中の場合は、既存のスレッドを再利用する

インターフェース

def flush(timeout: float | None = None) -> bool: ...

@spot.mark(save_sync=False)
def my_func(): ...

振る舞い

同期保存 (save_sync=True)

• 関数終了後、DBおよびBlobへの保存が完了するまで呼び出し元をブロックする。

非同期保存 (save_sync=False)

• 保存処理を BackgroundLoop に委譲し、即座に関数の戻り値を返す。

フラッシュ (flush)

• _inflight タスク（保存中）のリストを確認し、それら全てが完了するまで、またはタイムアウトまで待機する。

パラメータ詳細

エラーハンドリング

• タイムアウト: flush がタイムアウトした場合、False を返し、一部のデータが未保存である可能性を通知する。

エッジケース

• コンテキスト終了: with spot: 終了時に自動的に flush が呼ばれ、プロセス終了前のデータ保存を確実にする。

インターフェース

class LifecyclePolicy:
    def add_rule(self, pattern: str, retention: Any) -> None: ...
    def resolve_with_fallback(self, func_name: str) -> float | None: ...

振る舞い

1. 順次照合: 登録された順にルールを走査する

2. パターンマッチ: fnmatch.fnmatch を使用し、関数名がパターンに合致するか判定

3. 二段階解決:

   1. 最初に「完全修飾名 (module.qualname)」で照合

   2. マッチしなければ「短縮名 (qualname)」で照合

4. 結果返却: 最初にマッチしたルールの保持期間を秒数に変換して返す

パラメータ詳細

パラメータ例説明pattern"my_mod.*", "*_test"Glob形式のパターンretention"30d", 3600保持期間の定義

デフォルト保持期間

• default_retention パラメータで、どのルールにもマッチしない場合のデフォルト保持期間を指定できる
• LifecyclePolicy.default() のデフォルト値は "30d"（30日）
• None を指定すると無期限保持（従来互換）

エラーハンドリング

• マッチなし: 全てのルールにマッチしない場合は、default_retention で指定された保持期間を返す（デフォルト: 30日）

エッジケース

• 重複パターン: 先に登録されたルールが優先される。

インターフェース

def parse_retention(val: Any) -> float | None: ...

振る舞い

文字列パース

• "7d" (7日), "12h" (12時間), "30m" (30分), "10s" (10秒) 形式を秒数に変換する。大文字小文字は区別しない。

数値・オブジェクト

• int, float: 秒数としてそのまま扱う。
• timedelta: total_seconds() を取得。
• None: ポリシーに従う（または無期限）を意味する。

パラメータ詳細 (単位)

エラーハンドリング

• 不正な形式: パーサが理解できない形式の場合は ValidationError を送出する。
• 負の値: 過去の時間は指定できず、ValidationError となる。

エッジケース

• FOREVER: 特別な定数 Retention.FOREVER (無限大の秒数) を提供し、ポリシーに関わらず永続化を指示できる。

インターフェース

class TokenBucket(LimiterProtocol):
    def consume(self, cost: float = 1.0) -> None: ...
    async def consume_async(self, cost: float = 1.0) -> None: ...

振る舞い

GCRAアルゴリズム

• 各リクエストに対して「次回実行可能な理論時刻 (TAT: Theoretical Arrival Time)」を計算・更新する。
• 現在時刻が TAT - 許容バースト 以前の場合、リクエストを拒否または待機させる。

同期 vs 非同期

• consume: 待機が必要な場合は time.sleep() でブロック。
• consume_async: 待機が必要な場合は asyncio.sleep() で非占有待機。

パラメータ詳細

エラーハンドリング

• 即時拒否モード: 待機を許容しない設定でレート超過した場合、RateLimitError (ValueErrorのサブクラス) を送出する。

エッジケース

• 高コストリクエスト: max_burst を超えるコストを持つ単一リクエストは、常に拒否される。
• アイドル時間: 長時間リクエストがない場合も、TAT は現在時刻より過去に一定以上離れないよう補正される（過度なバーストの防止）。

インターフェース

class HookBase:
    def pre_execute(self, ctx: PreExecuteContext) -> None: ...
    def on_cache_hit(self, ctx: CacheHitContext) -> None: ...
    def on_cache_miss(self, ctx: CacheMissContext) -> None: ...

振る舞い

1. フック呼出: キャッシュエンジンの各ステート（実行前、ヒット、ミス）で登録されたフックを順番に実行する
2. スレッド安全化: ThreadSafeHookBase を継承した場合、メタクラスが自動的に各メソッドを with self._lock: で包む

パラメータ詳細 (Context)

エラーハンドリング

• フック内例外: フックの実行中に発生した例外はキャッチされ、エラーログとして出力されるが、主処理（関数の実行結果）は中断されない。

エッジケース

• フックの順序: hooks=[h1, h2] と指定した場合、h1 -> h2 の順で実行される。

インターフェース

class MaintenanceService:
    def clean_garbage(self, grace_period: int = 3600) -> GarbageStats: ...
    def prune(self, days: int, func_name: str | None = None) -> int: ...
    def clear(self, func_name: str | None = None) -> int: ...

振る舞い

ガベージコレクション (clean_garbage)

1. DB Flush: 保存中のタスクをDBへ反映させる
2. 期限切れ削除: delete_expired() でDBから古いタスクを除去
3. 孤立Blobスキャン: DB内の全 blob_key とストレージ内の全ファイルを比較
4. 物理削除: 参照のないBlobのうち、作成から grace_period 以上経過したものを削除
5. 構造清掃: ストレージ内の空ディレクトリを削除

パラメータ詳細

エラーハンドリング

• 削除失敗: 個別のファイル削除失敗は警告ログに留め、全体の処理は継続する。
• DBロック: メンテナンス中にDBがロックされた場合は、タイムアウトまで待機し失敗させる。

エッジケース

• 大規模ストレージ: Blobの列挙はジェネレータで行い、メモリ使用量を抑制する。
• 名前空間: clear(func_name) は前方一致ではなく、完全一致（または glob）で判定する。

インターフェース

beautyspot list [--db DB_PATH]
beautyspot gc [--name PROJECT_NAME] [--force]
beautyspot stats
beautyspot ui

振る舞い

1. コンテキスト解決: --db 指定がない場合、デフォルトの .beautyspot/ ディレクトリを探索する
2. コマンド実行: MaintenanceService または Spot インスタンスを生成し、対応するメソッドを呼び出す
3. フォーマット出力: Rich ライブラリを使用して、結果をテーブルやプログレスバーで表示する

サブコマンド詳細

エラーハンドリング

• DB不在: 指定されたパスにDBがない場合は、エラーメッセージを表示して終了する。
• 権限不足: ファイル削除権限がない場合は、スキップされた旨を表示する。

エッジケース

• 非対話環境: CI等での実行を考慮し、色の無効化や確認プロンプトのスキップ（--force）に対応する。

インターフェース

def Spot(
    name: str = "default",
    storage_path: str | Path | None = None,
    serializer: SerializerProtocol | None = None,
    limiter: LimiterProtocol | None = None,
    # ...その他の依存
) -> core.Spot: ...

振る舞い

1. パス解決: storage_path が未指定の場合、.beautyspot/ 以下のプロジェクト名ディレクトリを使用する
2. DB初期化: SQLiteTaskDB を生成し、マイグレーションを実行
3. ストレージ構成: URI（s3://等）を判定し、適切な BlobStorage 実装を生成
4. インスタンス化: 全てのコンポーネントを core.Spot のコンストラクタに注入して返す

パラメータ詳細

エラーハンドリング

• ディレクトリ作成失敗: 権限等の理由でストレージディレクトリが作成できない場合は、早期に OSError を送出する。

エッジケース

• グローバルシングルトン: 同じ name で複数回呼び出した場合も、原則として新しいインスタンスを返すが、DBファイルへの接続は共有される（SQLiteのWALモードを活用）。

インターフェース

class CacheManager:
    def get_or_create_inflight(
        self, key: str, is_async: bool
    ) -> ExecutionState: ...

振る舞い

同時実行制御

1. キャッシュキーをキーとした _inflight 辞書をチェックする
2. 存在しない場合: ExecutionState を新規作成し、自分が「実行者 (Executor)」となる
3. 存在する場合: 既存の ExecutionState を返し、自分は「待機者 (Waiter)」となる
4. 完了通知: 実行者が処理を終えたら、ExecutionState の Event (または Future) をセットし、結果を共有する

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

目的

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

データ構造

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

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

ライフサイクル

1. 作成: wait_herd_sync / wait_herd_async で _inflight にキーが存在しない場合、 _inflight_lock を保持した状態で新規タプルを挿入する
2. 保持: 実行者が処理を完了するまで、_inflight 辞書がタプルへの唯一の管理参照を保持する。 待機者はロック取得時にタプル要素への参照を取得するが、_inflight エントリが正規の所有者である
3. 解放: 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: ブロック内での 安全性が保証される。

パラメータ詳細

エラーハンドリング

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

エッジケース

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

インターフェース

class BlobStorageBase(ABC):
    @abstractmethod
    def save(self, key: str, data: ReadableBuffer) -> str: ...
    @abstractmethod
    def load(self, location: str) -> bytes: ...
    @abstractmethod
    def delete(self, location: str) -> None: ...
    @abstractmethod
    def list_keys(self) -> Iterator[str]: ...
    @abstractmethod
    def get_mtime(self, location: str) -> float: ...

振る舞い

• save: データを永続化し、一意な識別子（パスやURI）を返す。
• load: 識別子を基にバイトデータを復元する。
• delete: 識別子を基にデータを削除する。存在しない場合は無視する（冪等）。
• list_keys: メンテナンス用に、保存されている全オブジェクトの識別子を列挙する。
• get_mtime: 最終更新時刻を POSIX タイムスタンプで返す。

パラメータ詳細

エラーハンドリング

• 実装クラスは、データの欠損やアクセス不可時に CacheCorruptedError を送出すべきである。

エッジケース

• バイナリ互換: ReadableBuffer を受け入れることで、memoryview 等を使用したゼロコピー書き込みを可能にする。