Core Internals

HandlerPipeline

HandlerPipeline

Executes the full message processing pipeline.

on_receive() receives a RabbitMessage (transport builds it first).

The pipeline is responsible for: 1. Ack timing (per AckPolicy) 2. Deserialization (via serializer) 3. Parameter resolution (via DI resolver) 4. Handler invocation 5. Result serialization and publishing 6. Settlement (ack/nack/reject)

Both sync and async variants are provided.

Source code in src/rabbitkit/core/pipeline.py

class HandlerPipeline:
    """Executes the full message processing pipeline.

    on_receive() receives a RabbitMessage (transport builds it first).

    The pipeline is responsible for:
    1. Ack timing (per AckPolicy)
    2. Deserialization (via serializer)
    3. Parameter resolution (via DI resolver)
    4. Handler invocation
    5. Result serialization and publishing
    6. Settlement (ack/nack/reject)

    Both sync and async variants are provided.
    """

    def __init__(
        self,
        serializer: Serializer[Any] | None = None,
        di_resolver: DIResolver | None = None,
        context_repo: ContextRepo | None = None,
        reject_transient_on_redelivery: bool = False,
    ) -> None:
        self._serializer = serializer
        self._di_resolver = di_resolver
        self._context_repo = context_repo
        # M6: opt-in 2-strike cap on transient hot-loops (see ConsumerConfig).
        self._reject_transient_on_redelivery = reject_transient_on_redelivery
        # Per-handler caches so the hot path avoids inspect.signature() per message.
        self._body_type_cache: dict[Any, type | None] = {}
        self._sig_cache: dict[Any, Any] = {}  # handler -> inspect.Signature (fallback resolver)
        # Auto-DI: when no resolver is passed, handlers that use Depends/Header/Path/
        # Context markers get a lazily-created resolver; marker-free handlers keep the
        # fast fallback (so the simple-handler hot path and its behavior are unchanged).
        self._auto_resolver: Any | None = None
        self._needs_di_cache: dict[Any, bool] = {}
        # P3: cache serializer.decode availability per serializer.
        self._has_decode_cache: dict[Any, bool] = {}
        # Aggregate channel-loss settlement warnings (one bounce strands the
        # whole prefetch window; don't emit `prefetch` identical log lines).
        self._settlement_loss_warner = _SettlementLossWarner()
        # P4: precompute parameter binding plan per handler.
        self._binding_plan_cache: dict[Any, list[tuple[str, str]]] = {}
        # P5: cache whether handler is a coroutine function.
        self._is_async_handler_cache: dict[Any, bool] = {}
        # M-P1: cache the composed middleware chain per route — the chain depends
        # only on route.route_middlewares (fixed after registration), so rebuild
        # once per route instead of allocating N closures per message.
        self._consume_chain_cache: dict[int, Callable[[RabbitMessage], Any]] = {}
        self._consume_chain_async_cache: dict[int, Callable[[RabbitMessage], Awaitable[Any]]] = {}
        self._publish_chain_cache: dict[
            int, Callable[[MessageEnvelope, Callable[[MessageEnvelope], PublishOutcome]], Any]
        ] = {}
        self._publish_chain_async_cache: dict[
            int, Callable[[MessageEnvelope, Callable[[MessageEnvelope], Awaitable[PublishOutcome]]], Awaitable[Any]]
        ] = {}
        # C3: broker-level publish middleware chains — keyed by id(middlewares),
        # the SAME list object a broker stores once at construction (see
        # compose_broker_publish_sync/_async). Separate from the route-keyed
        # caches above because broker.publish() is not route-scoped.
        self._broker_publish_chain_cache: dict[
            int, Callable[[MessageEnvelope, Callable[[MessageEnvelope], PublishOutcome]], Any]
        ] = {}
        self._broker_publish_chain_async_cache: dict[
            int, Callable[[MessageEnvelope, Callable[[MessageEnvelope], Awaitable[PublishOutcome]]], Awaitable[Any]]
        ] = {}

    def clear_caches(self) -> None:
        """Drop all per-route caches.

        Clears the four route-keyed middleware-chain caches
        (``_consume_chain_cache``, ``_consume_chain_async_cache``,
        ``_publish_chain_cache``, ``_publish_chain_async_cache``).

        The caches are keyed by ``id(route)`` and are bounded by the number of
        registered routes, which is typically small and stable. They are only
        an eviction concern across reconnect/restart cycles where old
        ``RouteDefinition`` objects are dropped and replaced by new ones; in
        that case the stale entries (keyed by the old ``id``) would otherwise
        linger. Call this on reconnect/restart to reclaim them — the next
        message rebuilds the chain lazily.

        Does NOT clear the broker-level publish chain caches
        (``_broker_publish_chain_cache`` / ``_broker_publish_chain_async_cache``)
        — those are keyed by the broker's ``publish_middlewares`` list, which is
        set once at construction and never mutated in place, so they cannot go
        stale the way route-keyed caches can.
        """
        self._consume_chain_cache.clear()
        self._consume_chain_async_cache.clear()
        self._publish_chain_cache.clear()
        self._publish_chain_async_cache.clear()

    def compose_broker_publish_sync(
        self,
        middlewares: list[Any],
    ) -> Callable[[MessageEnvelope, Callable[[MessageEnvelope], PublishOutcome]], Any]:
        """Compose broker-level ``publish_scope`` middlewares into a reusable chain.

        Unlike :meth:`_compose_publish_sync` (which wraps a route's
        HANDLER-RESULT publishes per Contract 5), this wraps ``broker.publish()``
        itself — the primary producer API — so middleware such as signing or
        tracing actually applies to direct publishes, not just replies/results.

        Cached by ``id(middlewares)``: callers must pass the SAME list object on
        every call (e.g. a broker stores it once at construction) for the cache
        to hit — see :meth:`clear_caches`.
        """
        cached = self._broker_publish_chain_cache.get(id(middlewares))
        if cached is not None:
            return cached

        def leaf(env: MessageEnvelope, fn: Callable[[MessageEnvelope], PublishOutcome]) -> Any:
            return fn(env)

        chain: Callable[[MessageEnvelope, Callable[[MessageEnvelope], PublishOutcome]], Any] = leaf
        for mw in reversed(middlewares):
            nxt = chain

            def wrapped(
                env: MessageEnvelope,
                fn: Callable[[MessageEnvelope], PublishOutcome],
                _mw: Any = mw,
                _nxt: Any = nxt,
            ) -> Any:
                return _mw.publish_scope(lambda e: _nxt(e, fn), env)

            chain = wrapped
        self._broker_publish_chain_cache[id(middlewares)] = chain
        return chain

    def compose_broker_publish_async(
        self,
        middlewares: list[Any],
    ) -> Callable[[MessageEnvelope, Callable[[MessageEnvelope], Awaitable[PublishOutcome]]], Awaitable[Any]]:
        """Async variant of :meth:`compose_broker_publish_sync`."""
        cached = self._broker_publish_chain_async_cache.get(id(middlewares))
        if cached is not None:
            return cached

        async def leaf(
            env: MessageEnvelope,
            fn: Callable[[MessageEnvelope], Awaitable[PublishOutcome]],
        ) -> Any:
            return await fn(env)

        chain: Callable[[MessageEnvelope, Callable[[MessageEnvelope], Awaitable[PublishOutcome]]], Awaitable[Any]] = (
            leaf
        )
        for mw in reversed(middlewares):
            nxt = chain

            async def wrapped(
                env: MessageEnvelope,
                fn: Callable[[MessageEnvelope], Awaitable[PublishOutcome]],
                _mw: Any = mw,
                _nxt: Any = nxt,
            ) -> Any:
                return await _mw.publish_scope_async(lambda e: _nxt(e, fn), env)

            chain = wrapped
        self._broker_publish_chain_async_cache[id(middlewares)] = chain
        return chain

    def process_sync(
        self,
        route: RouteDefinition,
        message: RabbitMessage,
        publish_fn: Callable[[MessageEnvelope], PublishOutcome] | None = None,
    ) -> None:
        """Sync pipeline — calls msg.ack(), handler(), publish().

        Pipeline stages:
        1. ACK_FIRST: ack before handler
        2. Deserialize body → resolve params → call handler
        3. Process result (serialize + publish if applicable)
        4. Settle message (ack/nack/reject per AckPolicy)
        """
        # Filter check — reject before any processing
        if route.filter_fn is not None and not route.filter_fn(message):
            if not message.is_settled:
                message.nack(requeue=False)
            _emit_settlement_metric(route, message)
            return

        # M-P3: only bind contextvars when DEBUG is emitted — avoids per-message
        # dict/token churn on the hot path when structured logging isn't in DEBUG.
        debug = _stdlib_logger.isEnabledFor(logging.DEBUG)
        if debug:
            structlog.contextvars.bind_contextvars(
                message_id=message.message_id,
                routing_key=message.routing_key,
                queue=route.queue.name,
                handler=getattr(route.handler, "__qualname__", repr(route.handler)),
                correlation_id=message.correlation_id,
                retry_count=_retry_count_from_headers(message),
            )

        try:
            strategy = _ACK_STRATEGIES[route.ack_policy]

            # ACK_FIRST: ack before handler runs
            if strategy.acks_first:
                message.ack()

            try:
                # Resolve parameters and call handler (through the middleware chain)
                result = self._run_consume_sync(route, message)

                # Publish result if needed (Contract 5). M7: the
                # REQUEUED_FOR_RETRY sentinel is NOT a handler return value —
                # an inner RetryMiddleware requeued the message and already
                # settled it. Publishing it would serialize the sentinel as a
                # bogus RPC reply/result (once per retry attempt). Skip it.
                if (
                    result is not None
                    and result is not REQUEUED_FOR_RETRY
                    and not self._publish_result_sync(route, message, result, publish_fn)
                ):
                    # Result lost — don't ack. Nack+requeue for redelivery
                    # (handlers are idempotent under at-least-once delivery).
                    if not message.is_settled:
                        message.nack(requeue=True)
                    return

                # Settle on success
                strategy.on_success(message)

            except AckMessage:
                if not message.is_settled:
                    message.ack()

            except NackMessage as exc:
                if not message.is_settled:
                    message.nack(requeue=exc.requeue)

            except RejectMessage as exc:
                if not message.is_settled:
                    message.reject(requeue=exc.requeue)

            except Exception as exc:
                self._handle_sync_exception(route, message, exc)

        finally:
            _emit_settlement_metric(route, message)
            if debug:
                structlog.contextvars.clear_contextvars()

    async def process_async(
        self,
        route: RouteDefinition,
        message: RabbitMessage,
        publish_fn: Callable[[MessageEnvelope], Awaitable[PublishOutcome]] | None = None,
    ) -> None:
        """Async pipeline — calls await msg.ack_async(), await handler(), await publish().

        Same stages as sync, but async.
        """
        # Filter check — reject before any processing
        if route.filter_fn is not None and not route.filter_fn(message):
            if not message.is_settled:
                await message.nack_async(requeue=False)
            _emit_settlement_metric(route, message)
            return

        # M-P3: only bind contextvars when DEBUG is emitted.
        debug = _stdlib_logger.isEnabledFor(logging.DEBUG)
        if debug:
            structlog.contextvars.bind_contextvars(
                message_id=message.message_id,
                routing_key=message.routing_key,
                queue=route.queue.name,
                handler=getattr(route.handler, "__qualname__", repr(route.handler)),
                correlation_id=message.correlation_id,
                retry_count=_retry_count_from_headers(message),
            )

        try:
            strategy = _ACK_STRATEGIES_ASYNC[route.ack_policy]

            # ACK_FIRST: ack before handler runs
            if strategy.acks_first:
                await message.ack_async()

            try:
                # Resolve parameters and call handler (through the middleware chain)
                result = await self._run_consume_async(route, message)

                # Publish result if needed (Contract 5). M7: skip the
                # REQUEUED_FOR_RETRY sentinel (see the sync path above).
                if (
                    result is not None
                    and result is not REQUEUED_FOR_RETRY
                    and not await self._publish_result_async(route, message, result, publish_fn)
                ):
                    # Result lost — don't ack. Nack+requeue for redelivery
                    # (handlers are idempotent under at-least-once delivery).
                    if not message.is_settled:
                        await message.nack_async(requeue=True)
                    return

                # Settle on success
                await strategy.on_success(message)

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

            except NackMessage as exc:
                if not message.is_settled:
                    await message.nack_async(requeue=exc.requeue)

            except RejectMessage as exc:
                if not message.is_settled:
                    await message.reject_async(requeue=exc.requeue)

            except Exception as exc:
                await self._handle_async_exception(route, message, exc)

        finally:
            _emit_settlement_metric(route, message)
            if debug:
                structlog.contextvars.clear_contextvars()

    # ── Internal: middleware composition ─────────────────────────────────

    def _run_consume_sync(self, route: RouteDefinition, message: RabbitMessage) -> Any:
        """Run on_receive hooks, then the consume_scope chain around the handler.

        Middlewares are applied OUTER → INNER: the first item in
        ``route.route_middlewares`` is the outermost wrapper. Each middleware's
        ``consume_scope(call_next, message)`` wraps the next; the innermost
        ``call_next`` deserializes + resolves + invokes the handler.

        H7 — on_receive ordering and exception semantics (READ THIS before
        combining SigningMiddleware/CompressionMiddleware or writing your own
        on_receive-based transform):

        * on_receive hooks run in a FIXED, FLAT pre-pass — entirely BEFORE the
          consume_scope chain is entered. An exception raised here is NEVER
          seen by any middleware's consume_scope (RetryMiddleware included):
          it propagates straight to process_sync's own exception handler,
          which settles the message per the route's AckPolicy using the
          pipeline's default classifier — NOT RetryMiddleware's classifier
          or predicates, and NEVER via RetryMiddleware's delay-queue routing.
          A signing/decompression failure is not retried; it typically
          rejects straight to the DLQ. This is deliberate (an on_receive
          failure means "this delivery is untrustworthy or unreadable," not
          "the handler failed" — retrying doesn't make a bad signature or a
          corrupt payload become valid) and unlikely to change; if you need a
          retry-eligible on_receive-style check, put it in ``consume_scope``
          instead, where it participates in the same chain as everything else.
        * on_receive hooks run in REVERSE registration order — deliberately
          the mirror of publish_scope's OUTER→INNER composition order, so a
          receive-side "undo" (e.g. decompress) always runs relative to a
          publish-side "apply" (e.g. compress) in the mathematically correct
          order regardless of what's paired with what. Concretely: for
          ``middlewares=[A, B]``, publish applies A's transform then B's (A
          outer, B inner); on_receive runs B's hook then A's — the reverse —
          so whichever transform was applied LAST on publish is undone FIRST
          on receive. Before this fix, on_receive ran in the SAME (forward)
          order as publish_scope's apply order, so a receive-side hook always
          ran against a body/metadata state that had already been
          (or not yet been) transformed by the OTHER middleware — never
          matching what that middleware actually needs.
        * This fix does NOT make ``middlewares=[SigningMiddleware,
          CompressionMiddleware]`` order-independent — only ONE relative
          order works: ``middlewares=[CompressionMiddleware,
          SigningMiddleware]`` (compression outer, signing inner). Reason:
          SigningMiddleware's signature covers ``content_encoding`` (H3),
          which is a field CompressionMiddleware's ``publish_scope`` is what
          actually *sets*. If signing runs first (outer), it necessarily
          signs ``content_encoding=None`` (unset at that point) while
          compression sets it to e.g. ``"gzip"`` afterward — the delivered
          message's ``content_encoding`` then never matches what was signed,
          and verification fails unconditionally, independent of the
          on_receive reordering above. Compression outer / signing inner is
          the only order where signing sees the FINAL ``content_encoding``
          it needs to sign correctly. See
          ``tests/unit/core/test_pipeline_middleware.py::TestSigningCompressionComposition``
          for a worked example of both the working and the failing order.
        """
        middlewares = route.route_middlewares
        if not middlewares:
            return self._invoke_handler_sync(route, message)

        for mw in reversed(middlewares):
            mw.on_receive(message)

        # M-P1: cache the composed chain per route — closures are allocated once
        # per route, not per message.
        chain = self._consume_chain_cache.get(id(route))
        if chain is None:

            def call_next(msg: RabbitMessage) -> Any:
                return self._invoke_handler_sync(route, msg)

            for mw in reversed(middlewares):
                nxt = call_next

                def wrapped(msg: RabbitMessage, _mw: Any = mw, _nxt: Any = nxt) -> Any:
                    return _mw.consume_scope(_nxt, msg)

                call_next = wrapped

            chain = call_next
            self._consume_chain_cache[id(route)] = chain

        return chain(message)

    async def _run_consume_async(self, route: RouteDefinition, message: RabbitMessage) -> Any:
        """Async variant of :meth:`_run_consume_sync` — see its docstring
        (H7) for on_receive's ordering and exception-interception semantics,
        which apply identically here."""
        middlewares = route.route_middlewares
        if not middlewares:
            return await self._invoke_handler_async(route, message)

        for mw in reversed(middlewares):
            await mw.on_receive_async(message)

        chain = self._consume_chain_async_cache.get(id(route))
        if chain is None:

            async def call_next(msg: RabbitMessage) -> Any:
                return await self._invoke_handler_async(route, msg)

            for mw in reversed(middlewares):
                nxt = call_next

                async def wrapped(msg: RabbitMessage, _mw: Any = mw, _nxt: Any = nxt) -> Any:
                    return await _mw.consume_scope_async(_nxt, msg)

                call_next = wrapped

            chain = call_next
            self._consume_chain_async_cache[id(route)] = chain

        return await chain(message)

    # ── Internal: handler invocation ─────────────────────────────────────

    def _invoke_handler_sync(self, route: RouteDefinition, message: RabbitMessage) -> Any:
        """Deserialize, resolve params, call handler (sync).

        A ``DependencyScope`` is created whenever the *effective* resolver is
        non-None (explicit OR auto-DI), so generator dependencies are always
        tracked for teardown — including the documented zero-setup marker path.
        The scope wraps BOTH resolution and handler invocation, so a resolution
        failure still tears down any generators opened earlier in the same call.
        """
        # Deserialize body if serializer is available
        body = self._deserialize_body(route, message)

        # Create scope whenever the effective resolver (explicit or auto) is in
        # play — this is the fix for the auto-DI generator-teardown leak.
        resolver = self._effective_resolver(route.handler)
        scope: DependencyScope | None = None
        if resolver is not None and hasattr(resolver, "resolve"):
            scope = DependencyScope()

        # Resolve + invoke under a single try/finally so resolution failures
        # also run generator teardown (generators opened before the failing one).
        try:
            kwargs = self._resolve_params(route, message, body, scope=scope)
            return route.handler(**kwargs)
        finally:
            if scope is not None:
                try:
                    scope.cleanup()
                except Exception as cleanup_exc:
                    logger.error(
                        "DI generator cleanup raised an exception — possible resource leak: %s",
                        cleanup_exc,
                        exc_info=True,
                    )

    async def _invoke_handler_async(self, route: RouteDefinition, message: RabbitMessage) -> Any:
        """Deserialize, resolve params, call handler (async)."""
        body = await self._deserialize_body_async(route, message)

        resolver = self._effective_resolver(route.handler)
        scope: DependencyScope | None = None
        if resolver is not None and hasattr(resolver, "resolve_async"):
            scope = DependencyScope()

        try:
            kwargs = await self._resolve_params_async(route, message, body, scope=scope)
            result = route.handler(**kwargs)
            # P5: cached async detection — avoids per-message hasattr(result, "__await__").
            is_async = self._is_async_handler_cache.get(route.handler)
            if is_async is None:
                import inspect as _inspect
                is_async = _inspect.iscoroutinefunction(route.handler)
                self._is_async_handler_cache[route.handler] = is_async
            if is_async:
                result = await result
            return result
        finally:
            if scope is not None:
                try:
                    await scope.cleanup_async()
                except Exception as cleanup_exc:
                    logger.error(
                        "DI generator cleanup raised an exception — possible resource leak: %s",
                        cleanup_exc,
                        exc_info=True,
                    )

    def _deserialize_body(self, route: RouteDefinition, message: RabbitMessage) -> Any:
        """Deserialize message body using the route's serializer."""
        serializer = route.serializer_override or self._serializer
        if serializer is None:
            return message.body
        # P3: cached hasattr check — avoids a per-message attribute lookup.
        can_decode = self._has_decode_cache.get(serializer)
        if can_decode is None:
            can_decode = hasattr(serializer, "decode")
            self._has_decode_cache[serializer] = can_decode
        if can_decode:
            target_type = self._get_body_type(route)
            if target_type is not None and target_type is not bytes:
                return serializer.decode(message.body, target_type)
        return message.body

    async def _deserialize_body_async(self, route: RouteDefinition, message: RabbitMessage) -> Any:
        """Async body deserialization (M10).

        Identical to :meth:`_deserialize_body`, except a large body is decoded
        in a worker thread via ``asyncio.to_thread`` so a multi-MB
        JSON/msgspec/pydantic parse doesn't block the event loop — which would
        otherwise stall heartbeats, publisher confirms, and every other
        consumer sharing the loop. Small bodies decode inline (the thread hop
        costs more than the parse).
        """
        serializer = route.serializer_override or self._serializer
        if serializer is None:
            return message.body
        can_decode = self._has_decode_cache.get(serializer)
        if can_decode is None:
            can_decode = hasattr(serializer, "decode")
            self._has_decode_cache[serializer] = can_decode
        if can_decode:
            target_type = self._get_body_type(route)
            if target_type is not None and target_type is not bytes:
                if len(message.body) >= _DECODE_OFFLOAD_THRESHOLD_BYTES:
                    import asyncio

                    return await asyncio.to_thread(serializer.decode, message.body, target_type)
                return serializer.decode(message.body, target_type)
        return message.body

    def _get_body_type(self, route: RouteDefinition) -> type | None:
        """Get the body parameter type from the handler signature (cached per handler)."""
        handler = route.handler
        if handler in self._body_type_cache:
            return self._body_type_cache[handler]
        body_type = self._compute_body_type(route)
        self._body_type_cache[handler] = body_type
        return body_type

    def _compute_body_type(self, route: RouteDefinition) -> type | None:
        """Resolve the body parameter type. Returns None if none or if it is bytes.

        Uses ``typing.get_type_hints()`` so that string annotations produced by
        ``from __future__ import annotations`` are resolved to their real types
        before being handed to the serializer.  Falls back to the raw
        ``inspect.Parameter.annotation`` value when ``get_type_hints()`` cannot
        resolve the annotation (e.g. forward references that are not yet defined).
        """
        import inspect
        import typing

        try:
            hints = typing.get_type_hints(route.handler, include_extras=True)
        except Exception:
            hints = {}

        sig = inspect.signature(route.handler)
        for param_name, param in sig.parameters.items():
            # Prefer the resolved hint; fall back to the raw annotation.
            ann = hints.get(param_name, param.annotation)
            if ann is inspect.Parameter.empty:
                continue
            # Skip RabbitMessage type
            if is_rabbit_message_annotation(ann):
                continue
            # Skip Annotated types (DI marker)
            origin = getattr(ann, "__metadata__", None)
            if origin is not None:
                continue
            # First non-special parameter is the body type
            return ann  # type: ignore[no-any-return]
        return None

    def _resolve_params(
        self,
        route: RouteDefinition,
        message: RabbitMessage,
        body: Any,
        scope: Any | None = None,
    ) -> dict[str, Any]:
        """Resolve handler parameters.

        Uses DI resolver if available, otherwise falls back to simple
        body + message injection.
        """
        resolver = self._effective_resolver(route.handler)
        if resolver is not None and hasattr(resolver, "resolve"):
            return resolver.resolve(route.handler, message, self._context_repo, body, scope=scope)  # type: ignore[no-any-return]

        # P4: precomputed binding plan — avoids per-message inspect.signature iteration,
        # is_rabbit_message_annotation string checks, and param.default lookups.
        # The plan is a list of (param_name, action) where action is
        # "message", "body", or "skip" (use default). Computed once per handler.
        plan = self._binding_plan_cache.get(route.handler)
        if plan is None:
            import inspect
            sig = self._sig_cache.get(route.handler)
            if sig is None:
                sig = inspect.signature(route.handler)
                self._sig_cache[route.handler] = sig
            plan = []
            body_injected = False
            for param_name, param in sig.parameters.items():
                # Skip *args and **kwargs — they can't be passed via **kwargs
                if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
                    plan.append((param_name, "skip"))
                    continue
                ann = param.annotation
                if is_rabbit_message_annotation(ann):
                    plan.append((param_name, "message"))
                elif not body_injected:
                    plan.append((param_name, "body"))
                    body_injected = True
                elif param.default is not inspect.Parameter.empty:
                    plan.append((param_name, "skip"))
                else:
                    plan.append((param_name, "message"))
            self._binding_plan_cache[route.handler] = plan

        # Execute the plan — a tight loop with no per-message reflection.
        kwargs: dict[str, Any] = {}
        for param_name, action in plan:
            if action == "message":
                kwargs[param_name] = message
            elif action == "body":
                kwargs[param_name] = body
            # "skip" → use default (omit from kwargs)
        return kwargs

    async def _resolve_params_async(
        self,
        route: RouteDefinition,
        message: RabbitMessage,
        body: Any,
        scope: Any | None = None,
    ) -> dict[str, Any]:
        """Resolve handler parameters (async variant).

        Uses async DI resolver if available, otherwise falls back to sync resolve.
        """
        resolver = self._effective_resolver(route.handler)
        if resolver is not None and hasattr(resolver, "resolve_async"):
            return await resolver.resolve_async(  # type: ignore[no-any-return]
                route.handler, message, self._context_repo, body, scope=scope
            )
        # Fall back to sync resolve (fast path for marker-free handlers)
        return self._resolve_params(route, message, body, scope=scope)

    def _effective_resolver(self, handler: Any) -> Any | None:
        """Return the resolver to use for a handler.

        Explicit `di_resolver` always wins. Otherwise, handlers that use DI markers
        (Depends/Header/Path/Context) get a lazily-created default DIResolver, so the
        documented markers work without the caller wiring one. Marker-free handlers
        return None → the fast body/message fallback, unchanged.
        """
        if self._di_resolver is not None:
            return self._di_resolver
        if not self._handler_needs_di(handler):
            return None
        if self._auto_resolver is None:
            self._auto_resolver = DIResolver()
        return self._auto_resolver

    def _handler_needs_di(self, handler: Any) -> bool:
        """True if any parameter is Annotated with a DI marker. Cached per handler.

        L11: uses the SAME hint-resolution strength as ``DIResolver`` itself
        (``get_type_hints_with_fallback`` — includes the closure-``localns``
        retry) so this detector and the resolver it gates never diverge. A
        weaker, 2-attempt version used to live here; a closure-scoped
        ``Depends(...)`` annotation could resolve fine for ``DIResolver`` but
        still be mis-detected as "no DI needed" by this method, silently
        binding the marked parameter to the message body instead.
        """
        cached = self._needs_di_cache.get(handler)
        if cached is not None:
            return cached

        from rabbitkit.di.context import Context, Header, Path
        from rabbitkit.di.depends import Depends
        from rabbitkit.di.resolver import get_type_hints_with_fallback

        hints = get_type_hints_with_fallback(handler)

        markers = (Depends, Header, Path, Context)
        needs = any(any(isinstance(m, markers) for m in getattr(ann, "__metadata__", ())) for ann in hints.values())
        self._needs_di_cache[handler] = needs
        return needs

    # ── Internal: result publishing ──────────────────────────────────────

    def _compose_publish_sync(
        self,
        route: RouteDefinition,
        publish_fn: Callable[[MessageEnvelope], PublishOutcome],
    ) -> Callable[[MessageEnvelope, Callable[[MessageEnvelope], PublishOutcome]], Any]:
        """Compose this route's ``publish_scope`` middlewares into a reusable chain.

        So a route that carries e.g. a signing/tracing middleware applies it to
        the results it publishes. (Standalone producer publishes via
        ``broker.publish`` are not route-scoped and apply publish middlewares
        manually — see docs.)

        The composed chain is cached per route (keyed by ``id(route)``) — the
        middleware list is fixed after registration, so rebuild once per route
        instead of allocating N closures per message (mirrors the consume cache).

        L-1: ``publish_fn`` is NOT captured in the cached closure — it is threaded
        through at invocation time as the second argument, so a later call with a
        different ``publish_fn`` actually uses the new one (previously the first
        ``publish_fn`` was silently captured and reused forever).
        """
        cached = self._publish_chain_cache.get(id(route))
        if cached is not None:
            return cached

        # The innermost shim defers to ``publish_fn`` supplied at call time.
        def leaf(env: MessageEnvelope, fn: Callable[[MessageEnvelope], PublishOutcome]) -> Any:
            return fn(env)

        chain: Callable[[MessageEnvelope, Callable[[MessageEnvelope], PublishOutcome]], Any] = leaf
        for mw in reversed(route.route_middlewares):
            nxt = chain

            def wrapped(
                env: MessageEnvelope,
                fn: Callable[[MessageEnvelope], PublishOutcome],
                _mw: Any = mw,
                _nxt: Any = nxt,
            ) -> Any:
                # Bind the current publish_fn into the call_next shim so the
                # middleware's publish_scope(call_next, env) signature is unchanged.
                return _mw.publish_scope(lambda e: _nxt(e, fn), env)

            chain = wrapped
        self._publish_chain_cache[id(route)] = chain
        return chain

    def _publish_result_sync(
        self,
        route: RouteDefinition,
        message: RabbitMessage,
        result: Any,
        publish_fn: Callable[[MessageEnvelope], PublishOutcome] | None,
    ) -> bool:
        """Publish handler result (Contract 5 precedence).

        Returns False only when a publish was attempted and failed, so the
        caller can avoid acking a message whose result was lost — the
        caller nacks with ``requeue=True`` instead (see ``process_sync``).

        L1: a nack+requeue here re-runs the handler from scratch on
        redelivery, including any side effects it already performed —
        this is only safe if handlers are idempotent (the same baseline
        assumption at-least-once delivery already requires everywhere
        else; see ``docs/rabbitmq-retry-architecture.md``). If
        ``message.redelivered`` is already ``True``, this is not the
        first time this exact message has hit a failing result publish —
        logged at ERROR (vs. WARNING for a first attempt) so a sustained
        publish outage that would otherwise hot-loop silently is loud and
        alertable via log-based monitoring.
        """
        if publish_fn is None:
            return True

        envelope = self._build_result_envelope(route, message, result)
        if envelope is None:
            return True

        outcome = self._compose_publish_sync(route, publish_fn)(envelope, publish_fn)
        if not outcome.ok:
            _log_result_publish_failure(message, outcome)
            return False
        return True

    def _compose_publish_async(
        self,
        route: RouteDefinition,
        publish_fn: Callable[[MessageEnvelope], Awaitable[PublishOutcome]],
    ) -> Callable[[MessageEnvelope, Callable[[MessageEnvelope], Awaitable[PublishOutcome]]], Awaitable[Any]]:
        """Async variant of :meth:`_compose_publish_sync`.

        The composed chain is cached per route (keyed by ``id(route)``) — see
        :meth:`_compose_publish_sync` for the rationale.

        L-1: ``publish_fn`` is NOT captured in the cached closure — it is threaded
        through at invocation time as the second argument.
        """
        cached = self._publish_chain_async_cache.get(id(route))
        if cached is not None:
            return cached

        async def leaf(
            env: MessageEnvelope,
            fn: Callable[[MessageEnvelope], Awaitable[PublishOutcome]],
        ) -> Any:
            return await fn(env)

        chain: Callable[[MessageEnvelope, Callable[[MessageEnvelope], Awaitable[PublishOutcome]]], Awaitable[Any]] = (
            leaf
        )
        for mw in reversed(route.route_middlewares):
            nxt = chain

            async def wrapped(
                env: MessageEnvelope,
                fn: Callable[[MessageEnvelope], Awaitable[PublishOutcome]],
                _mw: Any = mw,
                _nxt: Any = nxt,
            ) -> Any:
                return await _mw.publish_scope_async(lambda e: _nxt(e, fn), env)

            chain = wrapped
        self._publish_chain_async_cache[id(route)] = chain
        return chain

    async def _publish_result_async(
        self,
        route: RouteDefinition,
        message: RabbitMessage,
        result: Any,
        publish_fn: Callable[[MessageEnvelope], Awaitable[PublishOutcome]] | None,
    ) -> bool:
        """Publish handler result (async, Contract 5 precedence).

        Returns False only when a publish was attempted and failed. See
        :meth:`_publish_result_sync` (L1) for the redelivery-escalation
        rationale — identical here.
        """
        if publish_fn is None:
            return True

        envelope = self._build_result_envelope(route, message, result)
        if envelope is None:
            return True

        outcome = await self._compose_publish_async(route, publish_fn)(envelope, publish_fn)
        if not outcome.ok:
            _log_result_publish_failure(message, outcome)
            return False
        return True

    def _build_result_envelope(
        self,
        route: RouteDefinition,
        message: RabbitMessage,
        result: Any,
    ) -> MessageEnvelope | None:
        """Build MessageEnvelope from handler result.

        Contract 5 precedence:
        1. None return → no publish
        2. reply_to → RPC reply (takes precedence)
        3. result_publisher → publish to configured exchange/routing_key
        4. Both → reply_to wins
        """
        if result is None:
            return None

        user_envelope = result if isinstance(result, MessageEnvelope) else None
        body = self._serialize_result(route, result)

        # Determine destination (Contract 5)
        if message.reply_to:
            # RPC reply takes precedence
            if user_envelope is not None:
                # Preserve user-provided fields (headers, message_id,
                # content_type, priority, expiration, ...); only the
                # precedence-driven destination is merged in.
                return dataclasses.replace(
                    user_envelope,
                    routing_key=message.reply_to,
                    exchange="",
                    correlation_id=message.correlation_id,
                )
            return MessageEnvelope(
                routing_key=message.reply_to,
                body=body,
                exchange="",
                correlation_id=message.correlation_id,
            )

        if route.result_publisher is not None:
            exchange_name = route.result_publisher.resolve_exchange_name()
            if user_envelope is not None:
                # Override only exchange/routing_key; keep user fields.
                return dataclasses.replace(
                    user_envelope,
                    routing_key=route.result_publisher.routing_key,
                    exchange=exchange_name,
                )
            return MessageEnvelope(
                routing_key=route.result_publisher.routing_key,
                body=body,
                exchange=exchange_name,
            )

        if user_envelope is not None:
            logger.warning(
                "handler returned a MessageEnvelope but route has no result_publisher"
                " and message has no reply_to; result dropped"
            )
        return None

    def _serialize_result(self, route: RouteDefinition, result: Any) -> bytes:
        """Serialize handler return value to bytes."""
        if isinstance(result, bytes):
            return result
        if isinstance(result, str):
            return result.encode("utf-8")
        if isinstance(result, MessageEnvelope):
            return result.body

        serializer = route.serializer_override or self._serializer
        if serializer is not None and hasattr(serializer, "encode"):
            return serializer.encode(result)

        # Fallback: JSON encode
        import json

        return json.dumps(result).encode("utf-8")

    # ── Internal: exception handling ─────────────────────────────────────

    def _handle_sync_exception(
        self,
        route: RouteDefinition,
        message: RabbitMessage,
        exc: Exception,
    ) -> None:
        """Handle exception in sync pipeline per AckPolicy (Contract 1)."""
        try:
            self._handle_sync_exception_inner(route, message, exc)
        except Exception as settle_exc:
            # The settlement attempt itself failed because the channel or
            # connection died (SIGTERM drain, broker restart, network cut).
            # Nothing further can be settled on a dead channel and the broker
            # will redeliver the unacked message — warn instead of letting a
            # secondary exception escape as a full ERROR traceback.
            if not _is_channel_gone(settle_exc):
                raise
            self._settlement_loss_warner.warn(type(settle_exc).__name__)

    def _handle_sync_exception_inner(
        self,
        route: RouteDefinition,
        message: RabbitMessage,
        exc: Exception,
    ) -> None:
        if message.is_settled:
            # Already settled (e.g., MANUAL mode handler settled then raised)
            logger.warning("Exception after settlement: %s", exc)
            return

        # RetryMiddleware tags exhausted/permanent failures as terminal. Dead-letter
        # them (reject → source-queue DLX → DLQ) rather than re-classifying: an
        # *exhausted transient* error would otherwise be re-classified TRANSIENT and
        # nack(requeue=True)'d straight back into a hot loop, never reaching the DLQ.
        # MANUAL is excluded — retry is incompatible with MANUAL (handler owns ack).
        if getattr(exc, "_rabbitkit_terminal", False) and route.ack_policy != AckPolicy.MANUAL:
            message.reject(requeue=False)
            return

        # M6: 2-strike cap on the transient hot-loop. A transient error on a
        # message the broker has already redelivered escalates to the DLQ
        # instead of an unbounded nack-requeue. Only AUTO requeues transients;
        # opt-in via ConsumerConfig.reject_transient_on_redelivery.
        if (
            self._reject_transient_on_redelivery
            and message.redelivered
            and route.ack_policy == AckPolicy.AUTO
            and classify_error(exc).severity == ErrorSeverity.TRANSIENT
        ):
            logger.warning(
                "Transient error on an already-redelivered message; rejecting to DLQ "
                "instead of requeuing again (reject_transient_on_redelivery)",
                exc_info=True,
            )
            message.reject(requeue=False)
            return

        _ACK_STRATEGIES[route.ack_policy].on_error(message, exc)

    async def _handle_async_exception(
        self,
        route: RouteDefinition,
        message: RabbitMessage,
        exc: Exception,
    ) -> None:
        """Handle exception in async pipeline per AckPolicy (Contract 1)."""
        try:
            await self._handle_async_exception_inner(route, message, exc)
        except Exception as settle_exc:
            # See _handle_sync_exception: a dead channel means nothing can be
            # settled and the broker redelivers — warn, don't raise.
            if not _is_channel_gone(settle_exc):
                raise
            self._settlement_loss_warner.warn(type(settle_exc).__name__)

    async def _handle_async_exception_inner(
        self,
        route: RouteDefinition,
        message: RabbitMessage,
        exc: Exception,
    ) -> None:
        if message.is_settled:
            logger.warning("Exception after settlement: %s", exc)
            return

        # See _handle_sync_exception: terminal (exhausted/permanent) failures
        # dead-letter directly instead of being re-classified into a hot loop.
        if getattr(exc, "_rabbitkit_terminal", False) and route.ack_policy != AckPolicy.MANUAL:
            await message.reject_async(requeue=False)
            return

        # M6: 2-strike cap on the transient hot-loop (see _handle_sync_exception).
        if (
            self._reject_transient_on_redelivery
            and message.redelivered
            and route.ack_policy == AckPolicy.AUTO
            and classify_error(exc).severity == ErrorSeverity.TRANSIENT
        ):
            logger.warning(
                "Transient error on an already-redelivered message; rejecting to DLQ "
                "instead of requeuing again (reject_transient_on_redelivery)",
                exc_info=True,
            )
            await message.reject_async(requeue=False)
            return

        await _ACK_STRATEGIES_ASYNC[route.ack_policy].on_error(message, exc)

Methods:

clear_caches() -> None

Drop all per-route caches.

Clears the four route-keyed middleware-chain caches (_consume_chain_cache, _consume_chain_async_cache, _publish_chain_cache, _publish_chain_async_cache).

The caches are keyed by id(route) and are bounded by the number of registered routes, which is typically small and stable. They are only an eviction concern across reconnect/restart cycles where old RouteDefinition objects are dropped and replaced by new ones; in that case the stale entries (keyed by the old id) would otherwise linger. Call this on reconnect/restart to reclaim them — the next message rebuilds the chain lazily.

Does NOT clear the broker-level publish chain caches (_broker_publish_chain_cache / _broker_publish_chain_async_cache) — those are keyed by the broker's publish_middlewares list, which is set once at construction and never mutated in place, so they cannot go stale the way route-keyed caches can.

Source code in src/rabbitkit/core/pipeline.py

def clear_caches(self) -> None:
    """Drop all per-route caches.

    Clears the four route-keyed middleware-chain caches
    (``_consume_chain_cache``, ``_consume_chain_async_cache``,
    ``_publish_chain_cache``, ``_publish_chain_async_cache``).

    The caches are keyed by ``id(route)`` and are bounded by the number of
    registered routes, which is typically small and stable. They are only
    an eviction concern across reconnect/restart cycles where old
    ``RouteDefinition`` objects are dropped and replaced by new ones; in
    that case the stale entries (keyed by the old ``id``) would otherwise
    linger. Call this on reconnect/restart to reclaim them — the next
    message rebuilds the chain lazily.

    Does NOT clear the broker-level publish chain caches
    (``_broker_publish_chain_cache`` / ``_broker_publish_chain_async_cache``)
    — those are keyed by the broker's ``publish_middlewares`` list, which is
    set once at construction and never mutated in place, so they cannot go
    stale the way route-keyed caches can.
    """
    self._consume_chain_cache.clear()
    self._consume_chain_async_cache.clear()
    self._publish_chain_cache.clear()
    self._publish_chain_async_cache.clear()

compose_broker_publish_sync(middlewares: list[Any]) -> Callable[[MessageEnvelope, Callable[[MessageEnvelope], PublishOutcome]], Any]

Compose broker-level publish_scope middlewares into a reusable chain.

Unlike :meth:_compose_publish_sync (which wraps a route's HANDLER-RESULT publishes per Contract 5), this wraps broker.publish() itself — the primary producer API — so middleware such as signing or tracing actually applies to direct publishes, not just replies/results.

Cached by id(middlewares): callers must pass the SAME list object on every call (e.g. a broker stores it once at construction) for the cache to hit — see :meth:clear_caches.

Source code in src/rabbitkit/core/pipeline.py

def compose_broker_publish_sync(
    self,
    middlewares: list[Any],
) -> Callable[[MessageEnvelope, Callable[[MessageEnvelope], PublishOutcome]], Any]:
    """Compose broker-level ``publish_scope`` middlewares into a reusable chain.

    Unlike :meth:`_compose_publish_sync` (which wraps a route's
    HANDLER-RESULT publishes per Contract 5), this wraps ``broker.publish()``
    itself — the primary producer API — so middleware such as signing or
    tracing actually applies to direct publishes, not just replies/results.

    Cached by ``id(middlewares)``: callers must pass the SAME list object on
    every call (e.g. a broker stores it once at construction) for the cache
    to hit — see :meth:`clear_caches`.
    """
    cached = self._broker_publish_chain_cache.get(id(middlewares))
    if cached is not None:
        return cached

    def leaf(env: MessageEnvelope, fn: Callable[[MessageEnvelope], PublishOutcome]) -> Any:
        return fn(env)

    chain: Callable[[MessageEnvelope, Callable[[MessageEnvelope], PublishOutcome]], Any] = leaf
    for mw in reversed(middlewares):
        nxt = chain

        def wrapped(
            env: MessageEnvelope,
            fn: Callable[[MessageEnvelope], PublishOutcome],
            _mw: Any = mw,
            _nxt: Any = nxt,
        ) -> Any:
            return _mw.publish_scope(lambda e: _nxt(e, fn), env)

        chain = wrapped
    self._broker_publish_chain_cache[id(middlewares)] = chain
    return chain

compose_broker_publish_async(middlewares: list[Any]) -> Callable[[MessageEnvelope, Callable[[MessageEnvelope], Awaitable[PublishOutcome]]], Awaitable[Any]]

Async variant of :meth:compose_broker_publish_sync.

Source code in src/rabbitkit/core/pipeline.py

def compose_broker_publish_async(
    self,
    middlewares: list[Any],
) -> Callable[[MessageEnvelope, Callable[[MessageEnvelope], Awaitable[PublishOutcome]]], Awaitable[Any]]:
    """Async variant of :meth:`compose_broker_publish_sync`."""
    cached = self._broker_publish_chain_async_cache.get(id(middlewares))
    if cached is not None:
        return cached

    async def leaf(
        env: MessageEnvelope,
        fn: Callable[[MessageEnvelope], Awaitable[PublishOutcome]],
    ) -> Any:
        return await fn(env)

    chain: Callable[[MessageEnvelope, Callable[[MessageEnvelope], Awaitable[PublishOutcome]]], Awaitable[Any]] = (
        leaf
    )
    for mw in reversed(middlewares):
        nxt = chain

        async def wrapped(
            env: MessageEnvelope,
            fn: Callable[[MessageEnvelope], Awaitable[PublishOutcome]],
            _mw: Any = mw,
            _nxt: Any = nxt,
        ) -> Any:
            return await _mw.publish_scope_async(lambda e: _nxt(e, fn), env)

        chain = wrapped
    self._broker_publish_chain_async_cache[id(middlewares)] = chain
    return chain

process_sync(route: RouteDefinition, message: RabbitMessage, publish_fn: Callable[[MessageEnvelope], PublishOutcome] | None = None) -> None

Sync pipeline — calls msg.ack(), handler(), publish().

Pipeline stages: 1. ACK_FIRST: ack before handler 2. Deserialize body → resolve params → call handler 3. Process result (serialize + publish if applicable) 4. Settle message (ack/nack/reject per AckPolicy)

Source code in src/rabbitkit/core/pipeline.py

def process_sync(
    self,
    route: RouteDefinition,
    message: RabbitMessage,
    publish_fn: Callable[[MessageEnvelope], PublishOutcome] | None = None,
) -> None:
    """Sync pipeline — calls msg.ack(), handler(), publish().

    Pipeline stages:
    1. ACK_FIRST: ack before handler
    2. Deserialize body → resolve params → call handler
    3. Process result (serialize + publish if applicable)
    4. Settle message (ack/nack/reject per AckPolicy)
    """
    # Filter check — reject before any processing
    if route.filter_fn is not None and not route.filter_fn(message):
        if not message.is_settled:
            message.nack(requeue=False)
        _emit_settlement_metric(route, message)
        return

    # M-P3: only bind contextvars when DEBUG is emitted — avoids per-message
    # dict/token churn on the hot path when structured logging isn't in DEBUG.
    debug = _stdlib_logger.isEnabledFor(logging.DEBUG)
    if debug:
        structlog.contextvars.bind_contextvars(
            message_id=message.message_id,
            routing_key=message.routing_key,
            queue=route.queue.name,
            handler=getattr(route.handler, "__qualname__", repr(route.handler)),
            correlation_id=message.correlation_id,
            retry_count=_retry_count_from_headers(message),
        )

    try:
        strategy = _ACK_STRATEGIES[route.ack_policy]

        # ACK_FIRST: ack before handler runs
        if strategy.acks_first:
            message.ack()

        try:
            # Resolve parameters and call handler (through the middleware chain)
            result = self._run_consume_sync(route, message)

            # Publish result if needed (Contract 5). M7: the
            # REQUEUED_FOR_RETRY sentinel is NOT a handler return value —
            # an inner RetryMiddleware requeued the message and already
            # settled it. Publishing it would serialize the sentinel as a
            # bogus RPC reply/result (once per retry attempt). Skip it.
            if (
                result is not None
                and result is not REQUEUED_FOR_RETRY
                and not self._publish_result_sync(route, message, result, publish_fn)
            ):
                # Result lost — don't ack. Nack+requeue for redelivery
                # (handlers are idempotent under at-least-once delivery).
                if not message.is_settled:
                    message.nack(requeue=True)
                return

            # Settle on success
            strategy.on_success(message)

        except AckMessage:
            if not message.is_settled:
                message.ack()

        except NackMessage as exc:
            if not message.is_settled:
                message.nack(requeue=exc.requeue)

        except RejectMessage as exc:
            if not message.is_settled:
                message.reject(requeue=exc.requeue)

        except Exception as exc:
            self._handle_sync_exception(route, message, exc)

    finally:
        _emit_settlement_metric(route, message)
        if debug:
            structlog.contextvars.clear_contextvars()

process_async(route: RouteDefinition, message: RabbitMessage, publish_fn: Callable[[MessageEnvelope], Awaitable[PublishOutcome]] | None = None) -> None async

Async pipeline — calls await msg.ack_async(), await handler(), await publish().

Same stages as sync, but async.

Source code in src/rabbitkit/core/pipeline.py

async def process_async(
    self,
    route: RouteDefinition,
    message: RabbitMessage,
    publish_fn: Callable[[MessageEnvelope], Awaitable[PublishOutcome]] | None = None,
) -> None:
    """Async pipeline — calls await msg.ack_async(), await handler(), await publish().

    Same stages as sync, but async.
    """
    # Filter check — reject before any processing
    if route.filter_fn is not None and not route.filter_fn(message):
        if not message.is_settled:
            await message.nack_async(requeue=False)
        _emit_settlement_metric(route, message)
        return

    # M-P3: only bind contextvars when DEBUG is emitted.
    debug = _stdlib_logger.isEnabledFor(logging.DEBUG)
    if debug:
        structlog.contextvars.bind_contextvars(
            message_id=message.message_id,
            routing_key=message.routing_key,
            queue=route.queue.name,
            handler=getattr(route.handler, "__qualname__", repr(route.handler)),
            correlation_id=message.correlation_id,
            retry_count=_retry_count_from_headers(message),
        )

    try:
        strategy = _ACK_STRATEGIES_ASYNC[route.ack_policy]

        # ACK_FIRST: ack before handler runs
        if strategy.acks_first:
            await message.ack_async()

        try:
            # Resolve parameters and call handler (through the middleware chain)
            result = await self._run_consume_async(route, message)

            # Publish result if needed (Contract 5). M7: skip the
            # REQUEUED_FOR_RETRY sentinel (see the sync path above).
            if (
                result is not None
                and result is not REQUEUED_FOR_RETRY
                and not await self._publish_result_async(route, message, result, publish_fn)
            ):
                # Result lost — don't ack. Nack+requeue for redelivery
                # (handlers are idempotent under at-least-once delivery).
                if not message.is_settled:
                    await message.nack_async(requeue=True)
                return

            # Settle on success
            await strategy.on_success(message)

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

        except NackMessage as exc:
            if not message.is_settled:
                await message.nack_async(requeue=exc.requeue)

        except RejectMessage as exc:
            if not message.is_settled:
                await message.reject_async(requeue=exc.requeue)

        except Exception as exc:
            await self._handle_async_exception(route, message, exc)

    finally:
        _emit_settlement_metric(route, message)
        if debug:
            structlog.contextvars.clear_contextvars()

SubscriberRegistry

SubscriberRegistry

Stores all @subscriber/@publisher registrations.

Used by RabbitApp and RabbitRouter to collect route definitions which are later wired by the broker.

Source code in src/rabbitkit/core/registry.py

class SubscriberRegistry:
    """Stores all @subscriber/@publisher registrations.

    Used by RabbitApp and RabbitRouter to collect route definitions
    which are later wired by the broker.
    """

    def __init__(self, broker_retry: RetryConfig | None = None) -> None:
        self._routes: list[RouteDefinition] = []
        self._queue_names: set[str] = set()
        self._pending_publishers: dict[int, ResultPublisher] = {}  # handler id → ResultPublisher
        self._broker_retry = broker_retry

    @property
    def routes(self) -> list[RouteDefinition]:
        """Return all registered routes."""
        return list(self._routes)

    def subscriber(
        self,
        queue: RabbitQueue | str,
        exchange: RabbitExchange | str | None = None,
        routing_key: str = "",
        ack_policy: AckPolicy = AckPolicy.AUTO,
        middlewares: list[BaseMiddleware] | None = None,
        serializer: Serializer[Any] | None = None,
        retry: RetryConfig | RetryDisabled | None = None,
        tags: frozenset[str] | set[str] | None = None,
        description: str = "",
        name: str | None = None,
        prefetch_count: int | None = None,
        filter_fn: Callable[[RabbitMessage], bool] | None = None,
        reject_without_dlx: str | None = None,
    ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
        """Decorator to register a message handler for a queue.

        Args:
            queue: Queue to consume from (str auto-creates RabbitQueue).
            exchange: Exchange to bind to (str auto-creates RabbitExchange).
            routing_key: Routing key for binding.
            ack_policy: Acknowledgment policy for this route.
            middlewares: Route-specific middleware list.
            serializer: Route-specific serializer override.
            retry: Per-route retry config (None=inherit, RETRY_DISABLED=opt-out).
            tags: Route tags for filtering/grouping.
            description: Human-readable route description.
            name: Explicit route name (auto-generated if None).
            prefetch_count: Per-route prefetch override (None=use global).
            reject_without_dlx: Per-route override of
                ``SafetyConfig.reject_without_dlx`` — 'auto_provision',
                'error', or 'discard' (None=inherit broker default).
        """
        # Normalize queue
        if isinstance(queue, str):
            queue = RabbitQueue(name=queue)

        # Apply routing_key to the queue if not already set. The queue is frozen,
        # so build a copy rather than mutating the caller's object (which may be
        # shared across brokers/routers — see topology freeze + include-router).
        if routing_key and not queue.routing_key:
            from dataclasses import replace as _replace

            queue = _replace(queue, routing_key=routing_key)

        # Normalize exchange
        if isinstance(exchange, str):
            exchange = RabbitExchange(name=exchange)

        # Normalize tags
        if tags is not None and not isinstance(tags, frozenset):
            tags = frozenset(tags)
        elif tags is None:
            tags = frozenset()

        def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
            # Validate handler signature at registration time (fail fast).
            # Covers *args/**kwargs and multiple body-like parameters (Contract 4).
            from rabbitkit.di.resolver import DIResolver

            DIResolver().validate_handler(func)

            # Check duplicate queue
            if queue.name in self._queue_names:
                raise DuplicateRouteError(
                    f"Queue '{queue.name}' already has a registered handler. "
                    "rabbitkit enforces one handler per queue. "
                    "Use multiple routing keys on the same queue for fan-in."
                )

            # Auto-generate name
            route_name = name or f"{queue.name}:{func.__qualname__}"

            # Check for pending @publisher
            result_publisher = self._pending_publishers.pop(id(func), None)

            # Create route
            route = RouteDefinition(
                name=route_name,
                queue=queue,
                exchange=exchange,
                handler=func,
                ack_policy=ack_policy,
                route_middlewares=middlewares or [],
                result_publisher=result_publisher,
                serializer_override=serializer,
                retry_override=retry,
                prefetch_count=prefetch_count,
                tags=tags,
                description=description,
                filter_fn=filter_fn,
                reject_without_dlx=reject_without_dlx,
            )

            # Validate at registration time (fail fast)
            route.validate(self._broker_retry)

            self._routes.append(route)
            self._queue_names.add(queue.name)

            # M-C6: detect dead-letter-exchange cycles across the growing route graph.
            self.validate_dlx_graph()

            return func

        return decorator

    def publisher(
        self,
        exchange: RabbitExchange | str | None = None,
        routing_key: str = "",
    ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
        """Decorator to configure result publishing for a handler.

        Must be applied BEFORE @subscriber on the same handler.
        If applied without @subscriber, the publisher info is stored
        and applied when @subscriber is later applied.

        Args:
            exchange: Target exchange for result publishing.
            routing_key: Routing key for result publishing.
        """
        # Normalize exchange
        if isinstance(exchange, str):
            exchange = RabbitExchange(name=exchange)

        result_pub = ResultPublisher(exchange=exchange, routing_key=routing_key)

        def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
            # Store pending publisher for this handler
            self._pending_publishers[id(func)] = result_pub
            return func

        return decorator

    def include_router(self, router: RabbitRouter, prefix: str = "") -> None:
        """Include routes from a RabbitRouter.

        Applies the router's prefix to routing keys and merges
        router-level defaults (exchange, middlewares, serializer, tags).
        """
        # Duck-typed check retained despite the static RabbitRouter annotation:
        # this is a public entry point Python won't enforce the type on at
        # runtime, and a clear TypeError beats an opaque AttributeError below.
        if not hasattr(router, "_registry"):
            raise TypeError(f"Expected a RabbitRouter, got {type(router).__name__}")

        for route in router._registry.routes:
            # Apply prefix to routing key WITHOUT mutating the included router's
            # RabbitQueue/RouteDefinition (frozen + shared-object safety). Build
            # fresh copies so re-including the same router under a different
            # prefix doesn't double-prefix or cross-contaminate.
            from dataclasses import replace as _replace

            if prefix:
                effective_rk = f"{prefix}.{route.queue.routing_key}" if route.queue.routing_key else prefix
                new_queue = _replace(route.queue, routing_key=effective_rk)
                route = _replace(route, queue=new_queue)

            # Check duplicate
            if route.queue.name in self._queue_names:
                raise DuplicateRouteError(
                    f"Queue '{route.queue.name}' already has a registered handler. Duplicate from included router."
                )

            # Validate with broker retry context
            route.validate(self._broker_retry)

            self._routes.append(route)
            self._queue_names.add(route.queue.name)

            # M-C6: detect DLX cycles after including router routes.
            self.validate_dlx_graph()

    def set_broker_retry(self, retry: RetryConfig | None) -> None:
        """Update broker retry default. Re-validates all existing routes."""
        self._broker_retry = retry
        for route in self._routes:
            route.validate(self._broker_retry)

    def validate_dlx_graph(self) -> None:
        """Detect dead-letter-exchange cycles across registered routes.

        A queue's ``dead_letter_exchange`` points at an exchange; if that exchange
        is the bind target of another route whose queue dead-letters back to an
        exchange reachable from the first, messages loop forever (until TTL/GC).
        This builds a queue→queue graph via DLX→exchange→bound-queue and rejects
        any cycle with :class:`ConfigurationError`. Best-effort: only exchanges
        named as a route's ``exchange`` are resolvable to a queue; unknown DLX
        targets (external exchanges) are treated as sinks.
        """
        from rabbitkit.core.errors import ConfigurationError as _CfgErr

        # exchange name → list of queue names bound to it via a registered route
        exchange_to_queues: dict[str, list[str]] = {}
        for r in self._routes:
            if r.exchange is not None:
                exchange_to_queues.setdefault(r.exchange.name, []).append(r.queue.name)

        # queue → next queues reachable via its DLX (through the exchange graph)
        adj: dict[str, list[str]] = {}
        for r in self._routes:
            dlx = r.queue.dead_letter_exchange
            if not dlx:
                continue
            nxts = exchange_to_queues.get(dlx, [])  # unknown DLX → sink (no edges)
            adj.setdefault(r.queue.name, []).extend(nxts)

        # DFS cycle detection
        white, gray, black = 0, 1, 2
        color: dict[str, int] = {}

        def dfs(node: str, path: list[str]) -> None:
            color[node] = gray
            for nxt in adj.get(node, []):
                if color.get(nxt, white) == gray:
                    cycle = " -> ".join([*path, node, nxt])
                    raise _CfgErr(
                        f"Dead-letter-exchange cycle detected: {cycle}. "
                        "Messages would loop forever. Break the cycle (e.g. let "
                        "retry own DLQ topology, or point the final DLX at a sink "
                        "exchange with no bound queue)."
                    )
                if color.get(nxt, white) == white:
                    dfs(nxt, [*path, node])
            color[node] = black

        for q in adj:
            if color.get(q, white) == white:
                dfs(q, [])

Attributes

routes: list[RouteDefinition] property

Return all registered routes.

Methods:

subscriber(queue: RabbitQueue | str, exchange: RabbitExchange | str | None = None, routing_key: str = '', ack_policy: AckPolicy = AckPolicy.AUTO, middlewares: list[BaseMiddleware] | None = None, serializer: Serializer[Any] | None = None, retry: RetryConfig | RetryDisabled | None = None, tags: frozenset[str] | set[str] | None = None, description: str = '', name: str | None = None, prefetch_count: int | None = None, filter_fn: Callable[[RabbitMessage], bool] | None = None, reject_without_dlx: str | None = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator to register a message handler for a queue.

Parameters:

Name	Type	Description	Default
queue	RabbitQueue | str	Queue to consume from (str auto-creates RabbitQueue).	required
exchange	RabbitExchange | str | None	Exchange to bind to (str auto-creates RabbitExchange).	None
routing_key	str	Routing key for binding.	''
ack_policy	AckPolicy	Acknowledgment policy for this route.	AUTO
middlewares	list[BaseMiddleware] | None	Route-specific middleware list.	None
serializer	Serializer[Any] | None	Route-specific serializer override.	None
retry	RetryConfig | RetryDisabled | None	Per-route retry config (None=inherit, RETRY_DISABLED=opt-out).	None
tags	frozenset[str] | set[str] | None	Route tags for filtering/grouping.	None
description	str	Human-readable route description.	''
name	str | None	Explicit route name (auto-generated if None).	None
prefetch_count	int | None	Per-route prefetch override (None=use global).	None
reject_without_dlx	str | None	Per-route override of SafetyConfig.reject_without_dlx — 'auto_provision', 'error', or 'discard' (None=inherit broker default).	None

Source code in src/rabbitkit/core/registry.py

def subscriber(
    self,
    queue: RabbitQueue | str,
    exchange: RabbitExchange | str | None = None,
    routing_key: str = "",
    ack_policy: AckPolicy = AckPolicy.AUTO,
    middlewares: list[BaseMiddleware] | None = None,
    serializer: Serializer[Any] | None = None,
    retry: RetryConfig | RetryDisabled | None = None,
    tags: frozenset[str] | set[str] | None = None,
    description: str = "",
    name: str | None = None,
    prefetch_count: int | None = None,
    filter_fn: Callable[[RabbitMessage], bool] | None = None,
    reject_without_dlx: str | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator to register a message handler for a queue.

    Args:
        queue: Queue to consume from (str auto-creates RabbitQueue).
        exchange: Exchange to bind to (str auto-creates RabbitExchange).
        routing_key: Routing key for binding.
        ack_policy: Acknowledgment policy for this route.
        middlewares: Route-specific middleware list.
        serializer: Route-specific serializer override.
        retry: Per-route retry config (None=inherit, RETRY_DISABLED=opt-out).
        tags: Route tags for filtering/grouping.
        description: Human-readable route description.
        name: Explicit route name (auto-generated if None).
        prefetch_count: Per-route prefetch override (None=use global).
        reject_without_dlx: Per-route override of
            ``SafetyConfig.reject_without_dlx`` — 'auto_provision',
            'error', or 'discard' (None=inherit broker default).
    """
    # Normalize queue
    if isinstance(queue, str):
        queue = RabbitQueue(name=queue)

    # Apply routing_key to the queue if not already set. The queue is frozen,
    # so build a copy rather than mutating the caller's object (which may be
    # shared across brokers/routers — see topology freeze + include-router).
    if routing_key and not queue.routing_key:
        from dataclasses import replace as _replace

        queue = _replace(queue, routing_key=routing_key)

    # Normalize exchange
    if isinstance(exchange, str):
        exchange = RabbitExchange(name=exchange)

    # Normalize tags
    if tags is not None and not isinstance(tags, frozenset):
        tags = frozenset(tags)
    elif tags is None:
        tags = frozenset()

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        # Validate handler signature at registration time (fail fast).
        # Covers *args/**kwargs and multiple body-like parameters (Contract 4).
        from rabbitkit.di.resolver import DIResolver

        DIResolver().validate_handler(func)

        # Check duplicate queue
        if queue.name in self._queue_names:
            raise DuplicateRouteError(
                f"Queue '{queue.name}' already has a registered handler. "
                "rabbitkit enforces one handler per queue. "
                "Use multiple routing keys on the same queue for fan-in."
            )

        # Auto-generate name
        route_name = name or f"{queue.name}:{func.__qualname__}"

        # Check for pending @publisher
        result_publisher = self._pending_publishers.pop(id(func), None)

        # Create route
        route = RouteDefinition(
            name=route_name,
            queue=queue,
            exchange=exchange,
            handler=func,
            ack_policy=ack_policy,
            route_middlewares=middlewares or [],
            result_publisher=result_publisher,
            serializer_override=serializer,
            retry_override=retry,
            prefetch_count=prefetch_count,
            tags=tags,
            description=description,
            filter_fn=filter_fn,
            reject_without_dlx=reject_without_dlx,
        )

        # Validate at registration time (fail fast)
        route.validate(self._broker_retry)

        self._routes.append(route)
        self._queue_names.add(queue.name)

        # M-C6: detect dead-letter-exchange cycles across the growing route graph.
        self.validate_dlx_graph()

        return func

    return decorator

publisher(exchange: RabbitExchange | str | None = None, routing_key: str = '') -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator to configure result publishing for a handler.

Must be applied BEFORE @subscriber on the same handler. If applied without @subscriber, the publisher info is stored and applied when @subscriber is later applied.

Parameters:

Name	Type	Description	Default
exchange	RabbitExchange | str | None	Target exchange for result publishing.	None
routing_key	str	Routing key for result publishing.	''

Source code in src/rabbitkit/core/registry.py

def publisher(
    self,
    exchange: RabbitExchange | str | None = None,
    routing_key: str = "",
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator to configure result publishing for a handler.

    Must be applied BEFORE @subscriber on the same handler.
    If applied without @subscriber, the publisher info is stored
    and applied when @subscriber is later applied.

    Args:
        exchange: Target exchange for result publishing.
        routing_key: Routing key for result publishing.
    """
    # Normalize exchange
    if isinstance(exchange, str):
        exchange = RabbitExchange(name=exchange)

    result_pub = ResultPublisher(exchange=exchange, routing_key=routing_key)

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        # Store pending publisher for this handler
        self._pending_publishers[id(func)] = result_pub
        return func

    return decorator

include_router(router: RabbitRouter, prefix: str = '') -> None

Include routes from a RabbitRouter.

Applies the router's prefix to routing keys and merges router-level defaults (exchange, middlewares, serializer, tags).

Source code in src/rabbitkit/core/registry.py

def include_router(self, router: RabbitRouter, prefix: str = "") -> None:
    """Include routes from a RabbitRouter.

    Applies the router's prefix to routing keys and merges
    router-level defaults (exchange, middlewares, serializer, tags).
    """
    # Duck-typed check retained despite the static RabbitRouter annotation:
    # this is a public entry point Python won't enforce the type on at
    # runtime, and a clear TypeError beats an opaque AttributeError below.
    if not hasattr(router, "_registry"):
        raise TypeError(f"Expected a RabbitRouter, got {type(router).__name__}")

    for route in router._registry.routes:
        # Apply prefix to routing key WITHOUT mutating the included router's
        # RabbitQueue/RouteDefinition (frozen + shared-object safety). Build
        # fresh copies so re-including the same router under a different
        # prefix doesn't double-prefix or cross-contaminate.
        from dataclasses import replace as _replace

        if prefix:
            effective_rk = f"{prefix}.{route.queue.routing_key}" if route.queue.routing_key else prefix
            new_queue = _replace(route.queue, routing_key=effective_rk)
            route = _replace(route, queue=new_queue)

        # Check duplicate
        if route.queue.name in self._queue_names:
            raise DuplicateRouteError(
                f"Queue '{route.queue.name}' already has a registered handler. Duplicate from included router."
            )

        # Validate with broker retry context
        route.validate(self._broker_retry)

        self._routes.append(route)
        self._queue_names.add(route.queue.name)

        # M-C6: detect DLX cycles after including router routes.
        self.validate_dlx_graph()

set_broker_retry(retry: RetryConfig | None) -> None

Update broker retry default. Re-validates all existing routes.

Source code in src/rabbitkit/core/registry.py

def set_broker_retry(self, retry: RetryConfig | None) -> None:
    """Update broker retry default. Re-validates all existing routes."""
    self._broker_retry = retry
    for route in self._routes:
        route.validate(self._broker_retry)

validate_dlx_graph() -> None

Detect dead-letter-exchange cycles across registered routes.

A queue's dead_letter_exchange points at an exchange; if that exchange is the bind target of another route whose queue dead-letters back to an exchange reachable from the first, messages loop forever (until TTL/GC). This builds a queue→queue graph via DLX→exchange→bound-queue and rejects any cycle with :class:ConfigurationError. Best-effort: only exchanges named as a route's exchange are resolvable to a queue; unknown DLX targets (external exchanges) are treated as sinks.

Source code in src/rabbitkit/core/registry.py

def validate_dlx_graph(self) -> None:
    """Detect dead-letter-exchange cycles across registered routes.

    A queue's ``dead_letter_exchange`` points at an exchange; if that exchange
    is the bind target of another route whose queue dead-letters back to an
    exchange reachable from the first, messages loop forever (until TTL/GC).
    This builds a queue→queue graph via DLX→exchange→bound-queue and rejects
    any cycle with :class:`ConfigurationError`. Best-effort: only exchanges
    named as a route's ``exchange`` are resolvable to a queue; unknown DLX
    targets (external exchanges) are treated as sinks.
    """
    from rabbitkit.core.errors import ConfigurationError as _CfgErr

    # exchange name → list of queue names bound to it via a registered route
    exchange_to_queues: dict[str, list[str]] = {}
    for r in self._routes:
        if r.exchange is not None:
            exchange_to_queues.setdefault(r.exchange.name, []).append(r.queue.name)

    # queue → next queues reachable via its DLX (through the exchange graph)
    adj: dict[str, list[str]] = {}
    for r in self._routes:
        dlx = r.queue.dead_letter_exchange
        if not dlx:
            continue
        nxts = exchange_to_queues.get(dlx, [])  # unknown DLX → sink (no edges)
        adj.setdefault(r.queue.name, []).extend(nxts)

    # DFS cycle detection
    white, gray, black = 0, 1, 2
    color: dict[str, int] = {}

    def dfs(node: str, path: list[str]) -> None:
        color[node] = gray
        for nxt in adj.get(node, []):
            if color.get(nxt, white) == gray:
                cycle = " -> ".join([*path, node, nxt])
                raise _CfgErr(
                    f"Dead-letter-exchange cycle detected: {cycle}. "
                    "Messages would loop forever. Break the cycle (e.g. let "
                    "retry own DLQ topology, or point the final DLX at a sink "
                    "exchange with no bound queue)."
                )
            if color.get(nxt, white) == white:
                dfs(nxt, [*path, node])
        color[node] = black

    for q in adj:
        if color.get(q, white) == white:
            dfs(q, [])

DuplicateRouteError

Bases: Exception

Raised when two routes target the same queue.

Source code in src/rabbitkit/core/registry.py

class DuplicateRouteError(Exception):
    """Raised when two routes target the same queue."""

RouteDefinition

RouteDefinition dataclass

Internal route model — produced by registry, consumed by broker.

Registration metadata is immutable (frozen) after creation. Runtime state (consumer_tag) lives in the separate mutable :class:RouteRuntimeState instance held by :attr:runtime_state, and is populated by the broker during start/reconnect.

A backward-compatible consumer_tag property is provided (read-only) that delegates to runtime_state.consumer_tag so that callers that still read route.consumer_tag (e.g. health.py) keep working without modification. Brokers write via route.runtime_state.consumer_tag = ....

Contains all metadata needed to: - declare topology - start consuming - process messages through pipeline - publish results

See Contracts 1, 4, 5 for semantic rules.

Source code in src/rabbitkit/core/route.py

@dataclass(frozen=True, slots=True)
class RouteDefinition:
    """Internal route model — produced by registry, consumed by broker.

    Registration metadata is immutable (frozen) after creation. Runtime
    state (``consumer_tag``) lives in the separate mutable
    :class:`RouteRuntimeState` instance held by :attr:`runtime_state`, and
    is populated by the broker during start/reconnect.

    A backward-compatible ``consumer_tag`` property is provided (read-only)
    that delegates to ``runtime_state.consumer_tag`` so that callers that
    still read ``route.consumer_tag`` (e.g. ``health.py``) keep working
    without modification. Brokers write via
    ``route.runtime_state.consumer_tag = ...``.

    Contains all metadata needed to:
    - declare topology
    - start consuming
    - process messages through pipeline
    - publish results

    See Contracts 1, 4, 5 for semantic rules.
    """

    # ── Registration metadata (fixed after creation) ──

    # Identity
    name: str

    # Consumer side
    queue: RabbitQueue
    exchange: RabbitExchange | None
    handler: Callable[..., Any]
    ack_policy: AckPolicy = AckPolicy.AUTO
    route_middlewares: list[BaseMiddleware] = field(default_factory=list)

    # Publisher side (optional)
    result_publisher: ResultPublisher | None = None

    # Overrides
    serializer_override: Serializer[Any] | None = None
    retry_override: RetryConfig | RetryDisabled | None = None
    prefetch_count: int | None = None  # Per-route prefetch override (None=use global)
    tags: frozenset[str] = field(default_factory=frozenset)
    description: str = ""

    # Filter predicate — reject messages before deserialization
    filter_fn: Callable[[RabbitMessage], bool] | None = None

    # Per-route override of SafetyConfig.reject_without_dlx (None = inherit)
    reject_without_dlx: str | None = None

    # ── Runtime state (mutable sub-object; populated by broker) ──
    runtime_state: RouteRuntimeState = field(default_factory=RouteRuntimeState)

    # ── Backward-compatible runtime accessor (read-only) ──

    @property
    def consumer_tag(self) -> str | None:
        """Active consumer tag for this route (delegates to runtime_state).

        Read-only (L10) — kept for callers (e.g. ``health.py``) that read
        ``route.consumer_tag``. To set it, write
        ``route.runtime_state.consumer_tag = ...`` directly;
        ``route.consumer_tag = ...`` raises (a ``TypeError``, not a clean
        ``FrozenInstanceError`` — a side effect of ``frozen=True`` combined
        with ``slots=True`` and a same-named property) just like assigning
        any other field on this frozen dataclass. There is no special-cased
        ``__setattr__`` here — the mutable runtime state lives entirely in
        the separate :class:`RouteRuntimeState` object, and this class is
        genuinely frozen.
        """
        return self.runtime_state.consumer_tag

    def has_retry_enabled(self, broker_retry: RetryConfig | None = None) -> bool:
        """Check if this route has retry enabled.

        Resolution order:
        1. retry_override=RetryDisabled → NO retry (explicit opt-out)
        2. retry_override=RetryConfig(...) → YES retry (per-route override)
        3. retry_override=None → inherit broker default (broker_retry)
        """
        if isinstance(self.retry_override, RetryDisabled):
            return False
        if isinstance(self.retry_override, RetryConfig):
            return True
        # None → inherit broker default
        return broker_retry is not None

    def effective_retry_config(self, broker_retry: RetryConfig | None = None) -> RetryConfig | None:
        """Get the effective retry config for this route.

        Returns None if retry is disabled.
        """
        if isinstance(self.retry_override, RetryDisabled):
            return None
        if isinstance(self.retry_override, RetryConfig):
            return self.retry_override
        return broker_retry

    def validate_retry_ack_compatibility(self, broker_retry: RetryConfig | None = None) -> None:
        """Validate that retry + ack policy are compatible.

        Raises ConfigurationError if retry is enabled on MANUAL or ACK_FIRST routes.
        Called at registration time — fail fast.
        """
        if not self.has_retry_enabled(broker_retry):
            return

        if self.ack_policy == AckPolicy.MANUAL:
            raise ConfigurationError(
                f"Route '{self.name}': retry is incompatible with MANUAL ack policy. "
                "MANUAL mode means the handler owns settlement — retry cannot interfere. "
                "Either set ack_policy=AUTO or disable retry via retry=RETRY_DISABLED."
            )

        if self.ack_policy == AckPolicy.ACK_FIRST:
            raise ConfigurationError(
                f"Route '{self.name}': retry is incompatible with ACK_FIRST ack policy. "
                "ACK_FIRST acks before the handler runs — retry cannot nack/reject. "
                "Either set ack_policy=AUTO or disable retry via retry=RETRY_DISABLED."
            )

    def validate_retry_dlx_conflict(self, broker_retry: RetryConfig | None = None) -> None:
        """Validate that retry and manual DLX config don't conflict.

        If retry is enabled, RetryRouter owns DLQ topology.
        User must not also set dead_letter_exchange on the queue.
        """
        if not self.has_retry_enabled(broker_retry):
            return

        if self.queue.dead_letter_exchange is not None:
            raise ConfigurationError(
                f"Route '{self.name}': retry is enabled but queue '{self.queue.name}' "
                "already has dead_letter_exchange set. RetryRouter owns DLQ topology "
                "when retry is enabled — do not set dead_letter_exchange manually. "
                "To use custom DLQ routing, disable retry via retry=RETRY_DISABLED."
            )

        if self.queue.dead_letter_routing_key is not None:
            raise ConfigurationError(
                f"Route '{self.name}': retry is enabled but queue '{self.queue.name}' "
                "already has dead_letter_routing_key set. RetryRouter owns DLQ topology "
                "when retry is enabled — do not set dead_letter_routing_key manually. "
                "To use custom DLQ routing, disable retry via retry=RETRY_DISABLED."
            )

    def validate_headers_binding(self) -> None:
        """Validate headers-exchange bindings at registration time (C4).

        A headers exchange routes on binding *arguments*, not routing keys —
        an argument-less binding matches everything (silent firehose /
        misrouting). Require ``bind_arguments`` with a valid ``x-match``.
        """
        if self.exchange is None or self.exchange.type != ExchangeType.HEADERS:
            return
        if not self.queue.bind_arguments:
            raise ConfigurationError(
                f"Route '{self.name}': queue '{self.queue.name}' binds to headers "
                f"exchange '{self.exchange.name}' without bind_arguments. A headers "
                "binding with no arguments matches EVERY message. Set "
                "RabbitQueue(bind_arguments={'x-match': 'all'|'any', ...})."
            )
        x_match = self.queue.bind_arguments.get("x-match", "all")
        if x_match not in ("all", "any", "all-with-x", "any-with-x"):
            raise ConfigurationError(
                f"Route '{self.name}': bind_arguments['x-match'] must be 'all', "
                f"'any', 'all-with-x', or 'any-with-x'; got {x_match!r}."
            )

    def validate_reject_without_dlx_value(self) -> None:
        """Validate the per-route reject_without_dlx override value."""
        if self.reject_without_dlx is None:
            return
        if self.reject_without_dlx not in ("auto_provision", "error", "discard"):
            raise ConfigurationError(
                f"Route '{self.name}': reject_without_dlx must be one of "
                f"'auto_provision', 'error', 'discard'; got {self.reject_without_dlx!r}."
            )

    def validate(self, broker_retry: RetryConfig | None = None) -> None:
        """Run all route validations.

        Called at registration time — fail fast on conflicts.
        """
        self.validate_retry_ack_compatibility(broker_retry)
        self.validate_retry_dlx_conflict(broker_retry)
        self.validate_headers_binding()
        self.validate_reject_without_dlx_value()

    def can_reject_without_dlx(self, broker_retry: RetryConfig | None = None) -> bool:
        """True when this route can ``reject(requeue=False)`` into a queue
        with no dead-letter path — i.e. RabbitMQ would discard the message.

        Retry-enabled routes get a DLX from RetryRouter; a manually
        configured ``dead_letter_exchange`` is a dead-letter path already.
        ACK_FIRST routes without a filter ack before the handler runs, so
        they never reject. Everything else (AUTO/NACK_ON_ERROR/MANUAL,
        filter_fn, permanent-error classification) can reject.
        """
        if self.effective_retry_config(broker_retry) is not None:
            return False
        if self.queue.dead_letter_exchange is not None:
            return False
        return not (self.ack_policy == AckPolicy.ACK_FIRST and self.filter_fn is None)

    def resolve_safety_dlq(
        self,
        safety: SafetyConfig,
        broker_retry: RetryConfig | None = None,
    ) -> str | None:
        """Apply the reject-without-DLX safety policy to this route (C3).

        Returns the DLQ name to auto-provision, or ``None`` when nothing is
        needed (route already has a dead-letter path, cannot reject, or the
        policy is ``discard``). Raises :class:`UnsafeTopologyError` under the
        ``error`` policy.

        Called by brokers during topology declaration, only under
        ``TopologyMode.AUTO_DECLARE``.
        """
        if not self.can_reject_without_dlx(broker_retry):
            return None

        policy = self.reject_without_dlx or safety.reject_without_dlx
        if policy == RejectWithoutDLXPolicy.AUTO_PROVISION:
            return f"{self.queue.name}{safety.dlq_suffix}"
        if policy == RejectWithoutDLXPolicy.ERROR:
            raise UnsafeTopologyError(
                f"Queue '{self.queue.name}' (route '{self.name}') can reject "
                "messages with requeue=False, but no dead-letter exchange is "
                "configured — RabbitMQ would discard them permanently. Configure "
                "a DLX/DLQ on the queue, use reject_without_dlx='auto_provision', "
                "or explicitly opt into loss with reject_without_dlx='discard'."
            )
        # policy == DISCARD: explicit opt-in to loss
        if safety.warn_on_discard:
            import warnings

            warnings.warn(
                f"Route {self.name!r}: reject_without_dlx='discard' — messages "
                f"rejected with requeue=False on queue {self.queue.name!r} will be "
                "permanently discarded by RabbitMQ (no dead-letter exchange). Use "
                "this only when message loss is acceptable.",
                RuntimeWarning,
                stacklevel=2,
            )
        return None

Attributes

consumer_tag: str | None property

Active consumer tag for this route (delegates to runtime_state).

Read-only (L10) — kept for callers (e.g. health.py) that read route.consumer_tag. To set it, write route.runtime_state.consumer_tag = ... directly; route.consumer_tag = ... raises (a TypeError, not a clean FrozenInstanceError — a side effect of frozen=True combined with slots=True and a same-named property) just like assigning any other field on this frozen dataclass. There is no special-cased __setattr__ here — the mutable runtime state lives entirely in the separate :class:RouteRuntimeState object, and this class is genuinely frozen.

Methods:

has_retry_enabled(broker_retry: RetryConfig | None = None) -> bool

Check if this route has retry enabled.

Resolution order: 1. retry_override=RetryDisabled → NO retry (explicit opt-out) 2. retry_override=RetryConfig(...) → YES retry (per-route override) 3. retry_override=None → inherit broker default (broker_retry)

Source code in src/rabbitkit/core/route.py

def has_retry_enabled(self, broker_retry: RetryConfig | None = None) -> bool:
    """Check if this route has retry enabled.

    Resolution order:
    1. retry_override=RetryDisabled → NO retry (explicit opt-out)
    2. retry_override=RetryConfig(...) → YES retry (per-route override)
    3. retry_override=None → inherit broker default (broker_retry)
    """
    if isinstance(self.retry_override, RetryDisabled):
        return False
    if isinstance(self.retry_override, RetryConfig):
        return True
    # None → inherit broker default
    return broker_retry is not None

effective_retry_config(broker_retry: RetryConfig | None = None) -> RetryConfig | None

Get the effective retry config for this route.

Returns None if retry is disabled.

Source code in src/rabbitkit/core/route.py

def effective_retry_config(self, broker_retry: RetryConfig | None = None) -> RetryConfig | None:
    """Get the effective retry config for this route.

    Returns None if retry is disabled.
    """
    if isinstance(self.retry_override, RetryDisabled):
        return None
    if isinstance(self.retry_override, RetryConfig):
        return self.retry_override
    return broker_retry

validate_retry_ack_compatibility(broker_retry: RetryConfig | None = None) -> None

Validate that retry + ack policy are compatible.

Raises ConfigurationError if retry is enabled on MANUAL or ACK_FIRST routes. Called at registration time — fail fast.

Source code in src/rabbitkit/core/route.py

def validate_retry_ack_compatibility(self, broker_retry: RetryConfig | None = None) -> None:
    """Validate that retry + ack policy are compatible.

    Raises ConfigurationError if retry is enabled on MANUAL or ACK_FIRST routes.
    Called at registration time — fail fast.
    """
    if not self.has_retry_enabled(broker_retry):
        return

    if self.ack_policy == AckPolicy.MANUAL:
        raise ConfigurationError(
            f"Route '{self.name}': retry is incompatible with MANUAL ack policy. "
            "MANUAL mode means the handler owns settlement — retry cannot interfere. "
            "Either set ack_policy=AUTO or disable retry via retry=RETRY_DISABLED."
        )

    if self.ack_policy == AckPolicy.ACK_FIRST:
        raise ConfigurationError(
            f"Route '{self.name}': retry is incompatible with ACK_FIRST ack policy. "
            "ACK_FIRST acks before the handler runs — retry cannot nack/reject. "
            "Either set ack_policy=AUTO or disable retry via retry=RETRY_DISABLED."
        )

validate_retry_dlx_conflict(broker_retry: RetryConfig | None = None) -> None

Validate that retry and manual DLX config don't conflict.

If retry is enabled, RetryRouter owns DLQ topology. User must not also set dead_letter_exchange on the queue.

Source code in src/rabbitkit/core/route.py

def validate_retry_dlx_conflict(self, broker_retry: RetryConfig | None = None) -> None:
    """Validate that retry and manual DLX config don't conflict.

    If retry is enabled, RetryRouter owns DLQ topology.
    User must not also set dead_letter_exchange on the queue.
    """
    if not self.has_retry_enabled(broker_retry):
        return

    if self.queue.dead_letter_exchange is not None:
        raise ConfigurationError(
            f"Route '{self.name}': retry is enabled but queue '{self.queue.name}' "
            "already has dead_letter_exchange set. RetryRouter owns DLQ topology "
            "when retry is enabled — do not set dead_letter_exchange manually. "
            "To use custom DLQ routing, disable retry via retry=RETRY_DISABLED."
        )

    if self.queue.dead_letter_routing_key is not None:
        raise ConfigurationError(
            f"Route '{self.name}': retry is enabled but queue '{self.queue.name}' "
            "already has dead_letter_routing_key set. RetryRouter owns DLQ topology "
            "when retry is enabled — do not set dead_letter_routing_key manually. "
            "To use custom DLQ routing, disable retry via retry=RETRY_DISABLED."
        )

validate_headers_binding() -> None

Validate headers-exchange bindings at registration time (C4).

A headers exchange routes on binding arguments, not routing keys — an argument-less binding matches everything (silent firehose / misrouting). Require bind_arguments with a valid x-match.

Source code in src/rabbitkit/core/route.py

def validate_headers_binding(self) -> None:
    """Validate headers-exchange bindings at registration time (C4).

    A headers exchange routes on binding *arguments*, not routing keys —
    an argument-less binding matches everything (silent firehose /
    misrouting). Require ``bind_arguments`` with a valid ``x-match``.
    """
    if self.exchange is None or self.exchange.type != ExchangeType.HEADERS:
        return
    if not self.queue.bind_arguments:
        raise ConfigurationError(
            f"Route '{self.name}': queue '{self.queue.name}' binds to headers "
            f"exchange '{self.exchange.name}' without bind_arguments. A headers "
            "binding with no arguments matches EVERY message. Set "
            "RabbitQueue(bind_arguments={'x-match': 'all'|'any', ...})."
        )
    x_match = self.queue.bind_arguments.get("x-match", "all")
    if x_match not in ("all", "any", "all-with-x", "any-with-x"):
        raise ConfigurationError(
            f"Route '{self.name}': bind_arguments['x-match'] must be 'all', "
            f"'any', 'all-with-x', or 'any-with-x'; got {x_match!r}."
        )

validate_reject_without_dlx_value() -> None

Validate the per-route reject_without_dlx override value.

Source code in src/rabbitkit/core/route.py

def validate_reject_without_dlx_value(self) -> None:
    """Validate the per-route reject_without_dlx override value."""
    if self.reject_without_dlx is None:
        return
    if self.reject_without_dlx not in ("auto_provision", "error", "discard"):
        raise ConfigurationError(
            f"Route '{self.name}': reject_without_dlx must be one of "
            f"'auto_provision', 'error', 'discard'; got {self.reject_without_dlx!r}."
        )

validate(broker_retry: RetryConfig | None = None) -> None

Run all route validations.

Called at registration time — fail fast on conflicts.

Source code in src/rabbitkit/core/route.py

def validate(self, broker_retry: RetryConfig | None = None) -> None:
    """Run all route validations.

    Called at registration time — fail fast on conflicts.
    """
    self.validate_retry_ack_compatibility(broker_retry)
    self.validate_retry_dlx_conflict(broker_retry)
    self.validate_headers_binding()
    self.validate_reject_without_dlx_value()

can_reject_without_dlx(broker_retry: RetryConfig | None = None) -> bool

True when this route can reject(requeue=False) into a queue with no dead-letter path — i.e. RabbitMQ would discard the message.

Retry-enabled routes get a DLX from RetryRouter; a manually configured dead_letter_exchange is a dead-letter path already. ACK_FIRST routes without a filter ack before the handler runs, so they never reject. Everything else (AUTO/NACK_ON_ERROR/MANUAL, filter_fn, permanent-error classification) can reject.

Source code in src/rabbitkit/core/route.py

def can_reject_without_dlx(self, broker_retry: RetryConfig | None = None) -> bool:
    """True when this route can ``reject(requeue=False)`` into a queue
    with no dead-letter path — i.e. RabbitMQ would discard the message.

    Retry-enabled routes get a DLX from RetryRouter; a manually
    configured ``dead_letter_exchange`` is a dead-letter path already.
    ACK_FIRST routes without a filter ack before the handler runs, so
    they never reject. Everything else (AUTO/NACK_ON_ERROR/MANUAL,
    filter_fn, permanent-error classification) can reject.
    """
    if self.effective_retry_config(broker_retry) is not None:
        return False
    if self.queue.dead_letter_exchange is not None:
        return False
    return not (self.ack_policy == AckPolicy.ACK_FIRST and self.filter_fn is None)

resolve_safety_dlq(safety: SafetyConfig, broker_retry: RetryConfig | None = None) -> str | None

Apply the reject-without-DLX safety policy to this route (C3).

Returns the DLQ name to auto-provision, or None when nothing is needed (route already has a dead-letter path, cannot reject, or the policy is discard). Raises :class:UnsafeTopologyError under the error policy.

Called by brokers during topology declaration, only under TopologyMode.AUTO_DECLARE.

Source code in src/rabbitkit/core/route.py

def resolve_safety_dlq(
    self,
    safety: SafetyConfig,
    broker_retry: RetryConfig | None = None,
) -> str | None:
    """Apply the reject-without-DLX safety policy to this route (C3).

    Returns the DLQ name to auto-provision, or ``None`` when nothing is
    needed (route already has a dead-letter path, cannot reject, or the
    policy is ``discard``). Raises :class:`UnsafeTopologyError` under the
    ``error`` policy.

    Called by brokers during topology declaration, only under
    ``TopologyMode.AUTO_DECLARE``.
    """
    if not self.can_reject_without_dlx(broker_retry):
        return None

    policy = self.reject_without_dlx or safety.reject_without_dlx
    if policy == RejectWithoutDLXPolicy.AUTO_PROVISION:
        return f"{self.queue.name}{safety.dlq_suffix}"
    if policy == RejectWithoutDLXPolicy.ERROR:
        raise UnsafeTopologyError(
            f"Queue '{self.queue.name}' (route '{self.name}') can reject "
            "messages with requeue=False, but no dead-letter exchange is "
            "configured — RabbitMQ would discard them permanently. Configure "
            "a DLX/DLQ on the queue, use reject_without_dlx='auto_provision', "
            "or explicitly opt into loss with reject_without_dlx='discard'."
        )
    # policy == DISCARD: explicit opt-in to loss
    if safety.warn_on_discard:
        import warnings

        warnings.warn(
            f"Route {self.name!r}: reject_without_dlx='discard' — messages "
            f"rejected with requeue=False on queue {self.queue.name!r} will be "
            "permanently discarded by RabbitMQ (no dead-letter exchange). Use "
            "this only when message loss is acceptable.",
            RuntimeWarning,
            stacklevel=2,
        )
    return None

ResultPublisher dataclass

Where to publish handler return values.

Set by @publisher decorator on a handler.

Source code in src/rabbitkit/core/route.py

@dataclass(frozen=True, slots=True)
class ResultPublisher:
    """Where to publish handler return values.

    Set by @publisher decorator on a handler.
    """

    exchange: RabbitExchange | str | None = None
    routing_key: str = ""

    def resolve_exchange_name(self) -> str:
        """Get the exchange name as string."""
        if self.exchange is None:
            return ""
        if isinstance(self.exchange, str):
            return self.exchange
        return self.exchange.name

Methods:

resolve_exchange_name() -> str

Get the exchange name as string.

Source code in src/rabbitkit/core/route.py

def resolve_exchange_name(self) -> str:
    """Get the exchange name as string."""
    if self.exchange is None:
        return ""
    if isinstance(self.exchange, str):
        return self.exchange
    return self.exchange.name

ConfigurationError

Bases: Exception

Raised for invalid configuration detected at registration time.

Single canonical class for all registration-time misconfigurations (route conflicts, invalid handler signatures, bad retry/ack combinations). Both core/route.py and di/resolver.py raise this; tests/users can catch one type regardless of import source.

Source code in src/rabbitkit/core/errors.py

class ConfigurationError(Exception):
    """Raised for invalid configuration detected at registration time.

    Single canonical class for all registration-time misconfigurations (route
    conflicts, invalid handler signatures, bad retry/ack combinations). Both
    ``core/route.py`` and ``di/resolver.py`` raise this; tests/users can catch
    one type regardless of import source.
    """

Error Classification

classify_error(exc: BaseException, *, predicates: Sequence[ErrorPredicate] = (), transient: tuple[type[BaseException], ...] = TRANSIENT_ERRORS, permanent: tuple[type[BaseException], ...] = PERMANENT_ERRORS, unknown_policy: ErrorSeverity = ErrorSeverity.PERMANENT) -> ClassifiedError

Classify exception severity.

Evaluation order: 1. User predicates (first non-None wins) 2. transient tuple (isinstance check) 3. permanent tuple (isinstance check) 4. unknown_policy (configurable, default=PERMANENT)

DEFAULT IS PERMANENT, NOT TRANSIENT. Reason: unknown errors include malformed payloads, business validation failures, handler bugs, and schema mismatches. Treating these as transient creates retry storms. Network/connection errors are already in the transient tuple.

Override per-route: RetryMiddleware accepts unknown_policy to change per route.

Parameters:

Name	Type	Description	Default
exc	BaseException	The exception to classify.	required
predicates	Sequence[ErrorPredicate]	User-provided classification functions. Return True=transient, False=permanent, None=no opinion.	()
transient	tuple[type[BaseException], ...]	Exception types considered transient.	TRANSIENT_ERRORS
permanent	tuple[type[BaseException], ...]	Exception types considered permanent.	PERMANENT_ERRORS
unknown_policy	ErrorSeverity	Severity for unclassified exceptions.	PERMANENT

Returns:

Type	Description
ClassifiedError	ClassifiedError with severity, original exception, and reason.

Source code in src/rabbitkit/core/errors.py

def classify_error(
    exc: BaseException,
    *,
    predicates: Sequence[ErrorPredicate] = (),
    transient: tuple[type[BaseException], ...] = TRANSIENT_ERRORS,
    permanent: tuple[type[BaseException], ...] = PERMANENT_ERRORS,
    unknown_policy: ErrorSeverity = ErrorSeverity.PERMANENT,
) -> ClassifiedError:
    """Classify exception severity.

    Evaluation order:
    1. User predicates (first non-None wins)
    2. transient tuple (isinstance check)
    3. permanent tuple (isinstance check)
    4. unknown_policy (configurable, default=PERMANENT)

    DEFAULT IS PERMANENT, NOT TRANSIENT.
    Reason: unknown errors include malformed payloads, business validation
    failures, handler bugs, and schema mismatches. Treating these as
    transient creates retry storms. Network/connection errors are already
    in the transient tuple.

    Override per-route: RetryMiddleware accepts unknown_policy to change
    per route.

    Args:
        exc: The exception to classify.
        predicates: User-provided classification functions.
            Return True=transient, False=permanent, None=no opinion.
        transient: Exception types considered transient.
        permanent: Exception types considered permanent.
        unknown_policy: Severity for unclassified exceptions.

    Returns:
        ClassifiedError with severity, original exception, and reason.
    """
    # 1. User predicates (first non-None wins)
    for predicate in predicates:
        result = predicate(exc)
        if result is True:
            return ClassifiedError(
                severity=ErrorSeverity.TRANSIENT,
                original=exc,
                reason=f"predicate classified as transient: {type(exc).__name__}",
            )
        if result is False:
            return ClassifiedError(
                severity=ErrorSeverity.PERMANENT,
                original=exc,
                reason=f"predicate classified as permanent: {type(exc).__name__}",
            )

    # 2. Transient tuple (isinstance check)
    if isinstance(exc, transient):
        return ClassifiedError(
            severity=ErrorSeverity.TRANSIENT,
            original=exc,
            reason=f"transient error: {type(exc).__name__}",
        )

    # 3. Permanent tuple (isinstance check)
    if isinstance(exc, permanent):
        return ClassifiedError(
            severity=ErrorSeverity.PERMANENT,
            original=exc,
            reason=f"permanent error: {type(exc).__name__}",
        )

    # 4. Unknown policy
    return ClassifiedError(
        severity=unknown_policy,
        original=exc,
        reason=f"unknown error classified as {unknown_policy.value}: {type(exc).__name__}",
    )

ErrorSeverity

Bases: str, Enum

Error classification severity levels.

Source code in src/rabbitkit/core/types.py

class ErrorSeverity(str, Enum):
    """Error classification severity levels."""

    TRANSIENT = "transient"
    PERMANENT = "permanent"

ClassifiedError dataclass

Result of error classification.

Source code in src/rabbitkit/core/types.py

@dataclass(frozen=True, slots=True)
class ClassifiedError:
    """Result of error classification."""

    severity: ErrorSeverity
    original: BaseException
    reason: str

Exceptions

All raised for caller-side mistakes; the validation errors dual-inherit ValueError and the runtime-misuse errors dual-inherit RuntimeError, so pre-existing builtin catches keep working. See the exception taxonomy for the full table and catch patterns.

ConfigurationError

Bases: Exception

Raised for invalid configuration detected at registration time.

Single canonical class for all registration-time misconfigurations (route conflicts, invalid handler signatures, bad retry/ack combinations). Both core/route.py and di/resolver.py raise this; tests/users can catch one type regardless of import source.

Source code in src/rabbitkit/core/errors.py

class ConfigurationError(Exception):
    """Raised for invalid configuration detected at registration time.

    Single canonical class for all registration-time misconfigurations (route
    conflicts, invalid handler signatures, bad retry/ack combinations). Both
    ``core/route.py`` and ``di/resolver.py`` raise this; tests/users can catch
    one type regardless of import source.
    """

ConfigValidationError

Bases: ConfigurationError, ValueError

An invalid value was passed to a rabbitkit config object or constructor — RabbitConfig sub-configs (RetryConfig(max_retries=-1), WorkerConfig(max_queue_size=-1), …) and AMQP short-string fields (queue/exchange/routing-key names that are too long or contain control characters, including on MessageEnvelope).

Raised at construction time (__post_init__), so a bad value fails where it's written, not at first use.

Source code in src/rabbitkit/core/errors.py

class ConfigValidationError(ConfigurationError, ValueError):
    """An invalid value was passed to a rabbitkit config object or
    constructor — ``RabbitConfig`` sub-configs (``RetryConfig(max_retries=-1)``,
    ``WorkerConfig(max_queue_size=-1)``, …) and AMQP short-string fields
    (queue/exchange/routing-key names that are too long or contain control
    characters, including on ``MessageEnvelope``).

    Raised at construction time (``__post_init__``), so a bad value fails
    where it's written, not at first use.
    """

TopologyValidationError

Bases: ConfigurationError, ValueError

An invalid RabbitQueue / RabbitExchange declaration — a combination RabbitMQ itself would reject or silently misbehave on (non-durable quorum queue, priorities on a stream, delivery_limit on a classic queue, non-positive consumer_timeout, …).

Raised at model construction time, before anything touches the broker.

Source code in src/rabbitkit/core/errors.py

class TopologyValidationError(ConfigurationError, ValueError):
    """An invalid ``RabbitQueue`` / ``RabbitExchange`` declaration — a
    combination RabbitMQ itself would reject or silently misbehave on
    (non-durable quorum queue, priorities on a stream, ``delivery_limit`` on
    a classic queue, non-positive ``consumer_timeout``, …).

    Raised at model construction time, before anything touches the broker.
    """

UnsafeTopologyError

Bases: ConfigurationError

Raised at startup when RejectWithoutDLXPolicy.ERROR is active and a route can reject(requeue=False) but its queue has no dead-letter exchange — RabbitMQ would silently discard rejected messages.

Subclasses :class:ConfigurationError so existing catch-alls for registration/startup misconfiguration keep working.

Source code in src/rabbitkit/core/errors.py

class UnsafeTopologyError(ConfigurationError):
    """Raised at startup when ``RejectWithoutDLXPolicy.ERROR`` is active and a
    route can ``reject(requeue=False)`` but its queue has no dead-letter
    exchange — RabbitMQ would silently discard rejected messages.

    Subclasses :class:`ConfigurationError` so existing catch-alls for
    registration/startup misconfiguration keep working.
    """

MessageTooLargeError

Bases: ValueError

A publish was rejected client-side because the body exceeds PublisherConfig.max_message_bytes (default 16 MiB, mirroring the server's max_message_size default).

Raised before the bytes hit the wire: the server would reject the message anyway, but with a channel exception that kills the (pooled) publisher channel and corrupts sibling in-flight publishes. Subclasses ValueError for backward compatibility with earlier releases that raised it directly.

Source code in src/rabbitkit/core/errors.py

class MessageTooLargeError(ValueError):
    """A publish was rejected client-side because the body exceeds
    ``PublisherConfig.max_message_bytes`` (default 16 MiB, mirroring the
    server's ``max_message_size`` default).

    Raised *before* the bytes hit the wire: the server would reject the
    message anyway, but with a channel exception that kills the (pooled)
    publisher channel and corrupts sibling in-flight publishes. Subclasses
    ``ValueError`` for backward compatibility with earlier releases that
    raised it directly.
    """

BrokerNotStartedError

Bases: RuntimeError

A broker method that needs a live transport (publish(), RPC, topology helpers) was called before start() (or after stop()).

Source code in src/rabbitkit/core/errors.py

class BrokerNotStartedError(RuntimeError):
    """A broker method that needs a live transport (``publish()``, RPC,
    topology helpers) was called before ``start()`` (or after ``stop()``).
    """

SettlementError

Bases: RuntimeError

Invalid message-settlement call: calling the sync settlement methods (ack() ...) on a message from an async-only transport, or the async variants (ack_async() ...) on a message with no settlement function wired at all.

Defined here rather than in core/errors.py because errors.py (indirectly, via types.py) imports this module -- it is re-exported from rabbitkit.core.errors and the package root, which are the import paths to use. Subclasses RuntimeError so pre-existing except RuntimeError handlers keep working.

Source code in src/rabbitkit/core/message.py

class SettlementError(RuntimeError):
    """Invalid message-settlement call: calling the sync settlement methods
    (``ack()`` ...) on a message from an async-only transport, or the async
    variants (``ack_async()`` ...) on a message with no settlement function
    wired at all.

    Defined here rather than in ``core/errors.py`` because ``errors.py``
    (indirectly, via ``types.py``) imports this module -- it is re-exported
    from ``rabbitkit.core.errors`` and the package root, which are the
    import paths to use. Subclasses ``RuntimeError`` so pre-existing
    ``except RuntimeError`` handlers keep working.
    """

MissingDependencyError

Bases: Exception

Raised at message-processing time when a required Header() / Path() / Context() marker's value is absent from the incoming message and no default is available — neither on the marker itself (Header("x-tenant", default=...)) nor the handler's own parameter default (tenant: Annotated[str | None, Header("x-tenant")] = None).

Names the specific parameter and marker so the failure is immediately actionable — unlike the bare KeyError this replaces, which looked identical to a handler bug (e.g. indexing a dict) and gave no indication which DI marker was the culprit. Classified PERMANENT by :func:classify_error (see PERMANENT_ERRORS below), matching the KeyError classification it replaces: a missing required value means the message itself is malformed for this handler — retrying will not fix it, so it settles straight to the DLQ rather than looping.

Source code in src/rabbitkit/core/errors.py

class MissingDependencyError(Exception):
    """Raised at message-processing time when a required ``Header()`` /
    ``Path()`` / ``Context()`` marker's value is absent from the incoming
    message and no default is available — neither on the marker itself
    (``Header("x-tenant", default=...)``) nor the handler's own parameter
    default (``tenant: Annotated[str | None, Header("x-tenant")] = None``).

    Names the specific parameter and marker so the failure is immediately
    actionable — unlike the bare ``KeyError`` this replaces, which looked
    identical to a handler bug (e.g. indexing a dict) and gave no indication
    which DI marker was the culprit. Classified PERMANENT by
    :func:`classify_error` (see ``PERMANENT_ERRORS`` below), matching the
    ``KeyError`` classification it replaces: a missing required value means
    the message itself is malformed for this handler — retrying will not fix
    it, so it settles straight to the DLQ rather than looping.
    """

    def __init__(self, marker_repr: str, param_name: str) -> None:
        super().__init__(
            f"{marker_repr} for parameter {param_name!r} is required but missing from the "
            "message, and no default is available (neither on the marker itself nor as a "
            "Python parameter default)."
        )
        self.param_name = param_name

BackpressureError

Bases: Exception

Raised when publish-side flow control blocks a publish attempt.

Only raised when BackpressureConfig.on_blocked == "raise".

Source code in src/rabbitkit/core/errors.py

class BackpressureError(Exception):
    """Raised when publish-side flow control blocks a publish attempt.

    Only raised when ``BackpressureConfig.on_blocked == "raise"``.
    """

PublishError

Bases: Exception

Raised by PublishOutcome.raise_for_status() on a failed publish.

broker.publish() never raises on its own — it returns a PublishOutcome so callers can decide how to handle NACKED / TIMEOUT / RETURNED / ERROR. Code that prefers exceptions can opt in with broker.publish(...).raise_for_status(). Carries the outcome for inspection (status, routing_key, underlying error).

Source code in src/rabbitkit/core/errors.py

class PublishError(Exception):
    """Raised by ``PublishOutcome.raise_for_status()`` on a failed publish.

    ``broker.publish()`` never raises on its own — it returns a
    ``PublishOutcome`` so callers can decide how to handle NACKED / TIMEOUT /
    RETURNED / ERROR. Code that prefers exceptions can opt in with
    ``broker.publish(...).raise_for_status()``. Carries the ``outcome`` for
    inspection (status, routing_key, underlying error).
    """

    def __init__(self, outcome: object) -> None:
        self.outcome = outcome
        super().__init__(str(outcome))

Protocols

Transport

Bases: Protocol

Sync transport — implemented by sync/transport.py.

Minimal interface for sync message broker I/O.

Source code in src/rabbitkit/core/protocols.py

@runtime_checkable
class Transport(Protocol):
    """Sync transport — implemented by sync/transport.py.

    Minimal interface for sync message broker I/O.
    """

    def connect(self) -> None:
        """Establish connection to broker."""
        ...

    def disconnect(self) -> None:
        """Close connection to broker."""
        ...

    def is_connected(self) -> bool:
        """Check if transport is connected."""
        ...

    def publish(self, envelope: MessageEnvelope) -> PublishOutcome:
        """Publish a message. Returns outcome with confirm status."""
        ...

    def consume(
        self,
        queue: str,
        callback: Callable[[RabbitMessage], None],
        prefetch: int = 10,
        *,
        no_ack: bool = False,
        declare: bool = True,
    ) -> str:
        """Start consuming from a queue. Returns consumer_tag.

        ``no_ack=True`` starts a no-ack consumer (the broker auto-acks on
        delivery; the built ``RabbitMessage`` gets no settlement functions).
        ``declare=False`` skips declaring/checking the queue first — required
        for AMQP pseudo-queues such as ``amq.rabbitmq.reply-to``, which the
        broker rejects any Queue.Declare for (active or passive).
        """
        ...

    def declare_exchange(self, exchange: RabbitExchange) -> None:
        """Declare an exchange on the broker."""
        ...

    def declare_queue(self, queue: RabbitQueue) -> None:
        """Declare a queue on the broker."""
        ...

    def bind_queue(
        self,
        queue: str,
        exchange: str,
        routing_key: str,
        arguments: dict[str, Any] | None = None,
    ) -> None:
        """Bind a queue to an exchange with a routing key.

        ``arguments`` carries header-match criteria for HEADERS exchanges.
        """
        ...

    def bind_exchange(
        self,
        destination: str,
        source: str,
        routing_key: str = "",
        arguments: dict[str, Any] | None = None,
    ) -> None:
        """Bind an exchange to another exchange (exchange-to-exchange binding)."""
        ...

    def cancel_consumer(self, consumer_tag: str) -> None:
        """Cancel a consumer by its tag."""
        ...

Methods:

connect() -> None

Establish connection to broker.

Source code in src/rabbitkit/core/protocols.py

def connect(self) -> None:
    """Establish connection to broker."""
    ...

disconnect() -> None

Close connection to broker.

Source code in src/rabbitkit/core/protocols.py

def disconnect(self) -> None:
    """Close connection to broker."""
    ...

is_connected() -> bool

Check if transport is connected.

Source code in src/rabbitkit/core/protocols.py

def is_connected(self) -> bool:
    """Check if transport is connected."""
    ...

publish(envelope: MessageEnvelope) -> PublishOutcome

Publish a message. Returns outcome with confirm status.

Source code in src/rabbitkit/core/protocols.py

def publish(self, envelope: MessageEnvelope) -> PublishOutcome:
    """Publish a message. Returns outcome with confirm status."""
    ...

consume(queue: str, callback: Callable[[RabbitMessage], None], prefetch: int = 10, *, no_ack: bool = False, declare: bool = True) -> str

Start consuming from a queue. Returns consumer_tag.

no_ack=True starts a no-ack consumer (the broker auto-acks on delivery; the built RabbitMessage gets no settlement functions). declare=False skips declaring/checking the queue first — required for AMQP pseudo-queues such as amq.rabbitmq.reply-to, which the broker rejects any Queue.Declare for (active or passive).

Source code in src/rabbitkit/core/protocols.py

def consume(
    self,
    queue: str,
    callback: Callable[[RabbitMessage], None],
    prefetch: int = 10,
    *,
    no_ack: bool = False,
    declare: bool = True,
) -> str:
    """Start consuming from a queue. Returns consumer_tag.

    ``no_ack=True`` starts a no-ack consumer (the broker auto-acks on
    delivery; the built ``RabbitMessage`` gets no settlement functions).
    ``declare=False`` skips declaring/checking the queue first — required
    for AMQP pseudo-queues such as ``amq.rabbitmq.reply-to``, which the
    broker rejects any Queue.Declare for (active or passive).
    """
    ...

declare_exchange(exchange: RabbitExchange) -> None

Declare an exchange on the broker.

Source code in src/rabbitkit/core/protocols.py

def declare_exchange(self, exchange: RabbitExchange) -> None:
    """Declare an exchange on the broker."""
    ...

declare_queue(queue: RabbitQueue) -> None

Declare a queue on the broker.

Source code in src/rabbitkit/core/protocols.py

def declare_queue(self, queue: RabbitQueue) -> None:
    """Declare a queue on the broker."""
    ...

bind_queue(queue: str, exchange: str, routing_key: str, arguments: dict[str, Any] | None = None) -> None

Bind a queue to an exchange with a routing key.

arguments carries header-match criteria for HEADERS exchanges.

Source code in src/rabbitkit/core/protocols.py

def bind_queue(
    self,
    queue: str,
    exchange: str,
    routing_key: str,
    arguments: dict[str, Any] | None = None,
) -> None:
    """Bind a queue to an exchange with a routing key.

    ``arguments`` carries header-match criteria for HEADERS exchanges.
    """
    ...

bind_exchange(destination: str, source: str, routing_key: str = '', arguments: dict[str, Any] | None = None) -> None

Bind an exchange to another exchange (exchange-to-exchange binding).

Source code in src/rabbitkit/core/protocols.py

def bind_exchange(
    self,
    destination: str,
    source: str,
    routing_key: str = "",
    arguments: dict[str, Any] | None = None,
) -> None:
    """Bind an exchange to another exchange (exchange-to-exchange binding)."""
    ...

cancel_consumer(consumer_tag: str) -> None

Cancel a consumer by its tag.

Source code in src/rabbitkit/core/protocols.py

def cancel_consumer(self, consumer_tag: str) -> None:
    """Cancel a consumer by its tag."""
    ...

AsyncTransport

Bases: Protocol

Async transport — implemented by async_/transport.py.

Minimal interface for async message broker I/O.

Source code in src/rabbitkit/core/protocols.py

@runtime_checkable
class AsyncTransport(Protocol):
    """Async transport — implemented by async_/transport.py.

    Minimal interface for async message broker I/O.
    """

    async def connect(self) -> None:
        """Establish connection to broker."""
        ...

    async def disconnect(self) -> None:
        """Close connection to broker."""
        ...

    def is_connected(self) -> bool:
        """Check if transport is connected."""
        ...

    async def publish(self, envelope: MessageEnvelope) -> PublishOutcome:
        """Publish a message. Returns outcome with confirm status."""
        ...

    async def consume(
        self,
        queue: str,
        callback: Callable[[RabbitMessage], Awaitable[None]],
        prefetch: int = 10,
        *,
        no_ack: bool = False,
        declare: bool = True,
    ) -> str:
        """Start consuming from a queue. Returns consumer_tag.

        ``no_ack=True`` starts a no-ack consumer (the broker auto-acks on
        delivery; the built ``RabbitMessage`` gets no settlement functions).
        ``declare=False`` skips declaring/checking the queue first — required
        for AMQP pseudo-queues such as ``amq.rabbitmq.reply-to``, which the
        broker rejects any Queue.Declare for (active or passive).
        """
        ...

    async def declare_exchange(self, exchange: RabbitExchange) -> None:
        """Declare an exchange on the broker."""
        ...

    async def declare_queue(self, queue: RabbitQueue) -> None:
        """Declare a queue on the broker."""
        ...

    async def bind_queue(
        self,
        queue: str,
        exchange: str,
        routing_key: str,
        arguments: dict[str, Any] | None = None,
    ) -> None:
        """Bind a queue to an exchange with a routing key.

        ``arguments`` carries header-match criteria for HEADERS exchanges.
        """
        ...

    async def bind_exchange(
        self,
        destination: str,
        source: str,
        routing_key: str = "",
        arguments: dict[str, Any] | None = None,
    ) -> None:
        """Bind an exchange to another exchange (exchange-to-exchange binding)."""
        ...

    async def cancel_consumer(self, consumer_tag: str) -> None:
        """Cancel a consumer by its tag."""
        ...

Methods:

connect() -> None async

Establish connection to broker.

Source code in src/rabbitkit/core/protocols.py

async def connect(self) -> None:
    """Establish connection to broker."""
    ...

disconnect() -> None async

Close connection to broker.

Source code in src/rabbitkit/core/protocols.py

async def disconnect(self) -> None:
    """Close connection to broker."""
    ...

is_connected() -> bool

Check if transport is connected.

Source code in src/rabbitkit/core/protocols.py

def is_connected(self) -> bool:
    """Check if transport is connected."""
    ...

publish(envelope: MessageEnvelope) -> PublishOutcome async

Publish a message. Returns outcome with confirm status.

Source code in src/rabbitkit/core/protocols.py

async def publish(self, envelope: MessageEnvelope) -> PublishOutcome:
    """Publish a message. Returns outcome with confirm status."""
    ...

consume(queue: str, callback: Callable[[RabbitMessage], Awaitable[None]], prefetch: int = 10, *, no_ack: bool = False, declare: bool = True) -> str async

Start consuming from a queue. Returns consumer_tag.

no_ack=True starts a no-ack consumer (the broker auto-acks on delivery; the built RabbitMessage gets no settlement functions). declare=False skips declaring/checking the queue first — required for AMQP pseudo-queues such as amq.rabbitmq.reply-to, which the broker rejects any Queue.Declare for (active or passive).

Source code in src/rabbitkit/core/protocols.py

async def consume(
    self,
    queue: str,
    callback: Callable[[RabbitMessage], Awaitable[None]],
    prefetch: int = 10,
    *,
    no_ack: bool = False,
    declare: bool = True,
) -> str:
    """Start consuming from a queue. Returns consumer_tag.

    ``no_ack=True`` starts a no-ack consumer (the broker auto-acks on
    delivery; the built ``RabbitMessage`` gets no settlement functions).
    ``declare=False`` skips declaring/checking the queue first — required
    for AMQP pseudo-queues such as ``amq.rabbitmq.reply-to``, which the
    broker rejects any Queue.Declare for (active or passive).
    """
    ...

declare_exchange(exchange: RabbitExchange) -> None async

Declare an exchange on the broker.

Source code in src/rabbitkit/core/protocols.py

async def declare_exchange(self, exchange: RabbitExchange) -> None:
    """Declare an exchange on the broker."""
    ...

declare_queue(queue: RabbitQueue) -> None async

Declare a queue on the broker.

Source code in src/rabbitkit/core/protocols.py

async def declare_queue(self, queue: RabbitQueue) -> None:
    """Declare a queue on the broker."""
    ...

bind_queue(queue: str, exchange: str, routing_key: str, arguments: dict[str, Any] | None = None) -> None async

Bind a queue to an exchange with a routing key.

arguments carries header-match criteria for HEADERS exchanges.

Source code in src/rabbitkit/core/protocols.py

async def bind_queue(
    self,
    queue: str,
    exchange: str,
    routing_key: str,
    arguments: dict[str, Any] | None = None,
) -> None:
    """Bind a queue to an exchange with a routing key.

    ``arguments`` carries header-match criteria for HEADERS exchanges.
    """
    ...

bind_exchange(destination: str, source: str, routing_key: str = '', arguments: dict[str, Any] | None = None) -> None async

Bind an exchange to another exchange (exchange-to-exchange binding).

Source code in src/rabbitkit/core/protocols.py

async def bind_exchange(
    self,
    destination: str,
    source: str,
    routing_key: str = "",
    arguments: dict[str, Any] | None = None,
) -> None:
    """Bind an exchange to another exchange (exchange-to-exchange binding)."""
    ...

cancel_consumer(consumer_tag: str) -> None async

Cancel a consumer by its tag.

Source code in src/rabbitkit/core/protocols.py

async def cancel_consumer(self, consumer_tag: str) -> None:
    """Cancel a consumer by its tag."""
    ...

SupportsBackpressure

Bases: Protocol

Transport supports connection.blocked/unblocked callbacks.

Source code in src/rabbitkit/core/protocols.py

@runtime_checkable
class SupportsBackpressure(Protocol):
    """Transport supports connection.blocked/unblocked callbacks."""

    def on_blocked(self, callback: Callable[[], None]) -> None:
        """Register callback for connection.blocked."""
        ...

    def on_unblocked(self, callback: Callable[[], None]) -> None:
        """Register callback for connection.unblocked."""
        ...

Methods:

on_blocked(callback: Callable[[], None]) -> None

Register callback for connection.blocked.

Source code in src/rabbitkit/core/protocols.py

def on_blocked(self, callback: Callable[[], None]) -> None:
    """Register callback for connection.blocked."""
    ...

on_unblocked(callback: Callable[[], None]) -> None

Register callback for connection.unblocked.

Source code in src/rabbitkit/core/protocols.py

def on_unblocked(self, callback: Callable[[], None]) -> None:
    """Register callback for connection.unblocked."""
    ...

Types

ExchangeType

Bases: str, Enum

AMQP exchange types.

Source code in src/rabbitkit/core/types.py

class ExchangeType(str, Enum):
    """AMQP exchange types."""

    DIRECT = "direct"
    FANOUT = "fanout"
    TOPIC = "topic"
    HEADERS = "headers"

QueueType

Bases: str, Enum

RabbitMQ queue types.

Source code in src/rabbitkit/core/types.py

class QueueType(str, Enum):
    """RabbitMQ queue types."""

    CLASSIC = "classic"
    QUORUM = "quorum"
    STREAM = "stream"

AppState

Bases: str, Enum

Application lifecycle states.

Canonical home for this enum is core/types.py per the project rule that types.py is the SINGLE canonical location for all enums and data types.

Source code in src/rabbitkit/core/types.py

class AppState(str, Enum):
    """Application lifecycle states.

    Canonical home for this enum is ``core/types.py`` per the project rule that
    ``types.py`` is the SINGLE canonical location for all enums and data types.
    """

    IDLE = "idle"
    STARTING = "starting"
    RUNNING = "running"
    STOPPING = "stopping"
    STOPPED = "stopped"

Logging

LoggingConfig dataclass

Structured logging configuration.

Attributes:

Name	Type	Description
render_json	bool	True for JSON output (prod), False for console (dev).
add_log_level	bool	Include log level in output.
timestamper_fmt	str	Timestamp format ("iso", "unix", or None to disable).
include_caller_info	bool	Add filename/line number to log events.
redact_keys	frozenset[str] | None	Key names to redact from log events -- checked at the top level and one level deep inside nested dict values (e.g. a headers={...} field). Matching is case-insensitive and normalizes AMQP-style x- prefixes/hyphens, so api_key also matches X-Api-Key. Defaults to :data:DEFAULT_REDACT_KEYS. Pass None to disable redaction entirely, or your own frozenset to customize it. See the module docstring ("Secrets and message content") for scope and limitations.
capture_warnings	bool	Route Python's warnings module (used for every rabbitkit safety warning -- topology drift, retry-without- confirms, unsafe TLS, dashboard auth, ...) through the standard logging module via logging.captureWarnings(). Without this, warnings.warn() writes directly to sys.stderr in its own format, completely bypassing whatever log pipeline render_json/handlers were set up for -- a "loud warning" is only actually loud if something is watching raw stderr in dev; in a production JSON-logging deployment it's invisible unless this is enabled. Default True. Set False if your application already manages captureWarnings itself.

Source code in src/rabbitkit/core/logging.py

@dataclass(frozen=True, slots=True)
class LoggingConfig:
    """Structured logging configuration.

    Attributes:
        render_json: True for JSON output (prod), False for console (dev).
        add_log_level: Include log level in output.
        timestamper_fmt: Timestamp format ("iso", "unix", or None to disable).
        include_caller_info: Add filename/line number to log events.
        redact_keys: Key names to redact from log events -- checked at the
            top level and one level deep inside nested dict values (e.g. a
            ``headers={...}`` field). Matching is case-insensitive and
            normalizes AMQP-style ``x-`` prefixes/hyphens, so ``api_key``
            also matches ``X-Api-Key``. Defaults to
            :data:`DEFAULT_REDACT_KEYS`. Pass ``None`` to disable redaction
            entirely, or your own ``frozenset`` to customize it. See the
            module docstring ("Secrets and message content") for scope and
            limitations.
        capture_warnings: Route Python's ``warnings`` module (used for every
            rabbitkit safety warning -- topology drift, retry-without-
            confirms, unsafe TLS, dashboard auth, ...) through the standard
            ``logging`` module via ``logging.captureWarnings()``. Without
            this, ``warnings.warn()`` writes directly to ``sys.stderr`` in
            its own format, completely bypassing whatever log pipeline
            ``render_json``/handlers were set up for -- a "loud warning" is
            only actually loud if something is watching raw stderr in dev;
            in a production JSON-logging deployment it's invisible unless
            this is enabled. Default ``True``. Set ``False`` if your
            application already manages ``captureWarnings`` itself.
    """

    render_json: bool = False
    add_log_level: bool = True
    timestamper_fmt: str = "iso"
    include_caller_info: bool = False
    redact_keys: frozenset[str] | None = DEFAULT_REDACT_KEYS
    capture_warnings: bool = True

configure_structlog(config: LoggingConfig | None = None) -> None

One-time structlog configuration.

Safe to call multiple times — last call wins. If config is None, uses defaults (console renderer, ISO timestamps).

Source code in src/rabbitkit/core/logging.py

def configure_structlog(config: LoggingConfig | None = None) -> None:
    """One-time structlog configuration.

    Safe to call multiple times — last call wins.
    If config is None, uses defaults (console renderer, ISO timestamps).
    """
    import logging

    import structlog

    if config is None:
        config = LoggingConfig()

    # L16 follow-up: bridge warnings.warn() (every rabbitkit safety warning)
    # into the standard logging module -- otherwise it bypasses this whole
    # pipeline entirely, writing straight to sys.stderr in its own format.
    logging.captureWarnings(config.capture_warnings)

    processors: list[structlog.types.Processor] = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
    ]

    if config.redact_keys:
        processors.append(_redact_processor(config.redact_keys))

    if config.add_log_level:
        processors.append(structlog.stdlib.add_log_level)

    if config.timestamper_fmt:
        fmt = config.timestamper_fmt if config.timestamper_fmt != "iso" else "iso"
        processors.append(structlog.processors.TimeStamper(fmt=fmt))

    if config.include_caller_info:
        processors.append(structlog.processors.CallsiteParameterAdder())

    processors.append(structlog.stdlib.PositionalArgumentsFormatter())
    processors.append(structlog.processors.StackInfoRenderer())
    processors.append(structlog.processors.UnicodeDecoder())

    if config.render_json:
        processors.append(structlog.processors.JSONRenderer())
    else:
        processors.append(structlog.dev.ConsoleRenderer())

    structlog.configure(
        processors=processors,
        wrapper_class=structlog.stdlib.BoundLogger,
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

Path extraction

extract_path(actual_routing_key: str, pattern: str) -> dict[str, str]

Extract named segments from a delivered routing key given the route pattern.

Positional: each {name} in the pattern maps to the same-index segment of the actual key. Stops at a # (which spans a variable number of words, so nothing after it has a fixed position). Returns {} when the pattern has no named segments — the broker already matched the binding, so no validation is needed here.

Source code in src/rabbitkit/core/path.py

def extract_path(actual_routing_key: str, pattern: str) -> dict[str, str]:
    """Extract named segments from a delivered routing key given the route pattern.

    Positional: each ``{name}`` in the pattern maps to the same-index segment of
    the actual key. Stops at a ``#`` (which spans a variable number of words, so
    nothing after it has a fixed position). Returns ``{}`` when the pattern has no
    named segments — the broker already matched the binding, so no validation is
    needed here.
    """
    if "{" not in pattern:
        return {}
    actual = actual_routing_key.split(".")
    out: dict[str, str] = {}
    for i, seg in enumerate(pattern.split(".")):
        if seg == "#":
            break
        if i >= len(actual):
            break
        if _is_named(seg):
            out[seg[1:-1]] = actual[i]
    return out

to_binding_key(routing_key: str) -> str

Translate {name} segments to the AMQP single-word wildcard *.

Routing keys without named segments are returned unchanged (fast path), so existing topic/direct routes are completely unaffected.

Source code in src/rabbitkit/core/path.py

def to_binding_key(routing_key: str) -> str:
    """Translate ``{name}`` segments to the AMQP single-word wildcard ``*``.

    Routing keys without named segments are returned unchanged (fast path), so
    existing topic/direct routes are completely unaffected.
    """
    if "{" not in routing_key:
        return routing_key
    return ".".join("*" if _is_named(s) else s for s in routing_key.split("."))