orchestrata_ultimate.kernel.performance.resources

Reference › orchestrata_ultimate › kernel › performance › resources

The resource-management building blocks of the performance package, SPEC-001 story US-010. manager uses LRUCache and BackpressureController directly; the others are available for callers that need pooling, batching or bounded buffers. Defaults come from _common.

• LRUCache is a bounded least-recently-used cache with a per-entry time-to-live: get, put (with an optional time-to-live), invalidate, clear and get_stats.
• ConnectionPool pools async connections made by a factory between a minimum and maximum size. acquire takes a pooled connection or creates one, or waits up to a timeout; release returns it; close closes all, using the structural protocols _AsyncCloseable (aclose) or _SyncCloseable (close).
• BatchProcessor collects items and passes them to a processor callable when the batch reaches a maximum size or a timed flush fires; flush forces it.
• RingBuffer is a fixed-capacity buffer with push (which returns the evicted item when full), pop, peek and to_list.
• BackpressureController turns a load between 0 and 1 into a state (normal, warning, critical, shedding) using thresholds, with a recovery threshold back to normal, and calls registered callbacks on a change. should_accept(priority) accepts everything in normal and warning states, only priority 5 or higher when critical, and only priority 8 or higher when shedding, counting each shed request while shedding. It is traced to REQ-001-US-010-020 and the graceful-degradation requirement.

At a glance

Class diagram

Dependency diagram

API

Classes

_AsyncCloseable (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 39

Structural type for pooled connections exposing an aclose().

_SyncCloseable (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 58

Structural type for pooled connections exposing a close().

LRUCache

Decorators: traces(spec='001', story='US-010', reqs=['REQ-001-US-010-011']) · Line 66

LRU cache for query results (REQ-001-US-010-011).

ConnectionPool

Decorators: traces(spec='001', story='US-010', reqs=['REQ-001-US-010-012', 'REQ-001-US-010-013']) · Line 186

Generic connection pool (REQ-001-US-010-012, REQ-001-US-010-013).

BatchProcessor

Decorators: traces(spec='001', story='US-010', reqs=['REQ-001-US-010-014', 'REQ-001-US-010-018']) · Line 365

Batch processor for message dispatches (REQ-001-US-010-014).

RingBuffer

Decorators: traces(spec='001', story='US-010', reqs=['REQ-001-US-010-017']) · Line 450

Fixed-size ring buffer (REQ-001-US-010-017).

BackpressureController

Decorators: traces(spec='001', story='US-002', reqs=['REQ-001-US-002-024', 'REQ-001-US-007-005', 'REQ-001-US-010-010', 'REQ-001-US-… · Line 550

Backpressure mechanism (REQ-001-US-010-020).

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/kernel/performance/resources.py

# mypy: ignore-errors
"""Performance subsystem: resources."""

import asyncio
import collections
import inspect
import threading
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from enum import Enum
from typing import Any, Generic, Protocol, TypeVar, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.kernel.performance._common import (
    BACKPRESSURE_CRITICAL_THRESHOLD,
    BACKPRESSURE_SHEDDING_THRESHOLD,
    BACKPRESSURE_WARNING_THRESHOLD,
    BATCH_PROCESSOR_DEFAULT_MAX_SIZE,
    BATCH_PROCESSOR_DEFAULT_MAX_WAIT_MS,
    CACHE_DEFAULT_MAX_SIZE,
    CACHE_DEFAULT_TTL_SECONDS,
    CONNECTION_POOL_DEFAULT_MAX_IDLE_SECONDS,
    CONNECTION_POOL_DEFAULT_MAX_SIZE,
    T,
    _logger,
)

# Dedicated TypeVar for LRUCache.CacheEntry: reusing the outer PEP-695 `T`
# (via `Generic[T]`) makes pyright treat `LRUCache.CacheEntry` as already
# fully specialized by the enclosing class, so `CacheEntry[T]` at the usage
# site below errors as "already specialized" (reportInvalidTypeArguments).
# A distinctly-named TypeVar keeps CacheEntry genuinely generic in its own
# right while still resolving to the enclosing `T@LRUCache` at each use.
_CacheEntryT = TypeVar("_CacheEntryT")


@runtime_checkable
class _AsyncCloseable(Protocol):
    """Structural type for pooled connections exposing an ``aclose()``.

    ``ConnectionPool[T]`` is generic over an unbounded ``T`` (any pooled
    resource), so pyright can't narrow ``hasattr(conn, "aclose")`` the way
    the original duck-typed code relied on -- ``conn`` stays typed as
    ``object*`` (reportAttributeAccessIssue). A ``runtime_checkable``
    Protocol gives ``isinstance()`` the same structural check with real
    narrowing. The method is intentionally NOT declared ``async def``: some
    pooled resources expose a synchronous ``aclose``/``close``, and this
    class only needs to assert the attribute exists, not its call
    convention (that is still resolved via ``inspect.iscoroutinefunction``
    below).
    """

    def aclose(self) -> Any: ...


@runtime_checkable
class _SyncCloseable(Protocol):
    """Structural type for pooled connections exposing a ``close()``. See
    ``_AsyncCloseable`` for why this exists."""

    def close(self) -> Any: ...


@traces(spec="001", story="US-010", reqs=["REQ-001-US-010-011"])
class LRUCache[T]:
    """LRU cache for query results (REQ-001-US-010-011).

    Thread-safe implementation with TTL support.
    """

    @dataclass
    class CacheEntry(Generic[_CacheEntryT]):
        """Cache entry with metadata."""

        value: _CacheEntryT
        created_at: float
        ttl_seconds: float
        hits: int = 0

    def __init__(
        self,
        max_size: int = CACHE_DEFAULT_MAX_SIZE,
        default_ttl: float = CACHE_DEFAULT_TTL_SECONDS,
    ) -> None:
        """Initialize cache.

        Args:
            max_size: Maximum cache entries
            default_ttl: Default TTL in seconds
        """
        self._max_size = max_size
        self._default_ttl = default_ttl
        self._cache: collections.OrderedDict[str, LRUCache.CacheEntry[T]] = (
            collections.OrderedDict()
        )
        self._lock = threading.Lock()
        self._hits = 0
        self._misses = 0

    def get(self, key: str) -> T | None:
        """Get value from cache.

        Args:
            key: Cache key

        Returns:
            Cached value or None if not found/expired
        """
        with self._lock:
            entry = self._cache.get(key)
            if entry is None:
                self._misses += 1
                return None

            # Check TTL
            if time.monotonic() - entry.created_at > entry.ttl_seconds:
                del self._cache[key]
                self._misses += 1
                return None

            # Move to end (most recently used)
            self._cache.move_to_end(key)
            entry.hits += 1
            self._hits += 1
            return entry.value

    def put(self, key: str, value: T, ttl: float | None = None) -> None:
        """Put value in cache.

        Args:
            key: Cache key
            value: Value to cache
            ttl: Optional TTL override
        """
        with self._lock:
            # Remove if exists
            if key in self._cache:
                del self._cache[key]

            # Evict oldest if at capacity
            while len(self._cache) >= self._max_size:
                self._cache.popitem(last=False)

            self._cache[key] = LRUCache.CacheEntry(
                value=value,
                created_at=time.monotonic(),
                ttl_seconds=ttl or self._default_ttl,
            )

    def invalidate(self, key: str) -> bool:
        """Invalidate cache entry.

        Args:
            key: Cache key

        Returns:
            True if entry was found and removed
        """
        with self._lock:
            if key in self._cache:
                del self._cache[key]
                return True
            return False

    def clear(self) -> None:
        """Clear entire cache."""
        with self._lock:
            self._cache.clear()

    def get_stats(self) -> dict[str, Any]:
        """Get cache statistics."""
        with self._lock:
            total = self._hits + self._misses
            hit_rate = (self._hits / total * 100) if total > 0 else 0
            return {
                "size": len(self._cache),
                "max_size": self._max_size,
                "hits": self._hits,
                "misses": self._misses,
                "hit_rate_percent": hit_rate,
            }


@traces(spec="001", story="US-010", reqs=["REQ-001-US-010-012", "REQ-001-US-010-013"])
class ConnectionPool[T]:
    """Generic connection pool (REQ-001-US-010-012, REQ-001-US-010-013).

    Provides async connection pooling with health checks.
    """

    def __init__(
        self,
        factory: Callable[[], Awaitable[T]],
        max_size: int = CONNECTION_POOL_DEFAULT_MAX_SIZE,
        min_size: int = 2,
        max_idle_seconds: float = CONNECTION_POOL_DEFAULT_MAX_IDLE_SECONDS,
        health_check: Callable[[T], Awaitable[bool]] | None = None,
    ) -> None:
        """Initialize connection pool.

        Args:
            factory: Async function to create new connections
            max_size: Maximum pool size
            min_size: Minimum connections to maintain
            max_idle_seconds: Max idle time before closing connection
            health_check: Optional async function to check connection health
        """
        self._factory = factory
        self._max_size = max_size
        self._min_size = min_size
        self._max_idle_seconds = max_idle_seconds
        self._health_check = health_check

        self._pool: asyncio.Queue[tuple[T, float]] = asyncio.Queue(maxsize=max_size)
        self._size = 0
        self._lock = asyncio.Lock()
        self._closed = False

        self._acquired_count = 0
        self._released_count = 0
        self._created_count = 0
        self._failed_count = 0

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-015"])
    async def acquire(self, timeout: float | None = None) -> T:
        """Acquire connection from pool (REQ-001-US-010-015 - async I/O).

        Args:
            timeout: Optional timeout in seconds

        Returns:
            Connection instance

        Raises:
            RuntimeError: If pool is closed
            asyncio.TimeoutError: If timeout exceeded
        """
        if self._closed:
            raise RuntimeError("Connection pool is closed")

        conn = await self._try_acquire_pooled()
        if conn is not None:
            return conn

        return await self._acquire_new_or_wait(timeout)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-015"])
    async def _try_acquire_pooled(self) -> T | None:
        """Try to get a healthy, non-stale connection from the pool without blocking.

        Returns:
            A ready connection, or None if none available.
        """
        try:
            conn, acquired_at = self._pool.get_nowait()
        except asyncio.QueueEmpty:
            return None

        if time.monotonic() - acquired_at > self._max_idle_seconds:
            self._size -= 1
            return None

        if self._health_check:
            try:
                if await self._health_check(conn):
                    self._acquired_count += 1
                    return conn
            except Exception as exc:
                _logger.warning(
                    "Connection health check failed: %s", exc, exc_info=True
                )
            self._size -= 1
            return None

        self._acquired_count += 1
        return conn

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-015"])
    async def _acquire_new_or_wait(self, timeout: float | None) -> T:
        """Create a new connection if under limit, otherwise wait for one.

        Args:
            timeout: Optional timeout in seconds

        Returns:
            Connection instance

        Raises:
            asyncio.TimeoutError: If timeout exceeded while waiting
        """
        async with self._lock:
            if self._size < self._max_size:
                try:
                    conn = await asyncio.wait_for(self._factory(), timeout=timeout)
                    self._size += 1
                    self._created_count += 1
                    self._acquired_count += 1
                    return conn
                except Exception as exc:
                    self._failed_count += 1
                    _logger.warning(
                        "Failed to create new pool connection: %s", exc, exc_info=True
                    )
                    raise

        try:
            conn, _ = await asyncio.wait_for(self._pool.get(), timeout=timeout)
            self._acquired_count += 1
            return conn
        except TimeoutError:
            self._failed_count += 1
            raise

    async def release(self, conn: T) -> None:
        """Release connection back to pool.

        Args:
            conn: Connection to release
        """
        if self._closed:
            return

        try:
            self._pool.put_nowait((conn, time.monotonic()))
            self._released_count += 1
        except asyncio.QueueFull:
            # Pool full, discard connection
            self._size -= 1

    async def close(self) -> None:
        """Close all connections in pool."""
        self._closed = True
        while not self._pool.empty():
            try:
                conn, _ = self._pool.get_nowait()
                # If connection has close method, call it
                if isinstance(conn, _AsyncCloseable):
                    if inspect.iscoroutinefunction(conn.aclose):
                        await conn.aclose()
                    else:
                        conn.aclose()
                elif isinstance(conn, _SyncCloseable):
                    if inspect.iscoroutinefunction(conn.close):
                        await conn.close()
                    else:
                        conn.close()
            except asyncio.QueueEmpty:  # pragma: no cover
                break  # pragma: no cover

    def get_stats(self) -> dict[str, Any]:
        """Get pool statistics."""
        return {
            "size": self._size,
            "max_size": self._max_size,
            "available": self._pool.qsize(),
            "acquired": self._acquired_count,
            "released": self._released_count,
            "created": self._created_count,
            "failed": self._failed_count,
        }


@traces(spec="001", story="US-010", reqs=["REQ-001-US-010-014", "REQ-001-US-010-018"])
class BatchProcessor[T]:
    """Batch processor for message dispatches (REQ-001-US-010-014).

    Accumulates items and processes in batches for efficiency.
    """

    def __init__(
        self,
        processor: Callable[[list[T]], Awaitable[None]],
        max_batch_size: int = BATCH_PROCESSOR_DEFAULT_MAX_SIZE,
        max_wait_ms: float = BATCH_PROCESSOR_DEFAULT_MAX_WAIT_MS,
    ) -> None:
        """Initialize batch processor.

        Args:
            processor: Async function to process a batch
            max_batch_size: Maximum items per batch
            max_wait_ms: Maximum wait time before flushing
        """
        self._processor = processor
        self._max_batch_size = max_batch_size
        self._max_wait_seconds = max_wait_ms / 1000.0

        self._buffer: list[T] = []
        self._lock = asyncio.Lock()
        self._last_flush = time.monotonic()
        self._flush_task: asyncio.Task | None = None

        self._batch_count = 0
        self._item_count = 0

    async def add(self, item: T) -> None:
        """Add item to batch.

        Args:
            item: Item to add
        """
        async with self._lock:
            self._buffer.append(item)
            self._item_count += 1

            # Flush if batch is full
            if len(self._buffer) >= self._max_batch_size:
                await self._flush_locked()
            elif self._flush_task is None or self._flush_task.done():
                # Schedule timed flush
                self._flush_task = asyncio.create_task(self._timed_flush())

    async def _timed_flush(self) -> None:
        """Flush after max wait time."""
        await asyncio.sleep(self._max_wait_seconds)
        async with self._lock:
            if self._buffer:
                await self._flush_locked()

    async def _flush_locked(self) -> None:
        """Flush buffer (must hold lock)."""
        if not self._buffer:
            return

        batch = self._buffer
        self._buffer = []
        self._last_flush = time.monotonic()
        self._batch_count += 1

        await self._processor(batch)

    async def flush(self) -> None:
        """Force flush any pending items."""
        async with self._lock:
            await self._flush_locked()

    def get_stats(self) -> dict[str, Any]:
        """Get processor statistics."""
        return {
            "pending": len(self._buffer),
            "batch_count": self._batch_count,
            "item_count": self._item_count,
            "avg_batch_size": (
                self._item_count / self._batch_count if self._batch_count > 0 else 0
            ),
        }


@traces(spec="001", story="US-010", reqs=["REQ-001-US-010-017"])
class RingBuffer[T]:
    """Fixed-size ring buffer (REQ-001-US-010-017).

    Efficient data structure for bounded collections.
    """

    def __init__(self, capacity: int) -> None:
        """Initialize ring buffer.

        Args:
            capacity: Maximum number of items
        """
        self._capacity = capacity
        self._buffer: list[T | None] = [None] * capacity
        self._head = 0  # Next write position
        self._tail = 0  # Next read position
        self._size = 0
        self._lock = threading.Lock()

    def push(self, item: T) -> T | None:
        """Push item to buffer.

        Args:
            item: Item to push

        Returns:
            Evicted item if buffer was full, None otherwise
        """
        with self._lock:
            evicted = None
            if self._size == self._capacity:
                evicted = self._buffer[self._tail]
                self._tail = (self._tail + 1) % self._capacity
            else:
                self._size += 1

            self._buffer[self._head] = item
            self._head = (self._head + 1) % self._capacity
            return evicted

    def pop(self) -> T | None:
        """Pop oldest item from buffer.

        Returns:
            Oldest item or None if empty
        """
        with self._lock:
            if self._size == 0:
                return None

            item = self._buffer[self._tail]
            self._buffer[self._tail] = None
            self._tail = (self._tail + 1) % self._capacity
            self._size -= 1
            return item

    def peek(self) -> T | None:
        """Peek at oldest item without removing.

        Returns:
            Oldest item or None if empty
        """
        with self._lock:
            if self._size == 0:
                return None
            return self._buffer[self._tail]

    def __len__(self) -> int:
        """Get current size."""
        return self._size

    def is_empty(self) -> bool:
        """Check if buffer is empty."""
        return self._size == 0

    def is_full(self) -> bool:
        """Check if buffer is full."""
        return self._size == self._capacity

    def to_list(self) -> list[T]:
        """Convert to list (oldest first)."""
        with self._lock:
            result = []
            idx = self._tail
            for _ in range(self._size):
                result.append(self._buffer[idx])
                idx = (idx + 1) % self._capacity
            return result


@traces(
    spec="001",
    story="US-002",
    reqs=[
        "REQ-001-US-002-024",
        "REQ-001-US-007-005",
        "REQ-001-US-010-010",
        "REQ-001-US-010-020",
    ],
)
class BackpressureController:
    """Backpressure mechanism (REQ-001-US-010-020).

    Implements load shedding and flow control.
    """

    class State(Enum):
        """Backpressure state."""

        NORMAL = "normal"
        WARNING = "warning"
        CRITICAL = "critical"
        SHEDDING = "shedding"

    def __init__(
        self,
        warning_threshold: float = BACKPRESSURE_WARNING_THRESHOLD,
        critical_threshold: float = BACKPRESSURE_CRITICAL_THRESHOLD,
        shedding_threshold: float = BACKPRESSURE_SHEDDING_THRESHOLD,
        recovery_threshold: float = 0.5,
    ) -> None:
        """Initialize backpressure controller.

        Args:
            warning_threshold: Load level for warning state (0-1)
            critical_threshold: Load level for critical state (0-1)
            shedding_threshold: Load level to start shedding (0-1)
            recovery_threshold: Load level to recover to normal (0-1)
        """
        self._warning_threshold = warning_threshold
        self._critical_threshold = critical_threshold
        self._shedding_threshold = shedding_threshold
        self._recovery_threshold = recovery_threshold

        self._current_load = 0.0
        self._state = BackpressureController.State.NORMAL
        self._shed_count = 0
        self._lock = threading.Lock()

        self._state_callbacks: list[Callable[[BackpressureController.State], None]] = []

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-020"])
    def update_load(self, load: float) -> None:
        """Update current load level.

        Args:
            load: Current load (0-1)
        """
        with self._lock:
            self._current_load = max(0.0, min(1.0, load))
            old_state = self._state

            if self._current_load >= self._shedding_threshold:
                self._state = BackpressureController.State.SHEDDING
            elif self._current_load >= self._critical_threshold:
                self._state = BackpressureController.State.CRITICAL
            elif self._current_load >= self._warning_threshold:
                self._state = BackpressureController.State.WARNING
            elif self._current_load <= self._recovery_threshold:
                self._state = BackpressureController.State.NORMAL

            if old_state != self._state:
                for callback in self._state_callbacks:
                    try:
                        callback(self._state)
                    except Exception as exc:
                        _logger.warning(
                            "Backpressure state callback failed: %s", exc, exc_info=True
                        )

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-024"])
    def should_accept(self, priority: int = 0) -> bool:
        """Check if request is accepted (REQ-001-US-010-010 - graceful degradation).

        Args:
            priority: Request priority (higher = more important)

        Returns:
            True if request is accepted
        """
        with self._lock:
            if self._state == BackpressureController.State.NORMAL:
                return True
            elif self._state == BackpressureController.State.WARNING:
                return True  # Accept all in warning
            elif self._state == BackpressureController.State.CRITICAL:
                return priority >= 5  # Only high priority
            else:  # SHEDDING
                self._shed_count += 1
                return priority >= 8  # Only critical

    def get_state(self) -> BackpressureController.State:
        """Get current state."""
        return self._state

    def get_load(self) -> float:
        """Get current load."""
        return self._current_load

    def get_stats(self) -> dict[str, Any]:
        """Get controller statistics."""
        return {
            "state": self._state.value,
            "load": self._current_load,
            "shed_count": self._shed_count,
        }

    def on_state_change(
        self, callback: Callable[[BackpressureController.State], None]
    ) -> None:
        """Register state change callback.

        Args:
            callback: Function to call on state change
        """
        self._state_callbacks.append(callback)