orchestrata_ultimate.kernel.audit.queue

Reference › orchestrata_ultimate › kernel › audit › queue

A Redis-backed queue that makes audit logging fire-and-forget (REQ-001-US-008-018). RedisAuditQueue lets the logger hand an AuditEntry (see model) to Redis immediately, while a background worker writes it to the file and graph writers later.

connect opens the async Redis client. enqueue serialises the entry to JSON, pushes it to the right of a Redis list and refreshes the list's expiry; it returns False when the client is not connected or Redis fails, so the caller can fall back to its in-memory buffer. dequeue does a blocking left pop with a timeout (1 second by default) and rebuilds the entry; it returns None when the queue is empty or an error occurs, logging the error. The host and port come from the shared resolvers unless passed in, the database defaults to 15, and the list key is made unique per process by appending the process id and a random suffix unless a key is supplied. close closes the client.

At a glance

Class diagram

Dependency diagram

API

Classes

RedisAuditQueue

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

Redis-backed audit queue for durable fire-and-forget logging.

Functions

Source

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

# mypy: ignore-errors
"""Redis-backed audit queue for durable fire-and-forget logging."""

import json
import os
import uuid
from collections.abc import Awaitable
from typing import cast

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.redis_factory import (
    resolve_redis_host,
    resolve_redis_port,
)
from orchestrata_ultimate.kernel.audit._common import (
    REDIS_AVAILABLE,
    _logger,
    redis,
)
from orchestrata_ultimate.kernel.audit.model import AuditEntry


@traces(spec="001", story="US-008", reqs=["REQ-001-US-008-018"])
def _redis_result[T](value: Awaitable[T] | T) -> Awaitable[T]:
    """Narrow a redis-py command result back to its true async type.

    redis-py types command methods (``rpush``, ``blpop``, ...) as
    ``Awaitable[T] | T`` because the same command mixin backs both the sync
    and async clients (see ``redis.typing.ResponseT``). ``RedisAuditQueue``
    only ever holds an async client, which always returns an awaitable at
    runtime -- this documents that fact for pyright instead of suppressing
    the check.
    """
    return cast(Awaitable[T], value)


@traces(spec="001", story="US-008", reqs=["REQ-001-US-008-018"])
class RedisAuditQueue:
    """Redis-backed audit queue for durable fire-and-forget logging.

    The queue is PER-INSTANCE by default, not one global key. ``dequeue`` is a
    destructive ``BLPOP``, so every ``AuditLogger`` sharing a key is a
    COMPETING CONSUMER of every other one's entries: with the fixed
    ``orchestrata:audit:queue``, the long-running kernel service that
    ``start_test_services.py`` boots for every suite run silently swallowed
    the audit entries produced by unrelated processes and wrote them into ITS
    OWN graph (measured live 2026-07-26 — an entry pushed by a standalone
    script vanished in under 0.5s and never appeared in any database, because
    that kernel still held pre-cutover connection config). Any process
    asserting on its own audit trail was racing a thief it never knew about.

    Pass an explicit ``queue_key`` to opt INTO a shared queue (a real
    producer/consumer fan-out, where a single drain is the point).
    """

    QUEUE_KEY = "orchestrata:audit:queue"
    KEY_TTL_S = 7 * 24 * 3600

    def __init__(
        self,
        redis_host: str | None = None,
        redis_port: int | None = None,
        redis_db: int = 15,
        queue_key: str | None = None,
        socket_connect_timeout: int = 30,
    ) -> None:
        self.redis_host = redis_host if redis_host is not None else resolve_redis_host()
        self.redis_port = redis_port if redis_port is not None else resolve_redis_port()
        self.redis_db = redis_db
        self.socket_connect_timeout = socket_connect_timeout
        self._client: redis.Redis | None = None
        # Per-instance by default (see the class docstring); an explicit
        # queue_key opts into a deliberately shared queue.
        self.QUEUE_KEY = (
            queue_key
            if queue_key is not None
            else f"{self.QUEUE_KEY}:{os.getpid()}:{uuid.uuid4().hex[:12]}"
        )

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-018"])
    async def connect(self) -> None:
        """Connect to Redis."""
        if not REDIS_AVAILABLE:
            _logger.warning("Redis not available for audit queue")
            return

        try:
            from orchestrata_ultimate.core.redis_factory import create_async_redis

            self._client = create_async_redis(
                host=self.redis_host,
                port=self.redis_port,
                db=self.redis_db,
                decode_responses=False,
            )
            await _redis_result(self._client.ping())
            _logger.info(
                f"Connected to Redis audit queue at {self.redis_host}:{self.redis_port}"
            )
        except Exception as e:
            _logger.error(f"Failed to connect to Redis: {e}")
            self._client = None

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-018"])
    async def enqueue(self, entry: AuditEntry) -> bool:
        """Enqueue an audit entry. Returns True on success."""
        if self._client is None:
            return False

        try:
            data = json.dumps(entry.to_dict())
            await _redis_result(self._client.rpush(self.QUEUE_KEY, data))
            # Per-instance keys are abandoned when their process dies, so the
            # queue carries a rolling TTL: an undrained one disappears instead
            # of accumulating in Redis forever. Refreshed on every push, so a
            # live queue never expires under its own consumer.
            await _redis_result(self._client.expire(self.QUEUE_KEY, self.KEY_TTL_S))
            return True
        except Exception as e:
            _logger.error(f"Failed to enqueue audit entry: {e}")
            return False

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-018"])
    async def dequeue(self, timeout: float = 1.0) -> AuditEntry | None:
        """Dequeue an audit entry with timeout. Returns None if queue is empty."""
        if self._client is None:
            return None

        try:
            result = await _redis_result(
                self._client.blpop(self.QUEUE_KEY, timeout=timeout)
            )
            if result is None:
                return None

            _, data = result
            entry_dict = json.loads(data)
            return AuditEntry.from_dict(entry_dict)
        except Exception as e:
            _logger.error(f"Failed to dequeue audit entry: {e}")
            return None

    async def close(self) -> None:
        """Close Redis connection."""
        if self._client:
            await self._client.aclose()
            self._client = None