# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-fallback-and-degradation-boundaries-must-an-instance-isolate-partial-upstream-outages-to-prevent-cascading-service-failure-and-thread-pool-saturation.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-fallback-and-degradation-boundaries-must-an-instance-isolate-partial-upstream-outages-to-prevent-cascading-service-failure-and-thread-pool-saturation.warrant.json
# Markdown SHA-256:   bbf24b7312b1cab284963520b5ee57bcaa2c105da6bc4fa0e32d96d90bd73d68
# Code AST SHA-256:   57158613d25c9e1c46f8ddb0c592d50f4189414980a882cb7edf9c91391bdc27
# Verified Classes:   UpstreamDegradedError
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import asyncio
import logging
import time
from typing import Any, Awaitable, Callable, TypeVar

T = TypeVar("T")
logger = logging.getLogger("corpusfetch.resilience")


class UpstreamDegradedError(Exception):
    '''Raised when an upstream service is unreachable and no fallback is available.'''
    pass


async def execute_with_fallback(
    operation_name: str,
    async_fn: Callable[[], Awaitable[T]],
    fallback_value: T,
    *,
    timeout_sec: float = 1.5,
    circuit_breaker: Any | None = None,
) -> T:
    '''
    Executes an optional or secondary asynchronous operation with a strict timeout,
    optional circuit-breaker check, and a deterministic fallback return value on failure.
    '''
    # Gate 1: Check upstream circuit breaker state if provided
    if circuit_breaker is not None and hasattr(circuit_breaker, "allow_request"):
        if not circuit_breaker.allow_request():
            logger.warning(
                "Circuit breaker OPEN for %s. Bypassing upstream call and returning fallback.",
                operation_name,
            )
            return fallback_value

    start_time = time.monotonic()
    try:
        # Gate 2: Enforce strict non-blocking timeout
        result = await asyncio.wait_for(async_fn(), timeout=timeout_sec)
        
        # Record success on circuit breaker if present
        if circuit_breaker is not None and hasattr(circuit_breaker, "record_success"):
            circuit_breaker.record_success()
            
        return result

    except (asyncio.TimeoutError, ConnectionError, OSError) as exc:
        duration = time.monotonic() - start_time
        logger.warning(
            "Upstream failure in %s after %.2fs (%s: %s). Falling back to default value.",
            operation_name,
            duration,
            type(exc).__name__,
            exc,
        )
        # Record failure on circuit breaker if present
        if circuit_breaker is not None and hasattr(circuit_breaker, "record_failure"):
            circuit_breaker.record_failure()

        return fallback_value

    except Exception as exc:
        duration = time.monotonic() - start_time
        logger.error(
            "Unexpected error in %s after %.2fs (%s: %s). Falling back.",
            operation_name,
            duration,
            type(exc).__name__,
            exc,
        )
        if circuit_breaker is not None and hasattr(circuit_breaker, "record_failure"):
            circuit_breaker.record_failure()
        return fallback_value

def verify_degradation_boundary_invariants(timeout_seconds: float = 2.0) -> bool:
    if timeout_seconds <= 0.0:
        raise ValueError("timeout_seconds must be strictly positive")
    res = execute_optional_enrichment(
        operation_name="test_failing_enrichment",
        fn=lambda: (_ for _ in ()).throw(TimeoutError("upstream timed out")),
        fallback_value={"degraded": True},
    )
    assert res == {"degraded": True}
    return True