# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-disambiguate-nested-timeouts-when-cascading-failures-are-indistinguishable.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-disambiguate-nested-timeouts-when-cascading-failures-are-indistinguishable.warrant.json # Markdown SHA-256: b280ab7b616dda2b88ce26377b33bacee09931d269137073d9d54533af105002 # Code AST SHA-256: e25fdd46936e5bc8314033484a409477a245da369606d733336729fdf9b70790 # Verified Classes: TimeoutVerdict, TimeoutEvent, TimeoutPolicy, TimeoutManifest, NestedTimeoutGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class TimeoutVerdict(Enum): INNERMOST_LAYER = auto() INNER_LAYER_CAUSED = auto() OUTER_LAYER_CAUSED = auto() CASCADE_ORIGIN_IDENTIFIED = auto() @dataclass(frozen=True) class TimeoutEvent: timeout_id: str observed_layer: str inner_layer_latencies: tuple[tuple[str, float], ...] outer_timeout_sec: float outer_processing_sec: float cascade_dependency: str | None latency_budget_pct: float @dataclass(frozen=True) class TimeoutPolicy: require_inner_layer_attribution: bool @dataclass(frozen=True) class TimeoutManifest: verdict: TimeoutVerdict directive: str manifest_digest: str class NestedTimeoutGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_timeout( cls, event: TimeoutEvent, policy: TimeoutPolicy ) -> TimeoutManifest: # Gate 1: Innermost layer if not event.inner_layer_latencies: digest = cls._hash(json.dumps({ "timeout": event.timeout_id, "layer": event.observed_layer }, sort_keys=True)) return TimeoutManifest( verdict=TimeoutVerdict.INNERMOST_LAYER, directive=f"PROCEED: Timeout {event.timeout_id} at innermost layer {event.observed_layer}.", manifest_digest=digest ) # Gate 2: Cascade origin if event.cascade_dependency: return TimeoutManifest( verdict=TimeoutVerdict.CASCADE_ORIGIN_IDENTIFIED, directive=f"TRACE: Timeout {event.timeout_id} cascades from {event.cascade_dependency}.", manifest_digest=cls._hash(f"CASCADE:{event.timeout_id}:{event.cascade_dependency}") ) # Gate 3: Inner latency sum inner_sum = sum(lat for _, lat in event.inner_layer_latencies) inner_dominant = inner_sum + event.outer_processing_sec >= event.outer_timeout_sec if policy.require_inner_layer_attribution and inner_dominant: worst_layer = max(event.inner_layer_latencies, key=lambda x: x[1]) budget_used = (worst_layer[1] / event.outer_timeout_sec) * 100 if event.outer_timeout_sec > 0 else 0 if budget_used > event.latency_budget_pct: return TimeoutManifest( verdict=TimeoutVerdict.INNER_LAYER_CAUSED, directive=f"ATTRIBUTE: Timeout {event.timeout_id} caused by {worst_layer[0]} ({worst_layer[1]:.2f}s of {event.outer_timeout_sec:.2f}s).", manifest_digest=cls._hash(f"INNER:{worst_layer[0]}:{event.timeout_id}") ) # Gate 4: Outer layer caused if event.outer_processing_sec > event.outer_timeout_sec * 0.5: return TimeoutManifest( verdict=TimeoutVerdict.OUTER_LAYER_CAUSED, directive=f"ATTRIBUTE: Timeout {event.timeout_id} caused by outer layer {event.observed_layer}.", manifest_digest=cls._hash(f"OUTER:{event.observed_layer}:{event.timeout_id}") ) # Gate 5: Ambiguous, inner dominant digest = cls._hash(json.dumps({ "timeout": event.timeout_id, "inner_sum": f"{inner_sum:.3f}", "outer_processing": f"{event.outer_processing_sec:.3f}", "budget": f"{event.outer_timeout_sec:.3f}" }, sort_keys=True)) return TimeoutManifest( verdict=TimeoutVerdict.INNER_LAYER_CAUSED, directive=f"ATTRIBUTE_WITH_CAUTION: Inner layers consumed {inner_sum:.2f}s of {event.outer_timeout_sec:.2f}s.", manifest_digest=digest )