# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-memory-invariants-must-survive-a-distributed-cache-eviction-storm.md # Remnant Warrant: https://corpusfetch.com/c/what-memory-invariants-must-survive-a-distributed-cache-eviction-storm.warrant.json # Markdown SHA-256: 8be2aa788ad9320e6031fd0e44f4cd18949570395a623cd650296142cc74b97f # Code AST SHA-256: da3222e685f5aeff662759cfb357ffed634d5b7d0423559a459aa68130462627 # Verified Classes: EvictionVerdict, CacheState, EvictionPolicy, EvictionManifest, EvictionStormGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class EvictionVerdict(Enum): NOMINAL_EVICTION = auto() STORM_DETECTED_THROTTLE = auto() STORM_DETECTED_HALT_REPOPULATION = auto() @dataclass(frozen=True) class CacheState: eviction_rate_per_sec: float safe_eviction_threshold: float backing_store_headroom_ratio: float @dataclass(frozen=True) class EvictionPolicy: critical_keys: frozenset[str] min_headroom_for_repopulation: float @dataclass(frozen=True) class EvictionManifest: verdict: EvictionVerdict directive: str manifest_digest: str class EvictionStormGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_storm( cls, state: CacheState, policy: EvictionPolicy ) -> EvictionManifest: # Gate 1: Eviction rate within safe bounds if state.eviction_rate_per_sec <= state.safe_eviction_threshold: digest = cls._hash(json.dumps({ "rate": f"{state.eviction_rate_per_sec:.2f}", "threshold": f"{state.safe_eviction_threshold:.2f}" }, sort_keys=True)) return EvictionManifest( verdict=EvictionVerdict.NOMINAL_EVICTION, directive="PROCEED: Eviction rate within normal bounds.", manifest_digest=digest ) # Gate 2: Backing store headroom check if state.backing_store_headroom_ratio < policy.min_headroom_for_repopulation: return EvictionManifest( verdict=EvictionVerdict.STORM_DETECTED_HALT_REPOPULATION, directive=f"HALT: Eviction storm detected. Headroom {state.backing_store_headroom_ratio:.2f} < {policy.min_headroom_for_repopulation:.2f}. Halt repopulation.", manifest_digest=cls._hash(f"HALT:{state.backing_store_headroom_ratio:.2f}") ) # Gate 3: Throttled repopulation digest = cls._hash(json.dumps({ "rate": f"{state.eviction_rate_per_sec:.2f}", "headroom": f"{state.backing_store_headroom_ratio:.2f}", "critical_keys": tuple(sorted(policy.critical_keys)) }, sort_keys=True)) return EvictionManifest( verdict=EvictionVerdict.STORM_DETECTED_THROTTLE, directive=f"THROTTLE: Repopulate only critical keys {sorted(policy.critical_keys)}. Throttle other fetches.", manifest_digest=digest )