orchestrata_ultimate.kernel.queue_manager

Reference › orchestrata_ultimate › kernel › queue_manager

The durable parking place for packets whose recipient is offline. QueueManager keeps one Redis list per recipient and priority, so a packet survives a restart and is delivered when the volk returns. It serves SPEC-001 story US-002 (REQ-001-US-002-021 to -030). It sits behind router, which marks packets for offline recipients as queued, and dispatcher, which parks undeliverable packets here. The Redis client is built lazily through core.redis_factory, with the host and port passed in or resolved from the environment. The postman queue manager is the separate queue layer of the broker.

enqueue first checks the total depth against max_queue_depth (10000 by default) and returns False when the queue is full (backpressure). Otherwise it serialises the packet with its queue time, priority, recipient and message id, pushes it to the list orchestrata:queue:<recipient>:p<priority>, sets the expiry from message_ttl_seconds (one day by default) and writes a metadata hash keyed by the packet id. dequeue probes the priority queues from CRITICAL down to LOWEST and pops the first message, using a blocking pop when a timeout is given; it updates the metadata, counts the message in flight and returns the rebuilt CXPPacket, or None when all queues are empty. record_delivery_success and record_delivery_failure move the in-flight counters. expire_stale_messages scans each queue, collects entries older than the TTL, removes them and moves them to the dead-letter list through move_to_dead_letter, which writes a reason and keeps the list for dead_letter_ttl_seconds (seven days by default). get_stats returns a QueueStats; inspect_queue lists messages for debugging.

Redis failures in these methods are logged and reported as False, None or 0 rather than raised.

At a glance

Class diagram

Dependency diagram

API

Classes

QueueStats (dataclass)

Decorators: traces(spec='001', story='US-002', reqs=['REQ-001-US-002-028']), dataclass · Line 50

Statistics for a message queue.

QueueManager

Decorators: traces(spec='001', story='US-002', reqs=['REQ-001-US-002-021', 'REQ-001-US-002-022', 'REQ-001-US-002-023', 'REQ-001-US-… · Line 98

Manages Redis-backed message queues for offline Volks.

Functions

Source

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

# mypy: ignore-errors
"""Redis-backed Message Queue Manager for Orchestration Kernel.

This module implements persistent message queuing for offline Volks
with priority support, dead-letter queue, and monitoring as specified in Stream 2A.

Requirements Implemented:
- REQ-001-US-002-021 to REQ-001-US-002-030: Queue management with Redis persistence
"""

import asyncio
import json
import logging
from collections.abc import Awaitable
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Any, cast

if TYPE_CHECKING:
    from types import TracebackType

import redis.asyncio as redis

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, DeliveryStatus, Priority
from orchestrata_ultimate.core.redis_factory import (
    resolve_redis_host,
    resolve_redis_port,
)

logger = logging.getLogger(__name__)


@traces(spec="001", story="US-002", reqs=["REQ-001-US-002-021"])
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``, ``hset``, ``llen``, ...) as
    ``Awaitable[T] | T`` because the same command mixin backs both the sync
    and async clients (see ``redis.typing.ResponseT``). This manager only
    ever holds an async client (``get_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-002", reqs=["REQ-001-US-002-028"])
@dataclass
class QueueStats:
    """Statistics for a message queue."""

    queue_name: str
    total_messages: int
    pending_messages: int
    in_flight_messages: int
    failed_messages: int
    dead_letter_messages: int
    average_latency_ms: float = 0.0
    delivery_success_rate: float = 0.0
    oldest_message_age_seconds: float | None = None
    last_delivery: datetime | None = None
    last_failure: datetime | None = None

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "queue_name": self.queue_name,
            "total_messages": self.total_messages,
            "pending_messages": self.pending_messages,
            "in_flight_messages": self.in_flight_messages,
            "failed_messages": self.failed_messages,
            "dead_letter_messages": self.dead_letter_messages,
            "average_latency_ms": self.average_latency_ms,
            "delivery_success_rate": self.delivery_success_rate,
            "oldest_message_age_seconds": self.oldest_message_age_seconds,
            "last_delivery": (
                self.last_delivery.isoformat() if self.last_delivery else None
            ),
            "last_failure": (
                self.last_failure.isoformat() if self.last_failure else None
            ),
        }


@traces(
    spec="001",
    story="US-002",
    reqs=[
        "REQ-001-US-002-021",
        "REQ-001-US-002-022",
        "REQ-001-US-002-023",
        "REQ-001-US-002-024",
        "REQ-001-US-002-027",
        "REQ-001-US-002-028",
    ],
)
class QueueManager:
    """Manages Redis-backed message queues for offline Volks.

    Provides durable queuing with priority support, backpressure,
    and dead-letter queue for failed messages.
    """

    def __init__(
        self,
        redis_host: str | None = None,
        redis_port: int | None = None,
        redis_db: int = 0,
        max_queue_depth: int = 10000,
        message_ttl_seconds: int = 86400,  # 24 hours
        dead_letter_ttl_seconds: int = 604800,  # 7 days
        socket_connect_timeout: float = 30.0,
        redis_client: redis.Redis | None = None,
    ) -> None:
        """Initialize queue manager.

        Args:
            redis_host: Redis server host
            redis_port: Redis server port
            redis_db: Redis database number
            max_queue_depth: Maximum messages per queue (backpressure threshold)
            message_ttl_seconds: Time-to-live for queued messages
            dead_letter_ttl_seconds: Time-to-live for dead-letter messages
            socket_connect_timeout: TCP connect timeout in seconds
            redis_client: Pre-built async client injected directly (the
                host/port/db/timeout args are ignored for connection
                purposes when provided). The seam for callers that need a
                non-default retry policy — e.g. error-path tests inject a
                fail-fast client because redis-py 7's default
                ``Retry(ExponentialWithJitterBackoff(), 3)`` turns each
                command against a dead address into a 4-13s backoff loop.
        """
        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.max_queue_depth = max_queue_depth
        self.message_ttl_seconds = message_ttl_seconds
        self.dead_letter_ttl_seconds = dead_letter_ttl_seconds
        self.socket_connect_timeout = socket_connect_timeout

        self._client: redis.Redis | None = redis_client
        self._stats_cache: dict[str, QueueStats] = {}
        self._in_flight: dict[str, int] = {}
        self._failed: dict[str, int] = {}
        self._lock = asyncio.Lock()

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-012"])
    async def get_client(self) -> redis.Redis:
        """Get or create async Redis client (lazy initialization)."""
        if self._client is None:
            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,
                socket_connect_timeout=self.socket_connect_timeout,
            )
        return self._client

    async def __aenter__(self) -> QueueManager:
        """Context manager entry."""
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> bool:
        """Context manager exit - close connection."""
        await self.aclose()
        return False

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

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-023"])
    def _queue_key(self, recipient: str, priority: int = 0) -> str:
        """Build Redis key for queue."""
        return f"orchestrata:queue:{recipient}:p{priority}"

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-027"])
    def _dlq_key(self, recipient: str) -> str:
        """Build Redis key for dead-letter queue."""
        return f"orchestrata:dlq:{recipient}"

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-037"])
    def _metadata_key(self, message_id: str) -> str:
        """Build Redis key for message metadata."""
        return f"orchestrata:msg_meta:{message_id}"

    @traces(
        spec="001",
        story="US-002",
        reqs=[
            "REQ-001-US-002-018",
            "REQ-001-US-002-021",
            "REQ-001-US-002-023",
            "REQ-001-US-002-030",
            "REQ-001-US-010-015",
        ],
    )
    async def enqueue(
        self, packet: CXPPacket, recipient: str, priority: int = 0
    ) -> bool:
        """Enqueue message for offline recipient.

        REQ-001-US-002-021: Queue messages when recipient unavailable
        REQ-001-US-002-022: Persist queue to Redis
        REQ-001-US-002-023: Support priority queuing
        REQ-001-US-002-024: Apply backpressure when queue full

        Args:
            packet: CXP packet to queue
            recipient: Recipient Volk alias
            priority: Message priority (0-5, higher = more urgent)

        Returns:
            True if message was queued, False if queue is full
        """
        # REQ-001-US-002-024: Check backpressure
        queue_depth = await self.get_queue_depth(recipient)
        if queue_depth >= self.max_queue_depth:
            logger.warning(
                f"Queue full for {recipient}: {queue_depth}/{self.max_queue_depth}"
            )
            return False

        # REQ-001-US-002-023: Use priority-based queue
        queue_key = self._queue_key(recipient, priority)

        # REQ-001-US-002-021: Mark as queued
        packet.delivery_status = DeliveryStatus.PENDING

        # Serialize packet with metadata
        message_data = {
            "packet": packet.to_dict(),
            "queued_at": datetime.now(UTC).isoformat(),
            "priority": priority,
            "recipient": recipient,
            "message_id": packet.header.packet_id,
        }

        try:
            # REQ-001-US-002-022: Persist to Redis (native async)
            client = await self.get_client()
            message_json = json.dumps(message_data)
            await _redis_result(client.rpush(queue_key, message_json))

            # Set TTL for message expiration (REQ-001-US-002-026)
            await client.expire(queue_key, self.message_ttl_seconds)

            # Store metadata for tracking
            metadata = {
                "recipient": recipient,
                "priority": str(priority),
                "queued_at": datetime.now(UTC).isoformat(),
                "status": "queued",
            }
            await _redis_result(
                client.hset(
                    self._metadata_key(packet.header.packet_id), mapping=metadata
                )
            )

            logger.info(
                f"Queued message {packet.header.packet_id} for {recipient} "
                f"(priority={priority}, depth={queue_depth + 1})"
            )

            return True

        except Exception as e:
            logger.error(
                f"Failed to enqueue message {packet.header.packet_id} for {recipient}: {e}",
                exc_info=True,
            )
            return False

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-021"])
    async def dequeue(self, recipient: str, timeout: float = 0) -> CXPPacket | None:
        """Dequeue message for recipient.

        REQ-001-US-002-021: Dequeue messages when recipient becomes available

        Args:
            recipient: Recipient Volk alias
            timeout: Blocking timeout in seconds (0 = non-blocking). Redis
                BLPOP has supported fractional-second timeouts since 6.0, so
                this accepts a float -- tests can shrink the real per-priority
                blocking wait (this method probes every priority queue from
                CRITICAL to LOWEST, so a naive timeout=1 costs up to 11
                REAL seconds against empty queues) without changing which
                branch runs.

        Returns:
            CXP packet or None if queue is empty
        """
        # Try priority queues from highest to lowest (REQ-001-US-002-023)
        try:
            client = await self.get_client()
            for priority in range(
                Priority.CRITICAL.value, Priority.LOWEST.value - 1, -1
            ):
                queue_key = self._queue_key(recipient, priority)

                # Native async Redis calls. blpop/lpop are typed by redis-py
                # as returning "list" resp. "str | list | None" (the same
                # command mixin backs count-based lpop variants), but with a
                # single key and no count this client always gets back
                # exactly a (key, value) tuple or None -- see redis-py's
                # BLPOP/BRPOP response callback (tuple(r) or None) and the
                # LPOP-without-count RESP reply (bulk string or nil).
                if timeout > 0:
                    result = await cast(
                        Awaitable[tuple[str, str] | None],
                        client.blpop(queue_key, timeout),
                    )
                else:
                    result = await cast(Awaitable[str | None], client.lpop(queue_key))

                if result:
                    # Handle blpop tuple result (queue_key, message) vs lpop string result
                    # blpop returns (key, value) tuple, lpop returns value directly
                    if isinstance(result, tuple):
                        # blpop always returns tuple (queue_key, message_json)
                        _, message_json = result
                    else:
                        # lpop returns message_json directly
                        message_json = result

                    message_data = json.loads(message_json)
                    packet = CXPPacket.from_dict(message_data["packet"])

                    # Update metadata (native async)
                    meta_key = self._metadata_key(packet.header.packet_id)
                    await _redis_result(client.hset(meta_key, "status", "dequeued"))
                    await _redis_result(
                        client.hset(
                            meta_key, "dequeued_at", datetime.now(UTC).isoformat()
                        )
                    )

                    # Track in-flight message
                    async with self._lock:
                        self._in_flight[recipient] = (
                            self._in_flight.get(recipient, 0) + 1
                        )

                    logger.info(
                        f"Dequeued message {packet.header.packet_id} for {recipient}"
                    )

                    return packet

            return None

        except Exception as e:
            logger.error(
                f"Failed to dequeue message for {recipient}: {e}", exc_info=True
            )
            return None

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-028"])
    async def record_delivery_success(self, recipient: str) -> None:
        """Record successful delivery, decrementing in-flight count.

        Args:
            recipient: Recipient Volk alias
        """
        async with self._lock:
            current = self._in_flight.get(recipient, 0)
            self._in_flight[recipient] = max(0, current - 1)

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-028"])
    async def record_delivery_failure(self, recipient: str) -> None:
        """Record failed delivery, moving from in-flight to failed.

        Args:
            recipient: Recipient Volk alias
        """
        async with self._lock:
            current = self._in_flight.get(recipient, 0)
            self._in_flight[recipient] = max(0, current - 1)
            self._failed[recipient] = self._failed.get(recipient, 0) + 1

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-025"])
    async def get_queue_depth(self, recipient: str) -> int:
        """Get total queue depth across all priorities.

        Args:
            recipient: Recipient Volk alias

        Returns:
            Total number of queued messages
        """
        try:
            client = await self.get_client()
            total = 0
            for priority in range(Priority.CRITICAL.value + 1):
                queue_key = self._queue_key(recipient, priority)
                # Native async llen call
                length = await _redis_result(client.llen(queue_key))
                total += length
            return total
        except Exception as e:
            logger.error(
                f"Failed to get queue depth for {recipient}: {e}", exc_info=True
            )
            return 0

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-027"])
    async def move_to_dead_letter(self, message_id: str, reason: str) -> bool:
        """Move failed message to dead-letter queue.

        REQ-001-US-002-027: Support dead-letter queue

        Args:
            message_id: Message ID to move
            reason: Failure reason

        Returns:
            True if message was moved to DLQ
        """
        try:
            # Get message metadata (native async)
            client = await self.get_client()
            meta_key = self._metadata_key(message_id)
            metadata = await _redis_result(client.hgetall(meta_key))

            if not metadata:
                logger.warning(f"No metadata found for message {message_id}")
                return False

            recipient = metadata.get("recipient")
            if not recipient:
                return False

            # Create DLQ entry
            dlq_entry = {
                "message_id": message_id,
                "recipient": recipient,
                "failed_at": datetime.now(UTC).isoformat(),
                "reason": reason,
                "metadata": metadata,
            }

            # Add to DLQ (native async)
            dlq_key = self._dlq_key(recipient)
            await _redis_result(client.rpush(dlq_key, json.dumps(dlq_entry)))

            # Set TTL for DLQ (native async)
            await client.expire(dlq_key, self.dead_letter_ttl_seconds)

            # Update metadata (native async)
            await _redis_result(client.hset(meta_key, "status", "dead_letter"))
            await _redis_result(
                client.hset(meta_key, "failed_at", datetime.now(UTC).isoformat())
            )
            await _redis_result(client.hset(meta_key, "failure_reason", reason))

            logger.info(f"Moved message {message_id} to DLQ: {reason}")

            return True

        except Exception as e:
            logger.error(
                f"Failed to move message {message_id} to DLQ: {e}", exc_info=True
            )
            return False

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-026"])
    async def expire_stale_messages(self, recipient: str) -> int:
        """Remove expired messages from queue.

        REQ-001-US-002-026: Expire stale messages

        Args:
            recipient: Recipient Volk alias

        Returns:
            Number of expired messages removed
        """
        try:
            client = await self.get_client()
            expired_count = 0
            cutoff_time = datetime.now(UTC) - timedelta(
                seconds=self.message_ttl_seconds
            )

            # Check all priority queues
            for priority in range(Priority.CRITICAL.value + 1):
                queue_key = self._queue_key(recipient, priority)
                # Native async llen call
                queue_length = await _redis_result(client.llen(queue_key))

                # Scan first, remove after. lrem() shifts later elements down,
                # so removing during the index walk skips the element that
                # shifts into the current position (it would miss ~half of any
                # run of consecutive expired messages). Collect, then remove.
                expired: list[tuple[str, dict[str, Any]]] = []
                for i in range(queue_length):
                    # Native async lindex call
                    message_json = await _redis_result(client.lindex(queue_key, i))
                    if not message_json:
                        continue

                    message_data = json.loads(message_json)
                    queued_at = datetime.fromisoformat(message_data["queued_at"])

                    if queued_at < cutoff_time:
                        expired.append((message_json, message_data))

                for message_json, message_data in expired:
                    # count=1 removes the first matching occurrence.
                    removed = await _redis_result(
                        client.lrem(queue_key, 1, message_json)
                    )
                    if not removed:
                        continue
                    expired_count += 1
                    # Move to DLQ
                    await self.move_to_dead_letter(
                        message_data["message_id"], "Message expired"
                    )

            if expired_count > 0:
                logger.info(f"Expired {expired_count} messages for {recipient}")

            return expired_count

        except Exception as e:
            logger.error(
                f"Failed to expire stale messages for {recipient}: {e}", exc_info=True
            )
            return 0

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-025", "REQ-001-US-002-028"],
    )
    async def get_stats(self, recipient: str) -> QueueStats:
        """Get queue statistics.

        REQ-001-US-002-025: Monitor queue depth and latency
        REQ-001-US-002-028: Provide queue statistics

        Args:
            recipient: Recipient Volk alias

        Returns:
            Queue statistics
        """
        try:
            client = await self.get_client()
            total_messages = await self.get_queue_depth(recipient)

            # Count DLQ messages (native async)
            dlq_key = self._dlq_key(recipient)
            dead_letter_count = await _redis_result(client.llen(dlq_key))

            # Calculate oldest message age
            oldest_age = None
            for priority in range(Priority.CRITICAL.value + 1):
                queue_key = self._queue_key(recipient, priority)
                # Native async lindex call
                oldest_json = await _redis_result(client.lindex(queue_key, 0))

                if oldest_json:
                    oldest_data = json.loads(oldest_json)
                    queued_at = datetime.fromisoformat(oldest_data["queued_at"])
                    age_seconds = (datetime.now(UTC) - queued_at).total_seconds()

                    if oldest_age is None or age_seconds > oldest_age:
                        oldest_age = age_seconds

            in_flight = self._in_flight.get(recipient, 0)
            failed = self._failed.get(recipient, 0)

            stats = QueueStats(
                queue_name=recipient,
                total_messages=total_messages,
                pending_messages=total_messages,
                in_flight_messages=in_flight,
                failed_messages=failed,
                dead_letter_messages=dead_letter_count,
                oldest_message_age_seconds=oldest_age,
            )

            # Cache stats
            self._stats_cache[recipient] = stats

            return stats

        except Exception as e:
            logger.error(f"Failed to get stats for {recipient}: {e}", exc_info=True)
            # Return empty stats on error
            return QueueStats(
                queue_name=recipient,
                total_messages=0,
                pending_messages=0,
                in_flight_messages=0,
                failed_messages=0,
                dead_letter_messages=0,
            )

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-029"])
    async def inspect_queue(
        self, recipient: str, limit: int = 10
    ) -> list[dict[str, Any]]:
        """Inspect queued messages for debugging.

        REQ-001-US-002-029: Support queue inspection

        Args:
            recipient: Recipient Volk alias
            limit: Maximum messages to return

        Returns:
            List of queued message summaries
        """
        try:
            client = await self.get_client()
            messages = []

            # Inspect all priority queues
            for priority in range(
                Priority.CRITICAL.value, Priority.LOWEST.value - 1, -1
            ):
                queue_key = self._queue_key(recipient, priority)
                # Native async llen call
                queue_length = await _redis_result(client.llen(queue_key))

                for i in range(min(queue_length, limit - len(messages))):
                    # Native async lindex call
                    message_json = await _redis_result(client.lindex(queue_key, i))
                    if not message_json:
                        continue

                    message_data = json.loads(message_json)

                    summary = {
                        "message_id": message_data["message_id"],
                        "priority": message_data["priority"],
                        "queued_at": message_data["queued_at"],
                        "sender": message_data["packet"]["header"]["sender"],
                        "packet_type": message_data["packet"]["header"]["type"],
                    }

                    messages.append(summary)

                    if len(messages) >= limit:
                        break

                if len(messages) >= limit:
                    break

            return messages

        except Exception as e:
            logger.error(f"Failed to inspect queue for {recipient}: {e}", exc_info=True)
            return []

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