Middleware

BaseMiddleware (base class)

BaseMiddleware

Base middleware with lifecycle hooks.

Hooks: - on_receive(msg): notification when message is received (before processing) - consume_scope(call_next, msg): wrap handler execution - after_processed(msg, exc): post-processing notification - publish_scope(call_next, envelope): wrap outgoing publish

All hooks have no-op defaults. Subclasses override what they need. Both sync and async variants provided.

Source code in src/rabbitkit/middleware/base.py

class BaseMiddleware:
    """Base middleware with lifecycle hooks.

    Hooks:
    - on_receive(msg): notification when message is received (before processing)
    - consume_scope(call_next, msg): wrap handler execution
    - after_processed(msg, exc): post-processing notification
    - publish_scope(call_next, envelope): wrap outgoing publish

    All hooks have no-op defaults. Subclasses override what they need.
    Both sync and async variants provided.
    """

    # ── Consume-side hooks ───────────────────────────────────────────────

    def on_receive(self, message: RabbitMessage) -> None:
        """Called when a message is received, before processing.

        H7 — two things every on_receive override must account for:

        1. Runs in a fixed, flat pre-pass entirely BEFORE consume_scope is
           entered for ANY middleware on the route. An exception raised here
           is NOT caught by any middleware's consume_scope — not even
           RetryMiddleware's — so it is never retry-eligible via the
           delay-queue mechanism; it settles per the route's AckPolicy using
           the pipeline's default classifier instead. If your check should be
           retryable, implement it in consume_scope, not here.
        2. Runs in REVERSE registration order — the mirror of publish_scope's
           outer→inner composition — so a receive-side transform that undoes
           a publish-side one (e.g. decompress undoing compress) runs in the
           mathematically correct relative order. This does NOT mean any two
           on_receive-based middlewares can be listed in either order and
           both work — e.g. SigningMiddleware + CompressionMiddleware only
           work as ``[CompressionMiddleware, SigningMiddleware]`` (compression
           outer), because the signature covers content_encoding, a field
           compression itself sets. See HandlerPipeline._run_consume_sync's
           docstring for the full explanation.
        """

    async def on_receive_async(self, message: RabbitMessage) -> None:
        """Async variant of on_receive."""
        self.on_receive(message)

    def consume_scope(
        self,
        call_next: Callable[[RabbitMessage], Any],
        message: RabbitMessage,
    ) -> Any:
        """Wrap handler execution (sync). Must call call_next(message)."""
        return call_next(message)

    async def consume_scope_async(
        self,
        call_next: Callable[[RabbitMessage], Awaitable[Any]],
        message: RabbitMessage,
    ) -> Any:
        """Wrap handler execution (async). Must call await call_next(message)."""
        return await call_next(message)

    def after_processed(
        self,
        message: RabbitMessage,
        exc: BaseException | None = None,
    ) -> None:
        """Called after message processing completes (success or failure)."""

    async def after_processed_async(
        self,
        message: RabbitMessage,
        exc: BaseException | None = None,
    ) -> None:
        """Async variant of after_processed."""
        self.after_processed(message, exc)

    # ── Publish-side hooks ───────────────────────────────────────────────

    def publish_scope(
        self,
        call_next: Callable[[MessageEnvelope], Any],
        envelope: MessageEnvelope,
    ) -> Any:
        """Wrap outgoing publish (sync). Must call call_next(envelope)."""
        return call_next(envelope)

    async def publish_scope_async(
        self,
        call_next: Callable[[MessageEnvelope], Awaitable[Any]],
        envelope: MessageEnvelope,
    ) -> Any:
        """Wrap outgoing publish (async). Must call await call_next(envelope)."""
        return await call_next(envelope)

Methods:

on_receive(message: RabbitMessage) -> None

Called when a message is received, before processing.

H7 — two things every on_receive override must account for:

1. Runs in a fixed, flat pre-pass entirely BEFORE consume_scope is entered for ANY middleware on the route. An exception raised here is NOT caught by any middleware's consume_scope — not even RetryMiddleware's — so it is never retry-eligible via the delay-queue mechanism; it settles per the route's AckPolicy using the pipeline's default classifier instead. If your check should be retryable, implement it in consume_scope, not here.
2. Runs in REVERSE registration order — the mirror of publish_scope's outer→inner composition — so a receive-side transform that undoes a publish-side one (e.g. decompress undoing compress) runs in the mathematically correct relative order. This does NOT mean any two on_receive-based middlewares can be listed in either order and both work — e.g. SigningMiddleware + CompressionMiddleware only work as [CompressionMiddleware, SigningMiddleware] (compression outer), because the signature covers content_encoding, a field compression itself sets. See HandlerPipeline._run_consume_sync's docstring for the full explanation.

Source code in src/rabbitkit/middleware/base.py

def on_receive(self, message: RabbitMessage) -> None:
    """Called when a message is received, before processing.

    H7 — two things every on_receive override must account for:

    1. Runs in a fixed, flat pre-pass entirely BEFORE consume_scope is
       entered for ANY middleware on the route. An exception raised here
       is NOT caught by any middleware's consume_scope — not even
       RetryMiddleware's — so it is never retry-eligible via the
       delay-queue mechanism; it settles per the route's AckPolicy using
       the pipeline's default classifier instead. If your check should be
       retryable, implement it in consume_scope, not here.
    2. Runs in REVERSE registration order — the mirror of publish_scope's
       outer→inner composition — so a receive-side transform that undoes
       a publish-side one (e.g. decompress undoing compress) runs in the
       mathematically correct relative order. This does NOT mean any two
       on_receive-based middlewares can be listed in either order and
       both work — e.g. SigningMiddleware + CompressionMiddleware only
       work as ``[CompressionMiddleware, SigningMiddleware]`` (compression
       outer), because the signature covers content_encoding, a field
       compression itself sets. See HandlerPipeline._run_consume_sync's
       docstring for the full explanation.
    """

on_receive_async(message: RabbitMessage) -> None async

Async variant of on_receive.

Source code in src/rabbitkit/middleware/base.py

async def on_receive_async(self, message: RabbitMessage) -> None:
    """Async variant of on_receive."""
    self.on_receive(message)

consume_scope(call_next: Callable[[RabbitMessage], Any], message: RabbitMessage) -> Any

Wrap handler execution (sync). Must call call_next(message).

Source code in src/rabbitkit/middleware/base.py

def consume_scope(
    self,
    call_next: Callable[[RabbitMessage], Any],
    message: RabbitMessage,
) -> Any:
    """Wrap handler execution (sync). Must call call_next(message)."""
    return call_next(message)

consume_scope_async(call_next: Callable[[RabbitMessage], Awaitable[Any]], message: RabbitMessage) -> Any async

Wrap handler execution (async). Must call await call_next(message).

Source code in src/rabbitkit/middleware/base.py

async def consume_scope_async(
    self,
    call_next: Callable[[RabbitMessage], Awaitable[Any]],
    message: RabbitMessage,
) -> Any:
    """Wrap handler execution (async). Must call await call_next(message)."""
    return await call_next(message)

after_processed(message: RabbitMessage, exc: BaseException | None = None) -> None

Called after message processing completes (success or failure).

Source code in src/rabbitkit/middleware/base.py

def after_processed(
    self,
    message: RabbitMessage,
    exc: BaseException | None = None,
) -> None:
    """Called after message processing completes (success or failure)."""

after_processed_async(message: RabbitMessage, exc: BaseException | None = None) -> None async

Async variant of after_processed.

Source code in src/rabbitkit/middleware/base.py

async def after_processed_async(
    self,
    message: RabbitMessage,
    exc: BaseException | None = None,
) -> None:
    """Async variant of after_processed."""
    self.after_processed(message, exc)

publish_scope(call_next: Callable[[MessageEnvelope], Any], envelope: MessageEnvelope) -> Any

Wrap outgoing publish (sync). Must call call_next(envelope).

Source code in src/rabbitkit/middleware/base.py

def publish_scope(
    self,
    call_next: Callable[[MessageEnvelope], Any],
    envelope: MessageEnvelope,
) -> Any:
    """Wrap outgoing publish (sync). Must call call_next(envelope)."""
    return call_next(envelope)

publish_scope_async(call_next: Callable[[MessageEnvelope], Awaitable[Any]], envelope: MessageEnvelope) -> Any async

Wrap outgoing publish (async). Must call await call_next(envelope).

Source code in src/rabbitkit/middleware/base.py

async def publish_scope_async(
    self,
    call_next: Callable[[MessageEnvelope], Awaitable[Any]],
    envelope: MessageEnvelope,
) -> Any:
    """Wrap outgoing publish (async). Must call await call_next(envelope)."""
    return await call_next(envelope)

NoOpMiddleware

Bases: BaseMiddleware

Null Object middleware — zero-overhead pass-through.

Use as a default when no middleware is configured, eliminating if collector is None: return call_next(...) branches on the hot path.

Source code in src/rabbitkit/middleware/base.py

class NoOpMiddleware(BaseMiddleware):
    """Null Object middleware — zero-overhead pass-through.

    Use as a default when no middleware is configured, eliminating
    ``if collector is None: return call_next(...)`` branches on the hot path.
    """

    def consume_scope(self, call_next: Any, message: Any) -> Any:
        return call_next(message)

    async def consume_scope_async(self, call_next: Any, message: Any) -> Any:
        return await call_next(message)

    def publish_scope(self, call_next: Any, envelope: Any) -> Any:
        return call_next(envelope)

    async def publish_scope_async(self, call_next: Any, envelope: Any) -> Any:
        return await call_next(envelope)

RetryMiddleware

RetryMiddleware

Bases: BaseMiddleware

Routes failed messages to delay queues for retry.

On exception: 1. Classify error (transient/permanent) 2. If transient + retries left → publish to delay queue + ack source 3. If permanent or retries exhausted → tag as terminal + re-raise

Source code in src/rabbitkit/middleware/retry.py

class RetryMiddleware(BaseMiddleware):
    """Routes failed messages to delay queues for retry.

    On exception:
    1. Classify error (transient/permanent)
    2. If transient + retries left → publish to delay queue + ack source
    3. If permanent or retries exhausted → tag as terminal + re-raise
    """

    def __init__(
        self,
        config: RetryConfig,
        *,
        publish_fn: Callable[[MessageEnvelope], Any] | None = None,
        publish_async_fn: Callable[[MessageEnvelope], Awaitable[Any]] | None = None,
        predicates: Sequence[ErrorPredicate] = (),
        metrics_collector: Any | None = None,
        metrics_config: Any | None = None,
    ) -> None:
        self._config = config
        # predicates run first (True=transient, False=permanent, None=defer to the
        # built-in type tuples, then unknown_policy). Lets callers classify by
        # something other than exception type (e.g. an HTTP status attribute).
        self._classifier = ErrorClassifierMiddleware(
            predicates=predicates,
            unknown_policy=config.unknown_policy,
        )
        self._publish_fn = publish_fn
        self._publish_async_fn = publish_async_fn
        # M2: optional -- wired by the broker from a MetricsMiddleware already
        # present on the same route, so retried/dead-lettered counts are
        # observable. None (the default) is a no-op; RetryMiddleware itself
        # has no metrics opinion otherwise.
        self._metrics_collector = metrics_collector
        self._metrics_config = metrics_config

    def ensure_publish_fns(
        self,
        publish_fn: Callable[[MessageEnvelope], Any] | None = None,
        publish_async_fn: Callable[[MessageEnvelope], Any] | None = None,
    ) -> None:
        """Fill in whichever publish function is not already set.

        Called by the broker at start() for USER-CONSTRUCTED RetryMiddleware
        instances found in ``middlewares=[...]``: previously such an instance
        with no publish fn silently fell into the ack-without-publish drop
        path on every transient failure (now a nack+RuntimeWarning, but still
        a hot loop). Injecting the broker's own confirmed-publish fn makes a
        manually-added ``RetryMiddleware(config)`` behave identically to the
        auto-wired one. Explicitly-passed fns are never overwritten.
        """
        if self._publish_fn is None and publish_fn is not None:
            self._publish_fn = publish_fn
        if self._publish_async_fn is None and publish_async_fn is not None:
            self._publish_async_fn = publish_async_fn

    def _record_metric(self, metric_name: str | None, message: RabbitMessage) -> None:
        if self._metrics_collector is None or metric_name is None:
            return
        queue = message.headers.get("x-rabbitkit-original-queue") or message.routing_key or "unknown"
        self._metrics_collector.inc_counter(metric_name, {"queue": str(queue)})

    @property
    def config(self) -> RetryConfig:
        return self._config

    def consume_scope(
        self,
        call_next: Callable[[RabbitMessage], Any],
        message: RabbitMessage,
    ) -> Any:
        """Sync retry scope.

        H8: on a caught, requeued failure, returns ``REQUEUED_FOR_RETRY``
        (never ``None``) — see that sentinel's docstring in ``core/types.py``.
        ``_handle_retry_sync`` either returns normally (requeued: routed to a
        delay queue, or nacked for immediate redelivery if that publish
        itself failed) or re-raises (terminal: permanent/exhausted) via
        ``_mark_terminal_and_raise``, so reaching this ``return`` unambiguously
        means "requeued" — an outer middleware (e.g. DeduplicationMiddleware)
        MUST NOT treat this the same as the handler actually succeeding.
        """
        try:
            return call_next(message)
        except Exception as exc:
            self._handle_retry_sync(exc, message)
            return REQUEUED_FOR_RETRY

    async def consume_scope_async(
        self,
        call_next: Callable[[RabbitMessage], Awaitable[Any]],
        message: RabbitMessage,
    ) -> Any:
        """Async retry scope. See :meth:`consume_scope` (H8) for why this
        returns ``REQUEUED_FOR_RETRY`` rather than ``None``."""
        try:
            return await call_next(message)
        except Exception as exc:
            await self._handle_retry_async(exc, message)
            return REQUEUED_FOR_RETRY

    def _handle_retry_sync(self, exc: Exception, message: RabbitMessage) -> None:
        """Handle exception in sync context."""
        classified = self._classifier.classify(exc)
        retry_count = self._get_retry_count(message)

        if classified.severity == ErrorSeverity.TRANSIENT and retry_count < self._config.max_retries:
            # Route to delay queue
            self._route_to_delay_queue_sync(message, retry_count, exc)
            return

        # Terminal: permanent or exhausted
        self._mark_terminal_and_raise(exc, classified.severity, retry_count, message)

    async def _handle_retry_async(self, exc: Exception, message: RabbitMessage) -> None:
        """Handle exception in async context."""
        classified = self._classifier.classify(exc)
        retry_count = self._get_retry_count(message)

        if classified.severity == ErrorSeverity.TRANSIENT and retry_count < self._config.max_retries:
            await self._route_to_delay_queue_async(message, retry_count, exc)
            return

        self._mark_terminal_and_raise(exc, classified.severity, retry_count, message)

    def _get_retry_count(self, message: RabbitMessage) -> int:
        """Get current retry count from message headers, clamped to
        ``[0, max_retries]`` (H5).

        The header is read verbatim from an inbound AMQP message — nothing
        distinguishes a value written by this middleware's own delay-queue
        round trip from one set directly by an untrusted producer (there is
        no broker-side attestation of provenance for a plain header). Without
        clamping, a producer could set it negative (``attempt = retry_count +
        1`` in :meth:`_build_retry_envelope` would then be <= 0, producing a
        delay-queue routing key like ``...retry.-4`` that was never declared
        — the retry publish silently targets a non-existent queue and the
        message is lost rather than retried) or absurdly large (forcing every
        message straight to the DLQ, skipping retries entirely). Clamping
        makes ``max_retries`` an enforced ceiling regardless of what the
        header claims, independent of its configured value being read from a
        trusted or untrusted source. A malformed (non-numeric) header value
        is treated the same as missing (0) rather than raising, so a garbage
        header degrades to "start of the retry sequence" instead of crashing
        the pipeline.

        For a broker-enforced backstop on top of this (e.g. against a
        misbehaving consumer that never expires/dead-letters a message),
        prefer quorum source queues with ``x-delivery-limit`` — see
        ``docs/retry-and-dlq.md``.
        """
        raw = message.headers.get(self._config.retry_header, 0)
        try:
            retry_count = int(raw)
        except (TypeError, ValueError):
            retry_count = 0
        return max(0, min(retry_count, self._config.max_retries))

    def _build_retry_envelope(self, message: RabbitMessage, retry_count: int, exc: Exception) -> MessageEnvelope:
        """Build envelope for the delay queue."""
        # Determine delay queue name
        attempt = retry_count + 1
        # Always per-queue (shared mode is rejected by RetryConfig — H3).
        source_queue = message.headers.get("x-rabbitkit-original-queue", "unknown")
        if self._config.jitter_mode == "sharded":
            shard = _shard_index(message.message_id or "", self._config.jitter_shards)
            delay_queue_rk = _shard_queue_name(str(source_queue), attempt, shard)
        else:
            delay_queue_rk = f"{source_queue}.retry.{attempt}"

        # Preserve original headers + add retry metadata
        headers = dict(message.headers)
        headers[self._config.retry_header] = retry_count + 1
        if "x-rabbitkit-original-exchange" not in headers:
            headers["x-rabbitkit-original-exchange"] = message.exchange
        if "x-rabbitkit-original-routing-key" not in headers:
            headers["x-rabbitkit-original-routing-key"] = message.routing_key
        if "x-rabbitkit-original-queue" not in headers:
            headers["x-rabbitkit-original-queue"] = ""  # set by broker at consume time

        # Item 7: DLQ-triage headers -- previously the only way to learn why
        # a message ended up on the DLQ was application logs, which may have
        # long since rotated out by the time anyone looks. error-type/message
        # describe THIS failure; first-failed-at is set once and preserved
        # across every subsequent retry (mirrors the original-* pattern
        # above); last-failed-at is overwritten on every retry.
        headers["x-rabbitkit-error-type"] = type(exc).__name__
        headers["x-rabbitkit-error-message"] = str(exc)[:_ERROR_MESSAGE_MAX_LEN]
        now = datetime.now(UTC).isoformat()
        if "x-rabbitkit-first-failed-at" not in headers:
            headers["x-rabbitkit-first-failed-at"] = now
        headers["x-rabbitkit-last-failed-at"] = now

        return MessageEnvelope(
            routing_key=delay_queue_rk,
            body=message.body,
            exchange="",  # direct to delay queue by name
            headers=headers,
            message_id=message.message_id or "",
            correlation_id=message.correlation_id,
            content_type=message.content_type or "application/octet-stream",
            content_encoding=message.content_encoding,
            # Preserve the remaining original message properties -- these used
            # to be silently dropped on every retry republish, so e.g. a
            # priority-queue message lost its priority on its first retry, and
            # an RPC request's reply_to/type/app_id/user_id never survived
            # long enough for the eventual (retried) reply to route back.
            reply_to=message.reply_to,
            priority=message.priority,
            # expiration is deliberately NOT preserved: a producer-set
            # per-message TTL shorter than the delay tier's x-message-ttl
            # would expire the message INSIDE the delay queue and dead-letter
            # it back to the source early, collapsing the whole backoff
            # ladder into a fast retry storm. The delay queue's own TTL is
            # the timing authority for the retry wait.
            type=message.type,
            app_id=message.app_id,
            user_id=message.user_id,
            # M4: mandatory so a runtime-deleted/missing delay queue comes back
            # as RETURNED (outcome not-ok) instead of being broker-confirmed
            # into the void. The route-to-delay-queue path checks outcome.ok
            # and nack-requeues on failure, so this turns silent loss into a
            # redelivery. (Requires publisher confirms + basic.return handling,
            # which both transports wire up.)
            mandatory=True,
        )

    def _route_to_delay_queue_sync(self, message: RabbitMessage, retry_count: int, exc: Exception) -> None:
        """Publish to delay queue and ack source (sync)."""
        envelope = self._build_retry_envelope(message, retry_count, exc)

        if self._publish_fn is None:
            # No sync publish fn wired (manual middleware construction, or a
            # RetryMiddleware carrying only publish_async_fn on a sync
            # broker). Acking here would DROP the message — nothing was ever
            # published to the delay queue. Nack for redelivery and warn
            # loudly; broker-wired routes never hit this (the broker injects
            # its own publish fn at start()).
            if not message.is_settled:
                message.nack(requeue=True)
            warnings.warn(
                "RetryMiddleware has no publish_fn wired for a sync route — the "
                "transient failure was nacked (requeued), NOT routed to a delay "
                "queue. Pass publish_fn=broker.publish, or set retry= on the "
                "route/broker and let rabbitkit wire it.",
                RuntimeWarning,
                stacklevel=2,
            )
            return

        outcome = self._publish_fn(envelope)
        # A None outcome (a custom/duck-typed publish fn that returns nothing)
        # is UNVERIFIED, not success — treat like a failed publish.
        if outcome is None or not outcome.ok:
            # Delay-queue publish failed — DO NOT ack, or the message is
            # lost forever (never retried, never dead-lettered). Nack with
            # requeue so the broker redelivers it.
            if not message.is_settled:
                message.nack(requeue=True)
            logger.warning(
                "Retry publish failed or unverified; nacked for redelivery: routing_key=%s",
                envelope.routing_key,
            )
            return

        # Ack source message (it's safely in the delay queue now)
        if not message.is_settled:
            message.ack()

        if self._metrics_config is not None:
            self._record_metric(self._metrics_config.messages_retried_total, message)

        logger.info(
            "Retrying message (attempt %d/%d): routing_key=%s",
            retry_count + 1,
            self._config.max_retries,
            envelope.routing_key,
        )

    async def _route_to_delay_queue_async(self, message: RabbitMessage, retry_count: int, exc: Exception) -> None:
        """Publish to delay queue and ack source (async)."""
        envelope = self._build_retry_envelope(message, retry_count, exc)

        if self._publish_async_fn is None:
            # See the sync variant: acking without a publish would DROP the
            # message. Nack for redelivery and warn loudly.
            if not message.is_settled:
                await message.nack_async(requeue=True)
            warnings.warn(
                "RetryMiddleware has no publish_async_fn wired for an async route — "
                "the transient failure was nacked (requeued), NOT routed to a delay "
                "queue. Pass publish_async_fn=broker.publish, or set retry= on the "
                "route/broker and let rabbitkit wire it.",
                RuntimeWarning,
                stacklevel=2,
            )
            return

        outcome = await self._publish_async_fn(envelope)
        # None outcome = unverified publish = failure (see sync variant).
        if outcome is None or not outcome.ok:
            # Delay-queue publish failed — DO NOT ack (see sync variant).
            if not message.is_settled:
                await message.nack_async(requeue=True)
            logger.warning(
                "Retry publish failed or unverified; nacked for redelivery: routing_key=%s",
                envelope.routing_key,
            )
            return

        if not message.is_settled:
            await message.ack_async()

        if self._metrics_config is not None:
            self._record_metric(self._metrics_config.messages_retried_total, message)

        logger.info(
            "Retrying message (attempt %d/%d): routing_key=%s",
            retry_count + 1,
            self._config.max_retries,
            envelope.routing_key,
        )

    def _mark_terminal_and_raise(
        self,
        exc: Exception,
        severity: ErrorSeverity,
        retry_count: int,
        message: RabbitMessage,
    ) -> None:
        """Mark exception as terminal and re-raise.

        M2: this is the point where a message is committed to being
        dead-lettered -- permanent errors dead-letter on the first attempt,
        exhausted-retry errors dead-letter after ``max_retries`` -- so
        ``messages_dead_lettered_total`` is recorded here rather than at the
        actual reject() call (which happens later, in the pipeline's
        exception handling, and doesn't know WHY the reject is happening).
        """
        exc._rabbitkit_terminal = True  # type: ignore[attr-defined]
        if self._metrics_config is not None:
            self._record_metric(self._metrics_config.messages_dead_lettered_total, message)
        logger.warning(
            "Terminal failure (%s, retries=%d/%d): %s: %s",
            severity.value,
            retry_count,
            self._config.max_retries,
            type(exc).__name__,
            exc,
        )
        raise

Methods:

ensure_publish_fns(publish_fn: Callable[[MessageEnvelope], Any] | None = None, publish_async_fn: Callable[[MessageEnvelope], Any] | None = None) -> None

Fill in whichever publish function is not already set.

Called by the broker at start() for USER-CONSTRUCTED RetryMiddleware instances found in middlewares=[...]: previously such an instance with no publish fn silently fell into the ack-without-publish drop path on every transient failure (now a nack+RuntimeWarning, but still a hot loop). Injecting the broker's own confirmed-publish fn makes a manually-added RetryMiddleware(config) behave identically to the auto-wired one. Explicitly-passed fns are never overwritten.

Source code in src/rabbitkit/middleware/retry.py

def ensure_publish_fns(
    self,
    publish_fn: Callable[[MessageEnvelope], Any] | None = None,
    publish_async_fn: Callable[[MessageEnvelope], Any] | None = None,
) -> None:
    """Fill in whichever publish function is not already set.

    Called by the broker at start() for USER-CONSTRUCTED RetryMiddleware
    instances found in ``middlewares=[...]``: previously such an instance
    with no publish fn silently fell into the ack-without-publish drop
    path on every transient failure (now a nack+RuntimeWarning, but still
    a hot loop). Injecting the broker's own confirmed-publish fn makes a
    manually-added ``RetryMiddleware(config)`` behave identically to the
    auto-wired one. Explicitly-passed fns are never overwritten.
    """
    if self._publish_fn is None and publish_fn is not None:
        self._publish_fn = publish_fn
    if self._publish_async_fn is None and publish_async_fn is not None:
        self._publish_async_fn = publish_async_fn

consume_scope(call_next: Callable[[RabbitMessage], Any], message: RabbitMessage) -> Any

Sync retry scope.

H8: on a caught, requeued failure, returns REQUEUED_FOR_RETRY (never None) — see that sentinel's docstring in core/types.py. _handle_retry_sync either returns normally (requeued: routed to a delay queue, or nacked for immediate redelivery if that publish itself failed) or re-raises (terminal: permanent/exhausted) via _mark_terminal_and_raise, so reaching this return unambiguously means "requeued" — an outer middleware (e.g. DeduplicationMiddleware) MUST NOT treat this the same as the handler actually succeeding.

Source code in src/rabbitkit/middleware/retry.py

def consume_scope(
    self,
    call_next: Callable[[RabbitMessage], Any],
    message: RabbitMessage,
) -> Any:
    """Sync retry scope.

    H8: on a caught, requeued failure, returns ``REQUEUED_FOR_RETRY``
    (never ``None``) — see that sentinel's docstring in ``core/types.py``.
    ``_handle_retry_sync`` either returns normally (requeued: routed to a
    delay queue, or nacked for immediate redelivery if that publish
    itself failed) or re-raises (terminal: permanent/exhausted) via
    ``_mark_terminal_and_raise``, so reaching this ``return`` unambiguously
    means "requeued" — an outer middleware (e.g. DeduplicationMiddleware)
    MUST NOT treat this the same as the handler actually succeeding.
    """
    try:
        return call_next(message)
    except Exception as exc:
        self._handle_retry_sync(exc, message)
        return REQUEUED_FOR_RETRY

consume_scope_async(call_next: Callable[[RabbitMessage], Awaitable[Any]], message: RabbitMessage) -> Any async

Async retry scope. See :meth:consume_scope (H8) for why this returns REQUEUED_FOR_RETRY rather than None.

Source code in src/rabbitkit/middleware/retry.py

async def consume_scope_async(
    self,
    call_next: Callable[[RabbitMessage], Awaitable[Any]],
    message: RabbitMessage,
) -> Any:
    """Async retry scope. See :meth:`consume_scope` (H8) for why this
    returns ``REQUEUED_FOR_RETRY`` rather than ``None``."""
    try:
        return await call_next(message)
    except Exception as exc:
        await self._handle_retry_async(exc, message)
        return REQUEUED_FOR_RETRY

RetryRouter

Declares delay queue topology at startup.

Called by broker.start() for each route that has retry enabled. RetryRouter is the SINGLE OWNER of all retry/DLQ topology for a route.

DLQ routing: - The source queue is re-declared with x-dead-letter-exchange="" and x-dead-letter-routing-key=<dlq_name> so that messages rejected/nacked with requeue=False are automatically routed to the DLQ by RabbitMQ — no application-level routing needed. - Use get_source_queue_dlq_arguments() to obtain the extra arguments that must be added to the source queue declaration.

Source code in src/rabbitkit/middleware/retry.py

class RetryRouter:
    """Declares delay queue topology at startup.

    Called by broker.start() for each route that has retry enabled.
    RetryRouter is the SINGLE OWNER of all retry/DLQ topology for a route.

    DLQ routing:
    - The source queue is re-declared with ``x-dead-letter-exchange=""``
      and ``x-dead-letter-routing-key=<dlq_name>`` so that messages
      rejected/nacked with ``requeue=False`` are automatically routed to
      the DLQ by RabbitMQ — no application-level routing needed.
    - Use ``get_source_queue_dlq_arguments()`` to obtain the extra arguments
      that must be added to the source queue declaration.
    """

    def __init__(self, config: RetryConfig) -> None:
        self._config = config

    def get_dlq_name(self, source_queue_name: str) -> str:
        """Return the DLQ name for a given source queue.

        Always per-queue — shared mode (per_queue=False) is rejected by
        RetryConfig (H3), so there is no shared-DLQ branch.
        """
        return f"{source_queue_name}.dlq"

    def get_source_queue_dlq_arguments(self, source_queue_name: str) -> dict[str, str]:
        """Return x-dead-letter arguments to add to the source queue declaration.

        When these arguments are present on the source queue, RabbitMQ
        automatically forwards messages that are rejected/nacked with
        requeue=False to the DLQ — making the DLQ actually reachable.
        """
        dlq_name = self.get_dlq_name(source_queue_name)
        return {
            "x-dead-letter-exchange": "",          # default exchange
            "x-dead-letter-routing-key": dlq_name, # route directly by queue name
        }

    def get_delay_queue_definitions(
        self,
        source_queue_name: str,
        source_exchange_name: str,  # kept for signature stability (M5) — see docstring
        *,
        source_queue_type: QueueType | None = None,
    ) -> list[RabbitQueue]:
        """Generate delay queue definitions for a source queue.

        Returns list of RabbitQueue objects for delay queues + DLQ.
        The DLQ is now reachable because ``get_source_queue_dlq_arguments()``
        wires the source queue's x-dead-letter-exchange to it.

        M5: on TTL expiry, a delay queue dead-letters back to the SOURCE
        QUEUE via the **default exchange** (``x-dead-letter-exchange=""``)
        with the queue's own name as the routing key — never the source
        queue's real exchange. On the default exchange a routing key that
        matches a queue's name always delivers directly to that queue,
        completely independent of how the queue is actually bound elsewhere.
        The previous version dead-lettered to ``source_exchange_name``
        using ``source_queue_name`` as the routing key — for a source queue
        bound to its real exchange via a topic pattern (e.g.
        ``orders.*.created``) rather than literally by its own name, that
        routing key almost never matches the binding, so the retried
        message silently vanished instead of coming back after the delay.
        ``source_exchange_name`` is intentionally unused now — kept as a
        parameter so existing call sites don't need updating.
        """
        queues: list[RabbitQueue] = []

        if self._config.jitter_mode == "sharded":
            multipliers = _shard_ttl_multipliers(self._config.jitter_shards, self._config.jitter_factor)
        else:
            multipliers = [1.0]

        for attempt in range(1, self._config.max_retries + 1):
            delay_ms = self._get_delay_ms(attempt - 1)

            # Always per-queue (shared mode is rejected by RetryConfig — H3).
            for shard, mult in enumerate(multipliers):
                queue = RabbitQueue(
                    name=_shard_queue_name(source_queue_name, attempt, shard),
                    durable=True,
                    arguments={
                        # Uniform per-queue TTL (head-of-line safety); shards
                        # stagger TTLs ACROSS queues, never within one (F4).
                        "x-message-ttl": max(1, int(delay_ms * mult)),
                        "x-dead-letter-exchange": "",  # default exchange (M5)
                        "x-dead-letter-routing-key": source_queue_name,
                        # Delay queues stay CLASSIC deliberately, even for a
                        # quorum source: they hold messages only for the delay
                        # window (bounded seconds/minutes), have no consumers,
                        # and quorum queues historically had weaker per-message
                        # TTL behavior. The DLQ below is different — it stores
                        # messages indefinitely, so it inherits quorum.
                        "x-queue-type": "classic",
                    },
                )
                queues.append(queue)

        # DLQ — declared as a plain durable queue.
        # The source queue's x-dead-letter-exchange (set via
        # get_source_queue_dlq_arguments) routes nacked/rejected messages here.
        dlq_name = self.get_dlq_name(source_queue_name)
        # The DLQ inherits QUORUM from a quorum source: it stores failed
        # messages indefinitely — the exact data the replicated source queue
        # was chosen to protect. Leaving it classic silently downgrades those
        # messages to unreplicated single-node storage. (Streams are not
        # inherited: basic_get-based DLQ inspection/replay doesn't apply to
        # stream semantics.)
        dlq = RabbitQueue(
            name=dlq_name,
            durable=True,
            queue_type=QueueType.QUORUM if source_queue_type == QueueType.QUORUM else QueueType.CLASSIC,
        )
        queues.append(dlq)

        return queues

    def _get_delay_ms(self, index: int) -> int:
        """Get delay in milliseconds for retry attempt."""
        delays = self._config.delays
        idx = min(index, len(delays) - 1)
        return delays[idx] * 1000

Methods:

get_dlq_name(source_queue_name: str) -> str

Return the DLQ name for a given source queue.

Always per-queue — shared mode (per_queue=False) is rejected by RetryConfig (H3), so there is no shared-DLQ branch.

Source code in src/rabbitkit/middleware/retry.py

def get_dlq_name(self, source_queue_name: str) -> str:
    """Return the DLQ name for a given source queue.

    Always per-queue — shared mode (per_queue=False) is rejected by
    RetryConfig (H3), so there is no shared-DLQ branch.
    """
    return f"{source_queue_name}.dlq"

get_source_queue_dlq_arguments(source_queue_name: str) -> dict[str, str]

Return x-dead-letter arguments to add to the source queue declaration.

When these arguments are present on the source queue, RabbitMQ automatically forwards messages that are rejected/nacked with requeue=False to the DLQ — making the DLQ actually reachable.

Source code in src/rabbitkit/middleware/retry.py

def get_source_queue_dlq_arguments(self, source_queue_name: str) -> dict[str, str]:
    """Return x-dead-letter arguments to add to the source queue declaration.

    When these arguments are present on the source queue, RabbitMQ
    automatically forwards messages that are rejected/nacked with
    requeue=False to the DLQ — making the DLQ actually reachable.
    """
    dlq_name = self.get_dlq_name(source_queue_name)
    return {
        "x-dead-letter-exchange": "",          # default exchange
        "x-dead-letter-routing-key": dlq_name, # route directly by queue name
    }

get_delay_queue_definitions(source_queue_name: str, source_exchange_name: str, *, source_queue_type: QueueType | None = None) -> list[RabbitQueue]

Generate delay queue definitions for a source queue.

Returns list of RabbitQueue objects for delay queues + DLQ. The DLQ is now reachable because get_source_queue_dlq_arguments() wires the source queue's x-dead-letter-exchange to it.

M5: on TTL expiry, a delay queue dead-letters back to the SOURCE QUEUE via the default exchange (x-dead-letter-exchange="") with the queue's own name as the routing key — never the source queue's real exchange. On the default exchange a routing key that matches a queue's name always delivers directly to that queue, completely independent of how the queue is actually bound elsewhere. The previous version dead-lettered to source_exchange_name using source_queue_name as the routing key — for a source queue bound to its real exchange via a topic pattern (e.g. orders.*.created) rather than literally by its own name, that routing key almost never matches the binding, so the retried message silently vanished instead of coming back after the delay. source_exchange_name is intentionally unused now — kept as a parameter so existing call sites don't need updating.

Source code in src/rabbitkit/middleware/retry.py

def get_delay_queue_definitions(
    self,
    source_queue_name: str,
    source_exchange_name: str,  # kept for signature stability (M5) — see docstring
    *,
    source_queue_type: QueueType | None = None,
) -> list[RabbitQueue]:
    """Generate delay queue definitions for a source queue.

    Returns list of RabbitQueue objects for delay queues + DLQ.
    The DLQ is now reachable because ``get_source_queue_dlq_arguments()``
    wires the source queue's x-dead-letter-exchange to it.

    M5: on TTL expiry, a delay queue dead-letters back to the SOURCE
    QUEUE via the **default exchange** (``x-dead-letter-exchange=""``)
    with the queue's own name as the routing key — never the source
    queue's real exchange. On the default exchange a routing key that
    matches a queue's name always delivers directly to that queue,
    completely independent of how the queue is actually bound elsewhere.
    The previous version dead-lettered to ``source_exchange_name``
    using ``source_queue_name`` as the routing key — for a source queue
    bound to its real exchange via a topic pattern (e.g.
    ``orders.*.created``) rather than literally by its own name, that
    routing key almost never matches the binding, so the retried
    message silently vanished instead of coming back after the delay.
    ``source_exchange_name`` is intentionally unused now — kept as a
    parameter so existing call sites don't need updating.
    """
    queues: list[RabbitQueue] = []

    if self._config.jitter_mode == "sharded":
        multipliers = _shard_ttl_multipliers(self._config.jitter_shards, self._config.jitter_factor)
    else:
        multipliers = [1.0]

    for attempt in range(1, self._config.max_retries + 1):
        delay_ms = self._get_delay_ms(attempt - 1)

        # Always per-queue (shared mode is rejected by RetryConfig — H3).
        for shard, mult in enumerate(multipliers):
            queue = RabbitQueue(
                name=_shard_queue_name(source_queue_name, attempt, shard),
                durable=True,
                arguments={
                    # Uniform per-queue TTL (head-of-line safety); shards
                    # stagger TTLs ACROSS queues, never within one (F4).
                    "x-message-ttl": max(1, int(delay_ms * mult)),
                    "x-dead-letter-exchange": "",  # default exchange (M5)
                    "x-dead-letter-routing-key": source_queue_name,
                    # Delay queues stay CLASSIC deliberately, even for a
                    # quorum source: they hold messages only for the delay
                    # window (bounded seconds/minutes), have no consumers,
                    # and quorum queues historically had weaker per-message
                    # TTL behavior. The DLQ below is different — it stores
                    # messages indefinitely, so it inherits quorum.
                    "x-queue-type": "classic",
                },
            )
            queues.append(queue)

    # DLQ — declared as a plain durable queue.
    # The source queue's x-dead-letter-exchange (set via
    # get_source_queue_dlq_arguments) routes nacked/rejected messages here.
    dlq_name = self.get_dlq_name(source_queue_name)
    # The DLQ inherits QUORUM from a quorum source: it stores failed
    # messages indefinitely — the exact data the replicated source queue
    # was chosen to protect. Leaving it classic silently downgrades those
    # messages to unreplicated single-node storage. (Streams are not
    # inherited: basic_get-based DLQ inspection/replay doesn't apply to
    # stream semantics.)
    dlq = RabbitQueue(
        name=dlq_name,
        durable=True,
        queue_type=QueueType.QUORUM if source_queue_type == QueueType.QUORUM else QueueType.CLASSIC,
    )
    queues.append(dlq)

    return queues

CompressionMiddleware

CompressionMiddleware

Bases: BaseMiddleware

Envelope/body transformation for compression.

Operates on MessageEnvelope (publish) and RabbitMessage.body (consume). NOT a handler-wrapping middleware — transforms data before/after serialize.

zstd contexts are not thread-safe, so a threading.local() holds a per-thread ZstdCompressor/ZstdDecompressor (created lazily), giving concurrent workers isolated contexts without a global lock. The decompressed size is capped via streaming decompression that aborts as soon as the running total exceeds max_decompressed_size (zip-bomb guard).

Source code in src/rabbitkit/middleware/compression.py

class CompressionMiddleware(BaseMiddleware):
    """Envelope/body transformation for compression.

    Operates on MessageEnvelope (publish) and RabbitMessage.body (consume).
    NOT a handler-wrapping middleware — transforms data before/after serialize.

    zstd contexts are **not** thread-safe, so a ``threading.local()`` holds a
    per-thread ``ZstdCompressor``/``ZstdDecompressor`` (created lazily), giving
    concurrent workers isolated contexts without a global lock. The
    decompressed size is capped via **streaming** decompression that aborts as
    soon as the running total exceeds ``max_decompressed_size`` (zip-bomb guard).
    """

    def __init__(
        self,
        config: CompressionConfig | None = None,
        *,
        max_decompressed_size: int = 64 * 1024 * 1024,
    ) -> None:
        self._config = config or CompressionConfig()
        self._max_decompressed_size = max_decompressed_size
        # Per-thread zstd contexts (zstandard contexts are not thread-safe).
        self._zstd_local = threading.local()

    def _get_cctx(self) -> Any:
        cctx = getattr(self._zstd_local, "cctx", None)
        if cctx is None:
            zstd = _get_zstd()
            cctx = zstd.ZstdCompressor(level=self._config.level)
            self._zstd_local.cctx = cctx
        return cctx

    def _get_dctx(self) -> Any:
        dctx = getattr(self._zstd_local, "dctx", None)
        if dctx is None:
            zstd = _get_zstd()
            try:
                dctx = zstd.ZstdDecompressor(max_window_size=self._max_decompressed_size)
            except TypeError:  # older zstandard has no max_window_size
                dctx = zstd.ZstdDecompressor()
            self._zstd_local.dctx = dctx
        return dctx

    def compress(self, data: bytes) -> tuple[bytes, str | None]:
        """Compress data if above threshold.

        Returns (compressed_data, content_encoding) or (original_data, None).
        """
        if len(data) < self._config.threshold:
            return data, None

        algorithm = self._config.algorithm
        if algorithm == "gzip":
            compressed = gzip.compress(data, compresslevel=self._config.level)
            return compressed, "gzip"
        elif algorithm == "zstd":
            cctx = self._get_cctx()
            compressed = cctx.compress(data)
            return compressed, "zstd"
        else:
            raise ValueError(f"Unknown compression algorithm: {algorithm}")

    def _decompress_gzip_streaming(self, data: bytes) -> bytes:
        """Streaming gzip decompression that aborts at the size cap.

        Uses ``zlib.decompressobj(16 + MAX_WBITS)`` and feeds ``data`` through
        ``decompress(..., _CHUNK)``. Because ``max_length`` limits *output*
        (not input consumed), the unconsumed input is re-fed via
        ``unconsumed_tail`` each iteration; the running total is checked every
        chunk — a zip bomb raises before a huge allocation is materialised.
        """
        decomp = zlib.decompressobj(16 + zlib.MAX_WBITS)
        out = bytearray()
        tail: bytes = data
        while True:
            chunk = decomp.decompress(tail, _CHUNK)
            out += chunk
            if len(out) > self._max_decompressed_size:
                raise ValueError(
                    f"Decompressed size ({len(out)}) exceeds max_decompressed_size ({self._max_decompressed_size})"
                )
            tail = decomp.unconsumed_tail
            if decomp.eof:
                break
            if not chunk and not tail:
                # No output and no input left to feed — avoid spinning.
                break
        out += decomp.flush()
        if len(out) > self._max_decompressed_size:
            raise ValueError(
                f"Decompressed size ({len(out)}) exceeds max_decompressed_size ({self._max_decompressed_size})"
            )
        return bytes(out)

    def _decompress_zstd_streaming(self, data: bytes) -> bytes:
        """Streaming zstd decompression that aborts at the size cap.

        ``max_window_size`` on the decompressor rejects frames whose window
        exceeds the cap (raising ``ZstdError`` before allocating); the streaming
        ``read`` loop enforces the running-total cap for high-ratio payloads. Both
        are surfaced as ``ValueError`` so callers see a single, consistent
        zip-bomb guard.
        """
        zstd = _get_zstd()
        dctx = self._get_dctx()
        reader = dctx.stream_reader(io.BytesIO(data))
        out = bytearray()
        try:
            while True:
                chunk = reader.read(_CHUNK)
                if not chunk:
                    break
                out += chunk
                if len(out) > self._max_decompressed_size:
                    raise ValueError(
                        f"Decompressed size ({len(out)}) exceeds max_decompressed_size ({self._max_decompressed_size})"
                    )
        except zstd.ZstdError as exc:
            # Frame too large for the configured window, or other decode error —
            # treat as a zip-bomb / oversized-payload rejection.
            raise ValueError(
                f"Decompressed size exceeds max_decompressed_size ({self._max_decompressed_size})"
            ) from exc
        return bytes(out)

    def decompress(self, data: bytes, content_encoding: str | None) -> bytes:
        """Decompress data based on content_encoding header.

        Raises ``ValueError`` as soon as the running decompressed total exceeds
        ``max_decompressed_size`` (streaming zip-bomb guard).
        """
        if content_encoding is None:
            return data

        if content_encoding == "gzip":
            return self._decompress_gzip_streaming(data)
        elif content_encoding == "zstd":
            return self._decompress_zstd_streaming(data)
        else:
            logger.warning("Unknown content_encoding: %s, returning raw data", content_encoding)
            return data

    def on_receive(self, message: RabbitMessage) -> None:
        """Decompress incoming message body if content_encoding is set."""
        if message.content_encoding:
            message.body = self.decompress(message.body, message.content_encoding)

    async def on_receive_async(self, message: RabbitMessage) -> None:
        """Async variant — offloads decompression to a worker thread.

        Offload whenever there is decompression work to do (content_encoding is
        set) OR the body is large, to avoid inline-decompressing a
        small-on-the-wire bomb on the event loop.
        """
        if not message.content_encoding:
            return
        # content_encoding is set: always offload to a worker thread. A
        # small-on-the-wire zip bomb can still produce a large decompressed
        # body, so never inline-decompress on the event loop.
        message.body = await asyncio.to_thread(self.decompress, message.body, message.content_encoding)

    def publish_scope(self, call_next: Any, envelope: MessageEnvelope) -> Any:
        """Compress the envelope body before publishing (sync).

        C4: without this override, ``transform_envelope`` had no caller —
        attaching ``CompressionMiddleware`` to a route or a broker's
        ``middlewares=[...]`` compressed nothing.
        """
        return call_next(self.transform_envelope(envelope))

    async def publish_scope_async(self, call_next: Any, envelope: MessageEnvelope) -> Any:
        """Async variant — compress the envelope body before publishing.

        Compression runs inline (not offloaded to a worker thread) — unlike
        ``on_receive_async``'s decompression, the body size here is caller-
        controlled, not attacker-controlled, so there is no zip-bomb-style
        amplification risk to guard against by offloading.
        """
        return await call_next(self.transform_envelope(envelope))

    def transform_envelope(self, envelope: MessageEnvelope) -> MessageEnvelope:
        """Compress envelope body and set content_encoding header."""
        compressed, encoding = self.compress(envelope.body)
        if encoding is not None:
            # Create new envelope with compressed body and encoding
            headers = dict(envelope.headers)
            return MessageEnvelope(
                routing_key=envelope.routing_key,
                body=compressed,
                exchange=envelope.exchange,
                headers=headers,
                message_id=envelope.message_id,
                correlation_id=envelope.correlation_id,
                reply_to=envelope.reply_to,
                timestamp=envelope.timestamp,
                content_type=envelope.content_type,
                content_encoding=encoding,
                expiration=envelope.expiration,
                priority=envelope.priority,
                mandatory=envelope.mandatory,
                delivery_mode=envelope.delivery_mode,
                type=envelope.type,
                user_id=envelope.user_id,
                app_id=envelope.app_id,
            )
        return envelope

Methods:

compress(data: bytes) -> tuple[bytes, str | None]

Compress data if above threshold.

Returns (compressed_data, content_encoding) or (original_data, None).

Source code in src/rabbitkit/middleware/compression.py

def compress(self, data: bytes) -> tuple[bytes, str | None]:
    """Compress data if above threshold.

    Returns (compressed_data, content_encoding) or (original_data, None).
    """
    if len(data) < self._config.threshold:
        return data, None

    algorithm = self._config.algorithm
    if algorithm == "gzip":
        compressed = gzip.compress(data, compresslevel=self._config.level)
        return compressed, "gzip"
    elif algorithm == "zstd":
        cctx = self._get_cctx()
        compressed = cctx.compress(data)
        return compressed, "zstd"
    else:
        raise ValueError(f"Unknown compression algorithm: {algorithm}")

decompress(data: bytes, content_encoding: str | None) -> bytes

Decompress data based on content_encoding header.

Raises ValueError as soon as the running decompressed total exceeds max_decompressed_size (streaming zip-bomb guard).

Source code in src/rabbitkit/middleware/compression.py

def decompress(self, data: bytes, content_encoding: str | None) -> bytes:
    """Decompress data based on content_encoding header.

    Raises ``ValueError`` as soon as the running decompressed total exceeds
    ``max_decompressed_size`` (streaming zip-bomb guard).
    """
    if content_encoding is None:
        return data

    if content_encoding == "gzip":
        return self._decompress_gzip_streaming(data)
    elif content_encoding == "zstd":
        return self._decompress_zstd_streaming(data)
    else:
        logger.warning("Unknown content_encoding: %s, returning raw data", content_encoding)
        return data

on_receive(message: RabbitMessage) -> None

Decompress incoming message body if content_encoding is set.

Source code in src/rabbitkit/middleware/compression.py

def on_receive(self, message: RabbitMessage) -> None:
    """Decompress incoming message body if content_encoding is set."""
    if message.content_encoding:
        message.body = self.decompress(message.body, message.content_encoding)

on_receive_async(message: RabbitMessage) -> None async

Async variant — offloads decompression to a worker thread.

Offload whenever there is decompression work to do (content_encoding is set) OR the body is large, to avoid inline-decompressing a small-on-the-wire bomb on the event loop.

Source code in src/rabbitkit/middleware/compression.py

async def on_receive_async(self, message: RabbitMessage) -> None:
    """Async variant — offloads decompression to a worker thread.

    Offload whenever there is decompression work to do (content_encoding is
    set) OR the body is large, to avoid inline-decompressing a
    small-on-the-wire bomb on the event loop.
    """
    if not message.content_encoding:
        return
    # content_encoding is set: always offload to a worker thread. A
    # small-on-the-wire zip bomb can still produce a large decompressed
    # body, so never inline-decompress on the event loop.
    message.body = await asyncio.to_thread(self.decompress, message.body, message.content_encoding)

publish_scope(call_next: Any, envelope: MessageEnvelope) -> Any

Compress the envelope body before publishing (sync).

C4: without this override, transform_envelope had no caller — attaching CompressionMiddleware to a route or a broker's middlewares=[...] compressed nothing.

Source code in src/rabbitkit/middleware/compression.py

def publish_scope(self, call_next: Any, envelope: MessageEnvelope) -> Any:
    """Compress the envelope body before publishing (sync).

    C4: without this override, ``transform_envelope`` had no caller —
    attaching ``CompressionMiddleware`` to a route or a broker's
    ``middlewares=[...]`` compressed nothing.
    """
    return call_next(self.transform_envelope(envelope))

publish_scope_async(call_next: Any, envelope: MessageEnvelope) -> Any async

Async variant — compress the envelope body before publishing.

Compression runs inline (not offloaded to a worker thread) — unlike on_receive_async's decompression, the body size here is caller- controlled, not attacker-controlled, so there is no zip-bomb-style amplification risk to guard against by offloading.

Source code in src/rabbitkit/middleware/compression.py

async def publish_scope_async(self, call_next: Any, envelope: MessageEnvelope) -> Any:
    """Async variant — compress the envelope body before publishing.

    Compression runs inline (not offloaded to a worker thread) — unlike
    ``on_receive_async``'s decompression, the body size here is caller-
    controlled, not attacker-controlled, so there is no zip-bomb-style
    amplification risk to guard against by offloading.
    """
    return await call_next(self.transform_envelope(envelope))

transform_envelope(envelope: MessageEnvelope) -> MessageEnvelope

Compress envelope body and set content_encoding header.

Source code in src/rabbitkit/middleware/compression.py

def transform_envelope(self, envelope: MessageEnvelope) -> MessageEnvelope:
    """Compress envelope body and set content_encoding header."""
    compressed, encoding = self.compress(envelope.body)
    if encoding is not None:
        # Create new envelope with compressed body and encoding
        headers = dict(envelope.headers)
        return MessageEnvelope(
            routing_key=envelope.routing_key,
            body=compressed,
            exchange=envelope.exchange,
            headers=headers,
            message_id=envelope.message_id,
            correlation_id=envelope.correlation_id,
            reply_to=envelope.reply_to,
            timestamp=envelope.timestamp,
            content_type=envelope.content_type,
            content_encoding=encoding,
            expiration=envelope.expiration,
            priority=envelope.priority,
            mandatory=envelope.mandatory,
            delivery_mode=envelope.delivery_mode,
            type=envelope.type,
            user_id=envelope.user_id,
            app_id=envelope.app_id,
        )
    return envelope

SigningMiddleware

SigningMiddleware

Bases: BaseMiddleware

Signs outgoing messages and verifies incoming signatures.

On publish: computes HMAC of body and adds signature to headers. On receive: verifies signature against body, rejects if invalid.

Source code in src/rabbitkit/middleware/signing.py

class SigningMiddleware(BaseMiddleware):
    """Signs outgoing messages and verifies incoming signatures.

    On publish: computes HMAC of body and adds signature to headers.
    On receive: verifies signature against body, rejects if invalid.
    """

    # ── Timestamp / nonce header names (replay protection) ──────────────
    _TIMESTAMP_HEADER = "x-rabbitkit-sign-timestamp"
    _NONCE_HEADER = "x-rabbitkit-sign-nonce"

    def __init__(self, config: SigningConfig) -> None:
        self._config = config
        self._key = config.secret_key.encode("utf-8") if isinstance(config.secret_key, str) else config.secret_key
        self._hash_name = "sha256" if config.algorithm == "hmac-sha256" else "sha512"
        # Default to the in-memory cache so replay protection is functional
        # out of the box for a single process; callers may inject a shared
        # (e.g. RedisNonceCache) cache — see the H4 warning below.
        self._nonce_cache: NonceCache = config.nonce_cache if config.nonce_cache is not None else TTLSetNonceCache()
        # H4: the default in-memory cache is per-process — a replay that
        # lands on a different worker/pod, or after a restart, is invisible
        # to it. This can't detect an actual multi-process deployment, so it
        # fires whenever the default is left in place with require_freshness
        # (the risky combination), rather than silently claiming replay
        # protection "works out of the box" for the common multi-replica case.
        if config.nonce_cache is None and config.require_freshness:
            import warnings

            warnings.warn(
                "SigningMiddleware is using the default in-memory TTLSetNonceCache "
                "with require_freshness=True. This nonce cache is per-process: in "
                "any multi-process or multi-pod deployment (or after a restart), a "
                "replayed message that lands on a different worker will NOT be "
                "detected. Pass nonce_cache=RedisNonceCache(...) (or your own "
                "NonceCache) to share the seen-set across processes before relying "
                "on this for security-sensitive traffic.",
                RuntimeWarning,
                stacklevel=2,
            )

    def _compute_signature(self, body: bytes) -> str:
        """Compute the legacy body-only HMAC signature (backward-compat).

        Deliberately NOT extended to cover routing metadata (H3): this path
        only runs when ``require_freshness=False`` and verifies signatures
        from producers that predate the freshness headers — external/legacy
        signers whose signing scheme this library cannot retroactively
        change. It carries no replay protection either; see the module
        docstring. Prefer the default ``require_freshness=True`` (the fresh
        path below) for any security-sensitive deployment.
        """
        return hmac.new(self._key, body, getattr(hashlib, self._hash_name)).hexdigest()

    @staticmethod
    def _canonical_route(
        exchange: str,
        routing_key: str,
        content_encoding: str | None,
        reply_to: str | None,
    ) -> bytes:
        """Canonical, NUL-delimited encoding of the routing metadata bound
        into the fresh signature (H3).

        Covers exactly ``exchange``, ``routing_key``, ``content_encoding``,
        and ``reply_to`` — the fields an attacker could otherwise flip on a
        captured, validly-signed message to re-route it, redirect an RPC
        reply, or hit a different decompression path, all without touching
        (or being able to forge) the body. Headers other than the
        signature/timestamp/nonce triplet itself are NOT covered — do not
        rely on freeform headers for security-critical routing decisions
        under this middleware. NUL (``\\x00``) is used as the field
        delimiter (including a trailing one) so concatenation cannot make
        two different field combinations hash identically.
        """
        return f"{exchange or ''}\x00{routing_key or ''}\x00{content_encoding or ''}\x00{reply_to or ''}\x00".encode()

    def _compute_fresh_signature(
        self,
        timestamp: float,
        nonce: str,
        body: bytes,
        *,
        exchange: str = "",
        routing_key: str = "",
        content_encoding: str | None = None,
        reply_to: str | None = None,
    ) -> str:
        """Compute the replay-protected, route-bound signature (H3).

        Signs ``timestamp:nonce:`` + the canonical routing metadata
        (:meth:`_canonical_route`) + ``body``. Binding the routing metadata
        means changing ``exchange``/``routing_key``/``content_encoding``/
        ``reply_to`` on a delivered message invalidates the signature even
        when the body, timestamp, and nonce are all byte-for-byte unchanged.
        """
        route = self._canonical_route(exchange, routing_key, content_encoding, reply_to)
        signed = f"{timestamp}:{nonce}:".encode() + route + body
        return hmac.new(self._key, signed, getattr(hashlib, self._hash_name)).hexdigest()

    def _verify_signature(self, body: bytes, signature: str | bytes) -> bool:
        """Verify HMAC signature using constant-time comparison."""
        expected = self._compute_signature(body)
        return _const_time_eq(expected, signature)

    def _verify_fresh_signature(
        self,
        timestamp: float,
        nonce: str,
        body: bytes,
        signature: str | bytes,
        *,
        exchange: str = "",
        routing_key: str = "",
        content_encoding: str | None = None,
        reply_to: str | None = None,
    ) -> bool:
        """Verify replay-protected, route-bound signature (H3) using
        constant-time comparison."""
        expected = self._compute_fresh_signature(
            timestamp,
            nonce,
            body,
            exchange=exchange,
            routing_key=routing_key,
            content_encoding=content_encoding,
            reply_to=reply_to,
        )
        return _const_time_eq(expected, signature)

    # ── Helpers ──────────────────────────────────────────────────────────

    def _sign_envelope(self, envelope: MessageEnvelope) -> MessageEnvelope:
        """Return a new envelope with signature (and freshness) headers added."""
        headers = dict(envelope.headers) if envelope.headers else {}
        # H4: always a fresh random nonce, independent of message_id. A
        # message_id is often caller-supplied and may be reused across
        # publishes (e.g. an at-least-once retry re-sending the "same"
        # message) — using it as the nonce would make the nonce predictable
        # and/or reused, weakening the seen-set's replay guarantee, which
        # depends on the nonce being unique per signing operation.
        nonce = uuid.uuid4().hex
        timestamp = time.time()
        sig = self._compute_fresh_signature(
            timestamp,
            nonce,
            envelope.body,
            exchange=envelope.exchange,
            routing_key=envelope.routing_key,
            content_encoding=envelope.content_encoding,
            reply_to=envelope.reply_to,
        )
        headers[self._config.header_name] = sig
        headers[self._TIMESTAMP_HEADER] = str(timestamp)
        headers[self._NONCE_HEADER] = nonce
        return MessageEnvelope(
            routing_key=envelope.routing_key,
            body=envelope.body,
            exchange=envelope.exchange,
            correlation_id=envelope.correlation_id,
            headers=headers,
            message_id=envelope.message_id,
            reply_to=envelope.reply_to,
            timestamp=envelope.timestamp,
            content_type=envelope.content_type,
            content_encoding=envelope.content_encoding,
            expiration=envelope.expiration,
            priority=envelope.priority,
            mandatory=envelope.mandatory,
            delivery_mode=envelope.delivery_mode,
            type=envelope.type,
            user_id=envelope.user_id,
            app_id=envelope.app_id,
        )

    # ── Publish: sign outgoing messages ──────────────────────────────────

    def publish_scope(self, call_next: Any, envelope: MessageEnvelope) -> Any:
        """Add HMAC signature to outgoing message headers."""
        signed = self._sign_envelope(envelope)
        return call_next(signed)

    async def publish_scope_async(self, call_next: Any, envelope: MessageEnvelope) -> Any:
        """Async variant — add HMAC signature."""
        signed = self._sign_envelope(envelope)
        return await call_next(signed)

    # ── Receive: verify incoming signatures ──────────────────────────────

    def on_receive(self, message: RabbitMessage) -> None:
        """Verify signature on incoming message.

        Replay protection rules (see module docstring):
        * Freshness headers present → skew + nonce always enforced.
        * Headers absent + require_freshness=True → rejected (strict).
        * Headers absent + require_freshness=False → legacy body-only verify + warn.
        """
        sig = message.headers.get(self._config.header_name)
        if sig is None:
            if self._config.reject_unsigned:
                raise InvalidSignatureError(f"Message has no {self._config.header_name} header")
            return
        # L-2: a non-str/bytes signature header would make hmac.compare_digest
        # raise TypeError; surface it as a permanent InvalidSignatureError instead.
        if not isinstance(sig, (str, bytes)):
            raise InvalidSignatureError("signature header is not a string/bytes")

        ts_raw = message.headers.get(self._TIMESTAMP_HEADER)
        nonce = message.headers.get(self._NONCE_HEADER)

        if ts_raw is not None and nonce is not None:
            # Fresh producer path — enforce skew + nonce unconditionally.
            try:
                timestamp = float(ts_raw)
            except (TypeError, ValueError) as exc:
                raise InvalidSignatureError("Invalid signature timestamp header") from exc
            # L-3: NaN/Inf would bypass the skew check (abs(now - NaN) is NaN,
            # which compares False to any threshold); reject non-finite values.
            if not math.isfinite(timestamp):
                raise InvalidSignatureError("non-finite timestamp")

            skew = abs(time.time() - timestamp)
            if skew > self._config.max_skew:
                raise InvalidSignatureError(
                    f"Signature timestamp outside max_skew ({skew:.1f}s > {self._config.max_skew}s)"
                )

            if self._config.reject_invalid and not self._verify_fresh_signature(
                timestamp,
                str(nonce),
                message.body,
                sig,
                exchange=message.exchange,
                routing_key=message.routing_key,
                content_encoding=message.content_encoding,
                reply_to=message.reply_to,
            ):
                raise InvalidSignatureError("Message signature verification failed")

            # Nonce replay check — mark after signature verifies so bogus
            # messages can't burn nonces. TTL covers the replay window.
            if not self._nonce_cache.seen(str(nonce), self._config.max_skew):
                # Duplicate nonce. H1: a BROKER REDELIVERY (redelivered=True) of
                # an unacked message legitimately carries the same nonce — a
                # transient handler failure → nack/requeue → redelivery must not
                # be destroyed as a "replay". Only a FRESH delivery
                # (redelivered=False) reusing a seen nonce is a real replay: an
                # attacker re-publishing a captured message arrives as a new
                # delivery, not a broker redelivery (the redelivered flag is set
                # by the broker, not the publisher, so it can't be forged).
                if not message.redelivered:
                    raise InvalidSignatureError("Replay detected: duplicate nonce")
                logger.debug(
                    "Duplicate nonce on a broker redelivery (redelivered=True) — "
                    "allowed as a legitimate redelivery, not a replay."
                )
            return

        # No freshness headers — legacy producer.
        if self._config.require_freshness:
            if self._config.reject_invalid:
                raise InvalidSignatureError("Missing freshness headers (require_freshness=True)")
            return

        logger.warning(
            "Message from %s without %s/%s headers — verifying body-only signature (no replay protection).",
            self._config.header_name,
            self._TIMESTAMP_HEADER,
            self._NONCE_HEADER,
        )
        if self._config.reject_invalid and not self._verify_signature(message.body, sig):
            raise InvalidSignatureError("Message signature verification failed")

    async def on_receive_async(self, message: RabbitMessage) -> None:
        """Async variant — verify signature."""
        self.on_receive(message)

Methods:

publish_scope(call_next: Any, envelope: MessageEnvelope) -> Any

Add HMAC signature to outgoing message headers.

Source code in src/rabbitkit/middleware/signing.py

def publish_scope(self, call_next: Any, envelope: MessageEnvelope) -> Any:
    """Add HMAC signature to outgoing message headers."""
    signed = self._sign_envelope(envelope)
    return call_next(signed)

publish_scope_async(call_next: Any, envelope: MessageEnvelope) -> Any async

Async variant — add HMAC signature.

Source code in src/rabbitkit/middleware/signing.py

async def publish_scope_async(self, call_next: Any, envelope: MessageEnvelope) -> Any:
    """Async variant — add HMAC signature."""
    signed = self._sign_envelope(envelope)
    return await call_next(signed)

on_receive(message: RabbitMessage) -> None

Verify signature on incoming message.

Replay protection rules (see module docstring): * Freshness headers present → skew + nonce always enforced. * Headers absent + require_freshness=True → rejected (strict). * Headers absent + require_freshness=False → legacy body-only verify + warn.

Source code in src/rabbitkit/middleware/signing.py

def on_receive(self, message: RabbitMessage) -> None:
    """Verify signature on incoming message.

    Replay protection rules (see module docstring):
    * Freshness headers present → skew + nonce always enforced.
    * Headers absent + require_freshness=True → rejected (strict).
    * Headers absent + require_freshness=False → legacy body-only verify + warn.
    """
    sig = message.headers.get(self._config.header_name)
    if sig is None:
        if self._config.reject_unsigned:
            raise InvalidSignatureError(f"Message has no {self._config.header_name} header")
        return
    # L-2: a non-str/bytes signature header would make hmac.compare_digest
    # raise TypeError; surface it as a permanent InvalidSignatureError instead.
    if not isinstance(sig, (str, bytes)):
        raise InvalidSignatureError("signature header is not a string/bytes")

    ts_raw = message.headers.get(self._TIMESTAMP_HEADER)
    nonce = message.headers.get(self._NONCE_HEADER)

    if ts_raw is not None and nonce is not None:
        # Fresh producer path — enforce skew + nonce unconditionally.
        try:
            timestamp = float(ts_raw)
        except (TypeError, ValueError) as exc:
            raise InvalidSignatureError("Invalid signature timestamp header") from exc
        # L-3: NaN/Inf would bypass the skew check (abs(now - NaN) is NaN,
        # which compares False to any threshold); reject non-finite values.
        if not math.isfinite(timestamp):
            raise InvalidSignatureError("non-finite timestamp")

        skew = abs(time.time() - timestamp)
        if skew > self._config.max_skew:
            raise InvalidSignatureError(
                f"Signature timestamp outside max_skew ({skew:.1f}s > {self._config.max_skew}s)"
            )

        if self._config.reject_invalid and not self._verify_fresh_signature(
            timestamp,
            str(nonce),
            message.body,
            sig,
            exchange=message.exchange,
            routing_key=message.routing_key,
            content_encoding=message.content_encoding,
            reply_to=message.reply_to,
        ):
            raise InvalidSignatureError("Message signature verification failed")

        # Nonce replay check — mark after signature verifies so bogus
        # messages can't burn nonces. TTL covers the replay window.
        if not self._nonce_cache.seen(str(nonce), self._config.max_skew):
            # Duplicate nonce. H1: a BROKER REDELIVERY (redelivered=True) of
            # an unacked message legitimately carries the same nonce — a
            # transient handler failure → nack/requeue → redelivery must not
            # be destroyed as a "replay". Only a FRESH delivery
            # (redelivered=False) reusing a seen nonce is a real replay: an
            # attacker re-publishing a captured message arrives as a new
            # delivery, not a broker redelivery (the redelivered flag is set
            # by the broker, not the publisher, so it can't be forged).
            if not message.redelivered:
                raise InvalidSignatureError("Replay detected: duplicate nonce")
            logger.debug(
                "Duplicate nonce on a broker redelivery (redelivered=True) — "
                "allowed as a legitimate redelivery, not a replay."
            )
        return

    # No freshness headers — legacy producer.
    if self._config.require_freshness:
        if self._config.reject_invalid:
            raise InvalidSignatureError("Missing freshness headers (require_freshness=True)")
        return

    logger.warning(
        "Message from %s without %s/%s headers — verifying body-only signature (no replay protection).",
        self._config.header_name,
        self._TIMESTAMP_HEADER,
        self._NONCE_HEADER,
    )
    if self._config.reject_invalid and not self._verify_signature(message.body, sig):
        raise InvalidSignatureError("Message signature verification failed")

on_receive_async(message: RabbitMessage) -> None async

Async variant — verify signature.

Source code in src/rabbitkit/middleware/signing.py

async def on_receive_async(self, message: RabbitMessage) -> None:
    """Async variant — verify signature."""
    self.on_receive(message)

SigningConfig dataclass

Configuration for message signing.

Attributes:

Name	Type	Description
secret_key	str | bytes	Shared secret for HMAC computation.
algorithm	str	Hash algorithm ("hmac-sha256" or "hmac-sha512").
header_name	str	AMQP header name for the signature.
reject_unsigned	bool	If True, reject messages without a signature.
reject_invalid	bool	If True, reject messages with invalid signatures.
max_skew	float	Max allowed |now - timestamp| skew in seconds (both past/future). Also the nonce's replay-window TTL (H4: tightened default of 60s — shrink further for high-value/financial traffic; a captured signature is replayable within this window on any process that has not already seen the nonce).
require_freshness	bool	If True (default), reject messages lacking freshness headers. If False, accept legacy body-only signatures with a warning.
nonce_cache	NonceCache | None	Pluggable nonce seen-set. None means a default in-memory TTLSetNonceCache is created lazily by the middleware (H4: per-process only — use :class:RedisNonceCache or your own shared implementation for any multi-process/multi-pod deployment; a RuntimeWarning is emitted when this default is left unset with require_freshness=True).

Source code in src/rabbitkit/middleware/signing.py

@dataclass(frozen=True, slots=True)
class SigningConfig:
    """Configuration for message signing.

    Attributes:
        secret_key: Shared secret for HMAC computation.
        algorithm: Hash algorithm ("hmac-sha256" or "hmac-sha512").
        header_name: AMQP header name for the signature.
        reject_unsigned: If True, reject messages without a signature.
        reject_invalid: If True, reject messages with invalid signatures.
        max_skew: Max allowed |now - timestamp| skew in seconds (both past/future).
            Also the nonce's replay-window TTL (H4: tightened default of 60s —
            shrink further for high-value/financial traffic; a captured
            signature is replayable within this window on any process that
            has not already seen the nonce).
        require_freshness: If True (default), reject messages lacking freshness
            headers. If False, accept legacy body-only signatures with a warning.
        nonce_cache: Pluggable nonce seen-set. ``None`` means a default
            in-memory ``TTLSetNonceCache`` is created lazily by the middleware
            (H4: per-process only — use :class:`RedisNonceCache` or your own
            shared implementation for any multi-process/multi-pod deployment;
            a ``RuntimeWarning`` is emitted when this default is left unset
            with ``require_freshness=True``).
    """

    secret_key: str | bytes
    algorithm: str = "hmac-sha256"
    header_name: str = "x-rabbitkit-signature"
    reject_unsigned: bool = False
    reject_invalid: bool = True
    # Replay protection. H4: default tightened from 300s to 60s — shrink
    # further for payments/high-value traffic.
    max_skew: float = 60.0
    require_freshness: bool = True
    nonce_cache: NonceCache | None = None

    def __post_init__(self) -> None:
        if self.algorithm not in ("hmac-sha256", "hmac-sha512"):
            raise ValueError(f"Unsupported algorithm: {self.algorithm}. Use 'hmac-sha256' or 'hmac-sha512'.")
        if self.max_skew <= 0:
            raise ValueError("max_skew must be positive")
        # L-1: the signature header must not collide with the freshness headers,
        # otherwise the timestamp/nonce would be overwritten by the signature.
        if self.header_name in (SigningMiddleware._TIMESTAMP_HEADER, SigningMiddleware._NONCE_HEADER):
            raise ValueError(
                f"header_name {self.header_name!r} collides with a freshness header; "
                "choose a different signature header name."
            )

InvalidSignatureError

Bases: Exception

Raised when a message signature is invalid.

Source code in src/rabbitkit/middleware/signing.py

class InvalidSignatureError(Exception):
    """Raised when a message signature is invalid."""

RateLimitMiddleware

RateLimitMiddleware

Bases: BaseMiddleware

Limits message processing rate using a token bucket.

Behavior when rate is exceeded (configurable via on_limited): - "wait": Sleep until a token is available (default) - "nack": Reject message with requeue=True (another consumer can try) - "drop": Reject message with requeue=False (message is lost/goes to DLQ)

L5 — per-process scoping: the token bucket lives in this process's memory. With worker_count > 1 consumer processes/pods sharing a queue, each gets its own independent bucket — the effective CLUSTER rate is workers x max_rate, not max_rate. Size max_rate accordingly, or put a cluster-wide limiter in front (e.g. a Redis-backed token bucket) if you need a hard global cap. Under sustained overload, "wait" blocks for up to _wait_deadline (30s by default) before falling back to a drop (nack, no requeue) — every nack/drop/wait-timeout is logged at WARNING and, if you pass metrics_collector, increments rate_limit_dropped_total labeled by reason so sustained overload is observable instead of silent.

Source code in src/rabbitkit/middleware/rate_limit.py

class RateLimitMiddleware(BaseMiddleware):
    """Limits message processing rate using a token bucket.

    Behavior when rate is exceeded (configurable via on_limited):
    - "wait": Sleep until a token is available (default)
    - "nack": Reject message with requeue=True (another consumer can try)
    - "drop": Reject message with requeue=False (message is lost/goes to DLQ)

    L5 — per-process scoping: the token bucket lives in this process's
    memory. With ``worker_count`` > 1 consumer processes/pods sharing a
    queue, each gets its own independent bucket — the effective CLUSTER
    rate is ``workers x max_rate``, not ``max_rate``. Size ``max_rate``
    accordingly, or put a cluster-wide limiter in front (e.g. a
    Redis-backed token bucket) if you need a hard global cap. Under
    sustained overload, "wait" blocks for up to ``_wait_deadline`` (30s by
    default) before falling back to a drop (nack, no requeue) — every
    nack/drop/wait-timeout is logged at WARNING and, if you pass
    ``metrics_collector``, increments ``rate_limit_dropped_total`` labeled
    by ``reason`` so sustained overload is observable instead of silent.
    """

    def __init__(
        self,
        config: RateLimitConfig,
        *,
        metrics_collector: Any | None = None,
        metrics_config: Any | None = None,
    ) -> None:
        self._config = config
        self._bucket = _TokenBucket(config.max_rate, config.burst)
        # Maximum time (s) a "wait" policy will block for a token before falling
        # back to drop semantics. Override per-instance if needed.
        self._wait_deadline: float = 30.0
        self._metrics_collector = metrics_collector
        self._metrics_config = metrics_config

    def _record_drop(self, reason: str) -> None:
        """L5: log + optional metric every time a message is settled without
        the handler running, so sustained rate-limit pressure is observable."""
        logger.warning("RateLimitMiddleware dropped a message (reason=%s)", reason)
        if self._metrics_collector is not None and self._metrics_config is not None:
            self._metrics_collector.inc_counter(
                self._metrics_config.rate_limit_dropped_total, {"reason": reason}
            )

    def consume_scope(
        self,
        call_next: Any,
        message: RabbitMessage,
    ) -> Any:
        """Rate-limit sync message processing.

        For ``on_limited="wait"`` the loop polls the token bucket until a token
        is acquired **or** ``wait_deadline`` (seconds, monotonic) expires. If the
        deadline elapses with no token, the message falls back to the configured
        drop/nack semantics so the handler is **never** invoked without a token.
        """
        if self._bucket.try_acquire():
            return call_next(message)

        if self._config.on_limited == "nack":
            self._record_drop("nack")
            if not message.is_settled:
                message.nack(requeue=True)
            return None
        if self._config.on_limited == "drop":
            self._record_drop("drop")
            if not message.is_settled:
                message.nack(requeue=False)
            return None

        # "wait" — bounded loop; only proceed once a token is actually acquired.
        deadline = time.monotonic() + self._wait_deadline
        while not self._bucket.try_acquire():
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                # No token within the deadline — fall back to drop semantics so
                # the handler is NOT called without a token.
                self._record_drop("wait_deadline_exceeded")
                if not message.is_settled:
                    message.nack(requeue=False)
                return None
            time.sleep(min(self._bucket.wait_time(), remaining))

        return call_next(message)

    async def consume_scope_async(
        self,
        call_next: Any,
        message: RabbitMessage,
    ) -> Any:
        """Rate-limit async message processing.

        Mirrors the sync logic: the "wait" loop is bounded by
        ``self._wait_deadline`` and falls back to drop semantics if no token is
        acquired in time, so the handler is never called without a token.
        """
        if self._bucket.try_acquire():
            return await call_next(message)

        if self._config.on_limited == "nack":
            self._record_drop("nack")
            if not message.is_settled:
                await message.nack_async(requeue=True)
            return None
        if self._config.on_limited == "drop":
            self._record_drop("drop")
            if not message.is_settled:
                await message.nack_async(requeue=False)
            return None

        # "wait" — bounded loop; only proceed once a token is actually acquired.
        deadline = time.monotonic() + self._wait_deadline
        while not self._bucket.try_acquire():
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                self._record_drop("wait_deadline_exceeded")
                if not message.is_settled:
                    await message.nack_async(requeue=False)
                return None
            await asyncio.sleep(min(self._bucket.wait_time(), remaining))

        return await call_next(message)

Methods:

consume_scope(call_next: Any, message: RabbitMessage) -> Any

Rate-limit sync message processing.

For on_limited="wait" the loop polls the token bucket until a token is acquired or wait_deadline (seconds, monotonic) expires. If the deadline elapses with no token, the message falls back to the configured drop/nack semantics so the handler is never invoked without a token.

Source code in src/rabbitkit/middleware/rate_limit.py

def consume_scope(
    self,
    call_next: Any,
    message: RabbitMessage,
) -> Any:
    """Rate-limit sync message processing.

    For ``on_limited="wait"`` the loop polls the token bucket until a token
    is acquired **or** ``wait_deadline`` (seconds, monotonic) expires. If the
    deadline elapses with no token, the message falls back to the configured
    drop/nack semantics so the handler is **never** invoked without a token.
    """
    if self._bucket.try_acquire():
        return call_next(message)

    if self._config.on_limited == "nack":
        self._record_drop("nack")
        if not message.is_settled:
            message.nack(requeue=True)
        return None
    if self._config.on_limited == "drop":
        self._record_drop("drop")
        if not message.is_settled:
            message.nack(requeue=False)
        return None

    # "wait" — bounded loop; only proceed once a token is actually acquired.
    deadline = time.monotonic() + self._wait_deadline
    while not self._bucket.try_acquire():
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            # No token within the deadline — fall back to drop semantics so
            # the handler is NOT called without a token.
            self._record_drop("wait_deadline_exceeded")
            if not message.is_settled:
                message.nack(requeue=False)
            return None
        time.sleep(min(self._bucket.wait_time(), remaining))

    return call_next(message)

consume_scope_async(call_next: Any, message: RabbitMessage) -> Any async

Rate-limit async message processing.

Mirrors the sync logic: the "wait" loop is bounded by self._wait_deadline and falls back to drop semantics if no token is acquired in time, so the handler is never called without a token.

Source code in src/rabbitkit/middleware/rate_limit.py

async def consume_scope_async(
    self,
    call_next: Any,
    message: RabbitMessage,
) -> Any:
    """Rate-limit async message processing.

    Mirrors the sync logic: the "wait" loop is bounded by
    ``self._wait_deadline`` and falls back to drop semantics if no token is
    acquired in time, so the handler is never called without a token.
    """
    if self._bucket.try_acquire():
        return await call_next(message)

    if self._config.on_limited == "nack":
        self._record_drop("nack")
        if not message.is_settled:
            await message.nack_async(requeue=True)
        return None
    if self._config.on_limited == "drop":
        self._record_drop("drop")
        if not message.is_settled:
            await message.nack_async(requeue=False)
        return None

    # "wait" — bounded loop; only proceed once a token is actually acquired.
    deadline = time.monotonic() + self._wait_deadline
    while not self._bucket.try_acquire():
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            self._record_drop("wait_deadline_exceeded")
            if not message.is_settled:
                await message.nack_async(requeue=False)
            return None
        await asyncio.sleep(min(self._bucket.wait_time(), remaining))

    return await call_next(message)

RateLimitConfig dataclass

Configuration for rate limiting.

Attributes:

Name	Type	Description
max_rate	float	Maximum messages per second.
burst	int	Maximum burst size above steady rate.
on_limited	str	Behavior when rate exceeded: "wait", "nack", or "drop".

Source code in src/rabbitkit/middleware/rate_limit.py

@dataclass(frozen=True, slots=True)
class RateLimitConfig:
    """Configuration for rate limiting.

    Attributes:
        max_rate: Maximum messages per second.
        burst: Maximum burst size above steady rate.
        on_limited: Behavior when rate exceeded: "wait", "nack", or "drop".
    """

    max_rate: float
    burst: int = 1
    on_limited: str = "wait"  # "wait" | "nack" | "drop"

    def __post_init__(self) -> None:
        if self.max_rate <= 0:
            raise ValueError("max_rate must be positive")
        if self.burst < 1:
            raise ValueError("burst must be >= 1")
        if self.on_limited not in ("wait", "nack", "drop"):
            raise ValueError(f"on_limited must be 'wait', 'nack', or 'drop', got '{self.on_limited}'")

DeduplicationMiddleware

DeduplicationConfig.mark_policy controls when the dedup key is recorded (accepts a DeduplicationMarkPolicy enum member or its string value):

Value	Prevents concurrent duplicates	Crash-safe	Behaviour
"on_success" (default)	No	Yes	Key checked (no write) before the handler, stored only after it succeeds. A consumer killed mid-handler leaves no mark, so the redelivery is processed. Concurrent duplicates may both process (at-least-once). Use for most consumers.
"claim"	Yes	Yes¹	Two-state: atomic in-flight claim (TTL = processing_timeout) before the handler, flipped to completed (full ttl) on success. A concurrent copy that sees a live claim is nack-requeued (default on_in_flight="nack_requeue"; "ack_skip" available) so it retries if the claiming consumer dies. A crash lets the claim expire. Use for sensitive side effects.
"on_start"	Yes	No — can lose messages	Key stored before the handler runs. If the process crashes after marking but before the handler finishes, the redelivery is skipped as a duplicate and the message is lost. Use only when duplicate execution is worse than message loss.

¹ Provided processing_timeout comfortably exceeds the worst-case handler duration — a handler that outlives its claim lets a duplicate start while it is still running.

Deduplication is not a replacement for idempotent business logic: RabbitMQ is an at-least-once system, and a handler can still run twice in some failure scenarios (e.g. side effect completes but the ack fails, or a claim expires under a slow handler). Keep a business-level guard — unique DB constraint, payment_id/transaction_id, outbox or processed_events table — for anything non-idempotent.

DeduplicationMiddleware

Bases: BaseMiddleware

Idempotent consumer middleware backed by Redis SETNX.

Usage::

import redis
mw = DeduplicationMiddleware(
    redis_client=redis.Redis(),
    config=DeduplicationConfig(key_source="message_id", ttl=86400),
)

To prevent concurrent duplicate processing at the cost of retry safety::

mw = DeduplicationMiddleware(
    redis_client=redis.Redis(),
    config=DeduplicationConfig(mark_policy="on_start"),
)

Source code in src/rabbitkit/middleware/deduplication.py

class DeduplicationMiddleware(BaseMiddleware):
    """Idempotent consumer middleware backed by Redis SETNX.

    Usage::

        import redis
        mw = DeduplicationMiddleware(
            redis_client=redis.Redis(),
            config=DeduplicationConfig(key_source="message_id", ttl=86400),
        )

    To prevent concurrent duplicate processing at the cost of retry safety::

        mw = DeduplicationMiddleware(
            redis_client=redis.Redis(),
            config=DeduplicationConfig(mark_policy="on_start"),
        )
    """

    def __init__(
        self,
        redis_client: Any,
        config: DeduplicationConfig | None = None,
        *,
        key_fn: Callable[[RabbitMessage], str] | None = None,
        metrics_collector: Any | None = None,
        metrics_config: Any | None = None,
    ) -> None:
        self._redis = redis_client
        self._config = config or DeduplicationConfig()
        self._key_fn = key_fn
        # M9: optional -- emits `dedup_fallback_total` every time a Redis
        # error causes this middleware to skip idempotency enforcement for a
        # message (fallback_on_redis_error=True, the default). None is a no-op.
        self._metrics_collector = metrics_collector
        self._metrics_config = metrics_config
        # Optional in-process LRU pre-filter — short-circuits Redis for keys we've
        # already confirmed as processed. Only allocated when local_cache_size > 0.
        # Evicts the oldest entry (FIFO) when capacity is reached.
        self._local_cache: OrderedDict[str, None] | None = (
            OrderedDict() if self._config.local_cache_size > 0 else None
        )
        # The cache is mutated from sync worker-pool daemon threads; OrderedDict
        # mutation is not atomic (move_to_end/popitem can corrupt mid-eviction).
        self._local_lock = threading.Lock()

    def _record_fallback(self, message: RabbitMessage) -> None:
        """M9: log at ERROR (not WARNING — idempotency being silently
        disabled for a message is an operational event worth alerting on,
        not routine noise) and emit `dedup_fallback_total` if a metrics
        collector is wired in."""
        logger.error(
            "Redis error during dedup check/mark; processing message anyway "
            "(fallback_on_redis_error=True) — idempotency is NOT enforced for "
            "this message. For workloads where a duplicate is unacceptable "
            "(e.g. financial), set fallback_on_redis_error=False to fail closed instead.",
            exc_info=True,
        )
        if self._metrics_collector is not None and self._metrics_config is not None:
            queue = message.headers.get("x-rabbitkit-original-queue") or message.routing_key or "unknown"
            self._metrics_collector.inc_counter(
                self._metrics_config.dedup_fallback_total, {"queue": str(queue)}
            )

    # ── Local LRU helpers ─────────────────────────────────────────────────

    def _local_is_dup(self, key: str) -> bool:
        """True if key is already in the local cache (= confirmed processed)."""
        if self._local_cache is None:
            return False
        with self._local_lock:
            return key in self._local_cache

    def _local_mark(self, key: str) -> None:
        """Record key in the local LRU; evicts oldest when at capacity."""
        if self._local_cache is None:
            return
        with self._local_lock:
            self._local_cache[key] = None
            self._local_cache.move_to_end(key)
            if len(self._local_cache) > self._config.local_cache_size:
                self._local_cache.popitem(last=False)

    def _local_remove(self, key: str) -> None:
        """Remove key from local cache (called when handler fails, key deleted from Redis)."""
        if self._local_cache is not None:
            with self._local_lock:
                self._local_cache.pop(key, None)

    def _cleanup_key_after_non_success(self, key: str) -> None:
        """Delete *key* from Redis + the local cache after a non-success
        (handler exception, or a requeue signaled via REQUEUED_FOR_RETRY —
        H8) so a later redelivery of the SAME message is not treated as a
        duplicate and dropped."""
        try:
            self._redis.delete(key)
        except Exception:
            logger.warning("Redis error during dedup key cleanup after non-success", exc_info=True)
        self._local_remove(key)

    async def _cleanup_key_after_non_success_async(self, key: str) -> None:
        """Async variant of :meth:`_cleanup_key_after_non_success`."""
        try:
            await self._redis.delete(key)
        except Exception:
            logger.warning("Redis error during dedup key cleanup after non-success", exc_info=True)
        self._local_remove(key)

    # ── Key extraction ────────────────────────────────────────────────────

    def _extract_key(self, message: RabbitMessage) -> str:
        """Build the dedup key from the message.

        Resolution:
        1. Custom ``key_fn`` (highest priority)
        2. ``config.key_source``:
           - ``"message_id"`` → ``message.message_id``
           - ``"correlation_id"`` → ``message.correlation_id``
           - ``"body_hash"`` → SHA-256 hex digest of ``message.body``
        If the selected id field is empty/None we fall back to a SHA-256 body
        hash (with a warning) instead of collapsing every id-less message to a
        single constant key.
        """
        if self._key_fn is not None:
            raw = self._key_fn(message)
        elif self._config.key_source == "message_id":
            raw = message.message_id or ""
        elif self._config.key_source == "correlation_id":
            raw = message.correlation_id or ""
        elif self._config.key_source == "body_hash":
            raw = hashlib.sha256(message.body).hexdigest()
        else:
            # Unknown key_source — fall back to message_id
            logger.warning(
                "Unknown key_source %r, falling back to message_id",
                self._config.key_source,
            )
            raw = message.message_id or ""

        # Empty raw key (id field missing) → fall back to body hash so distinct
        # id-less messages are NOT collapsed onto a single constant key.
        if not raw:
            logger.warning(
                "key_source=%r resolved to an empty id for message (routing_key=%r); falling back to body hash.",
                self._config.key_source,
                message.routing_key,
            )
            raw = hashlib.sha256(message.body).hexdigest()

        return f"{self._config.key_prefix}:{raw}"

    def _mark_key(self, key: str, message: RabbitMessage) -> bool:
        """Attempt to mark key as processed (sync). Returns True if this is a new key.

        Checks the local LRU cache first to avoid a Redis round-trip for keys
        we've already confirmed as processed in this process.
        """
        if self._local_is_dup(key):
            return False
        try:
            result = bool(self._redis.set(key, "1", nx=True, ex=self._config.ttl))
        except Exception:
            if not self._config.fallback_on_redis_error:
                raise
            self._record_fallback(message)
            return True
        if result:
            self._local_mark(key)
        return result

    async def _mark_key_async(self, key: str, message: RabbitMessage) -> bool:
        """Attempt to mark key as processed (async). Returns True if this is a new key.

        Checks the local LRU cache first to avoid a Redis round-trip for keys
        we've already confirmed as processed in this process.
        """
        if self._local_is_dup(key):
            return False
        try:
            result = bool(await self._redis.set(key, "1", nx=True, ex=self._config.ttl))
        except Exception:
            if not self._config.fallback_on_redis_error:
                raise
            self._record_fallback(message)
            return True
        if result:
            self._local_mark(key)
        return result

    # ── Consume-side hooks ────────────────────────────────────────────────

    def consume_scope(
        self,
        call_next: Callable[[RabbitMessage], Any],
        message: RabbitMessage,
    ) -> Any:
        """Sync: check dedup → skip duplicate → call handler."""
        key = self._extract_key(message)

        if self._config.mark_policy == "claim":
            return self._consume_claim_sync(call_next, message, key)

        if self._config.mark_policy == "on_start":
            # _mark_key checks local cache first, then Redis
            is_new = self._mark_key(key, message)
            if not is_new:
                logger.debug("Duplicate message detected (key=%s); acking and skipping", key)
                if not message.is_settled:
                    message.ack()
                return None
            result = call_next(message)
            if result is REQUEUED_FOR_RETRY:
                # H8: an inner RetryMiddleware requeued the failed handler
                # rather than succeeding — undo the premature on_start mark
                # so the retry redelivery is not dropped as a duplicate.
                self._cleanup_key_after_non_success(key)
            return result

        # mark_policy == "on_success": check (no write) → handler → mark.
        # The key is written only AFTER the handler returns successfully, so a
        # consumer killed mid-handler (OOM/SIGKILL) leaves no mark and the
        # broker's redelivery is processed instead of dropped as a duplicate.
        # A handler exception likewise leaves nothing behind — no cleanup
        # needed (and deleting here could erase a concurrent delivery's
        # legitimate success-mark).
        if self._local_is_dup(key):
            logger.debug("Duplicate message detected in local cache (key=%s); acking and skipping", key)
            if not message.is_settled:
                message.ack()
            return None

        try:
            already_seen = bool(self._redis.exists(key))
        except Exception:
            if not self._config.fallback_on_redis_error:
                raise
            self._record_fallback(message)
            return call_next(message)

        if already_seen:
            logger.debug("Duplicate message detected (key=%s); acking and skipping", key)
            if not message.is_settled:
                message.ack()
            return None

        result = call_next(message)
        if result is REQUEUED_FOR_RETRY:
            # H8: an inner RetryMiddleware requeued the failed handler instead
            # of it actually succeeding. Nothing was marked yet, so the retry
            # redelivery (same dedup key) passes the dedup check — just skip
            # the mark.
            return result
        # Handler succeeded — mark now. Never raise past this point, even with
        # fallback_on_redis_error=False: the handler's side effects are done,
        # and raising would nack → redeliver → a GUARANTEED duplicate
        # execution, worse than the unmarked-key window it would signal.
        try:
            self._redis.set(key, "1", nx=True, ex=self._config.ttl)
        except Exception:
            self._record_fallback(message)
            return result
        self._local_mark(key)
        return result

    async def consume_scope_async(
        self,
        call_next: Callable[[RabbitMessage], Awaitable[Any]],
        message: RabbitMessage,
    ) -> Any:
        """Async: check dedup → skip duplicate → call handler."""
        key = self._extract_key(message)

        if self._config.mark_policy == "claim":
            return await self._consume_claim_async(call_next, message, key)

        if self._config.mark_policy == "on_start":
            # _mark_key_async checks local cache first, then Redis
            is_new = await self._mark_key_async(key, message)
            if not is_new:
                logger.debug("Duplicate message detected (key=%s); acking and skipping", key)
                if not message.is_settled:
                    await message.ack_async()
                return None
            result = await call_next(message)
            if result is REQUEUED_FOR_RETRY:
                # H8: an inner RetryMiddleware requeued the failed handler
                # rather than succeeding — undo the premature on_start mark
                # so the retry redelivery is not dropped as a duplicate.
                await self._cleanup_key_after_non_success_async(key)
            return result

        # mark_policy == "on_success": check (no write) → handler → mark.
        # See the sync variant above for the crash-safety rationale.
        if self._local_is_dup(key):
            logger.debug("Duplicate message detected in local cache (key=%s); acking and skipping", key)
            if not message.is_settled:
                await message.ack_async()
            return None

        try:
            already_seen = bool(await self._redis.exists(key))
        except Exception:
            if not self._config.fallback_on_redis_error:
                raise
            self._record_fallback(message)
            return await call_next(message)

        if already_seen:
            logger.debug("Duplicate message detected (key=%s); acking and skipping", key)
            if not message.is_settled:
                await message.ack_async()
            return None

        result = await call_next(message)
        if result is REQUEUED_FOR_RETRY:
            # H8: an inner RetryMiddleware requeued the failed handler instead
            # of it actually succeeding. Nothing was marked yet, so the retry
            # redelivery (same dedup key) passes the dedup check — just skip
            # the mark.
            return result
        # Handler succeeded — mark now. Never raise past this point, even with
        # fallback_on_redis_error=False: raising would nack → redeliver → a
        # GUARANTEED duplicate execution.
        try:
            await self._redis.set(key, "1", nx=True, ex=self._config.ttl)
        except Exception:
            self._record_fallback(message)
            return result
        self._local_mark(key)
        return result

    # ── mark_policy == "claim" ────────────────────────────────────────────

    @staticmethod
    def _is_in_flight(raw: Any) -> bool:
        """True when a GET result is a live in-flight claim.

        ``None`` (the key expired between the failed SET NX and this GET)
        also counts — requeueing lets the redelivery claim it cleanly.
        Any other value (``"completed"``, or the legacy ``"1"`` written by
        on_success/on_start deployments) means completed.
        """
        if raw is None:
            return True
        value = raw.decode() if isinstance(raw, bytes) else raw
        return bool(value == _IN_FLIGHT)

    def _handle_in_flight_duplicate_sync(self, message: RabbitMessage, key: str) -> None:
        """A concurrent copy hit another consumer's live claim (sync)."""
        if self._config.on_in_flight == "ack_skip":
            logger.debug("Duplicate of in-flight message (key=%s); acking and skipping", key)
            if not message.is_settled:
                message.ack()
            return
        # "nack_requeue" (default): the copy comes back and retries, so it is
        # NOT lost if the claiming consumer dies mid-handler.
        # ponytail: immediate requeue — the duplicate redelivers in a tight
        # loop until the claim resolves, bounded by prefetch and the
        # handler's duration; add a delay queue if that churn ever matters.
        logger.debug("Duplicate of in-flight message (key=%s); nack-requeueing", key)
        if not message.is_settled:
            message.nack(requeue=True)

    def _consume_claim_sync(
        self,
        call_next: Callable[[RabbitMessage], Any],
        message: RabbitMessage,
        key: str,
    ) -> Any:
        """Sync claim flow: atomically claim in-flight → handler → flip to
        completed. Crash mid-handler lets the claim expire; the redelivery
        then re-claims and processes."""
        if self._local_is_dup(key):
            logger.debug("Duplicate message detected in local cache (key=%s); acking and skipping", key)
            if not message.is_settled:
                message.ack()
            return None

        try:
            claimed = bool(
                self._redis.set(key, _IN_FLIGHT, nx=True, ex=self._config.processing_timeout)
            )
        except Exception:
            if not self._config.fallback_on_redis_error:
                raise
            self._record_fallback(message)
            return call_next(message)

        if not claimed:
            try:
                raw = self._redis.get(key)
            except Exception:
                if not self._config.fallback_on_redis_error:
                    raise
                self._record_fallback(message)
                return call_next(message)
            if self._is_in_flight(raw):
                self._handle_in_flight_duplicate_sync(message, key)
                return None
            if self._config.store_results:
                has_result, stored = _decode_stored_result(raw)
                if has_result:
                    # F5: replay — returning the stored result makes the
                    # pipeline re-publish it to the route's result publisher /
                    # reply_to, so a duplicate (e.g. redelivered RPC request)
                    # gets a byte-identical answer without re-running the
                    # handler's side effects.
                    logger.debug("Duplicate (key=%s); replaying stored result", key)
                    if not message.is_settled:
                        message.ack()
                    return stored
            logger.debug("Duplicate message detected (key=%s); acking and skipping", key)
            if not message.is_settled:
                message.ack()
            return None

        try:
            result = call_next(message)
        except Exception:
            # Release the claim so a retry redelivery can re-claim immediately
            # instead of waiting out processing_timeout.
            self._cleanup_key_after_non_success(key)
            raise
        if result is REQUEUED_FOR_RETRY:
            # H8: release the claim for the delayed retry redelivery.
            self._cleanup_key_after_non_success(key)
            return result
        # Handler succeeded — flip the claim to completed with the full TTL.
        # Never raise past this point (side effects are committed; raising
        # would nack → redeliver → a guaranteed duplicate execution).
        value = _COMPLETED
        if self._config.store_results and result is not None:
            encoded = _encode_completed_with_result(result, self._config.max_result_bytes)
            if encoded is not None:
                value = encoded
        try:
            self._redis.set(key, value, ex=self._config.ttl)
        except Exception:
            self._record_fallback(message)
            return result
        self._local_mark(key)
        return result

    async def _handle_in_flight_duplicate_async(self, message: RabbitMessage, key: str) -> None:
        """A concurrent copy hit another consumer's live claim (async)."""
        if self._config.on_in_flight == "ack_skip":
            logger.debug("Duplicate of in-flight message (key=%s); acking and skipping", key)
            if not message.is_settled:
                await message.ack_async()
            return
        logger.debug("Duplicate of in-flight message (key=%s); nack-requeueing", key)
        if not message.is_settled:
            await message.nack_async(requeue=True)

    async def _consume_claim_async(
        self,
        call_next: Callable[[RabbitMessage], Awaitable[Any]],
        message: RabbitMessage,
        key: str,
    ) -> Any:
        """Async variant of :meth:`_consume_claim_sync`."""
        if self._local_is_dup(key):
            logger.debug("Duplicate message detected in local cache (key=%s); acking and skipping", key)
            if not message.is_settled:
                await message.ack_async()
            return None

        try:
            claimed = bool(
                await self._redis.set(key, _IN_FLIGHT, nx=True, ex=self._config.processing_timeout)
            )
        except Exception:
            if not self._config.fallback_on_redis_error:
                raise
            self._record_fallback(message)
            return await call_next(message)

        if not claimed:
            try:
                raw = await self._redis.get(key)
            except Exception:
                if not self._config.fallback_on_redis_error:
                    raise
                self._record_fallback(message)
                return await call_next(message)
            if self._is_in_flight(raw):
                await self._handle_in_flight_duplicate_async(message, key)
                return None
            if self._config.store_results:
                has_result, stored = _decode_stored_result(raw)
                if has_result:
                    # F5: replay the stored result -- see the sync counterpart.
                    logger.debug("Duplicate (key=%s); replaying stored result", key)
                    if not message.is_settled:
                        await message.ack_async()
                    return stored
            logger.debug("Duplicate message detected (key=%s); acking and skipping", key)
            if not message.is_settled:
                await message.ack_async()
            return None

        try:
            result = await call_next(message)
        except Exception:
            await self._cleanup_key_after_non_success_async(key)
            raise
        if result is REQUEUED_FOR_RETRY:
            await self._cleanup_key_after_non_success_async(key)
            return result
        value = _COMPLETED
        if self._config.store_results and result is not None:
            encoded = _encode_completed_with_result(result, self._config.max_result_bytes)
            if encoded is not None:
                value = encoded
        try:
            await self._redis.set(key, value, ex=self._config.ttl)
        except Exception:
            self._record_fallback(message)
            return result
        self._local_mark(key)
        return result

Methods:

consume_scope(call_next: Callable[[RabbitMessage], Any], message: RabbitMessage) -> Any

Sync: check dedup → skip duplicate → call handler.

Source code in src/rabbitkit/middleware/deduplication.py

def consume_scope(
    self,
    call_next: Callable[[RabbitMessage], Any],
    message: RabbitMessage,
) -> Any:
    """Sync: check dedup → skip duplicate → call handler."""
    key = self._extract_key(message)

    if self._config.mark_policy == "claim":
        return self._consume_claim_sync(call_next, message, key)

    if self._config.mark_policy == "on_start":
        # _mark_key checks local cache first, then Redis
        is_new = self._mark_key(key, message)
        if not is_new:
            logger.debug("Duplicate message detected (key=%s); acking and skipping", key)
            if not message.is_settled:
                message.ack()
            return None
        result = call_next(message)
        if result is REQUEUED_FOR_RETRY:
            # H8: an inner RetryMiddleware requeued the failed handler
            # rather than succeeding — undo the premature on_start mark
            # so the retry redelivery is not dropped as a duplicate.
            self._cleanup_key_after_non_success(key)
        return result

    # mark_policy == "on_success": check (no write) → handler → mark.
    # The key is written only AFTER the handler returns successfully, so a
    # consumer killed mid-handler (OOM/SIGKILL) leaves no mark and the
    # broker's redelivery is processed instead of dropped as a duplicate.
    # A handler exception likewise leaves nothing behind — no cleanup
    # needed (and deleting here could erase a concurrent delivery's
    # legitimate success-mark).
    if self._local_is_dup(key):
        logger.debug("Duplicate message detected in local cache (key=%s); acking and skipping", key)
        if not message.is_settled:
            message.ack()
        return None

    try:
        already_seen = bool(self._redis.exists(key))
    except Exception:
        if not self._config.fallback_on_redis_error:
            raise
        self._record_fallback(message)
        return call_next(message)

    if already_seen:
        logger.debug("Duplicate message detected (key=%s); acking and skipping", key)
        if not message.is_settled:
            message.ack()
        return None

    result = call_next(message)
    if result is REQUEUED_FOR_RETRY:
        # H8: an inner RetryMiddleware requeued the failed handler instead
        # of it actually succeeding. Nothing was marked yet, so the retry
        # redelivery (same dedup key) passes the dedup check — just skip
        # the mark.
        return result
    # Handler succeeded — mark now. Never raise past this point, even with
    # fallback_on_redis_error=False: the handler's side effects are done,
    # and raising would nack → redeliver → a GUARANTEED duplicate
    # execution, worse than the unmarked-key window it would signal.
    try:
        self._redis.set(key, "1", nx=True, ex=self._config.ttl)
    except Exception:
        self._record_fallback(message)
        return result
    self._local_mark(key)
    return result

consume_scope_async(call_next: Callable[[RabbitMessage], Awaitable[Any]], message: RabbitMessage) -> Any async

Async: check dedup → skip duplicate → call handler.

Source code in src/rabbitkit/middleware/deduplication.py

async def consume_scope_async(
    self,
    call_next: Callable[[RabbitMessage], Awaitable[Any]],
    message: RabbitMessage,
) -> Any:
    """Async: check dedup → skip duplicate → call handler."""
    key = self._extract_key(message)

    if self._config.mark_policy == "claim":
        return await self._consume_claim_async(call_next, message, key)

    if self._config.mark_policy == "on_start":
        # _mark_key_async checks local cache first, then Redis
        is_new = await self._mark_key_async(key, message)
        if not is_new:
            logger.debug("Duplicate message detected (key=%s); acking and skipping", key)
            if not message.is_settled:
                await message.ack_async()
            return None
        result = await call_next(message)
        if result is REQUEUED_FOR_RETRY:
            # H8: an inner RetryMiddleware requeued the failed handler
            # rather than succeeding — undo the premature on_start mark
            # so the retry redelivery is not dropped as a duplicate.
            await self._cleanup_key_after_non_success_async(key)
        return result

    # mark_policy == "on_success": check (no write) → handler → mark.
    # See the sync variant above for the crash-safety rationale.
    if self._local_is_dup(key):
        logger.debug("Duplicate message detected in local cache (key=%s); acking and skipping", key)
        if not message.is_settled:
            await message.ack_async()
        return None

    try:
        already_seen = bool(await self._redis.exists(key))
    except Exception:
        if not self._config.fallback_on_redis_error:
            raise
        self._record_fallback(message)
        return await call_next(message)

    if already_seen:
        logger.debug("Duplicate message detected (key=%s); acking and skipping", key)
        if not message.is_settled:
            await message.ack_async()
        return None

    result = await call_next(message)
    if result is REQUEUED_FOR_RETRY:
        # H8: an inner RetryMiddleware requeued the failed handler instead
        # of it actually succeeding. Nothing was marked yet, so the retry
        # redelivery (same dedup key) passes the dedup check — just skip
        # the mark.
        return result
    # Handler succeeded — mark now. Never raise past this point, even with
    # fallback_on_redis_error=False: raising would nack → redeliver → a
    # GUARANTEED duplicate execution.
    try:
        await self._redis.set(key, "1", nx=True, ex=self._config.ttl)
    except Exception:
        self._record_fallback(message)
        return result
    self._local_mark(key)
    return result

CircuitBreakerMiddleware

CircuitBreakerMiddleware

Bases: BaseMiddleware

Wraps handler execution and publish operations with circuit breaker.

If circuit breaker is not provided (None), all operations pass through without any wrapping (no-op mode).

Usage::

# any CircuitBreakerProtocol-compatible implementation, e.g. pybreaker
cb = MyCircuitBreaker(name="rabbitmq", fail_max=5, reset_timeout=60)
middleware = CircuitBreakerMiddleware(circuit_breaker=cb)

# Or with separate publish CB:
middleware = CircuitBreakerMiddleware(
    circuit_breaker=consume_cb,
    publish_circuit_breaker=publish_cb,
)

Parameters:

Name	Type	Description	Default
circuit_breaker	Any | None	Circuit breaker for consume operations (handler execution). Must satisfy CircuitBreakerProtocol. None for no-op.	None
publish_circuit_breaker	Any | None	Circuit breaker for publish operations. If None, uses circuit_breaker for both. If circuit_breaker is also None, publish operations pass through.	None
async_circuit_breaker	Any | None	Async circuit breaker for consume operations. Must satisfy AsyncCircuitBreakerProtocol. If None, falls back to wrapping the sync circuit_breaker in async context.	None
async_publish_circuit_breaker	Any | None	Async circuit breaker for publish operations.	None

Source code in src/rabbitkit/middleware/circuit_breaker.py

class CircuitBreakerMiddleware(BaseMiddleware):
    """Wraps handler execution and publish operations with circuit breaker.

    If circuit breaker is not provided (None), all operations pass through
    without any wrapping (no-op mode).

    Usage::

        # any CircuitBreakerProtocol-compatible implementation, e.g. pybreaker
        cb = MyCircuitBreaker(name="rabbitmq", fail_max=5, reset_timeout=60)
        middleware = CircuitBreakerMiddleware(circuit_breaker=cb)

        # Or with separate publish CB:
        middleware = CircuitBreakerMiddleware(
            circuit_breaker=consume_cb,
            publish_circuit_breaker=publish_cb,
        )

    Args:
        circuit_breaker: Circuit breaker for consume operations (handler execution).
            Must satisfy CircuitBreakerProtocol. None for no-op.
        publish_circuit_breaker: Circuit breaker for publish operations.
            If None, uses circuit_breaker for both. If circuit_breaker is also
            None, publish operations pass through.
        async_circuit_breaker: Async circuit breaker for consume operations.
            Must satisfy AsyncCircuitBreakerProtocol. If None, falls back to
            wrapping the sync circuit_breaker in async context.
        async_publish_circuit_breaker: Async circuit breaker for publish operations.
    """

    def __init__(
        self,
        circuit_breaker: Any | None = None,
        publish_circuit_breaker: Any | None = None,
        *,
        async_circuit_breaker: Any | None = None,
        async_publish_circuit_breaker: Any | None = None,
    ) -> None:
        self._cb = circuit_breaker
        self._publish_cb = publish_circuit_breaker or circuit_breaker
        self._async_cb = async_circuit_breaker
        self._async_publish_cb = async_publish_circuit_breaker or async_circuit_breaker

    # ── Consume-side ──────────────────────────────────────────────────────

    def consume_scope(
        self,
        call_next: Callable[[RabbitMessage], Any],
        message: RabbitMessage,
    ) -> Any:
        """Wrap handler execution with circuit breaker (sync)."""
        if self._cb is None:
            return call_next(message)
        return self._cb.call(call_next, message)

    async def consume_scope_async(
        self,
        call_next: Callable[[RabbitMessage], Awaitable[Any]],
        message: RabbitMessage,
    ) -> Any:
        """Wrap handler execution with circuit breaker (async)."""
        if self._async_cb is not None:
            return await self._async_cb.call_async(call_next, message)
        if self._cb is not None:
            # Sync CB cannot safely wrap an async handler — the sync call()
            # would receive a coroutine object, not the result. Raise at call
            # time so the misconfiguration surfaces immediately rather than
            # silently skipping every handler invocation.
            raise TypeError(
                "CircuitBreakerMiddleware: async handler requires "
                "async_circuit_breaker=. Providing only a sync circuit_breaker "
                "with an async broker is not supported (the handler would never "
                "run). Pass async_circuit_breaker= instead."
            )
        return await call_next(message)

    # ── Publish-side ──────────────────────────────────────────────────────

    def publish_scope(
        self,
        call_next: Callable[[MessageEnvelope], Any],
        envelope: MessageEnvelope,
    ) -> Any:
        """Wrap publish with circuit breaker (sync)."""
        if self._publish_cb is None:
            return call_next(envelope)
        return self._publish_cb.call(call_next, envelope)

    async def publish_scope_async(
        self,
        call_next: Callable[[MessageEnvelope], Awaitable[Any]],
        envelope: MessageEnvelope,
    ) -> Any:
        """Wrap publish with circuit breaker (async)."""
        if self._async_publish_cb is not None:
            return await self._async_publish_cb.call_async(call_next, envelope)
        if self._publish_cb is not None:
            raise TypeError(
                "CircuitBreakerMiddleware: async publish requires "
                "async_publish_circuit_breaker=. Providing only a sync "
                "publish_circuit_breaker with an async broker is not supported."
            )
        return await call_next(envelope)

Methods:

consume_scope(call_next: Callable[[RabbitMessage], Any], message: RabbitMessage) -> Any

Wrap handler execution with circuit breaker (sync).

Source code in src/rabbitkit/middleware/circuit_breaker.py

def consume_scope(
    self,
    call_next: Callable[[RabbitMessage], Any],
    message: RabbitMessage,
) -> Any:
    """Wrap handler execution with circuit breaker (sync)."""
    if self._cb is None:
        return call_next(message)
    return self._cb.call(call_next, message)

consume_scope_async(call_next: Callable[[RabbitMessage], Awaitable[Any]], message: RabbitMessage) -> Any async

Wrap handler execution with circuit breaker (async).

Source code in src/rabbitkit/middleware/circuit_breaker.py

async def consume_scope_async(
    self,
    call_next: Callable[[RabbitMessage], Awaitable[Any]],
    message: RabbitMessage,
) -> Any:
    """Wrap handler execution with circuit breaker (async)."""
    if self._async_cb is not None:
        return await self._async_cb.call_async(call_next, message)
    if self._cb is not None:
        # Sync CB cannot safely wrap an async handler — the sync call()
        # would receive a coroutine object, not the result. Raise at call
        # time so the misconfiguration surfaces immediately rather than
        # silently skipping every handler invocation.
        raise TypeError(
            "CircuitBreakerMiddleware: async handler requires "
            "async_circuit_breaker=. Providing only a sync circuit_breaker "
            "with an async broker is not supported (the handler would never "
            "run). Pass async_circuit_breaker= instead."
        )
    return await call_next(message)

publish_scope(call_next: Callable[[MessageEnvelope], Any], envelope: MessageEnvelope) -> Any

Wrap publish with circuit breaker (sync).

Source code in src/rabbitkit/middleware/circuit_breaker.py

def publish_scope(
    self,
    call_next: Callable[[MessageEnvelope], Any],
    envelope: MessageEnvelope,
) -> Any:
    """Wrap publish with circuit breaker (sync)."""
    if self._publish_cb is None:
        return call_next(envelope)
    return self._publish_cb.call(call_next, envelope)

publish_scope_async(call_next: Callable[[MessageEnvelope], Awaitable[Any]], envelope: MessageEnvelope) -> Any async

Wrap publish with circuit breaker (async).

Source code in src/rabbitkit/middleware/circuit_breaker.py

async def publish_scope_async(
    self,
    call_next: Callable[[MessageEnvelope], Awaitable[Any]],
    envelope: MessageEnvelope,
) -> Any:
    """Wrap publish with circuit breaker (async)."""
    if self._async_publish_cb is not None:
        return await self._async_publish_cb.call_async(call_next, envelope)
    if self._publish_cb is not None:
        raise TypeError(
            "CircuitBreakerMiddleware: async publish requires "
            "async_publish_circuit_breaker=. Providing only a sync "
            "publish_circuit_breaker with an async broker is not supported."
        )
    return await call_next(envelope)

CircuitBreakerOpenError

Bases: Exception

Raised when circuit breaker is open and rejects an operation.

Source code in src/rabbitkit/middleware/circuit_breaker.py

class CircuitBreakerOpenError(Exception):
    """Raised when circuit breaker is open and rejects an operation."""

TimeoutMiddleware

TimeoutMiddleware

Bases: BaseMiddleware

Enforces a maximum processing time per message.

Async: uses asyncio.wait_for() — clean cancellation. Sync: runs call_next in a daemon threading.Thread and raises HandlerTimeoutError if the thread is still alive at the deadline.

.. warning:: CPython cannot safely kill a thread, so a sync handler that exceeds the timeout is abandoned — it keeps running detached until it finishes naturally. This can leak threads and resources for long-running / blocked IO handlers. Use an async handler for true cooperative cancellation. When a sync timeout fires, a CRITICAL log record is emitted and the optional on_timeout callback (if configured) is invoked so the abandonment is observable (e.g. increment a metric counter).

Source code in src/rabbitkit/middleware/timeout.py

class TimeoutMiddleware(BaseMiddleware):
    """Enforces a maximum processing time per message.

    Async: uses ``asyncio.wait_for()`` — clean cancellation.
    Sync: runs ``call_next`` in a daemon ``threading.Thread`` and raises
    ``HandlerTimeoutError`` if the thread is still alive at the deadline.

    .. warning::
        CPython cannot safely kill a thread, so a sync handler that exceeds the
        timeout is **abandoned** — it keeps running detached until it finishes
        naturally. This can leak threads and resources for long-running / blocked
        IO handlers. Use an **async** handler for true cooperative cancellation.
        When a sync timeout fires, a CRITICAL log record is emitted and the
        optional ``on_timeout`` callback (if configured) is invoked so the
        abandonment is observable (e.g. increment a metric counter).
    """

    def __init__(
        self,
        config: TimeoutConfig | None = None,
        *,
        on_timeout: Callable[[RabbitMessage, float], None] | None = None,
    ) -> None:
        self._config = config or TimeoutConfig()
        self._on_timeout = on_timeout
        # Observable counter of sync threads abandoned due to timeout.
        self.abandoned_threads: int = 0

    def consume_scope(
        self,
        call_next: Any,
        message: RabbitMessage,
    ) -> Any:
        """Sync timeout — runs call_next in a thread with timeout.

        If the deadline elapses with the thread still alive, the thread is
        abandoned (CPython cannot kill it), a CRITICAL log is emitted, the
        ``abandoned_threads`` counter is incremented, ``on_timeout`` (if any)
        is invoked, and ``HandlerTimeoutError`` is raised.

        H9 — settlement is exclusively from the consumer (this) thread, never
        the background thread, even under ``AckPolicy.MANUAL`` (where the
        handler itself calls ``message.ack()``/``nack()``/``reject()``):
        while the background thread is running, ``message``'s settlement
        functions are swapped for stand-ins that CAPTURE (rather than
        execute) any settlement attempt made from that specific thread —
        calling the real pika-backed function from there, while THIS thread
        is blocked in ``thread.join()`` (i.e. not pumping the connection's
        I/O loop), can deadlock: the settlement call would marshal onto this
        thread and wait for it to drain the callback, but this thread is
        itself waiting on the background thread to finish. If the background
        thread finishes within the deadline, any settlement it captured is
        replayed for real on this (consumer/owner) thread, safely, after
        ``join()`` returns. If it does not, the guards stay installed
        (deliberately NOT restored — the background thread may still call
        ack()/nack()/reject() at any point later) but their thread-identity
        check means any FUTURE call specifically from that thread is
        discarded, while THIS thread's own subsequent settlement (e.g. via
        AckPolicy/RetryMiddleware after ``HandlerTimeoutError`` propagates
        below) is routed straight to the real fn — safe, since it's a
        same-thread call.
        """
        result_holder: list[Any] = []
        exception_holder: list[BaseException] = []
        captured_settlement: list[Callable[[], None]] = []
        timed_out = threading.Event()

        real_ack, real_nack, real_reject = message._ack_fn, message._nack_fn, message._reject_fn

        def _guarded_ack() -> None:
            if threading.get_ident() != thread.ident:
                if real_ack is not None:
                    real_ack()
                return
            if timed_out.is_set():
                logger.warning("Discarding ack() from an abandoned timed-out handler thread")
                # Raise (rather than return) so RabbitMessage.ack() does not
                # proceed to set _disposition="acked" for a call that never
                # actually touched the channel — that would silently block
                # the consumer thread's own later, real settlement.
                raise _DiscardedSettlement
            if real_ack is not None:
                captured_settlement.append(real_ack)

        def _guarded_nack(requeue: bool = True) -> None:
            if threading.get_ident() != thread.ident:
                if real_nack is not None:
                    real_nack(requeue)
                return
            if timed_out.is_set():
                logger.warning("Discarding nack() from an abandoned timed-out handler thread")
                raise _DiscardedSettlement
            if real_nack is not None:
                captured_settlement.append(lambda: real_nack(requeue))

        def _guarded_reject(requeue: bool = False) -> None:
            if threading.get_ident() != thread.ident:
                if real_reject is not None:
                    real_reject(requeue)
                return
            if timed_out.is_set():
                logger.warning("Discarding reject() from an abandoned timed-out handler thread")
                raise _DiscardedSettlement
            if real_reject is not None:
                captured_settlement.append(lambda: real_reject(requeue))

        # Only install a guard where a real fn exists — leaving an already-None
        # fn as None preserves message.ack()/nack()/reject()'s own "no
        # settlement fn set" RuntimeError for e.g. no-ack deliveries, instead
        # of silently swallowing it inside the guard.
        if real_ack is not None:
            message._ack_fn = _guarded_ack
        if real_nack is not None:
            message._nack_fn = _guarded_nack
        if real_reject is not None:
            message._reject_fn = _guarded_reject

        def _run() -> None:
            try:
                result_holder.append(call_next(message))
            except BaseException as exc:
                exception_holder.append(exc)

        thread = threading.Thread(target=_run, daemon=True)
        thread.start()
        thread.join(timeout=self._config.timeout_seconds)

        if thread.is_alive():
            # CPython cannot safely kill a thread — the handler keeps running
            # detached. Make the abandonment explicit and observable.
            #
            # Deliberately do NOT restore the real settlement fns here: the
            # background thread is still running and may call
            # ack()/nack()/reject() at any point after this method returns —
            # if the real fn were restored, that eventual call would hit the
            # pika channel directly from a non-owner thread. The guards stay
            # installed; their threading.get_ident() check already routes
            # THIS thread's own subsequent settlement (AckPolicy/
            # RetryMiddleware, after the exception below propagates) straight
            # to the real fn (safe, same-thread), while any FUTURE call
            # specifically from the background thread is discarded now that
            # timed_out is set.
            timed_out.set()
            self.abandoned_threads += 1
            logger.critical(
                "Sync handler exceeded %.1fs timeout; thread abandoned (still running). "
                "Use an async handler for real cancellation. "
                "abandoned_threads=%d",
                self._config.timeout_seconds,
                self.abandoned_threads,
            )
            if self._on_timeout is not None:
                try:
                    self._on_timeout(message, self._config.timeout_seconds)
                except Exception:  # pragma: no cover - callback must not break flow
                    logger.exception("on_timeout callback raised")
            raise HandlerTimeoutError(self._config.timeout_seconds)

        # Finished within the deadline — restore the real fns, then replay
        # any settlement the background thread captured, for real, on this
        # (consumer/owner) thread.
        message._ack_fn, message._nack_fn, message._reject_fn = real_ack, real_nack, real_reject
        for replay in captured_settlement:
            replay()

        if exception_holder:
            raise exception_holder[0]

        return result_holder[0] if result_holder else None

    async def consume_scope_async(
        self,
        call_next: Any,
        message: RabbitMessage,
    ) -> Any:
        """Async timeout — uses asyncio.wait_for()."""
        try:
            return await asyncio.wait_for(
                call_next(message),
                timeout=self._config.timeout_seconds,
            )
        except TimeoutError:
            raise HandlerTimeoutError(self._config.timeout_seconds) from None

Methods:

consume_scope(call_next: Any, message: RabbitMessage) -> Any

Sync timeout — runs call_next in a thread with timeout.

If the deadline elapses with the thread still alive, the thread is abandoned (CPython cannot kill it), a CRITICAL log is emitted, the abandoned_threads counter is incremented, on_timeout (if any) is invoked, and HandlerTimeoutError is raised.

H9 — settlement is exclusively from the consumer (this) thread, never the background thread, even under AckPolicy.MANUAL (where the handler itself calls message.ack()/nack()/reject()): while the background thread is running, message's settlement functions are swapped for stand-ins that CAPTURE (rather than execute) any settlement attempt made from that specific thread — calling the real pika-backed function from there, while THIS thread is blocked in thread.join() (i.e. not pumping the connection's I/O loop), can deadlock: the settlement call would marshal onto this thread and wait for it to drain the callback, but this thread is itself waiting on the background thread to finish. If the background thread finishes within the deadline, any settlement it captured is replayed for real on this (consumer/owner) thread, safely, after join() returns. If it does not, the guards stay installed (deliberately NOT restored — the background thread may still call ack()/nack()/reject() at any point later) but their thread-identity check means any FUTURE call specifically from that thread is discarded, while THIS thread's own subsequent settlement (e.g. via AckPolicy/RetryMiddleware after HandlerTimeoutError propagates below) is routed straight to the real fn — safe, since it's a same-thread call.

Source code in src/rabbitkit/middleware/timeout.py

def consume_scope(
    self,
    call_next: Any,
    message: RabbitMessage,
) -> Any:
    """Sync timeout — runs call_next in a thread with timeout.

    If the deadline elapses with the thread still alive, the thread is
    abandoned (CPython cannot kill it), a CRITICAL log is emitted, the
    ``abandoned_threads`` counter is incremented, ``on_timeout`` (if any)
    is invoked, and ``HandlerTimeoutError`` is raised.

    H9 — settlement is exclusively from the consumer (this) thread, never
    the background thread, even under ``AckPolicy.MANUAL`` (where the
    handler itself calls ``message.ack()``/``nack()``/``reject()``):
    while the background thread is running, ``message``'s settlement
    functions are swapped for stand-ins that CAPTURE (rather than
    execute) any settlement attempt made from that specific thread —
    calling the real pika-backed function from there, while THIS thread
    is blocked in ``thread.join()`` (i.e. not pumping the connection's
    I/O loop), can deadlock: the settlement call would marshal onto this
    thread and wait for it to drain the callback, but this thread is
    itself waiting on the background thread to finish. If the background
    thread finishes within the deadline, any settlement it captured is
    replayed for real on this (consumer/owner) thread, safely, after
    ``join()`` returns. If it does not, the guards stay installed
    (deliberately NOT restored — the background thread may still call
    ack()/nack()/reject() at any point later) but their thread-identity
    check means any FUTURE call specifically from that thread is
    discarded, while THIS thread's own subsequent settlement (e.g. via
    AckPolicy/RetryMiddleware after ``HandlerTimeoutError`` propagates
    below) is routed straight to the real fn — safe, since it's a
    same-thread call.
    """
    result_holder: list[Any] = []
    exception_holder: list[BaseException] = []
    captured_settlement: list[Callable[[], None]] = []
    timed_out = threading.Event()

    real_ack, real_nack, real_reject = message._ack_fn, message._nack_fn, message._reject_fn

    def _guarded_ack() -> None:
        if threading.get_ident() != thread.ident:
            if real_ack is not None:
                real_ack()
            return
        if timed_out.is_set():
            logger.warning("Discarding ack() from an abandoned timed-out handler thread")
            # Raise (rather than return) so RabbitMessage.ack() does not
            # proceed to set _disposition="acked" for a call that never
            # actually touched the channel — that would silently block
            # the consumer thread's own later, real settlement.
            raise _DiscardedSettlement
        if real_ack is not None:
            captured_settlement.append(real_ack)

    def _guarded_nack(requeue: bool = True) -> None:
        if threading.get_ident() != thread.ident:
            if real_nack is not None:
                real_nack(requeue)
            return
        if timed_out.is_set():
            logger.warning("Discarding nack() from an abandoned timed-out handler thread")
            raise _DiscardedSettlement
        if real_nack is not None:
            captured_settlement.append(lambda: real_nack(requeue))

    def _guarded_reject(requeue: bool = False) -> None:
        if threading.get_ident() != thread.ident:
            if real_reject is not None:
                real_reject(requeue)
            return
        if timed_out.is_set():
            logger.warning("Discarding reject() from an abandoned timed-out handler thread")
            raise _DiscardedSettlement
        if real_reject is not None:
            captured_settlement.append(lambda: real_reject(requeue))

    # Only install a guard where a real fn exists — leaving an already-None
    # fn as None preserves message.ack()/nack()/reject()'s own "no
    # settlement fn set" RuntimeError for e.g. no-ack deliveries, instead
    # of silently swallowing it inside the guard.
    if real_ack is not None:
        message._ack_fn = _guarded_ack
    if real_nack is not None:
        message._nack_fn = _guarded_nack
    if real_reject is not None:
        message._reject_fn = _guarded_reject

    def _run() -> None:
        try:
            result_holder.append(call_next(message))
        except BaseException as exc:
            exception_holder.append(exc)

    thread = threading.Thread(target=_run, daemon=True)
    thread.start()
    thread.join(timeout=self._config.timeout_seconds)

    if thread.is_alive():
        # CPython cannot safely kill a thread — the handler keeps running
        # detached. Make the abandonment explicit and observable.
        #
        # Deliberately do NOT restore the real settlement fns here: the
        # background thread is still running and may call
        # ack()/nack()/reject() at any point after this method returns —
        # if the real fn were restored, that eventual call would hit the
        # pika channel directly from a non-owner thread. The guards stay
        # installed; their threading.get_ident() check already routes
        # THIS thread's own subsequent settlement (AckPolicy/
        # RetryMiddleware, after the exception below propagates) straight
        # to the real fn (safe, same-thread), while any FUTURE call
        # specifically from the background thread is discarded now that
        # timed_out is set.
        timed_out.set()
        self.abandoned_threads += 1
        logger.critical(
            "Sync handler exceeded %.1fs timeout; thread abandoned (still running). "
            "Use an async handler for real cancellation. "
            "abandoned_threads=%d",
            self._config.timeout_seconds,
            self.abandoned_threads,
        )
        if self._on_timeout is not None:
            try:
                self._on_timeout(message, self._config.timeout_seconds)
            except Exception:  # pragma: no cover - callback must not break flow
                logger.exception("on_timeout callback raised")
        raise HandlerTimeoutError(self._config.timeout_seconds)

    # Finished within the deadline — restore the real fns, then replay
    # any settlement the background thread captured, for real, on this
    # (consumer/owner) thread.
    message._ack_fn, message._nack_fn, message._reject_fn = real_ack, real_nack, real_reject
    for replay in captured_settlement:
        replay()

    if exception_holder:
        raise exception_holder[0]

    return result_holder[0] if result_holder else None

consume_scope_async(call_next: Any, message: RabbitMessage) -> Any async

Async timeout — uses asyncio.wait_for().

Source code in src/rabbitkit/middleware/timeout.py

async def consume_scope_async(
    self,
    call_next: Any,
    message: RabbitMessage,
) -> Any:
    """Async timeout — uses asyncio.wait_for()."""
    try:
        return await asyncio.wait_for(
            call_next(message),
            timeout=self._config.timeout_seconds,
        )
    except TimeoutError:
        raise HandlerTimeoutError(self._config.timeout_seconds) from None

MetricsMiddleware

MetricsMiddleware

Bases: BaseMiddleware

Tracks consume and publish metrics via a pluggable MetricsCollector.

If collector is None, all operations pass through without any overhead (no-op mode).

Usage::

collector = PrometheusCollector()
middleware = MetricsMiddleware(collector)

Or with a custom collector::

class MyCollector:
    def inc_counter(self, name, labels, value=1.0): ...
    def observe_histogram(self, name, labels, value): ...

middleware = MetricsMiddleware(MyCollector())

Parameters:

Name	Type	Description	Default
collector	MetricsCollector | None	Any object satisfying the MetricsCollector protocol. None for no-op (passthrough) mode.	None

Source code in src/rabbitkit/middleware/metrics.py

class MetricsMiddleware(BaseMiddleware):
    """Tracks consume and publish metrics via a pluggable MetricsCollector.

    If ``collector`` is None, all operations pass through without
    any overhead (no-op mode).

    Usage::

        collector = PrometheusCollector()
        middleware = MetricsMiddleware(collector)

    Or with a custom collector::

        class MyCollector:
            def inc_counter(self, name, labels, value=1.0): ...
            def observe_histogram(self, name, labels, value): ...

        middleware = MetricsMiddleware(MyCollector())

    Args:
        collector: Any object satisfying the MetricsCollector protocol.
            None for no-op (passthrough) mode.
    """

    def __init__(
        self,
        collector: MetricsCollector | None = None,
        config: MetricsConfig | None = None,
    ) -> None:
        self._collector = collector
        self._cfg = config or MetricsConfig()

    @property
    def collector(self) -> MetricsCollector | None:
        """The configured collector (None in no-op mode). Read by
        ``HandlerPipeline``/``RetryMiddleware`` to emit settlement/retry
        metrics they observe but this middleware itself cannot (M2) --
        settlement happens in the pipeline's own ack-orchestration code,
        outside this middleware's ``consume_scope``."""
        return self._collector

    @property
    def config(self) -> MetricsConfig:
        return self._cfg

    def record_settlement(self, message: RabbitMessage, disposition: str) -> None:
        """Emit the ack/nack/reject counter for a settled message (M2).

        ``consume_scope``/``consume_scope_async`` only wrap handler
        execution -- final settlement (ack/nack/reject per AckPolicy) is
        decided by the pipeline's own ack-orchestration code, which runs
        AFTER this middleware's wrapped call returns. ``HandlerPipeline``
        calls this directly once a route's message is settled, if a
        ``MetricsMiddleware`` is present on that route.
        """
        if self._collector is None:
            return
        name = {
            "acked": self._cfg.messages_acked_total,
            "nacked": self._cfg.messages_nacked_total,
            "rejected": self._cfg.messages_rejected_total,
        }.get(disposition)
        if name is None:
            return  # pragma: no cover - defensive, disposition is always one of the three
        self._collector.inc_counter(name, {"queue": _queue_label(message)})

    # ── Consume-side ──────────────────────────────────────────────────

    def consume_scope(
        self,
        call_next: Callable[[RabbitMessage], Any],
        message: RabbitMessage,
    ) -> Any:
        """Wrap handler execution with metrics tracking (sync)."""
        if self._collector is None:
            return call_next(message)

        queue = _queue_label(message)
        if message.redelivered:
            # Broker-redelivery rate: a sustained rise means handlers are
            # dying/timing out before acking, which the success/error
            # consume counters alone can't distinguish from normal traffic.
            self._collector.inc_counter(
                self._cfg.messages_redelivered_total,
                {"queue": queue},
            )
        start = time.monotonic()
        try:
            result = call_next(message)
        except BaseException:
            self._collector.inc_counter(
                self._cfg.consumed_total,
                {"queue": queue, "status": "error"},
            )
            self._collector.observe_histogram(
                self._cfg.processing_seconds,
                {"queue": queue},
                time.monotonic() - start,
            )
            raise
        else:
            self._collector.inc_counter(
                self._cfg.consumed_total,
                {"queue": queue, "status": "success"},
            )
            self._collector.observe_histogram(
                self._cfg.processing_seconds,
                {"queue": queue},
                time.monotonic() - start,
            )
            return result

    async def consume_scope_async(
        self,
        call_next: Callable[[RabbitMessage], Awaitable[Any]],
        message: RabbitMessage,
    ) -> Any:
        """Wrap handler execution with metrics tracking (async)."""
        if self._collector is None:
            return await call_next(message)

        queue = _queue_label(message)
        if message.redelivered:
            # Broker-redelivery rate — see consume_scope.
            self._collector.inc_counter(
                self._cfg.messages_redelivered_total,
                {"queue": queue},
            )
        start = time.monotonic()
        try:
            result = await call_next(message)
        except BaseException:
            self._collector.inc_counter(
                self._cfg.consumed_total,
                {"queue": queue, "status": "error"},
            )
            self._collector.observe_histogram(
                self._cfg.processing_seconds,
                {"queue": queue},
                time.monotonic() - start,
            )
            raise
        else:
            self._collector.inc_counter(
                self._cfg.consumed_total,
                {"queue": queue, "status": "success"},
            )
            self._collector.observe_histogram(
                self._cfg.processing_seconds,
                {"queue": queue},
                time.monotonic() - start,
            )
            return result

    # ── Publish-side ──────────────────────────────────────────────────

    def publish_scope(
        self,
        call_next: Callable[[MessageEnvelope], Any],
        envelope: MessageEnvelope,
    ) -> Any:
        """Wrap outgoing publish with metrics tracking (sync)."""
        if self._collector is None:
            return call_next(envelope)

        exchange = envelope.exchange or "default"
        start = time.monotonic()
        try:
            result = call_next(envelope)
        except BaseException:
            self._collector.inc_counter(
                self._cfg.published_total,
                {"exchange": exchange, "status": "error"},
            )
            self._collector.observe_histogram(
                self._cfg.publish_seconds,
                {"exchange": exchange},
                time.monotonic() - start,
            )
            raise
        else:
            self._collector.inc_counter(
                self._cfg.published_total,
                {"exchange": exchange, "status": _publish_status_label(result)},
            )
            self._collector.observe_histogram(
                self._cfg.publish_seconds,
                {"exchange": exchange},
                time.monotonic() - start,
            )
            return result

    async def publish_scope_async(
        self,
        call_next: Callable[[MessageEnvelope], Awaitable[Any]],
        envelope: MessageEnvelope,
    ) -> Any:
        """Wrap outgoing publish with metrics tracking (async)."""
        if self._collector is None:
            return await call_next(envelope)

        exchange = envelope.exchange or "default"
        start = time.monotonic()
        try:
            result = await call_next(envelope)
        except BaseException:
            self._collector.inc_counter(
                self._cfg.published_total,
                {"exchange": exchange, "status": "error"},
            )
            self._collector.observe_histogram(
                self._cfg.publish_seconds,
                {"exchange": exchange},
                time.monotonic() - start,
            )
            raise
        else:
            self._collector.inc_counter(
                self._cfg.published_total,
                {"exchange": exchange, "status": _publish_status_label(result)},
            )
            self._collector.observe_histogram(
                self._cfg.publish_seconds,
                {"exchange": exchange},
                time.monotonic() - start,
            )
            return result

Attributes

collector: MetricsCollector | None property

The configured collector (None in no-op mode). Read by HandlerPipeline/RetryMiddleware to emit settlement/retry metrics they observe but this middleware itself cannot (M2) -- settlement happens in the pipeline's own ack-orchestration code, outside this middleware's consume_scope.

Methods:

record_settlement(message: RabbitMessage, disposition: str) -> None

Emit the ack/nack/reject counter for a settled message (M2).

consume_scope/consume_scope_async only wrap handler execution -- final settlement (ack/nack/reject per AckPolicy) is decided by the pipeline's own ack-orchestration code, which runs AFTER this middleware's wrapped call returns. HandlerPipeline calls this directly once a route's message is settled, if a MetricsMiddleware is present on that route.

Source code in src/rabbitkit/middleware/metrics.py

def record_settlement(self, message: RabbitMessage, disposition: str) -> None:
    """Emit the ack/nack/reject counter for a settled message (M2).

    ``consume_scope``/``consume_scope_async`` only wrap handler
    execution -- final settlement (ack/nack/reject per AckPolicy) is
    decided by the pipeline's own ack-orchestration code, which runs
    AFTER this middleware's wrapped call returns. ``HandlerPipeline``
    calls this directly once a route's message is settled, if a
    ``MetricsMiddleware`` is present on that route.
    """
    if self._collector is None:
        return
    name = {
        "acked": self._cfg.messages_acked_total,
        "nacked": self._cfg.messages_nacked_total,
        "rejected": self._cfg.messages_rejected_total,
    }.get(disposition)
    if name is None:
        return  # pragma: no cover - defensive, disposition is always one of the three
    self._collector.inc_counter(name, {"queue": _queue_label(message)})

consume_scope(call_next: Callable[[RabbitMessage], Any], message: RabbitMessage) -> Any

Wrap handler execution with metrics tracking (sync).

Source code in src/rabbitkit/middleware/metrics.py

def consume_scope(
    self,
    call_next: Callable[[RabbitMessage], Any],
    message: RabbitMessage,
) -> Any:
    """Wrap handler execution with metrics tracking (sync)."""
    if self._collector is None:
        return call_next(message)

    queue = _queue_label(message)
    if message.redelivered:
        # Broker-redelivery rate: a sustained rise means handlers are
        # dying/timing out before acking, which the success/error
        # consume counters alone can't distinguish from normal traffic.
        self._collector.inc_counter(
            self._cfg.messages_redelivered_total,
            {"queue": queue},
        )
    start = time.monotonic()
    try:
        result = call_next(message)
    except BaseException:
        self._collector.inc_counter(
            self._cfg.consumed_total,
            {"queue": queue, "status": "error"},
        )
        self._collector.observe_histogram(
            self._cfg.processing_seconds,
            {"queue": queue},
            time.monotonic() - start,
        )
        raise
    else:
        self._collector.inc_counter(
            self._cfg.consumed_total,
            {"queue": queue, "status": "success"},
        )
        self._collector.observe_histogram(
            self._cfg.processing_seconds,
            {"queue": queue},
            time.monotonic() - start,
        )
        return result

consume_scope_async(call_next: Callable[[RabbitMessage], Awaitable[Any]], message: RabbitMessage) -> Any async

Wrap handler execution with metrics tracking (async).

Source code in src/rabbitkit/middleware/metrics.py

async def consume_scope_async(
    self,
    call_next: Callable[[RabbitMessage], Awaitable[Any]],
    message: RabbitMessage,
) -> Any:
    """Wrap handler execution with metrics tracking (async)."""
    if self._collector is None:
        return await call_next(message)

    queue = _queue_label(message)
    if message.redelivered:
        # Broker-redelivery rate — see consume_scope.
        self._collector.inc_counter(
            self._cfg.messages_redelivered_total,
            {"queue": queue},
        )
    start = time.monotonic()
    try:
        result = await call_next(message)
    except BaseException:
        self._collector.inc_counter(
            self._cfg.consumed_total,
            {"queue": queue, "status": "error"},
        )
        self._collector.observe_histogram(
            self._cfg.processing_seconds,
            {"queue": queue},
            time.monotonic() - start,
        )
        raise
    else:
        self._collector.inc_counter(
            self._cfg.consumed_total,
            {"queue": queue, "status": "success"},
        )
        self._collector.observe_histogram(
            self._cfg.processing_seconds,
            {"queue": queue},
            time.monotonic() - start,
        )
        return result

publish_scope(call_next: Callable[[MessageEnvelope], Any], envelope: MessageEnvelope) -> Any

Wrap outgoing publish with metrics tracking (sync).

Source code in src/rabbitkit/middleware/metrics.py

def publish_scope(
    self,
    call_next: Callable[[MessageEnvelope], Any],
    envelope: MessageEnvelope,
) -> Any:
    """Wrap outgoing publish with metrics tracking (sync)."""
    if self._collector is None:
        return call_next(envelope)

    exchange = envelope.exchange or "default"
    start = time.monotonic()
    try:
        result = call_next(envelope)
    except BaseException:
        self._collector.inc_counter(
            self._cfg.published_total,
            {"exchange": exchange, "status": "error"},
        )
        self._collector.observe_histogram(
            self._cfg.publish_seconds,
            {"exchange": exchange},
            time.monotonic() - start,
        )
        raise
    else:
        self._collector.inc_counter(
            self._cfg.published_total,
            {"exchange": exchange, "status": _publish_status_label(result)},
        )
        self._collector.observe_histogram(
            self._cfg.publish_seconds,
            {"exchange": exchange},
            time.monotonic() - start,
        )
        return result

publish_scope_async(call_next: Callable[[MessageEnvelope], Awaitable[Any]], envelope: MessageEnvelope) -> Any async

Wrap outgoing publish with metrics tracking (async).

Source code in src/rabbitkit/middleware/metrics.py

async def publish_scope_async(
    self,
    call_next: Callable[[MessageEnvelope], Awaitable[Any]],
    envelope: MessageEnvelope,
) -> Any:
    """Wrap outgoing publish with metrics tracking (async)."""
    if self._collector is None:
        return await call_next(envelope)

    exchange = envelope.exchange or "default"
    start = time.monotonic()
    try:
        result = await call_next(envelope)
    except BaseException:
        self._collector.inc_counter(
            self._cfg.published_total,
            {"exchange": exchange, "status": "error"},
        )
        self._collector.observe_histogram(
            self._cfg.publish_seconds,
            {"exchange": exchange},
            time.monotonic() - start,
        )
        raise
    else:
        self._collector.inc_counter(
            self._cfg.published_total,
            {"exchange": exchange, "status": _publish_status_label(result)},
        )
        self._collector.observe_histogram(
            self._cfg.publish_seconds,
            {"exchange": exchange},
            time.monotonic() - start,
        )
        return result

MetricsCollector

Bases: Protocol

Protocol for metrics collection — works with Prometheus, StatsD, etc.

Source code in src/rabbitkit/middleware/metrics.py

@runtime_checkable
class MetricsCollector(Protocol):
    """Protocol for metrics collection — works with Prometheus, StatsD, etc."""

    def inc_counter(self, name: str, labels: dict[str, str], value: float = 1.0) -> None:
        """Increment a counter metric."""
        ...

    def observe_histogram(self, name: str, labels: dict[str, str], value: float) -> None:
        """Observe a value on a histogram metric."""
        ...

    def set_gauge(self, name: str, labels: dict[str, str], value: float) -> None:
        """Set a gauge metric to an absolute value (e.g. queue depth)."""
        ...

Methods:

inc_counter(name: str, labels: dict[str, str], value: float = 1.0) -> None

Increment a counter metric.

Source code in src/rabbitkit/middleware/metrics.py

def inc_counter(self, name: str, labels: dict[str, str], value: float = 1.0) -> None:
    """Increment a counter metric."""
    ...

observe_histogram(name: str, labels: dict[str, str], value: float) -> None

Observe a value on a histogram metric.

Source code in src/rabbitkit/middleware/metrics.py

def observe_histogram(self, name: str, labels: dict[str, str], value: float) -> None:
    """Observe a value on a histogram metric."""
    ...

set_gauge(name: str, labels: dict[str, str], value: float) -> None

Set a gauge metric to an absolute value (e.g. queue depth).

Source code in src/rabbitkit/middleware/metrics.py

def set_gauge(self, name: str, labels: dict[str, str], value: float) -> None:
    """Set a gauge metric to an absolute value (e.g. queue depth)."""
    ...

PrometheusCollector

Concrete MetricsCollector that wraps the prometheus_client library.

The prometheus_client import is lazy — the library is only required when this class is instantiated, not when the module is imported.

Usage::

collector = PrometheusCollector()
middleware = MetricsMiddleware(collector)

Source code in src/rabbitkit/middleware/metrics.py

class PrometheusCollector:
    """Concrete MetricsCollector that wraps the ``prometheus_client`` library.

    The ``prometheus_client`` import is lazy — the library is only required
    when this class is instantiated, not when the module is imported.

    Usage::

        collector = PrometheusCollector()
        middleware = MetricsMiddleware(collector)
    """

    def __init__(self) -> None:
        try:
            import prometheus_client
        except ImportError as exc:
            msg = (
                "PrometheusCollector requires the 'prometheus_client' package. "
                "Install it with: pip install prometheus-client"
            )
            raise ImportError(msg) from exc

        self._counters: dict[str, Any] = {}
        self._histograms: dict[str, Any] = {}
        self._gauges: dict[str, Any] = {}
        self._prometheus_client = prometheus_client

    def _get_counter(self, name: str, label_names: tuple[str, ...]) -> Any:
        if name not in self._counters:
            self._counters[name] = self._prometheus_client.Counter(
                name,
                f"rabbitkit {name}",
                label_names,
            )
        return self._counters[name]

    def _get_histogram(self, name: str, label_names: tuple[str, ...]) -> Any:
        if name not in self._histograms:
            self._histograms[name] = self._prometheus_client.Histogram(
                name,
                f"rabbitkit {name}",
                label_names,
            )
        return self._histograms[name]

    def _get_gauge(self, name: str, label_names: tuple[str, ...]) -> Any:
        if name not in self._gauges:
            self._gauges[name] = self._prometheus_client.Gauge(
                name,
                f"rabbitkit {name}",
                label_names,
            )
        return self._gauges[name]

    def inc_counter(self, name: str, labels: dict[str, str], value: float = 1.0) -> None:
        """Increment a Prometheus counter."""
        label_names = tuple(sorted(labels.keys()))
        counter = self._get_counter(name, label_names)
        counter.labels(**labels).inc(value)

    def observe_histogram(self, name: str, labels: dict[str, str], value: float) -> None:
        """Observe a value on a Prometheus histogram."""
        label_names = tuple(sorted(labels.keys()))
        histogram = self._get_histogram(name, label_names)
        histogram.labels(**labels).observe(value)

    def set_gauge(self, name: str, labels: dict[str, str], value: float) -> None:
        """Set a Prometheus gauge to an absolute value."""
        label_names = tuple(sorted(labels.keys()))
        gauge = self._get_gauge(name, label_names)
        gauge.labels(**labels).set(value)

Methods:

inc_counter(name: str, labels: dict[str, str], value: float = 1.0) -> None

Increment a Prometheus counter.

Source code in src/rabbitkit/middleware/metrics.py

def inc_counter(self, name: str, labels: dict[str, str], value: float = 1.0) -> None:
    """Increment a Prometheus counter."""
    label_names = tuple(sorted(labels.keys()))
    counter = self._get_counter(name, label_names)
    counter.labels(**labels).inc(value)

observe_histogram(name: str, labels: dict[str, str], value: float) -> None

Observe a value on a Prometheus histogram.

Source code in src/rabbitkit/middleware/metrics.py

def observe_histogram(self, name: str, labels: dict[str, str], value: float) -> None:
    """Observe a value on a Prometheus histogram."""
    label_names = tuple(sorted(labels.keys()))
    histogram = self._get_histogram(name, label_names)
    histogram.labels(**labels).observe(value)

set_gauge(name: str, labels: dict[str, str], value: float) -> None

Set a Prometheus gauge to an absolute value.

Source code in src/rabbitkit/middleware/metrics.py

def set_gauge(self, name: str, labels: dict[str, str], value: float) -> None:
    """Set a Prometheus gauge to an absolute value."""
    label_names = tuple(sorted(labels.keys()))
    gauge = self._get_gauge(name, label_names)
    gauge.labels(**labels).set(value)

metrics_app() -> Callable[[Any, Any, Any], Awaitable[None]]

Return a minimal ASGI app that exposes /metrics in Prometheus text format.

Requires prometheus_client (lazy-imported on first request). Mount it behind your existing ASGI server (uvicorn, hypercorn) or the dashboard::

from rabbitkit.middleware.metrics import metrics_app
app = metrics_app()
# uvicorn rabbitkit.middleware.metrics:metrics_app  (after binding the factory)

For a stdlib one-liner without an ASGI server, see :func:start_metrics_server.

Source code in src/rabbitkit/middleware/metrics.py

def metrics_app() -> Callable[[Any, Any, Any], Awaitable[None]]:
    """Return a minimal ASGI app that exposes ``/metrics`` in Prometheus text format.

    Requires ``prometheus_client`` (lazy-imported on first request). Mount it
    behind your existing ASGI server (uvicorn, hypercorn) or the dashboard::

        from rabbitkit.middleware.metrics import metrics_app
        app = metrics_app()
        # uvicorn rabbitkit.middleware.metrics:metrics_app  (after binding the factory)

    For a stdlib one-liner without an ASGI server, see :func:`start_metrics_server`.
    """
    try:
        from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
    except ImportError as exc:  # pragma: no cover
        raise ImportError(
            "metrics_app() requires the 'prometheus_client' package. Install it with: pip install prometheus-client"
        ) from exc

    async def app(scope: Any, receive: Any, send: Any) -> None:
        if scope["type"] != "http" or scope.get("path") != "/metrics":
            await send({"type": "http.response.start", "status": 404, "headers": []})
            await send({"type": "http.response.body", "body": b"not found", "more_body": False})
            return
        body = generate_latest()
        headers = [[b"content-type", CONTENT_TYPE_LATEST.encode()]]
        await send({"type": "http.response.start", "status": 200, "headers": headers})
        await send({"type": "http.response.body", "body": body, "more_body": False})

    return app

start_metrics_server(port: int = 8000, host: str = '127.0.0.1') -> None

Start a background HTTP server exposing /metrics on port.

Thin wrapper around prometheus_client.start_http_server. Call once at process startup (e.g. in a RabbitApp.on_startup hook). For k8s, scrape this port with a ServiceMonitor / PodMonitor.

The default host is 127.0.0.1 (loopback only) so the metrics endpoint is not exposed to the network by default. For k8s / multi-host scrapers pass host="0.0.0.0" explicitly and restrict access with a NetworkPolicy (the metrics endpoint is unauthenticated and exposes broker topology/throughput — never expose it publicly without authn in front).

Requires prometheus_client.

Source code in src/rabbitkit/middleware/metrics.py

def start_metrics_server(port: int = 8000, host: str = "127.0.0.1") -> None:
    """Start a background HTTP server exposing ``/metrics`` on ``port``.

    Thin wrapper around ``prometheus_client.start_http_server``. Call once at
    process startup (e.g. in a ``RabbitApp.on_startup`` hook). For k8s, scrape
    this port with a ``ServiceMonitor`` / ``PodMonitor``.

    The default ``host`` is ``127.0.0.1`` (loopback only) so the metrics
    endpoint is not exposed to the network by default. For k8s / multi-host
    scrapers pass ``host="0.0.0.0"`` explicitly and restrict access with a
    NetworkPolicy (the metrics endpoint is unauthenticated and exposes broker
    topology/throughput — never expose it publicly without authn in front).

    Requires ``prometheus_client``.
    """
    try:
        from prometheus_client import start_http_server
    except ImportError as exc:  # pragma: no cover
        raise ImportError(
            "start_metrics_server() requires the 'prometheus_client' package. "
            "Install it with: pip install prometheus-client"
        ) from exc
    start_http_server(port, host)
    logger.info("Prometheus metrics server on http://%s:%d/metrics", host, port)

ExceptionMiddleware

ExceptionMiddleware

Bases: BaseMiddleware

Outermost middleware. Catches exceptions after retry gives up.

Features: - Register exception handlers with fallback values - Terminal exceptions (from retry exhaustion) are re-raised by default - swallow_permanent=True opts in to swallowing permanent/exhausted failures

MANUAL mode restriction: - MAY log the error - MUST NOT auto-publish fallback to result_publisher unless msg.is_settled - MUST NOT settle the message

Source code in src/rabbitkit/middleware/exception.py

class ExceptionMiddleware(BaseMiddleware):
    """Outermost middleware. Catches exceptions after retry gives up.

    Features:
    - Register exception handlers with fallback values
    - Terminal exceptions (from retry exhaustion) are re-raised by default
    - swallow_permanent=True opts in to swallowing permanent/exhausted failures

    MANUAL mode restriction:
    - MAY log the error
    - MUST NOT auto-publish fallback to result_publisher unless msg.is_settled
    - MUST NOT settle the message
    """

    def __init__(self, *, swallow_permanent: bool = False) -> None:
        self._handlers: dict[type[BaseException], Callable[[BaseException], Any]] = {}
        self._swallow_permanent = swallow_permanent

    def add_handler(
        self,
        exc_type: type[BaseException],
        handler: Callable[[BaseException], Any],
    ) -> None:
        """Register an exception handler with a fallback return value."""
        self._handlers[exc_type] = handler

    def consume_scope(
        self,
        call_next: Callable[[RabbitMessage], Any],
        message: RabbitMessage,
    ) -> Any:
        """Wrap handler — catch exceptions, provide fallback values."""
        try:
            return call_next(message)
        except Exception as exc:
            return self._handle_exception(exc, message)

    async def consume_scope_async(
        self,
        call_next: Callable[[RabbitMessage], Awaitable[Any]],
        message: RabbitMessage,
    ) -> Any:
        """Async variant — catch exceptions, provide fallback values."""
        try:
            return await call_next(message)
        except Exception as exc:
            return self._handle_exception(exc, message)

    def _handle_exception(self, exc: Exception, message: RabbitMessage) -> Any:
        """Handle an exception with registered handlers or re-raise.

        Terminal exceptions (tagged with _rabbitkit_terminal=True by RetryMiddleware)
        are only swallowed if swallow_permanent=True.
        """
        is_terminal = getattr(exc, "_rabbitkit_terminal", False)

        if is_terminal and not self._swallow_permanent:
            logger.error(
                "Terminal exception (permanent/exhausted): %s: %s",
                type(exc).__name__,
                exc,
            )
            raise

        # Try registered handlers
        for exc_type, handler in self._handlers.items():
            if isinstance(exc, exc_type):
                logger.warning(
                    "Exception handled by %s handler: %s",
                    exc_type.__name__,
                    exc,
                )
                return handler(exc)

        # No handler found
        if is_terminal and self._swallow_permanent:
            logger.warning(
                "Swallowing terminal exception (swallow_permanent=True): %s: %s",
                type(exc).__name__,
                exc,
            )
            return None

        # Re-raise unhandled non-terminal exceptions
        raise

Methods:

add_handler(exc_type: type[BaseException], handler: Callable[[BaseException], Any]) -> None

Register an exception handler with a fallback return value.

Source code in src/rabbitkit/middleware/exception.py

def add_handler(
    self,
    exc_type: type[BaseException],
    handler: Callable[[BaseException], Any],
) -> None:
    """Register an exception handler with a fallback return value."""
    self._handlers[exc_type] = handler

consume_scope(call_next: Callable[[RabbitMessage], Any], message: RabbitMessage) -> Any

Wrap handler — catch exceptions, provide fallback values.

Source code in src/rabbitkit/middleware/exception.py

def consume_scope(
    self,
    call_next: Callable[[RabbitMessage], Any],
    message: RabbitMessage,
) -> Any:
    """Wrap handler — catch exceptions, provide fallback values."""
    try:
        return call_next(message)
    except Exception as exc:
        return self._handle_exception(exc, message)

consume_scope_async(call_next: Callable[[RabbitMessage], Awaitable[Any]], message: RabbitMessage) -> Any async

Async variant — catch exceptions, provide fallback values.

Source code in src/rabbitkit/middleware/exception.py

async def consume_scope_async(
    self,
    call_next: Callable[[RabbitMessage], Awaitable[Any]],
    message: RabbitMessage,
) -> Any:
    """Async variant — catch exceptions, provide fallback values."""
    try:
        return await call_next(message)
    except Exception as exc:
        return self._handle_exception(exc, message)

OTelTracingMiddleware

Native OpenTelemetry tracing (pip install rabbitkit[otel]) — W3C traceparent propagation over AMQP headers, CONSUMER/PRODUCER spans, no org-internal packages required.

OTelTracingMiddleware

Bases: BaseMiddleware

Wrap handler execution and publishes in standard OpenTelemetry spans.

• Consume: extracts W3C trace context from message headers and starts a CONSUMER-kind span as its child. Handler exceptions are recorded on the span (status ERROR) and re-raised.
• Publish: starts a PRODUCER-kind span and injects the current trace context into a COPY of the envelope's headers (envelopes are frozen).

Span names follow the OTel messaging convention: {destination} receive / {destination} send.

Source code in src/rabbitkit/middleware/otel.py

class OTelTracingMiddleware(BaseMiddleware):
    """Wrap handler execution and publishes in standard OpenTelemetry spans.

    - Consume: extracts W3C trace context from message headers and starts a
      ``CONSUMER``-kind span as its child. Handler exceptions are recorded on
      the span (status ``ERROR``) and re-raised.
    - Publish: starts a ``PRODUCER``-kind span and injects the current trace
      context into a COPY of the envelope's headers (envelopes are frozen).

    Span names follow the OTel messaging convention: ``{destination} receive``
    / ``{destination} send``.
    """

    def __init__(self, service_name: str = "rabbitkit") -> None:
        self._service_name = service_name
        otel = _get_otel()
        if otel is None:
            self._trace: Any = None
            self._propagate: Any = None
            self._tracer: Any = None
            logger.warning(
                "OTelTracingMiddleware(service_name=%r) added but opentelemetry is not "
                "installed -- every consume/publish span will be a silent no-op. "
                "Install with `pip install rabbitkit[otel]`, or remove this middleware.",
                service_name,
            )
        else:
            self._trace, self._propagate = otel
            self._tracer = self._trace.get_tracer("rabbitkit", instrumenting_library_version="1")

    @property
    def is_available(self) -> bool:
        """True if the opentelemetry API is importable."""
        return self._tracer is not None

    # ── Internal helpers ──────────────────────────────────────────────────

    @staticmethod
    def _str_carrier(headers: dict[str, Any]) -> dict[str, str]:
        """AMQP headers may hold non-str values; propagators want str→str."""
        return {k: v for k, v in headers.items() if isinstance(v, str)}

    def _consume_attributes(self, message: RabbitMessage) -> dict[str, str]:
        attrs: dict[str, str] = {
            "messaging.system": "rabbitmq",
            "messaging.operation": "receive",
        }
        if message.routing_key:
            attrs["messaging.rabbitmq.destination.routing_key"] = message.routing_key
        queue = message.headers.get("x-rabbitkit-original-queue", "")
        if queue:
            attrs["messaging.destination.name"] = str(queue)
        if message.message_id:
            attrs["messaging.message.id"] = message.message_id
        if message.correlation_id:
            attrs["messaging.message.conversation_id"] = message.correlation_id
        retry_count = message.headers.get("x-rabbitkit-retry-count")
        if retry_count is not None:
            attrs["messaging.rabbitmq.retry_count"] = str(retry_count)
        return attrs

    def _publish_attributes(self, envelope: MessageEnvelope) -> dict[str, str]:
        attrs: dict[str, str] = {
            "messaging.system": "rabbitmq",
            "messaging.operation": "send",
        }
        if envelope.routing_key:
            attrs["messaging.rabbitmq.destination.routing_key"] = envelope.routing_key
        if envelope.exchange:
            attrs["messaging.destination.name"] = envelope.exchange
        if envelope.message_id:
            attrs["messaging.message.id"] = envelope.message_id
        if envelope.correlation_id:
            attrs["messaging.message.conversation_id"] = envelope.correlation_id
        return attrs

    def _consume_span(self, message: RabbitMessage) -> Any:
        ctx = self._propagate.extract(self._str_carrier(message.headers))
        name = f"{message.headers.get('x-rabbitkit-original-queue', message.routing_key) or 'queue'} receive"
        return self._tracer.start_as_current_span(
            name,
            context=ctx,
            kind=self._trace.SpanKind.CONSUMER,
            attributes=self._consume_attributes(message),
        )

    def _publish_span(self, envelope: MessageEnvelope) -> Any:
        name = f"{envelope.exchange or envelope.routing_key or 'exchange'} send"
        return self._tracer.start_as_current_span(
            name,
            kind=self._trace.SpanKind.PRODUCER,
            attributes=self._publish_attributes(envelope),
        )

    def _record_failure(self, span: Any, exc: BaseException) -> None:
        span.record_exception(exc)
        span.set_status(self._trace.Status(self._trace.StatusCode.ERROR, str(exc)))

    def _envelope_with_context(self, envelope: MessageEnvelope) -> MessageEnvelope:
        """Copy of *envelope* with the CURRENT trace context injected."""
        carrier: dict[str, str] = {}
        self._propagate.inject(carrier)
        if not carrier:
            return envelope
        return replace(envelope, headers={**envelope.headers, **carrier})

    # ── Consume-side hooks ────────────────────────────────────────────────

    def consume_scope(
        self,
        call_next: Callable[[RabbitMessage], Any],
        message: RabbitMessage,
    ) -> Any:
        if self._tracer is None:
            return call_next(message)
        with self._consume_span(message) as span:
            try:
                return call_next(message)
            except BaseException as exc:
                self._record_failure(span, exc)
                raise

    async def consume_scope_async(
        self,
        call_next: Callable[[RabbitMessage], Awaitable[Any]],
        message: RabbitMessage,
    ) -> Any:
        if self._tracer is None:
            return await call_next(message)
        with self._consume_span(message) as span:
            try:
                return await call_next(message)
            except BaseException as exc:
                self._record_failure(span, exc)
                raise

    # ── Publish-side hooks ────────────────────────────────────────────────

    def publish_scope(
        self,
        call_next: Callable[[MessageEnvelope], Any],
        envelope: MessageEnvelope,
    ) -> Any:
        if self._tracer is None:
            return call_next(envelope)
        with self._publish_span(envelope) as span:
            try:
                return call_next(self._envelope_with_context(envelope))
            except BaseException as exc:
                self._record_failure(span, exc)
                raise

    async def publish_scope_async(
        self,
        call_next: Callable[[MessageEnvelope], Awaitable[Any]],
        envelope: MessageEnvelope,
    ) -> Any:
        if self._tracer is None:
            return await call_next(envelope)
        with self._publish_span(envelope) as span:
            try:
                return await call_next(self._envelope_with_context(envelope))
            except BaseException as exc:
                self._record_failure(span, exc)
                raise

Attributes

is_available: bool property

True if the opentelemetry API is importable.