baldur_pro.services.replay — Replay Queue

Backpressure-aware replay of stored failures: ReplayQueueService with its BackpressureStatus and RateLimitStatus signals.

🔒 PRO Feature — requires a baldur-pro license

These symbols ship in the baldur-pro distribution. PRO modules import normally — there is no ImportError. PRO features activate only when baldur.init() runs with a valid BALDUR_LICENSE_KEY; without it the system runs with OSS defaults and register_pro_services() logs entitlement.pro_registration_skipped.

replay

Replay Queue Package.

Provides replay queue infrastructure with rate limiting and backpressure support for the baldur framework.

Usage

from baldur_pro.services.replay import ( get_replay_queue_service, reset_replay_queue_service, )

service = get_replay_queue_service() success = service.enqueue_raw(payload, source="my_component")

Status: Public

BackpressureStatus dataclass

BackpressureStatus(
    active: bool = False,
    queue_depth: int = 0,
    max_depth: int = 10000,
    consumer_lag: float = 0.0,
    throttle_percent: float = 0.0,
)

Status of backpressure mechanisms for the replay queue.

Attributes:

Name	Type	Description
active	bool	Whether backpressure is currently being applied.
queue_depth	int	Current number of items in the queue.
max_depth	int	Maximum queue depth before backpressure activates.
consumer_lag	float	Estimated consumer lag in seconds.
throttle_percent	float	Current throttle percentage (0-100).

RateLimitStatus dataclass

RateLimitStatus(
    enabled: bool = True,
    current_rate: int = 0,
    limit: int = 1000,
    rejected_count: int = 0,
    window_seconds: int = 60,
)

Status of the rate limiter for the replay queue.

Attributes:

Name	Type	Description
enabled	bool	Whether rate limiting is currently active.
current_rate	int	Number of requests accepted in the current window.
limit	int	Maximum allowed requests per window.
rejected_count	int	Total number of requests rejected due to rate limiting.
window_seconds	int	Duration of the sliding window in seconds.

ReplayQueueService

ReplayQueueService(
    max_depth: int = DEFAULT_MAX_DEPTH,
    rate_limit: int = DEFAULT_RATE_LIMIT,
    rate_window_seconds: int = DEFAULT_RATE_WINDOW_SECONDS,
    backpressure_threshold: float = DEFAULT_BACKPRESSURE_THRESHOLD,
    rate_limiting_enabled: bool = True,
)

In-memory replay queue with rate limiting and backpressure.

This service provides a standalone replay queue that works without external dependencies (Redis, a message broker, etc.). It is designed for environments where the full replay infrastructure may not be available, such as isolated testing scenarios.

Thread-safety

All public methods are protected by a threading.Lock, making this class safe for concurrent use from multiple threads.

Fail-safe defaults

• Rate limiting enabled with generous defaults (1000 req/60s)
• Max queue depth of 10,000 to prevent unbounded memory growth
• Backpressure activates at 80% queue capacity
• All operations degrade gracefully on internal errors

Initialize the replay queue service.

Parameters:

Name	Type	Description	Default
max_depth	int	Maximum queue depth before rejecting new items.	DEFAULT_MAX_DEPTH
rate_limit	int	Maximum number of enqueues per window.	DEFAULT_RATE_LIMIT
rate_window_seconds	int	Sliding window duration in seconds.	DEFAULT_RATE_WINDOW_SECONDS
backpressure_threshold	float	Queue fullness ratio (0.0-1.0) at which backpressure activates.	DEFAULT_BACKPRESSURE_THRESHOLD
rate_limiting_enabled	bool	Whether to enforce rate limiting.	True

enqueue_raw

enqueue_raw(
    payload: bytes, source: str = "unknown"
) -> bool

Enqueue a raw payload for replay processing.

Parameters:

Name	Type	Description	Default
payload	bytes	Raw bytes payload to enqueue.	required
source	str	Identifier of the component that produced the payload.	'unknown'

Returns:

Type	Description
bool	True if the payload was accepted, False if rejected due to
bool	queue depth limit or rate limiting.

dequeue

dequeue() -> _QueueEntry | None

Dequeue the next entry for processing.

Returns:

Type	Description
_QueueEntry | None	The next queue entry, or None if the queue is empty.

get_rate_limit_status

get_rate_limit_status() -> RateLimitStatus

Get current rate limiting status.

Returns:

Type	Description
RateLimitStatus	RateLimitStatus with current rate limiting metrics.

get_backpressure_status

get_backpressure_status() -> BackpressureStatus

Get current backpressure status.

Returns:

Type	Description
BackpressureStatus	BackpressureStatus with current queue depth and throttle metrics.

get_queue_depth

get_queue_depth() -> int

Get the current number of items in the queue.

Returns:

Type	Description
int	Current queue depth.

get_stats

get_stats() -> dict[str, Any]

Get overall queue statistics.

Returns:

Type	Description
dict[str, Any]	Dict with enqueue/dequeue counts, depth, and status info.

get_replay_queue_service

get_replay_queue_service() -> ReplayQueueService

Get the singleton ReplayQueueService instance.

Thread-safe lazy initialization. Creates the service with default configuration on first access.

Returns:

Type	Description
ReplayQueueService	The singleton ReplayQueueService instance.

reset_replay_queue_service

reset_replay_queue_service() -> None

Reset the singleton instance (for testing).

This clears the singleton so the next call to get_replay_queue_service() creates a fresh instance.