# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-bounded-queue-admission-control-and-load-shedding-gates-must-an-instance-manage-concurrency-to-prevent-unbounded-memory-growth-and-tail-latency-collapse.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-bounded-queue-admission-control-and-load-shedding-gates-must-an-instance-manage-concurrency-to-prevent-unbounded-memory-growth-and-tail-latency-collapse.warrant.json
# Markdown SHA-256:   df59f8fed9039e5393c1357870b3ee2f3f48d5639ac71e60d35729989deedb27
# Code AST SHA-256:   274b341e070469505337ac08a69155b9e28f88c626206d1527e0ecc9e144c365
# Verified Classes:   LoadSheddingError, BoundedAdmissionController
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import asyncio
import time
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, TypeVar

T = TypeVar("T")


class LoadSheddingError(Exception):
    '''Raised immediately when the admission queue or concurrency pool is saturated.'''
    def __init__(self, retry_after_sec: float = 5.0) -> None:
        self.retry_after_sec = max(1.0, retry_after_sec)
        super().__init__(f"Server overloaded; load shedding active (retry_after={self.retry_after_sec}s)")


class BoundedAdmissionController:
    '''
    Enforces backpressure and load shedding through:
    1. Strict queue size bounding (`maxsize=N`).
    2. Non-blocking or short-timeout admission (`put_nowait` / short `wait_for`).
    3. Concurrent execution throttling via `asyncio.Semaphore(K)`.
    4. Fast-fail `LoadSheddingError` raising `429 / 503` responses instantly.
    '''

    def __init__(
        self,
        *,
        max_queue_size: int = 100,
        max_concurrent_workers: int = 10,
        admission_timeout_sec: float = 0.05,
    ) -> None:
        self.max_queue_size = max(1, max_queue_size)
        self.max_concurrent_workers = max(1, max_concurrent_workers)
        self.admission_timeout_sec = max(0.01, admission_timeout_sec)

        self._queue: asyncio.Queue[tuple[Callable[..., Awaitable[T]], tuple[Any, ...], dict[str, Any], asyncio.Future[T]]] = asyncio.Queue(maxsize=self.max_queue_size)
        self._semaphore = asyncio.Semaphore(self.max_concurrent_workers)
        self._workers_running = False
        self._worker_task: asyncio.Task[None] | None = None

    async def start_workers(self, loop: asyncio.AbstractEventLoop | None = None) -> None:
        '''Starts background worker consumers if not already running.'''
        if self._workers_running:
            return
        self._workers_running = True
        self._worker_task = asyncio.create_task(self._consumer_loop())

    async def stop_workers(self) -> None:
        self._workers_running = False
        if self._worker_task is not None:
            self._worker_task.cancel()
            try:
                await self._worker_task
            except asyncio.CancelledError:
                pass

    async def _consumer_loop(self) -> None:
        while self._workers_running:
            try:
                fn, args, kwargs, fut = await self._queue.get()
            except asyncio.CancelledError:
                break

            # Execute task under concurrency semaphore
            async def run_task() -> None:
                async with self._semaphore:
                    try:
                        res = await fn(*args, **kwargs)
                        if not fut.done():
                            fut.set_result(res)
                    except BaseException as exc:
                        if not fut.done():
                            fut.set_exception(exc)
                    finally:
                        self._queue.task_done()

            asyncio.create_task(run_task())

    async def submit(
        self,
        fn: Callable[..., Awaitable[T]],
        *args: Any,
        **kwargs: Any,
    ) -> T:
        '''
        Gate 2 & 3: Submits work into the bounded queue. If queue is full,
        sheds load instantly by raising `LoadSheddingError` within `admission_timeout_sec`.
        '''
        if not self._workers_running:
            await self.start_workers()

        fut: asyncio.Future[T] = asyncio.get_running_loop().create_future()
        item = (fn, args, kwargs, fut)

        try:
            # Attempt non-blocking or short-timeout admission
            self._queue.put_nowait(item)
        except asyncio.QueueFull:
            try:
                await asyncio.wait_for(
                    self._queue.put(item),
                    timeout=self.admission_timeout_sec,
                )
            except asyncio.TimeoutError:
                # Gate 4: Fast-fail load shedding
                raise LoadSheddingError(retry_after_sec=5.0)

        return await fut