Routing Substrate — Exchange Types and the Topic Trie

Routing Substrate — Exchange Types and the Topic Trie #

The routing substrate is the only one with no dedicated process. Routing is a pure function: given an exchange record and a message, return a list of destination queue names. It executes entirely in the calling rabbit_channel process, reading from ETS tables that are kept current by Khepri projections.

The rabbit_exchange_type Behaviour #

Every exchange type implements a common behaviour:

-callback route(rabbit_types:exchange(), mc:state()) ->
    rabbit_router:match_result().

-callback route(rabbit_types:exchange(), mc:state(), rabbit_exchange:route_opts()) ->
    rabbit_router:match_result().

-callback add_binding(Tx, rabbit_types:exchange(), rabbit_types:binding()) -> ok.
-callback remove_bindings(Tx, rabbit_types:exchange(), [rabbit_types:binding()]) -> ok.
-callback validate_binding(rabbit_types:exchange(), rabbit_types:binding()) ->
    rabbit_types:ok_or_error(rabbit_types:amqp_error()).

The route/2 and route/3 callbacks are the hot path. Everything else (add_binding, remove_bindings, validate_binding, create, delete) is called only on topology changes.

Direct Exchange: Hash Lookup #

rabbit_exchange_type_direct routing is an ETS lookup by exact routing key:

route(#exchange{name = XName}, Msg, _Opts) ->
    RKeys = mc:routing_keys(Msg),
    lists:append([rabbit_router:match_routing_key(XName, [RKey]) || RKey <- RKeys]).

rabbit_router:match_routing_key/2 does an ETS lookup in the rabbit_route table (keyed by {ExchangeName, RoutingKey}). This is O(1) per routing key.

Fanout Exchange: Scan All Bindings #

rabbit_exchange_type_fanout ignores the routing key entirely:

route(#exchange{name = XName}, _Msg, _Opts) ->
    rabbit_router:match_routing_key(XName, ['_']).

'_' is an ETS wildcard — it matches all bindings for this exchange name, returning all destination queues. The cost is O(bindings) rather than O(1), but fanout exchanges typically have few enough bindings that this is not a bottleneck.

Headers Exchange: Table Attribute Matching #

rabbit_exchange_type_headers route iterates over all bindings for the exchange and evaluates each binding’s arguments table against the message’s headers property:

route(#exchange{name = XName}, Msg, _Opts) ->
    Headers = mc:headers(Msg),
    rabbit_router:match_bindings(XName, fun(#binding{arguments = Args}) ->
        headers_match(Args, Headers)
    end).

headers_match(Args, Headers) ->
    case rabbit_misc:table_lookup(Args, <<"x-match">>) of
        {longstr, <<"all">>} -> all_match(Args, Headers);
        {longstr, <<"any">>} -> any_match(Args, Headers);
        _                    -> all_match(Args, Headers)  %% default: all
    end.

This is O(bindings × attributes_per_binding) — the slowest exchange type. Headers exchanges are appropriate for low-traffic routing with complex multi-attribute filtering but should not be used on high-throughput publish paths.

Topic Exchange: The Trie #

rabbit_exchange_type_topic delegates entirely to rabbit_db_topic_exchange:match/3:

route(#exchange{name = XName}, Msg, Opts) ->
    RKeys = mc:routing_keys(Msg),
    lists:append([rabbit_db_topic_exchange:match(XName, RKey, Opts) || RKey <- RKeys]).

rabbit_db_topic_exchange:match/3 splits the routing key into words using a compiled binary pattern (cached in persistent_term for zero-cost retrieval):

match(XName, RoutingKey, Opts) ->
    BKeys = maps:get(return_binding_keys, Opts, false),
    Words = split_topic_key_binary(RoutingKey),
    trie_match_in_khepri(XName, Words, BKeys).

split_topic_key_binary(RoutingKey) ->
    Pattern = persistent_term:get(?COMPILED_TOPIC_SPLIT_PATTERN, undefined),
    binary:split(RoutingKey, Pattern, [global]).

The trie walk is a depth-first search over a Khepri projection stored in ETS as rabbit_khepri_topic_trie_v3:

trie_match_in_khepri(X, Node, [W | RestW] = Words, BKeys, ResAcc) ->
    lists:foldl(
        fun({Search, MatchFun, Rest}, Acc) ->
            trie_match_part_in_khepri(X, Node, Search, MatchFun, Rest, BKeys, Acc)
        end,
        ResAcc,
        %% Try three paths: exact word, '*' (one word), '#' (zero or more)
        [{W,    fun trie_match_in_khepri/5,           RestW},
         {<<"*">>, fun trie_match_in_khepri/5,        RestW},
         {<<"#">>, fun trie_match_skip_any_in_khepri/5, Words}]
    ).

For each word position, the algorithm tries three trie edges:

1. The literal word W — exact match at this position.
2. * — matches any single word.
3. # — matches zero or more words (handled by trie_match_skip_any_in_khepri which recurses with the same word list and the next).

trie_match_skip_any_in_khepri(X, Node, [], BKeys, ResAcc) ->
    trie_match_in_khepri(X, Node, [], BKeys, ResAcc);
trie_match_skip_any_in_khepri(X, Node, [_ | RestW] = Words, BKeys, ResAcc) ->
    trie_match_skip_any_in_khepri(
        X, Node, RestW, BKeys,
        trie_match_in_khepri(X, Node, Words, BKeys, ResAcc)).

trie_child_in_khepri does the actual ETS lookup:

trie_child_in_khepri(X, Node, Word) ->
    case ets:lookup(
        ?KHEPRI_PROJECTION,
        #trie_edge{exchange_name = X, node_id = Node, word = Word}) of
        [#topic_trie_edge_v2{node_id = NextNode}] -> {ok, NextNode};
        []                                         -> error
    end.

One ETS lookup per trie edge traversal. The ETS table is an ordered_set indexed by {ExchangeName, NodeId, Word} — range queries are cheap. The projection is updated whenever a binding is added or removed via Khepri change notifications.

The Khepri Projection: Why ETS for Routing #

Routing must be fast enough to not be on the critical path of basic.publish. If routing required a Khepri read (a Raft-coordinated operation), every publish would pay a distributed consensus cost. Instead, Khepri maintains an ETS projection of the topic trie:

-define(KHEPRI_PROJECTION, rabbit_khepri_topic_trie_v3).

A Khepri projection is a function that Khepri calls whenever its tree changes under a watched path. The projection function updates an ETS table to reflect the new state. Reads from the ETS table are local, in-memory, and require no locking — they read directly from the table without coordination.

binding added to Khepri (Raft write, replicated)
    → Khepri notifies projection functions on all nodes
    → projection function updates ETS trie on this node
    → next routing call sees the new binding (ETS lookup)

There is a brief window between a binding being committed to Khepri and the ETS projection being updated — during this window, a publish could miss the new binding. This is intentional: routing consistency is eventual (ETS), while binding durability is strong (Khepri/Raft). The window is typically microseconds.

Exchange and Binding Metadata in Khepri #

Exchange declarations and binding configurations are stored in Khepri via rabbit_db_exchange and rabbit_db_binding:

%% rabbit_db_exchange.erl: storing an exchange definition
rabbit_khepri:handle_fallback(#{
    mnesia => fun() -> mnesia_store_exchange(X) end,
    khepri => fun() -> khepri_store_exchange(X) end
})

The handle_fallback pattern appears throughout rabbit_db_* modules — it dispatches to either Mnesia (legacy) or Khepri based on the khepri_db feature flag. This is the migration abstraction layer.

Exchange and binding records stored in Khepri are the authoritative source. The ETS routing tables are derived projections. If ETS is lost (node restart), Khepri re-populates all projections from its authoritative state.

rabbit_exchange:route/2 as the Entry Point #

The full routing call from rabbit_channel:

%% In rabbit_channel.erl
Exchange = rabbit_exchange:lookup_or_die(ExchangeName),
Queues = rabbit_exchange:route(Exchange, Message, #{}),

rabbit_exchange:route/2 calls the exchange type module’s route/2, collects results, deduplicates (topic exchange can return duplicates when multiple patterns match the same queue), and returns the final list.

The entire path — from rabbit_exchange:route/2 call to returned queue list — is pure function execution in the calling process. No message-passing, no process hops, no locks. Just ETS reads and pattern matching.

Summary #

The routing substrate is a pure function over ETS. rabbit_exchange_type is a behaviour with four implementations: direct (hash lookup O(1)), fanout (scan all bindings), headers (attribute matching per binding), topic (trie DFS over Khepri ETS projection). The topic trie uses persistent_term for compiled split patterns, ETS for trie edges (keyed {Exchange, NodeId, Word}), and Khepri projections to keep the trie current without Raft round-trips on every publish. Routing has no process boundary — it executes in rabbit_channel directly.