# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-deadline-propagation-and-cancellation-cascade-gates-must-an-instance-govern-grpc-and-http2-calls-to-prevent-zombie-worker-exhaustion-and-wasted-compute.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-deadline-propagation-and-cancellation-cascade-gates-must-an-instance-govern-grpc-and-http2-calls-to-prevent-zombie-worker-exhaustion-and-wasted-compute.warrant.json
# Markdown SHA-256:   d9355f4b155f81799fe9078a4db6a1757261533bbbc5add29c462fe26cec01b5
# Code AST SHA-256:   7fe405c2f17019d27dde8d59c64cb35bda6e4f6958927d4d15b63e0e107d2861
# Verified Classes:   DeadlineExceededError, UpstreamCancelledError, DeadlinePropagationManager
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import asyncio
import time
from contextlib import asynccontextmanager
from typing import Any, AsyncIterator, Awaitable, Callable, TypeVar

T = TypeVar("T")


class DeadlineExceededError(TimeoutError):
    '''Raised when an absolute request deadline has expired before or during execution.'''
    def __init__(self, message: str = "Request deadline exceeded") -> None:
        super().__init__(message)


class UpstreamCancelledError(asyncio.CancelledError):
    '''Raised when an upstream client disconnects or cancels the active stream.'''
    pass


class DeadlinePropagationManager:
    '''
    Manages absolute deadlines and cancellation cascading for distributed RPC calls:
    1. Extracts and tracks absolute deadline epochs (`deadline_epoch`).
    2. Computes shrinking remaining budgets (`remaining_sec`) across microservice hops.
    3. Fails fast (`DeadlineExceededError`) if remaining budget <= 0.
    4. Automatically cascades cancellations to child tasks upon upstream abort.
    '''

    def __init__(self, *, default_timeout_sec: float = 10.0) -> None:
        self.default_timeout_sec = max(0.1, default_timeout_sec)

    @asynccontextmanager
    async def request_scope(
        self,
        *,
        incoming_deadline_epoch: float | None = None,
        override_timeout_sec: float | None = None,
    ) -> AsyncIterator[dict[str, Any]]:
        '''
        Establishes an absolute deadline scope for an incoming request.
        Calculates remaining budget and links cancellation to child coroutines.
        '''
        now = time.monotonic()
        if incoming_deadline_epoch is not None:
            deadline = incoming_deadline_epoch
        else:
            timeout = override_timeout_sec if override_timeout_sec is not None else self.default_timeout_sec
            deadline = now + timeout

        remaining = deadline - now
        if remaining <= 0:
            raise DeadlineExceededError("Deadline already expired upon ingress entry")

        scope_meta = {
            "deadline_epoch": deadline,
            "allocated_timeout_sec": remaining,
        }

        # Track child tasks spawned under this request for cascading cancellation
        child_tasks: set[asyncio.Task[Any]] = set()

        async def spawn_child(fn: Callable[..., Awaitable[Any]], *args: Any, **kwargs: Any) -> Any:
            rem = deadline - time.monotonic()
            if rem <= 0:
                raise DeadlineExceededError("Deadline expired before child task execution")
            
            task = asyncio.create_task(fn(*args, **kwargs))
            child_tasks.add(task)
            try:
                return await asyncio.wait_for(task, timeout=rem)
            except asyncio.TimeoutError:
                task.cancel()
                raise DeadlineExceededError("Child task timed out against absolute deadline")
            finally:
                child_tasks.discard(task)

        scope_meta["spawn_child"] = spawn_child

        try:
            yield scope_meta
        except (asyncio.CancelledError, GeneratorExit):
            # Gate 4: Upstream disconnect cascades cancellation to all active child tasks instantly
            for task in child_tasks:
                if not task.done():
                    task.cancel()
            raise