orchestrata_ultimate.postman.queue

Reference › orchestrata_ultimate › postman › queue

The queue backends of the broker, SPEC-002 story FR-004 (REQ-002-FR4-001 to -015) with the Redis queue traced to SPEC-001 REQ-001-US-002-022. Every class implements IQueue from interfaces and holds QueueEntry values from types, so the caller can swap them. queue_manager is the importer: it builds queues from a QueueConfig. Errors are QueueError and QueueFullError from exceptions. The kernel has its own, separate Redis queue for offline volks in kernel queue_manager.

• BaseQueue is the abstract base: a lock and a deque of entries, with enqueue, dequeue, peek, size, clear and close implemented and hooks _on_enqueue, _on_dequeue and _on_clear; checkpoint and recover are left abstract.
• InMemoryQueue is a volatile queue whose checkpoint and recover do nothing.
• DiskBackedQueue survives restarts: its hooks write a JSON file after changes, checkpoint saves it, and recover reloads the entries and raises QueueError on an unreadable file.
• BoundedQueue wraps another queue with a maximum size (10000 by default) and an overflow policy: reject raises QueueFullError, and drop_oldest removes the head first. An invalid policy raises ValueError.
• PriorityQueue inserts each entry before the first entry with a larger priority number, keeping first-in-first-out order within one priority. A heap-based version with starvation control is in priority.
• RedisQueue keeps entries in the Redis list queue:<name>. The constructor resolves the host and port with the shared resolvers, pings Redis and raises QueueError if it cannot connect; enqueue and dequeue run the blocking client in a worker thread and wrap failures in QueueError.

At a glance

Class diagram

Dependency diagram

API

Classes

BaseQueue (abstract)

Bases: IQueue, ABC · Decorators: traces(spec='002', story='FR-004', reqs=['REQ-002-FR4-005', 'REQ-002-FR4-006', 'REQ-002-FR4-007', 'REQ-002-FR4-008', 'R… · Line 44

Base queue protocol with common functionality.

InMemoryQueue

Bases: BaseQueue · Decorators: traces(spec='002', story='FR-004', reqs=['REQ-002-FR4-001']) · Line 184

Fast, volatile queue backed by in-memory list.

DiskBackedQueue

Bases: BaseQueue · Decorators: traces(spec='002', story='FR-004', reqs=['REQ-002-FR4-002', 'REQ-002-FR4-013', 'REQ-002-FR4-014', 'REQ-002-FR4-015']) · Line 217

Persistent queue that survives restarts using JSON file storage.

BoundedQueue

Bases: IQueue · Decorators: traces(spec='002', story='FR-004', reqs=['REQ-002-FR4-004', 'REQ-002-US-003-016']) · Line 376

Queue wrapper with size limits and overflow handling.

PriorityQueue

Bases: BaseQueue · Decorators: traces(spec='002', story='FR-004', reqs=['REQ-002-FR4-003']) · Line 488

Queue with priority-based ordering.

RedisQueue

Bases: IQueue · Decorators: traces(spec='001', story='US-002', reqs=['REQ-001-US-002-022']) · Line 546

Redis-backed queue for distributed, persistent message storage.

Source

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

"""Queue implementations for Postman message broker.

Provides multiple queue implementations with different durability and performance
characteristics:
- BaseQueue: Abstract base with common functionality
- InMemoryQueue: Fast, volatile (list-based)
- DiskBackedQueue: Persistent, survives restarts (JSON file-based)
- BoundedQueue: With size limits and overflow handling
"""

import asyncio
import json
import logging
from abc import ABC, abstractmethod
from collections import deque
from datetime import UTC, datetime
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.constants import DEFAULT_QUEUE_MAX_SIZE
from orchestrata_ultimate.postman.exceptions import QueueError, QueueFullError
from orchestrata_ultimate.postman.interfaces import IQueue
from orchestrata_ultimate.postman.types import Priority, QueueEntry

# Spec: 002 - Postman Message Broker


logger = logging.getLogger(__name__)


@traces(
    spec="002",
    story="FR-004",
    reqs=[
        "REQ-002-FR4-005",
        "REQ-002-FR4-006",
        "REQ-002-FR4-007",
        "REQ-002-FR4-008",
        "REQ-002-FR4-009",
        "REQ-002-FR4-010",
        "REQ-002-US-003-006",
    ],
)
class BaseQueue(IQueue, ABC):
    """Base queue protocol with common functionality."""

    def __init__(self, name: str) -> None:
        """Initialize base queue.

        Args:
            name: Queue identifier
        """
        self.name = name
        self._lock = asyncio.Lock()
        self._entries: deque[QueueEntry] = deque()

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-006"])
    async def enqueue(self, entry: QueueEntry) -> str:
        """Add entry to queue.

        Args:
            entry: Queue entry to add

        Returns:
            Entry ID

        Raises:
            QueueError: If enqueue fails
        """
        async with self._lock:
            self._entries.append(entry)
            await self._on_enqueue(entry)
            return entry.entry_id

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-007"])
    async def dequeue(self) -> QueueEntry | None:
        """Remove and return next entry.

        Returns:
            Next queue entry or None if empty

        Raises:
            QueueError: If dequeue fails
        """
        async with self._lock:
            if not self._entries:
                return None
            entry = self._entries.popleft()
            await self._on_dequeue(entry)
            return entry

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-008"])
    async def peek(self) -> QueueEntry | None:
        """View next entry without removing.

        Returns:
            Next queue entry or None if empty
        """
        async with self._lock:
            if not self._entries:
                return None
            return self._entries[0]

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-009"])
    async def size(self) -> int:
        """Get current queue size.

        Returns:
            Number of entries in queue
        """
        async with self._lock:
            return len(self._entries)

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-010"])
    async def clear(self) -> int:
        """Clear all entries.

        Returns:
            Number of entries cleared
        """
        async with self._lock:
            count = len(self._entries)
            self._entries.clear()
            await self._on_clear()
            return count

    async def is_empty(self) -> bool:
        """Check if queue is empty.

        Returns:
            True if queue has no entries
        """
        async with self._lock:
            return len(self._entries) == 0

    async def close(self) -> None:
        """Close queue and release resources.

        Override in subclasses that need cleanup.
        """

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-013"])
    @abstractmethod
    async def checkpoint(self) -> None:
        """Save queue state."""
        ...

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-014"])
    @abstractmethod
    async def recover(self) -> int:
        """Restore queue state.

        Returns:
            Number of entries recovered
        """
        ...

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-006"])
    async def _on_enqueue(self, entry: QueueEntry) -> None:
        """Hook called after enqueue.

        Args:
            entry: Entry that was enqueued
        """

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-007"])
    async def _on_dequeue(self, entry: QueueEntry) -> None:
        """Hook called after dequeue.

        Args:
            entry: Entry that was dequeued
        """

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-010"])
    async def _on_clear(self) -> None:
        """Hook called after clear."""


@traces(
    spec="002",
    story="FR-004",
    reqs=["REQ-002-FR4-001"],
)
class InMemoryQueue(BaseQueue):
    """Fast, volatile queue backed by in-memory list."""

    def __init__(self, name: str) -> None:
        """Initialize in-memory queue.

        Args:
            name: Queue identifier
        """
        super().__init__(name)

    async def checkpoint(self) -> None:
        """No-op for in-memory queue (no persistence)."""

    async def recover(self) -> int:
        """No-op for in-memory queue (no persistence).

        Returns:
            Always 0 (nothing to recover)
        """
        return 0


@traces(
    spec="002",
    story="FR-004",
    reqs=[
        "REQ-002-FR4-002",
        "REQ-002-FR4-013",
        "REQ-002-FR4-014",
        "REQ-002-FR4-015",
    ],
)
class DiskBackedQueue(BaseQueue):
    """Persistent queue that survives restarts using JSON file storage."""

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-002"])
    def __init__(self, name: str, storage_path: str = "./queue_data") -> None:
        """Initialize disk-backed queue.

        Args:
            name: Queue identifier
            storage_path: Directory for queue files
        """
        super().__init__(name)
        self.storage_path = Path(storage_path)
        self.storage_path.mkdir(parents=True, exist_ok=True)
        self._file_path = self.storage_path / f"{name}.json"

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-013"])
    async def checkpoint(self) -> None:
        """Save current queue state to disk.

        Raises:
            QueueError: If checkpoint fails
        """
        async with self._lock:
            await self._checkpoint_unlocked()

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-013"])
    async def _checkpoint_unlocked(self) -> None:
        """Save queue state without acquiring lock (caller must hold lock).

        Raises:
            QueueError: If checkpoint fails
        """
        try:
            data = {
                "name": self.name,
                "timestamp": datetime.now(UTC).isoformat(),
                "entries": [self._serialize_entry(e) for e in self._entries],
            }

            # Write to temp file then rename for atomicity
            temp_path = self._file_path.with_suffix(".tmp")
            with open(temp_path, "w") as f:
                json.dump(data, f, indent=2)
            temp_path.replace(self._file_path)

        except (OSError, ValueError, TypeError) as e:
            raise QueueError(f"Checkpoint failed: {e}")

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-013"])
    async def close(self) -> None:
        """Close queue and clean up resources."""
        # Ensure final checkpoint
        async with self._lock:
            await self._checkpoint_unlocked()

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-014", "REQ-002-FR4-015"])
    async def recover(self) -> int:
        """Restore queue state from disk.

        Returns:
            Number of entries recovered

        Raises:
            QueueError: If recovery fails
        """
        async with self._lock:
            if not self._file_path.exists():
                return 0

            try:
                with open(self._file_path) as f:
                    data = json.load(f)

                self._entries.clear()
                for entry_data in data.get("entries", []):
                    entry = self._deserialize_entry(entry_data)
                    self._entries.append(entry)

                return len(self._entries)

            except (OSError, json.JSONDecodeError, KeyError, ValueError) as e:
                raise QueueError(f"Recovery failed: {e}")

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-013"])
    async def _on_enqueue(self, entry: QueueEntry) -> None:
        """Auto-checkpoint after enqueue (lock already held by caller).

        Args:
            entry: Entry that was enqueued
        """
        await self._checkpoint_unlocked()

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-013"])
    async def _on_dequeue(self, entry: QueueEntry) -> None:
        """Auto-checkpoint after dequeue (lock already held by caller).

        Args:
            entry: Entry that was dequeued
        """
        await self._checkpoint_unlocked()

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-013"])
    async def _on_clear(self) -> None:
        """Auto-checkpoint after clear (lock already held by caller)."""
        await self._checkpoint_unlocked()

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-002", "REQ-002-FR4-013"])
    def _serialize_entry(self, entry: QueueEntry) -> dict:
        """Serialize queue entry to dict.

        Args:
            entry: Entry to serialize

        Returns:
            Serialized entry data
        """
        return {
            "entry_id": entry.entry_id,
            "message_id": entry.message_id,
            "priority": entry.priority.value,
            "created_at": entry.created_at.isoformat(),
            "retries": entry.retries,
            "max_retries": entry.max_retries,
            "ttl_expires_at": (
                entry.ttl_expires_at.isoformat() if entry.ttl_expires_at else None
            ),
        }

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-002", "REQ-002-FR4-014"])
    def _deserialize_entry(self, data: dict) -> QueueEntry:
        """Deserialize queue entry from dict.

        Args:
            data: Serialized entry data

        Returns:
            Queue entry
        """
        return QueueEntry(
            entry_id=data["entry_id"],
            message_id=data["message_id"],
            priority=Priority(data["priority"]),
            created_at=datetime.fromisoformat(data["created_at"]),
            retries=data["retries"],
            max_retries=data["max_retries"],
            ttl_expires_at=(
                datetime.fromisoformat(data["ttl_expires_at"])
                if data["ttl_expires_at"]
                else None
            ),
        )


@traces(
    spec="002",
    story="FR-004",
    reqs=["REQ-002-FR4-004", "REQ-002-US-003-016"],
)
class BoundedQueue(IQueue):
    """Queue wrapper with size limits and overflow handling.

    Can wrap any IQueue implementation to add size limits.
    """

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-004"])
    def __init__(
        self,
        name: str | None = None,
        max_size: int = DEFAULT_QUEUE_MAX_SIZE,
        overflow_policy: str = "reject",
        base_queue: IQueue | None = None,
    ) -> None:
        """Initialize bounded queue.

        Args:
            name: Queue identifier (required if no base_queue)
            max_size: Maximum queue size
            overflow_policy: How to handle overflow ("reject" or "drop_oldest")
            base_queue: Optional queue to wrap (if None, uses InMemoryQueue)

        Raises:
            ValueError: If invalid overflow_policy or missing name
        """
        if overflow_policy not in ("reject", "drop_oldest"):
            raise ValueError(f"Invalid overflow_policy: {overflow_policy}")

        self._max_size = max_size
        self.overflow_policy = overflow_policy

        # Use provided base queue or create in-memory queue
        if base_queue is not None:
            self._queue = base_queue
            self.name = base_queue.name
        elif name is not None:
            self._queue = InMemoryQueue(name)
            self.name = name
        else:
            raise ValueError("Either name or base_queue must be provided")

    @property
    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-009"])
    def max_size(self) -> int:
        """Get maximum queue size."""
        return self._max_size

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-004"])
    async def enqueue(self, entry: QueueEntry) -> str:
        """Add entry to queue with size limit enforcement.

        Args:
            entry: Queue entry to add

        Returns:
            Entry ID

        Raises:
            QueueFullError: If queue is full and policy is "reject"
            QueueError: If enqueue fails
        """
        current_size = await self._queue.size()
        if current_size >= self._max_size:
            if self.overflow_policy == "reject":
                raise QueueFullError(
                    f"Queue {self.name} is full (max_size={self._max_size})"
                )
            elif self.overflow_policy == "drop_oldest":
                # Remove oldest entry to make room
                await self._queue.dequeue()

        return await self._queue.enqueue(entry)

    async def dequeue(self) -> QueueEntry | None:
        """Remove and return next entry."""
        return await self._queue.dequeue()

    async def peek(self) -> QueueEntry | None:
        """View next entry without removing."""
        return await self._queue.peek()

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-009"])
    async def size(self) -> int:
        """Get current queue size."""
        return await self._queue.size()

    async def clear(self) -> int:
        """Clear all entries."""
        return await self._queue.clear()

    async def is_empty(self) -> bool:
        """Check if queue is empty."""
        return await self._queue.is_empty()

    async def close(self) -> None:
        """Close underlying queue."""
        return await self._queue.close()

    async def checkpoint(self) -> None:
        """Delegate checkpoint to underlying queue."""
        return await self._queue.checkpoint()

    async def recover(self) -> int:
        """Delegate recover to underlying queue."""
        return await self._queue.recover()


@traces(
    spec="002",
    story="FR-004",
    reqs=["REQ-002-FR4-003"],
)
class PriorityQueue(BaseQueue):
    """Queue with priority-based ordering."""

    def __init__(self, name: str) -> None:
        """Initialize priority queue.

        Args:
            name: Queue identifier
        """
        super().__init__(name)

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-003"])
    async def enqueue(self, entry: QueueEntry) -> str:
        """Add entry to queue maintaining priority order.

        Higher priority (lower numeric value) entries are dequeued first.
        Within same priority, FIFO ordering is maintained.

        Args:
            entry: Queue entry to add

        Returns:
            Entry ID
        """
        async with self._lock:
            # Find insertion point to maintain priority order
            insert_pos = len(self._entries)
            for i, existing in enumerate(self._entries):
                if entry.priority.value < existing.priority.value:
                    insert_pos = i
                    break

            # Insert at correct position
            if insert_pos == len(self._entries):
                self._entries.append(entry)
            else:
                self._entries.insert(insert_pos, entry)

            await self._on_enqueue(entry)
            return entry.entry_id

    async def checkpoint(self) -> None:
        """No-op for priority queue (no persistence by default)."""

    async def recover(self) -> int:
        """No-op for priority queue (no persistence by default).

        Returns:
            Always 0 (nothing to recover)
        """
        return 0


@traces(
    spec="001",
    story="US-002",
    reqs=["REQ-001-US-002-022"],
)
class RedisQueue(IQueue):
    """Redis-backed queue for distributed, persistent message storage.

    Requires Redis connection. Fails immediately if Redis is unavailable.
    """

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-022"])
    def __init__(
        self,
        name: str,
        redis_host: str | None = None,
        redis_port: int | None = None,
        redis_db: int = 0,
    ) -> None:
        """Initialize Redis-backed queue.

        Args:
            name: Queue identifier
            redis_host: Redis server host; resolved from REDIS_HOST when None
            redis_port: Redis server port; resolved from REDIS_PORT when None
            redis_db: Redis database number

        Raises:
            QueueError: If Redis connection fails
        """
        self.name = name
        self._key = f"queue:{name}"

        try:
            from orchestrata_ultimate.core.redis_factory import (
                create_sync_redis,
                resolve_redis_host,
                resolve_redis_port,
            )

            self._redis = create_sync_redis(
                host=redis_host if redis_host is not None else resolve_redis_host(),
                port=redis_port if redis_port is not None else resolve_redis_port(),
                db=redis_db,
            )
            # Test connection immediately
            self._redis.ping()
        except Exception as e:
            raise QueueError(f"Redis connection failed: {e}")

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-006"])
    async def enqueue(self, entry: QueueEntry) -> str:
        """Add entry to Redis queue.

        Args:
            entry: Queue entry to add

        Returns:
            Entry ID

        Raises:
            QueueError: If Redis operation fails
        """
        try:
            data = json.dumps(self._serialize_entry(entry))
            await asyncio.to_thread(self._redis.rpush, self._key, data)
            return entry.entry_id
        except Exception as e:
            raise QueueError(f"Redis enqueue failed: {e}")

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-007"])
    async def dequeue(self) -> QueueEntry | None:
        """Remove and return next entry from Redis queue.

        Returns:
            Next queue entry or None if empty

        Raises:
            QueueError: If Redis operation fails
        """
        try:
            data: str | bytes | None = await asyncio.to_thread(self._redis.lpop, self._key)  # type: ignore[arg-type]
            if data is None:
                return None
            return self._deserialize_entry(json.loads(data))
        except Exception as e:
            raise QueueError(f"Redis dequeue failed: {e}")

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-008"])
    async def peek(self) -> QueueEntry | None:
        """View next entry without removing.

        Returns:
            Next queue entry or None if empty
        """
        try:
            data: str | bytes | None = await asyncio.to_thread(self._redis.lindex, self._key, 0)  # type: ignore[arg-type]
            if data is None:
                return None
            return self._deserialize_entry(json.loads(data))
        except Exception as e:
            raise QueueError(f"Redis peek failed: {e}")

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-009"])
    async def size(self) -> int:
        """Get current queue size.

        Returns:
            Number of entries in queue
        """
        try:
            size: int = await asyncio.to_thread(self._redis.llen, self._key)  # type: ignore[arg-type]
            return size
        except Exception as e:
            raise QueueError(f"Redis size failed: {e}")

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-010"])
    async def clear(self) -> int:
        """Clear all entries.

        Returns:
            Number of entries cleared
        """
        try:
            count = await self.size()
            await asyncio.to_thread(self._redis.delete, self._key)
            return count
        except Exception as e:
            raise QueueError(f"Redis clear failed: {e}")

    async def is_empty(self) -> bool:
        """Check if queue is empty."""
        return await self.size() == 0

    async def close(self) -> None:
        """Close Redis connection."""
        if self._redis:
            self._redis.close()

    async def checkpoint(self) -> None:
        """No-op for Redis queue (Redis is always persistent)."""

    async def recover(self) -> int:
        """No-op for Redis queue (data persists in Redis).

        Returns:
            Current queue size (already recovered)
        """
        return await self.size()

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-006"])
    def _serialize_entry(self, entry: QueueEntry) -> dict:
        """Serialize queue entry to dict."""
        return {
            "entry_id": entry.entry_id,
            "message_id": entry.message_id,
            "priority": entry.priority.value,
            "created_at": entry.created_at.isoformat(),
            "retries": entry.retries,
            "max_retries": entry.max_retries,
            "ttl_expires_at": (
                entry.ttl_expires_at.isoformat() if entry.ttl_expires_at else None
            ),
        }

    @traces(spec="002", story="FR-004", reqs=["REQ-002-FR4-007", "REQ-002-FR4-008"])
    def _deserialize_entry(self, data: dict) -> QueueEntry:
        """Deserialize queue entry from dict."""
        return QueueEntry(
            entry_id=data["entry_id"],
            message_id=data["message_id"],
            priority=Priority(data["priority"]),
            created_at=datetime.fromisoformat(data["created_at"]),
            retries=data["retries"],
            max_retries=data["max_retries"],
            ttl_expires_at=(
                datetime.fromisoformat(data["ttl_expires_at"])
                if data["ttl_expires_at"]
                else None
            ),
        )