orchestrata_ultimate.gateway.ring_buffer

Reference › orchestrata_ultimate › gateway › ring_buffer

ring_buffer keeps a bounded history of gateway events in Redis streams so that clients can replay what they missed (SPEC-014 US-008, REQ-014-125). RedisRingBuffer is built from config, which supplies the Redis connection and the RING_BUFFER_MAX_LEN cap. server owns it, rpc receives it, and the gateway handler and state handler read it for gateway.replay and state queries. See gateway.

connect opens an async Redis client through core.redis_factory. write flattens a dict with serialize_for_redis (dicts and lists become JSON, None becomes an empty string, other values use str) and appends it to a stream named with a fixed key prefix plus the stream name, with an exact maxlen cap, returning the entry id. read returns recent entries, and read_since returns entries from a millisecond timestamp, oldest first, each as a dict with an entry_id. A call before connect, or any Redis failure, raises RingBufferError. close releases the client.

At a glance

Class diagram

Dependency diagram

API

Classes

RedisRingBuffer

Decorators: traces(spec='014', story='US-008', reqs=['REQ-014-121']) · Line 53

Read/write ring buffer backed by Redis streams.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/ring_buffer.py

# mypy: ignore-errors
"""Redis ring buffer with read/write using streams."""

import json
import logging
from collections.abc import Awaitable
from typing import TYPE_CHECKING, Any, cast

import redis.asyncio as redis

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.redis_factory import create_async_redis
from orchestrata_ultimate.gateway.config import GatewayConfig
from orchestrata_ultimate.gateway.exceptions import (
    RingBufferError,
)

if TYPE_CHECKING:
    from redis.typing import EncodableT, FieldT

logger = logging.getLogger(__name__)

STREAM_KEY_PREFIX = "gw:ringbuf:"


@traces(spec="014", story="US-008", reqs=["REQ-014-125"])
def serialize_for_redis(
    data: dict[str, Any],
) -> dict[str, str]:
    """Flatten *data* values to strings for Redis hashes.

    Dicts and lists become JSON strings; None becomes
    the empty string; everything else uses str().

    Args:
        data: Dictionary to serialize.

    Returns:
        Dictionary with all values as strings.
    """
    result: dict[str, str] = {}
    for key, value in data.items():
        if isinstance(value, (dict, list)):
            result[key] = json.dumps(value)
        elif value is None:
            result[key] = ""
        else:
            result[key] = str(value)
    return result


@traces(spec="014", story="US-008", reqs=["REQ-014-121"])
class RedisRingBuffer:
    """Read/write ring buffer backed by Redis streams."""

    def __init__(self, config: GatewayConfig) -> None:
        """Initialise from gateway configuration."""
        self._config = config
        self._max_len = config.ring_buffer_max_len
        self._client: redis.Redis | None = None

    async def connect(self) -> None:
        """Open connection to Redis and verify with PING."""
        self._client = create_async_redis(
            host=self._config.redis_host,
            port=self._config.redis_port,
            db=self._config.redis_db,
            decode_responses=True,
        )
        # redis-py's command mixins are shared between the sync and async
        # clients, so `ping()` is stub-typed `Awaitable[bool] | bool`;
        # `self._client` here is always the async client.
        await cast("Awaitable[bool]", self._client.ping())
        logger.info("Ring buffer connected to Redis")

    @traces(spec="014", story="US-008", reqs=["REQ-014-125"])
    async def write(
        self,
        stream_name: str,
        data: dict[str, Any],
    ) -> str:
        """Append *data* to the capped stream.

        Args:
            stream_name: Redis stream name suffix.
            data: Key-value pairs to write.

        Returns:
            Redis stream entry ID string.

        Raises:
            RingBufferError: If not connected or write fails.
        """
        if not self._client:
            raise RingBufferError("Not connected to Redis")
        key = f"{STREAM_KEY_PREFIX}{stream_name}"
        serialized = serialize_for_redis(data)
        try:
            # `dict[str, str]` isn't assignable to `Dict[FieldT, EncodableT]`
            # despite `str` being a member of both unions: `Dict`'s type
            # parameters are invariant, so pyright requires an exact type
            # match rather than accepting a narrower concrete dict.
            entry_id: str = await self._client.xadd(
                key,
                cast("dict[FieldT, EncodableT]", serialized),
                maxlen=self._max_len,
                approximate=False,
            )
            return entry_id
        except Exception as exc:
            raise RingBufferError(f"Stream write failed: {exc}") from exc

    @traces(spec="014", story="US-008", reqs=["REQ-014-127", "REQ-014-128"])
    async def read(
        self,
        stream_name: str,
        count: int = 100,
    ) -> list[dict[str, Any]]:
        """Read the last *count* events from a stream.

        Args:
            stream_name: Redis stream name suffix.
            count: Maximum number of entries to return.

        Returns:
            List of event dictionaries, oldest first.

        Raises:
            RingBufferError: If not connected or read fails.
        """
        if not self._client:
            raise RingBufferError("Not connected to Redis")
        key = f"{STREAM_KEY_PREFIX}{stream_name}"
        try:
            entries = await self._client.xrevrange(key, count=count)
            result: list[dict[str, Any]] = []
            for entry_id, fields in reversed(entries):
                event: dict[str, Any] = {
                    "entry_id": entry_id,
                }
                event.update(fields)
                result.append(event)
            return result
        except Exception as exc:
            raise RingBufferError(f"Stream read failed: {exc}") from exc

    async def read_since(
        self,
        stream_name: str,
        timestamp_ms: int,
    ) -> list[dict[str, Any]]:
        """Read events since a Unix timestamp (milliseconds).

        Args:
            stream_name: Redis stream name suffix.
            timestamp_ms: Unix timestamp in ms to start from.

        Returns:
            List of event dictionaries, oldest first.

        Raises:
            RingBufferError: If not connected or read fails.
        """
        if not self._client:
            raise RingBufferError("Not connected to Redis")
        key = f"{STREAM_KEY_PREFIX}{stream_name}"
        min_id = f"{timestamp_ms}-0"
        try:
            entries = await self._client.xrange(key, min=min_id)
            result: list[dict[str, Any]] = []
            for entry_id, fields in entries:
                event: dict[str, Any] = {
                    "entry_id": entry_id,
                }
                event.update(fields)
                result.append(event)
            return result
        except Exception as exc:
            raise RingBufferError(f"Stream read failed: {exc}") from exc

    @property
    def client(self) -> redis.Redis | None:
        """Expose the Redis client for direct operations."""
        return self._client

    async def close(self) -> None:
        """Close the Redis connection."""
        if self._client:
            await self._client.aclose()
            self._client = None
            logger.info("Ring buffer disconnected")