Metadata-Version: 2.1
Name: mrsal
Version: 3.12.0
Summary: Production-ready AMQP message broker abstraction with advanced retry logic, dead letter exchanges, and high availability features.
Home-page: https://github.com/NeoMedSys/mrsal
License: GPL-3.0-or-later
Keywords: message broker,RabbitMQ,Pika,Mrsal,AMQP,async,dead letter exchange
Author: Jon E. Nesvold
Author-email: jnesvold@neomedsys.io
Maintainer: Jon E Nesvold
Maintainer-email: jnesvold@neomedsys.io
Requires-Python: >=3.10,<3.15
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Communications
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Networking
Requires-Dist: aio-pika (>=9.4.3,<10.0.0)
Requires-Dist: colorlog (>=6.7.0,<7.0.0)
Requires-Dist: pika (>=1.3.0,<2.0.0)
Requires-Dist: pydantic (>=2.11.5,<3.0.0)
Requires-Dist: tenacity (>=9.0.0,<10.0.0)
Project-URL: Documentation, https://neomedsys.github.io/mrsal/
Project-URL: Repository, https://github.com/NeoMedSys/mrsal
Description-Content-Type: text/markdown

# MRSAL AMQP
[![Release](https://img.shields.io/badge/release-3.12.0-blue.svg)](https://pypi.org/project/mrsal/) 
[![Python 3.10+](https://img.shields.io/badge/python-3.10%7C3.11%7C3.12-blue.svg)](https://www.python.org/downloads/)
[![Mrsal Workflow](https://github.com/NeoMedSys/mrsal/actions/workflows/mrsal.yaml/badge.svg?branch=main)](https://github.com/NeoMedSys/mrsal/actions/workflows/mrsal.yaml)
[![Coverage](https://neomedsys.github.io/mrsal/reports/badges/coverage-badge.svg)](https://neomedsys.github.io/mrsal/reports/coverage/htmlcov/)

## New in 3.12.0

- **Push-based metrics hooks** (additive, no breaking changes). Install a set of
  callables with `mrsal.set_metrics_hooks(MetricsHooks(...))` to instrument the
  publish / consume / DLX hot paths — `on_publish`, `on_consume`, `on_retry`,
  `on_dlx_final` — without parsing log lines. mrsal ships no metrics client; the
  hook receives plain values, so you wire them to Prometheus, StatsD, or plain
  counters. Parity with sonic's `types.MetricsHooks`. Unset hooks cost a single
  `None` check; the core install gains no dependency. See
  [§4.6](#46-metrics-hooks).

## New in 3.11.0

- **In-memory test broker** (additive, no breaking changes). `mrsal.testing`
  ships `TestMrsalBroker` / `TestMrsalAsyncBroker` to unit-test consumers and
  publishers without a live RabbitMQ. It runs mrsal's real validate → callback →
  ack/DLX code against an in-process broker — routing, DLX topology and retry
  cycles included — so tests exercise actual behaviour instead of mocks and run
  in milliseconds. Time is modeled: `br.advance(minutes=...)` fires the `.retry`
  queue TTL to drive retry cycles deterministically. Internally, `start_consumer`
  was split into a setup step and the consume loop (`_prepare_consumer` /
  `_run_consume_loop`); the public API is unchanged. See
  [§5](#5-testing-handlers-without-rabbitmq).

## New in 3.10.0

- **Reusable / pooled blocking publishers** (additive, no breaking changes).
  `MrsalBlockingPublisher` keeps one connection and one confirm-enabled channel
  warm across publishes, and caches declared topology so the passive
  `Exchange.Declare` / `Queue.Declare` round-trips run only once per target
  instead of on every publish. `MrsalBlockingPublisherPool` hands out one
  publisher per concurrent caller so the non-thread-safe pika connection is
  never shared. Existing `MrsalBlockingAMQP.publish_message` is unchanged. See
  [§2.2](#22-reusable-publishers--connection-pooling).

## Breaking changes in 3.9.0

- **DLX retry cycles now use exponential backoff by default**
  (`retry_backoff="exponential"`). Per-cycle delay grows as
  `retry_cycle_interval * 2**cycle_count` minutes (with ±20% jitter), clamped
  at `retry_backoff_max` (default 60m). The previous flat-interval behaviour
  is still available with `retry_backoff="fixed"`.
- **Topology change requires deleting the `.retry` queue on upgrade.** In
  exponential mode the `.retry` queue is declared with
  `x-message-ttl = retry_backoff_max * 60_000` (1h by default) instead of
  `retry_cycle_interval * 60_000` (10m by default). RabbitMQ will reject the
  redeclare with `inequivalent arg 'x-message-ttl'`, raising
  `MrsalAbortedSetup`. **Before redeploying, delete each `<queue>.retry`
  queue on the broker** (e.g. `rabbitmqctl delete_queue <queue>.retry`).
  The same applies if you switch `retry_backoff` modes later.
- **`max_retry_time_limit` default raised from 60m to 480m (8h)** so the
  longer per-cycle delays still have room to actually retry. If you depended
  on the 1h budget, pass `max_retry_time_limit=60` explicitly.
- **Internal `_log_dlx_result` signature changed.** Second positional arg is
  now `next_delay_ms: int | None` instead of `retry_cycle_interval: int`.
  Subclasses that overrode the method need to update.

## Breaking changes in 3.8.0

- **`MrsalBlockingAMQP.start_consumer` default `auto_ack` flipped from `True` to
  `False`,** matching the async consumer and every example in this README.
  Callers that relied on the previous default were silently opting out of DLX
  accountability while the library advertised reliability features. To keep the
  old behaviour, pass `auto_ack=True, dlx_enable=False` explicitly and accept
  that failed callbacks are dropped.
- **`start_consumer` now rejects incompatible flag combinations at setup** by
  raising `MrsalAbortedSetup`:
  - `auto_ack=True, dlx_enable=True` -- on both consumers. Once the broker has
    acked at delivery, failed messages cannot be routed to the DLX, so the
    combination is meaningless.
  - `auto_ack=True, threaded=True` -- on the blocking consumer. The executor's
    submit queue is unbounded, so a slow callback grows pending tasks until OOM.
  See §4.1.1 for full rationale and remediation paths.

## Breaking changes in 3.7.0

- **`MrsalAsyncAMQP.start_consumer` now wraps the queue iterator in
  `async with queue.iterator(...) as it:`** so consumer cancellation is
  deterministically delivered to the broker on exception or generator GC.
  Subclasses or tests that monkey-patched `queue.iterator()` to return a
  bare async generator must update their mocks to also support the async
  context-manager protocol (`__aenter__` / `__aexit__`). See
  `tests/conftest.py::AsyncIteratorMock` for the reference shape.
- **New `async def stop()` lifecycle method.** Once `stop()` has been
  called on a `MrsalAsyncAMQP` instance, that instance cannot be restarted —
  the internal stop event remains set so future `start_consumer` calls
  exit on the first iteration. To restart, construct a new instance. The
  persistent state is deliberate: it preserves a `stop()` request that
  arrives during a tenacity retry backoff, which would otherwise be
  silently dropped.
- **New `max_concurrent_tasks: int | None` parameter on the async
  `start_consumer`.** Default `None` preserves prior behaviour (sequential
  processing). When set to `N > 0`, up to N messages are dispatched as
  concurrent `asyncio` tasks bounded by a semaphore. Note that
  `prefetch_count` is broker-side buffering and does **not** parallelize
  the consumer — combine with `prefetch_count >= max_concurrent_tasks` for
  steady throughput.
- **New `drain_timeout: float | None` parameter on the async
  `start_consumer`.** Bounds how long the consumer waits for in-flight
  tasks to finish after a graceful stop. On timeout, remaining tasks are
  cancelled and their messages will be redelivered by the broker.

## Breaking changes in 3.6.0

- `validate_payload` now **returns** the validated model instance (it previously
  returned `None`). When `payload_model` is passed to `start_consumer`, the
  callback receives the **validated model instance** as its body argument
  instead of the raw `bytes`. Callbacks that called `json.loads(body)` /
  `Model(**json.loads(body))` internally must drop that step and treat the
  third argument as an instance of `payload_model`. See the example in §4.5.
- Publishers (`publish_message`, `publish_messages`) and DLX republishes now
  always enable publisher confirms, so `NackError` / `UnroutableError` are
  raised (and retried) instead of silently dropped. DLX republishes also
  honor an explicit `dlx_routing_key` instead of falling back to the original
  routing key — fixing silent loss when the DLX bind used a different key.
- Internal-only: `_process_single_message` now reads its `runtime_config`
  dict with `[]` instead of `.get()`. Callers that build their own
  `runtime_config` (e.g. in tests) must include all keys produced by
  `start_consumer`: `callback`, `callback_args`, `auto_ack`, `payload_model`,
  `threaded`, `dlx_enable`, `enable_retry_cycles`, `retry_cycle_interval`,
  `max_retry_time_limit`, `exchange_name`, `routing_key`, `dlx_exchange_name`,
  `dlx_routing_key`, `queue_name`. Missing keys now raise `KeyError` instead
  of silently being `None`.

## Intro
Mrsal is a **production-ready** message broker abstraction on top of [RabbitMQ](https://www.rabbitmq.com/), [aio-pika](https://aio-pika.readthedocs.io/en/latest/) and [Pika](https://pika.readthedocs.io/en/stable/index.html). 

**Why Mrsal?** Setting up robust AMQP in production is complex. You need dead letter exchanges, retry logic, quorum queues, proper error handling, queue management, and more. Mrsal gives you **enterprise-grade messaging** out of the box with just a few lines of code.

**What makes Mrsal production-ready:**

- **Dead Letter Exchange**: Automatic DLX setup with configurable retry cycles  
- **High Availability**: Quorum queues for data safety across cluster nodes  
- **Performance Tuning**: Queue limits, overflow behavior, lazy queues, prefetch control  
- **Zero Configuration**: Sensible defaults that work in production  
- **Full Observability**: Comprehensive logging and retry tracking  
- **Type Safety**: Pydantic integration for payload validation  
- **Async & Sync**: Both blocking and async implementations  
- **Threaded Consumers**: Bounded thread pool for long-running callbacks  
- **Resource Safety**: Context manager support for clean connection lifecycle  

The goal is to make Mrsal **trivial to re-use** across all services in your distributed system and to make advanced message queuing protocols **easy and safe**. No more big chunks of repetitive code across your services or bespoke solutions to handle dead letters.

**Perfect for:**
- Microservices communication
- Event-driven architectures  
- Background job processing
- Real-time data pipelines
- Mission-critical message processing

###### Mrsal is Arabic for a small arrow and is used to describe something that performs a task with lightness and speed.

## Quick Start guide

### 0. Requirements
1. RabbitMQ server up and running
2. python 3.10 >=
3. tested on linux only

### 1. Installing
First things first:

```bash
poetry add mrsal
```

Next set the default username, password and servername for your RabbitMQ setup. It's advisable to use a \`.env\` script or \`(.zsh)rc\` file for persistence.

```bash
[RabbitEnvVars]
RABBITMQ_USER=******
RABBITMQ_PASSWORD=******
RABBITMQ_VHOST=******
RABBITMQ_DOMAIN=******
RABBITMQ_PORT=******

# FOR TLS
RABBITMQ_CAFILE=/path/to/file
RABBITMQ_CERT=/path/to/file
RABBITMQ_KEY=/path/to/file
```

###### Mrsal was first developed by NeoMedSys and the research group \[CRAI\](https://crai.no/) at the university hospital of Oslo.

### 2. Setup and connect
- Example 1: Lets create a blocking connection on localhost with no TLS encryption

```python
from mrsal.amqp.subclass import MrsalBlockingAMQP

mrsal = MrsalBlockingAMQP(
    host=RABBITMQ_DOMAIN,  # Use a custom domain if you are using SSL e.g. mrsal.on-example.com
    port=int(RABBITMQ_PORT),
    credentials=(RABBITMQ_USER, RABBITMQ_PASSWORD),
    virtual_host=RABBITMQ_VHOST,
    ssl=False # Set this to True for SSL/TLS (you will need to set the cert paths if you do so)
)

# boom you are staged for connection. This instantiation stages for connection only
# When done, call mrsal.close() to clean up — or use a context manager:
# with MrsalBlockingAMQP(...) as mrsal:
```

#### 2.1 Publish
Now lets publish our message of friendship on the friendship exchange.
Note: When `auto_declare=True` means that MrsalAMQP will create the specified `exchange` and `queue`, then bind them together using `routing_key` in one go. If you want to customize each step then turn off auto_declare and specify each step yourself with custom arguments etc.

```python
# BasicProperties is used to set the message properties
prop = pika.BasicProperties(
        app_id='zoomer_app',
        message_id='zoomer_msg',
        content_type=' application/json',
        content_encoding='utf-8',
        delivery_mode=pika.DeliveryMode.Persistent,
        headers=None)

message_body = {'zoomer_message': 'Get it yia bish'}

# For publishers, use a context manager so the connection is cleaned up after sending
with MrsalBlockingAMQP(
    host=RABBITMQ_DOMAIN,
    port=int(RABBITMQ_PORT),
    credentials=(RABBITMQ_USER, RABBITMQ_PASSWORD),
    virtual_host=RABBITMQ_VHOST,
    ssl=False
) as mrsal:
    mrsal.publish_message(exchange_name='zoomer_x',
                            exchange_type='direct',
                            queue_name='zoomer_q',
                            routing_key='zoomer_key',
                            message=message_body,
                            prop=prop,
                            auto_declare=True)
# Connection is automatically closed here
```

#### 2.2 Reusable publishers & connection pooling

The context-manager pattern above opens a fresh connection (TCP + TLS + AMQP
handshake) for every send, and `publish_message` re-runs a passive
`Exchange.Declare` / `Queue.Declare` on each call. For services that publish
frequently — or per HTTP request — that handshake-and-declare churn adds up.

`MrsalBlockingPublisher` keeps the connection and a confirm-enabled channel warm
across publishes, and declares each target's topology only once:

```python
from mrsal.amqp.subclass import MrsalBlockingPublisher

publisher = MrsalBlockingPublisher(
    host=RABBITMQ_DOMAIN,
    port=int(RABBITMQ_PORT),
    credentials=(RABBITMQ_USER, RABBITMQ_PASSWORD),
    virtual_host=RABBITMQ_VHOST,
    ssl=False,
)

# First call declares/verifies the topology; later calls skip straight to
# basic_publish on the same warm channel. Publisher confirms stay on, so an
# unroutable target still raises (UnroutableError / NackError).
publisher.publish(exchange_name='zoomer_x',
                  exchange_type='direct',
                  queue_name='zoomer_q',
                  routing_key='zoomer_key',
                  message=message_body,
                  prop=prop)
# ... reuse `publisher` for the lifetime of the service, then:
publisher.close()
```

A `MrsalBlockingPublisher` is **not** thread-safe — pika's blocking connection
is owned by a single thread. For concurrent callers (e.g. a sync web framework
running route handlers in a thread pool), use `MrsalBlockingPublisherPool`,
which hands out one warm publisher per concurrent caller and returns it to the
pool afterwards:

```python
from mrsal.amqp.subclass import MrsalBlockingPublisherPool

pool = MrsalBlockingPublisherPool(
    size=4,                       # max concurrent connections kept warm
    host=RABBITMQ_DOMAIN,
    port=int(RABBITMQ_PORT),
    credentials=(RABBITMQ_USER, RABBITMQ_PASSWORD),
    virtual_host=RABBITMQ_VHOST,
    ssl=False,
)

# Borrow for the duration of the block; the connection is returned (kept warm)
# on exit. `acquire()` blocks when the pool is saturated — pass timeout=<sec>
# to raise queue.Empty instead of waiting indefinitely.
with pool.acquire() as publisher:
    publisher.publish(exchange_name='zoomer_x',
                      exchange_type='direct',
                      queue_name='zoomer_q',
                      routing_key='zoomer_key',
                      message=message_body,
                      prop=prop)

# On shutdown:
pool.close_all()
```

A connection that drops while idle is transparently reopened on its next use
(the topology cache is cleared on reconnect so the new connection re-verifies).

#### 2.3 Consume

Now lets setup a consumer that will listen to our very important messages. If you are using scripts rather than notebooks then it's advisable to run consume and publish separately. We are going to need a callback function which is triggered upon receiving the message from the queue we subscribe to. You can use the callback function to activate something in your system.

Note:
- Your callback is invoked as `callback(*callback_args, method_frame, properties, body)`. Anything you pass via `callback_args` is prepended; `method_frame`, `properties`, and `body` are always supplied by mrsal.
- When you pass `payload_model=YourModel`, `body` is the validated model instance instead of raw bytes (see §4.5).
- We can use pydantic BaseModel classes to enforce types in the body

```python
from pydantic import BaseModel

class ZoomerNRJ(BaseModel):
    zoomer_message: str

def consumer_callback_with_delivery_info(
     method_frame: pika.spec.Basic.Deliver,
     properties: pika.spec.BasicProperties,
     body: str
     ):
    if 'Get it' in body:
        app_id = properties.app_id
        msg_id = properties.message_id
        print(f'app_id={app_id}, msg_id={msg_id}')
        print('Slay with main character vibe')
    else:
        raise SadZoomerEnergyError('Zoomer sad now')

mrsal.start_consumer(
        queue_name='zoomer_q',
        exchange_name='zoomer_x',
        callback_args=None,  # no need to specifiy if you do not need it
        callback=consumer_callback_with_delivery_info,
        auto_declare=True,
        auto_ack=False
    )
```

Done! Your first message of zommerism has been sent to the zoomer queue on the exchange of Zoomeru in a blocking connection. Lets see how we can do it in async in the next step.

### 3. Setup and Connect Async
Its usually the best practise to use async consumers if high throughput is expected. We can easily do this by adjusting the code a little bit to fit the framework of async connection in python.
```python
from mrsal.amqp.subclass import MrsalAsyncAMQP

mrsal = MrsalAsyncAMQP(
    host=RABBITMQ_DOMAIN,  # Use a custom domain if you are using SSL e.g. mrsal.on-example.com
    port=int(RABBITMQ_PORT),
    credentials=(RABBITMQ_USER, RABBITMQ_PASSWORD),
    virtual_host=RABBITMQ_VHOST,
    ssl=False # Set this to True for SSL/TLS (you will need to set the cert paths if you do so)
)

# boom you are staged for async connection.
# When done, call await mrsal.close() to clean up — or use an async context manager:
# async with MrsalAsyncAMQP(...) as mrsal:
```

#### 3.1 Consume
Lets go turbo and set up the consumer in async for efficient AMQP handling
```python
import asyncio
from pydantic import BaseModel

class ZoomerNRJ(BaseModel):
    zoomer_message: str

async def consumer_callback_with_delivery_info(
     method_frame: pika.spec.Basic.Deliver,
     properties: pika.spec.BasicProperties,
     body: str
     ):
    if 'Get it' in body:
        app_id = properties.app_id
        msg_id = properties.message_id
        print(f'app_id={app_id}, msg_id={msg_id}')
        print('Slay with main character vibe')
    else:
        raise SadZoomerEnergyError('Zoomer sad now')

asyncio.run(mrsal.start_consumer(
        queue_name='zoomer_q',
        exchange_name='zoomer_x',
        callback_args=None,  # no need to specifiy if you do not need it
        callback=consumer_callback_with_delivery_info,
        auto_declare=True,
        auto_ack=False
    ))
```

That simple! You have now setups for full advanced message queueing protocols that you can use to promote friendship or other necessary communication between your services in both blocking or async connections.

###### Note! There are many parameters and settings that you can use to set up a more sophisticated communication protocol in both blocking or async connection with pydantic BaseModels to enforce data types in the expected payload.

### 4. Advanced Features

#### 4.1 Dead Letter Exchange & Retry Logic with Cycles

Mrsal provides time-delayed retry cycles via a broker-managed `<queue>.retry` queue, with a terminal `<queue>.dlx` parking lot for messages that exhaust the retry budget:

```python
mrsal = MrsalBlockingAMQP(
    host=RABBITMQ_DOMAIN,
    port=int(RABBITMQ_PORT),
    credentials=(RABBITMQ_USER, RABBITMQ_PASSWORD),
    virtual_host=RABBITMQ_VHOST,
    dlx_enable=True,        # Default: creates '<exchange>.dlx' (+ '<queue>.retry' when retry cycles are on)
)

# Advanced retry configuration with cycles
mrsal.start_consumer(
    queue_name='critical_queue',
    exchange_name='critical_exchange',
    exchange_type='direct',
    routing_key='critical_key',
    callback=my_callback,
    auto_ack=False,                    # Required for retry logic
    dlx_enable=True,                   # Enable DLX for this queue
    dlx_exchange_name='custom_dlx',    # Optional: custom DLX name
    dlx_routing_key='dlx_key',         # Optional: custom DLX routing
    enable_retry_cycles=True,          # Enable time-delayed retry cycles
    retry_cycle_interval=10,           # Minutes between retry cycles
    max_retry_time_limit=60,           # Total minutes before permanent failure
)
```

**How the advanced retry logic works:**

When `enable_retry_cycles=True`, Mrsal declares two queues alongside the DLX exchange:

- `<queue>.retry` — a delay queue with a broker-side `x-message-ttl` of `retry_cycle_interval` minutes and `x-dead-letter-exchange` / `x-dead-letter-routing-key` queue arguments pointing back to the original exchange/routing key. Failed messages land here, sit for the TTL, and are dead-lettered back to the original queue automatically by the broker.
- `<queue>.dlx` — the terminal parking lot. Messages whose `max_retry_time_limit` is exhausted are published here and stay until a human replays them.

1. **First failure**: Message is published to `<queue>.retry` with retry-tracking headers (`x-cycle-count`, `x-first-failure`, `x-total-elapsed`, `x-processing-error`).
2. **Retry Cycles**: After `retry_cycle_interval` minutes, RabbitMQ dead-letters the message back to the original queue. The consumer reprocesses it; if it fails again and `total_elapsed < max_retry_time_limit`, it cycles again.
3. **Permanent Failure**: Once `max_retry_time_limit` is exceeded, the message is published to `<queue>.dlx` with `x-retry-exhausted=True` and stays there for manual review.

**Benefits:**
- Handles longer outages with time-delayed cycles  
- Full observability with retry tracking  
- Manual intervention capability for persistent failures

> **Operational note:** `retry_cycle_interval` is baked into the `<queue>.retry` queue declaration as `x-message-ttl`. Changing it between deployments will trip RabbitMQ's `PRECONDITION_FAILED - inequivalent arg 'x-message-ttl'` error on the existing queue, and Mrsal will **abort startup with `MrsalAbortedSetup`** rather than silently fall through to a misconfigured retry path. To roll out a new interval: delete the `<queue>.retry` queue on the broker, then redeploy.

**Constraints (rejected at `start_consumer` time):**

Because the two-queue retry topology relies on the broker honoring distinct binding keys for `<queue>.retry` and `<queue>.dlx`, some configurations would silently drop cycled messages and are rejected with `MrsalAbortedSetup`:

- `exchange_type='fanout'` or `'headers'` with `enable_retry_cycles=True` — fanout/headers exchanges ignore routing keys, so cycling and parking collapse into the same queue and exhausted messages re-cycle indefinitely. Use `enable_retry_cycles=False` for these exchange types (terminal DLX still works).
- `exchange_type='topic'` with a wildcard `routing_key` (`*` or `#` segments) and `enable_retry_cycles=True` — the broker would dead-letter expired messages back to the original exchange with the literal wildcard string as the routing key, which matches no binding. Use a concrete routing key, or `enable_retry_cycles=False`.
- `retry_cycle_interval <= 0` — produces a zero or negative TTL, which either makes the broker reject the declare or dead-letters every message immediately into a tight retry loop.

#### 4.1.1 `auto_ack` and reliability

`auto_ack=True` tells the broker to ack each message at delivery, before the consumer has done anything with it. That means **you opt out of every reliability feature in this library**:

- **`auto_ack=True` is incompatible with `dlx_enable=True`.** Once the broker has acked, the message no longer exists from RabbitMQ's perspective — there is nothing left to route to the DLX on failure. Mrsal rejects this combination at `start_consumer` time with `MrsalAbortedSetup`. To use `auto_ack=True`, pass `dlx_enable=False` explicitly.
- **`auto_ack=True` is incompatible with `threaded=True` (blocking consumer).** With both flags, the consume loop hands each delivery to the `ThreadPoolExecutor` and immediately moves on — the broker has already acked, so `prefetch_count` no longer provides backpressure. A slow callback plus a fast broker grows the executor's pending-task queue without bound until the process runs out of memory. Mrsal rejects this combination at setup.
- **`auto_ack=True` with `dlx_enable=False` is allowed** and is fire-and-forget: failed callbacks are logged and the message is gone. Use it only when message loss is acceptable.
- **`auto_ack=True` on the async consumer has no broker-side backpressure.** AMQP's `no_ack=true` mode (what `auto_ack=True` translates to on the wire) tells the broker to disregard `prefetch_count` and push deliveries as fast as the TCP connection allows. aio-pika buffers those internally in an unbounded queue between the broker connection and the `async for ... in iterator` loop, so a slow callback plus a fast broker grows that buffer until the process runs out of memory. `max_concurrent_tasks` bounds mrsal's task set but does *not* bound aio-pika's receive buffer. The blocking consumer hits the same shape of bug under `threaded=True`, which is why that combination is rejected outright -- the async equivalent is allowed for parity with the original API but should be considered unsafe for production. Use `auto_ack=False` if you need both async and reliability.

For production, use `auto_ack=False` (the default) so the consumer acks on success and routes failures through DLX/retry as configured.

#### 4.2 Queue Management & Performance

Configure queues for optimal performance and resource management:

```python
mrsal.start_consumer(
    queue_name='high_performance_queue',
    exchange_name='perf_exchange',
    exchange_type='direct',
    routing_key='perf_key',
    callback=my_callback,
    
    # Queue limits and overflow behavior
    max_queue_length=10000,              # Max messages before overflow
    max_queue_length_bytes=None,         # Optional: max queue size in bytes
    queue_overflow="drop-head",          # "drop-head" or "reject-publish"
    
    # Performance settings
    single_active_consumer=False,        # Allow parallel processing
    lazy_queue=False,                    # Keep messages in RAM for speed
    use_quorum_queues=True,              # High availability
    
    # Memory optimization (for low-priority queues)
    lazy_queue=True,                     # Store messages on disk
    single_active_consumer=True          # Sequential processing
)
```

**Queue Configuration Options:**

- **\`max_queue_length\`**: Limit queue size to prevent memory issues
- **\`queue_overflow\`**: 
  - \`"drop-head"\`: Remove oldest messages when full
  - \`"reject-publish"\`: Reject new messages when full
- **\`single_active_consumer\`**: Only one consumer processes at a time (good for ordered processing)
- **\`lazy_queue\`**: Store messages on disk instead of RAM (memory efficient)
- **\`use_quorum_queues\`**: Enhanced durability and performance in clusters

#### 4.3 Quorum Queues

Quorum queues provide better data safety and performance for production environments:

```python
mrsal = MrsalBlockingAMQP(
    host=RABBITMQ_DOMAIN,
    port=int(RABBITMQ_PORT),
    credentials=(RABBITMQ_USER, RABBITMQ_PASSWORD),
    virtual_host=RABBITMQ_VHOST,
    use_quorum_queues=True  # Default: enables quorum queues
)

# Per-queue configuration
mrsal.start_consumer(
    queue_name='high_availability_queue',
    exchange_name='ha_exchange',
    exchange_type='direct',
    routing_key='ha_key',
    callback=my_callback,
    use_quorum_queues=True  # This queue will be highly available
)
```

**Benefits:**
- Better data replication across RabbitMQ cluster nodes  
- Improved performance under high load  
- Automatic leader election and failover  
- Works great in Kubernetes and bare metal deployments

#### 4.4 Threaded Consumer

For long-running callbacks that would otherwise block the heartbeat, use `threaded=True`. Messages are processed in a bounded thread pool instead of the main thread:

```python
mrsal.start_consumer(
    queue_name='heavy_queue',
    exchange_name='heavy_exchange',
    exchange_type='direct',
    routing_key='heavy_key',
    callback=slow_callback,
    auto_ack=False,
    threaded=True,              # Process messages in a thread pool
    max_workers=10,             # Pool size (defaults to prefetch_count)
)
```

#### 4.5 Production-Ready Example

```python
from mrsal.amqp.subclass import MrsalBlockingAMQP
from pydantic import BaseModel

class OrderMessage(BaseModel):
    order_id: str
    customer_id: str
    amount: float

def process_order(method_frame, properties, order: OrderMessage):
    # When ``payload_model`` is set on ``start_consumer``, mrsal validates the
    # raw bytes against the model and passes the validated instance here as the
    # third argument. Validation failures are routed to DLX before the callback
    # runs, so this function only sees well-formed payloads.
    print(f"Processing order {order.order_id} for customer {order.customer_id}")

    if order.amount < 0:
        raise ValueError("Invalid order amount")  # triggers retry/DLX

# Production-ready setup with full retry cycles
mrsal = MrsalBlockingAMQP(
    host=RABBITMQ_DOMAIN,
    port=int(RABBITMQ_PORT),
    credentials=(RABBITMQ_USER, RABBITMQ_PASSWORD),
    virtual_host=RABBITMQ_VHOST,
    dlx_enable=True,         # Automatic DLX for failed orders
    use_quorum_queues=True,  # High availability
    prefetch_count=10        # Process up to 10 messages concurrently
)

mrsal.start_consumer(
    queue_name='orders_queue',
    exchange_name='orders_exchange',
    exchange_type='direct',
    routing_key='new_order',
    callback=process_order,
    payload_model=OrderMessage,        # Automatic validation
    auto_ack=False,                    # Manual ack for reliability
    auto_declare=True,                 # Auto-create exchange/queue/DLX
    
    # Advanced retry configuration
    enable_retry_cycles=True,          # Enable retry cycles
    retry_cycle_interval=15,           # 15 minutes between cycles
    max_retry_time_limit=120,          # 2 hours total retry time
    
    # Queue performance settings
    max_queue_length=50000,            # Handle large order volumes
    queue_overflow="reject-publish",   # Reject when full (backpressure)
    single_active_consumer=False       # Parallel processing for speed
)
```

**Note!** There are many parameters and settings that you can use to set up a more sophisticated communication protocol in both blocking or async connection with pydantic BaseModels to enforce data types in the expected payload.

#### 4.6 Metrics Hooks

mrsal can push operational signal to your own instrumentation without you parsing
log lines. Install a `MetricsHooks` set and mrsal calls the hooks you provide on
the publish / consume / DLX hot paths. This mirrors sonic's `types.MetricsHooks`.

mrsal bundles **no** metrics client — each hook is a plain callable, so you wire
it to whatever you already use (Prometheus, StatsD, or in-memory counters). The
broker's own `rabbitmq_prometheus` plugin covers *queue-level* metrics (depth,
rates); these hooks cover the *handler-level* signal it can't see.

```python
from mrsal.amqp.subclass import MrsalBlockingAMQP
from mrsal.metrics import MetricsHooks

mrsal = MrsalBlockingAMQP(host=..., port=5672, credentials=(...), virtual_host="/")

mrsal.set_metrics_hooks(MetricsHooks(
    # success=False on a NackError / UnroutableError / confirm timeout.
    on_publish=lambda success, duration_s: ...,
    # Fires once per delivery. success=False when payload validation OR the
    # callback failed; duration_s spans validation + callback.
    on_consume=lambda success, duration_s: ...,
    # Fires when a failed message is republished to <queue>.retry.
    on_retry=lambda cycle, delay_s: ...,
    # Fires when a message is parked in the terminal <queue>.dlx.
    on_dlx_final=lambda queue_name: ...,
))
```

Each hook is optional — pass only the ones you want; an unset hook costs a single
`None` check on the hot path. Pass `None` to `set_metrics_hooks` to clear the set.
`MrsalBlockingPublisherPool.set_metrics_hooks(...)` cascades the set to every
publisher it hands out.

**Contract — fast, non-blocking, no exceptions.** Hooks run synchronously where
the event happens, including inside the `threaded=True` executor, so a hook that
blocks or does I/O stalls message processing — increment a counter and let a
separate task export it. Make hooks thread-safe (e.g. `prometheus_client`
collectors already are). On exceptions, mrsal mirrors sonic: a raise in a
consume-path hook (`on_consume` / `on_retry` / `on_dlx_final`) is caught and
logged so it never kills the consumer, but a raise in `on_publish` is **not**
recovered: it propagates to the publish caller and, if the publish itself was
already failing, *supersedes* that error. Keep `on_publish` exception-free.

A worked example wiring these to `prometheus_client`:

```python
from prometheus_client import Counter, Histogram, start_http_server
from mrsal.metrics import MetricsHooks

consume_seconds = Histogram("mrsal_message_processing_seconds", "Handler wall-clock")
consumed_total = Counter("mrsal_messages_consumed_total", "Consumed messages", ["outcome"])
dlx_final_total = Counter("mrsal_messages_dlx_final_total", "Messages parked in terminal DLX")

def on_consume(success, duration_s):
    consume_seconds.observe(duration_s)
    consumed_total.labels(outcome="processed" if success else "failed").inc()

def on_dlx_final(queue_name):
    dlx_final_total.inc()

mrsal.set_metrics_hooks(MetricsHooks(on_consume=on_consume, on_dlx_final=on_dlx_final))
start_http_server(8000)  # expose /metrics — mrsal is a library, you own exposition
```

---

### 5. Testing handlers without RabbitMQ

`mrsal.testing` ships an in-memory broker so you can unit-test consumers and
publishers without a live RabbitMQ. It runs mrsal's real validate → callback →
ack/DLX code against an in-process broker (routing, DLX topology and retry
cycles included), so the tests exercise actual behaviour instead of mocks, and
they run in milliseconds with no infrastructure.

`TestMrsalBroker` wraps a `MrsalBlockingAMQP` instance, swaps its connection for
the in-memory one, and delivers each published message straight to your handler —
no background thread, so assertions are deterministic.

```python
from pydantic.dataclasses import dataclass

from mrsal.amqp.subclass import MrsalBlockingAMQP
from mrsal.testing import TestMrsalBroker


@dataclass
class OrderEvent:
    order_id: str
    amount: float


def test_order_handler():
    seen = []

    def handle_order(method_frame, properties, body):
        seen.append(body)  # body is the validated OrderEvent

    consumer = MrsalBlockingAMQP(
        host="localhost", port=5672, credentials=("guest", "guest"), virtual_host="/",
    )
    with TestMrsalBroker(consumer) as br:
        br.register_consumer(
            queue_name="orders",
            exchange_name="orders.exchange",
            exchange_type="direct",
            routing_key="orders.new",
            callback=handle_order,
            payload_model=OrderEvent,
        )
        br.publish(
            {"order_id": "abc", "amount": 10.0},
            exchange="orders.exchange",
            routing_key="orders.new",
        )

    assert seen[0].order_id == "abc"
    assert br.message_count("orders") == 0  # processed and acked
```

A payload that fails validation (or a callback that raises) routes through the
real DLX path. Time is modeled, not real — call `br.advance(minutes=...)` to fire
the `.retry` queue's TTL and drive the retry cycle deterministically.

> **Note on retry exhaustion.** `advance()` fires `.retry` queue TTLs (re-delivery),
> but mrsal computes retry-*budget* exhaustion (`max_retry_time_limit`) from real
> wall-clock time, so advancing the modeled clock re-cycles a message without ever
> exhausting it. To assert that a message lands in the **terminal** `.dlx` queue,
> force it with a low `max_retry_time_limit` (e.g. `0`) rather than advancing the
> clock and waiting for exhaustion.

```python
def test_invalid_payload_parks_in_dlx():
    consumer = MrsalBlockingAMQP(
        host="localhost", port=5672, credentials=("guest", "guest"), virtual_host="/",
    )
    with TestMrsalBroker(consumer) as br:
        br.register_consumer(
            queue_name="orders",
            exchange_name="orders.exchange",
            exchange_type="direct",
            routing_key="orders.new",
            callback=lambda *a: None,
            payload_model=OrderEvent,
            max_retry_time_limit=0,   # first failure parks straight in .dlx
        )
        br.publish({"bad": "data"}, exchange="orders.exchange", routing_key="orders.new")

    assert br.message_count("orders.dlx") == 1
```

The async consumer has a parallel harness, `TestMrsalAsyncBroker`, with
`await br.register_consumer(...)`, `await br.publish(...)` and
`await br.advance(...)`.

---

