Replace Khepri topic routing projection with trie + ordered_set (v4)

Resolves #15588.

The previous Khepri topic routing projection (v3) stored topic bindings
as sets:set(#binding{}) inside trie leaf nodes. This design had a
major performance drawback:

On the insertion/deletion path (in the single Khepri Ra process),
every binding change required a read-modify-write of the entire
sets:set(), making it O(N) in the number of bindings at that leaf.
With many MQTT clients connecting concurrently (each subscribing to
the same topic filter), this made the Ra process a bottleneck.

Another less severe performance issue was that the entire binding
was being copied including the binding arguments containing the MQTT 5.0
subscription options such as:
```
{<<"x-mqtt-subscription-opts">>,table,
 [{<<"id">>,unsignedint,1},
  {<<"no-local">>,bool,false},
  {<<"qos">>,unsignedbyte,0},
  {<<"retain-as-published">>,bool,false},
  {<<"retain-handling">>,unsignedbyte,0}]}]
```

Replace the single ETS projection table with two purpose-built tables:

1. Trie edges table (ETS set, read_concurrency=true):
   - Row: `{{XSrc, ParentNodeId, Word}, ChildNodeId, ChildCount}`
   - XSrc = {VHost, ExchangeName} (compact 2-tuple of binaries)
   - NodeId = root | reference()
   - ChildCount tracks outgoing edges for garbage collection

2. Leaf bindings table (ETS ordered_set, read_concurrency=true):
   - Key: {NodeId, BindingKey, Dest}
   - Stored as 1-tuples: {{NodeId, BindingKey, Dest}}
   - No value column; all data is in the key to minimize copying

The trie structure preserves O(depth * 3) routing complexity regardless
of the number of overall bindings or wildcard filters. At each trie level, we
probe at most 3 edges (literal word, <<"*">>, <<"#">>), each via
ets:lookup_element/4 which copies only the ChildNodeId (a reference).

The ordered_set for bindings provides:
- O(log N) insert and delete per binding (no read-modify-write)
- The binding key (needed for MQTT subscription identifiers and topic
  aliases) is part of the key, so it is returned directly during
  destination collection without additional lookups

Collecting destinations at a matched trie leaf uses a hybrid strategy:
- Fanout 0-2 (the common case: unicast, device + stream): up to 3
  ets:next/2 probes. Each ets:next/2 call costs O(log N) because the
  CATree (used with read_concurrency) allocates a fresh tree traversal
  stack on each call.
- Fanout > 2: ets:select/2 with a partially bound key does an O(log N)
  seek followed by an O(F) range scan. The match spec compilation
  overhead amortises over the larger result set.

ets:lookup_element/4 (OTP 26+) returns a default value on miss
instead of throwing badarg, and copies only the requested element
on hit. This avoids both exception overhead (misses are common during
trie traversal of <<"*">> and <<"#">> branches) and unnecessary data
copying (we only need the ChildNodeId, not the full row).

Trie node IDs are ephemeral (the tables are rebuilt when the Khepri
projection is re-registered). make_ref() is fast, globally unique
within a node, and has good hash distribution for the ETS set table.

When a binding is deleted, the trie path from root to leaf is collected
in a single downward walk (trie_follow_down_get_path). Empty nodes are
then pruned bottom-up: a node is empty when its ChildCount is 0 and it
has no bindings in the ordered_set table.

Benchmarks below were run with 500K routing operations per scenario
(on the same machine, back-to-back between main (v3) and this commit.

Significant insert/delete improvements:

  Churn insert (8K bindings, 4 filters/client): ~1,120 vs ~810 ops/s (+38%)
    v3 did a read-modify-write of sets:set() per binding; v4 does
    a single ets:insert into the ordered_set plus trie edge updates.

  MQTT device insert (20K bindings): ~650 vs ~420 ops/s (+55%)
    Same mechanism as churn insert. Particularly impactful when many
    clients share the same wildcard filter (e.g. "broadcast.#"),
    since v3's sets:set() grew with each client while v4 inserts
    are O(log N) regardless.

  Same-key fanout insert (10K): ~415 vs ~290 ops/s (+43%)
    The worst case for v3: all 10K bindings share the same key,
    so each insert copies and rebuilds the growing sets:set().

Routing improvements:

  MQTT unicast (10K devices, 20K bindings): ~460K vs ~250K ops/s (+80%)
    Each route matches 1 queue among 10K unique exact keys plus
    10K queues sharing "broadcast.#". v3 stored bindings in the same
    ETS row as the trie edge, so every trie lookup copied the entire
    sets:set(). v4 separates trie edges (small rows, set table) from
    bindings (ordered_set), so the trie walk copies only references.

  Large fanout (10K queues, same key): ~3,100 vs ~1,170 ops/s (+165%)
    v3 copied a 10K-element sets:set() out of ETS in a single
    ets:lookup, then called sets:to_list/1. v4 uses ets:select/2
    with a partially bound key, which does an O(log N) seek and
    then an efficient O(F) range scan without intermediate set
    conversion.

  MQTT broadcast (10K fanout): ~0.6 vs ~0.9 ms/route (+50%)
    Same mechanism as above.

Scenarios with no significant change (within benchmark noise):

  Exact match, wildcard *, wildcard #, mixed wildcards, and many
  wildcard filters showed no clear difference. Both v3 and v4 use
  a trie walk, so routing speed is comparable when the fanout is
  small and the bottleneck is trie traversal rather than destination
  collection.

Author: ansd

deps/rabbit/include/rabbit_khepri.hrl

@@ -56,9 +56,9 @@
         ?RABBITMQ_KHEPRI_EXCHANGE_PATH(
           VHost, Exchange, [user_permissions, Username])).
 
--define(RABBITMQ_KHEPRI_ROUTE_PATH(VHost, SrcName, Kind, DstName, RoutingKey),
+-define(RABBITMQ_KHEPRI_ROUTE_PATH(VHost, SrcName, Kind, DstName, BindingKey),
         ?RABBITMQ_KHEPRI_EXCHANGE_PATH(
-          VHost, SrcName, [bindings, Kind, DstName, RoutingKey])).
+          VHost, SrcName, [bindings, Kind, DstName, BindingKey])).
 
 -define(RABBITMQ_KHEPRI_QUEUE_PATH(VHost, Name),
         ?RABBITMQ_KHEPRI_VHOST_PATH(VHost, [queues, Name])).

deps/rabbit/src/rabbit_db_binding.erl

@@ -635,11 +635,11 @@ khepri_route_path(
            key = RoutingKey}) ->
     khepri_route_path(VHost, SrcName, Kind, DstName, RoutingKey).
 
-khepri_route_path(VHost, SrcName, Kind, DstName, RoutingKey)
+khepri_route_path(VHost, SrcName, Kind, DstName, BindingKey)
   when ?IS_KHEPRI_PATH_CONDITION(Kind) andalso
        ?IS_KHEPRI_PATH_CONDITION(DstName) andalso
-       ?IS_KHEPRI_PATH_CONDITION(RoutingKey) ->
-    ?RABBITMQ_KHEPRI_ROUTE_PATH(VHost, SrcName, Kind, DstName, RoutingKey).
+       ?IS_KHEPRI_PATH_CONDITION(BindingKey) ->
+    ?RABBITMQ_KHEPRI_ROUTE_PATH(VHost, SrcName, Kind, DstName, BindingKey).
 
 khepri_route_path_to_args(Path) ->
     Pattern = khepri_route_path(

deps/rabbit/src/rabbit_db_topic_exchange.erl

@@ -9,122 +9,36 @@
 
 -include_lib("rabbit_common/include/rabbit.hrl").
 
--export([set/1, delete_all_for_exchange/1, delete/1, match/3]).
+-export([match/3]).
 
 %% Used by Khepri projections and the Mnesia-to-Khepri migration.
--export([
-         split_topic_key/1,
-         split_topic_key_binary/1
-        ]).
+-export([split_topic_key_binary/1,
+         split_topic_key/1]).
 
-%% For testing
--export([clear/0]).
-
--define(KHEPRI_PROJECTION, rabbit_khepri_topic_trie_v3).
+-define(TOPIC_TRIE_PROJECTION, rabbit_khepri_topic_trie_v4).
+-define(TOPIC_BINDING_PROJECTION, rabbit_khepri_topic_binding_v4).
 
 -type match_result() :: [rabbit_types:binding_destination() |
                          {rabbit_amqqueue:name(), rabbit_types:binding_key()}].
 
 -define(COMPILED_TOPIC_SPLIT_PATTERN, cp_dot).
 
-%% -------------------------------------------------------------------
-%% set().
-%% -------------------------------------------------------------------
-
--spec set(Binding) -> ok when
-      Binding :: rabbit_types:binding().
-%% @doc Sets a topic binding.
-%%
-%% @private
-
-set(#binding{}) ->
-    ok.
-
-%% -------------------------------------------------------------------
-%% delete_all_for_exchange().
-%% -------------------------------------------------------------------
-
--spec delete_all_for_exchange(ExchangeName) -> ok when
-      ExchangeName :: rabbit_exchange:name().
-%% @doc Deletes all topic bindings for the exchange named `ExchangeName'
-%%
-%% @private
-
-delete_all_for_exchange(_XName) ->
-    ok.
-
-%% -------------------------------------------------------------------
-%% delete().
-%% -------------------------------------------------------------------
-
--spec delete([Binding]) -> ok when
-      Binding :: rabbit_types:binding().
-%% @doc Deletes all given topic bindings
-%%
-%% @private
-
-delete(Bs) when is_list(Bs) ->
-    ok.
-
-%% -------------------------------------------------------------------
-%% match().
-%% -------------------------------------------------------------------
-
 -spec match(rabbit_exchange:name(),
             rabbit_types:routing_key(),
             rabbit_exchange:route_opts()) -> match_result().
-%% @doc Finds the topic bindings matching the given exchange and routing key and returns
-%% the destination of the bindings potentially with the matched binding key.
-%%
-%% @returns destinations with matched binding key
-%%
-%% @private
-
-match(XName, RoutingKey, Opts) ->
+match(#resource{virtual_host = VHost, name = XName}, RoutingKey, Opts) ->
     BKeys = maps:get(return_binding_keys, Opts, false),
     Words = split_topic_key_binary(RoutingKey),
-    trie_match_in_khepri(XName, Words, BKeys).
-
-%% -------------------------------------------------------------------
-%% clear().
-%% -------------------------------------------------------------------
-
--spec clear() -> ok.
-%% @doc Deletes all topic bindings
-%%
-%% @private
-
-clear() ->
-    ok.
-
-%% --------------------------------------------------------------
-%% split_topic_key().
-%% --------------------------------------------------------------
-
--spec split_topic_key(RoutingKey) -> Words when
-      RoutingKey :: binary(),
-      Words :: [[byte()]].
-
-split_topic_key(Key) ->
-    split_topic_key(Key, [], []).
-
-split_topic_key(<<>>, [], []) ->
-    [];
-split_topic_key(<<>>, RevWordAcc, RevResAcc) ->
-    lists:reverse([lists:reverse(RevWordAcc) | RevResAcc]);
-split_topic_key(<<$., Rest/binary>>, RevWordAcc, RevResAcc) ->
-    split_topic_key(Rest, [], [lists:reverse(RevWordAcc) | RevResAcc]);
-split_topic_key(<<C:8, Rest/binary>>, RevWordAcc, RevResAcc) ->
-    split_topic_key(Rest, [C | RevWordAcc], RevResAcc).
-
-%% --------------------------------------------------------------
-%% split_topic_key_binary().
-%% --------------------------------------------------------------
+    try
+        trie_match({VHost, XName}, root, Words, BKeys, [])
+    catch
+        error:badarg ->
+            []
+    end.
 
 -spec split_topic_key_binary(RoutingKey) -> Words when
       RoutingKey :: binary(),
       Words :: [binary()].
-
 split_topic_key_binary(<<>>) ->
     [];
 split_topic_key_binary(RoutingKey) ->
@@ -139,92 +53,120 @@ split_topic_key_binary(RoutingKey) ->
     end,
     binary:split(RoutingKey, Pattern, [global]).
 
-%% --------------------------------------------------------------
-%% Internal
-%% --------------------------------------------------------------
-
--spec add_matched([rabbit_types:binding_destination() |
-                   {rabbit_types:binding_destination(), BindingArgs :: list()}],
-                  ReturnBindingKeys :: boolean(),
-                  match_result()) ->
-    match_result().
-add_matched(Destinations, false, Acc) ->
-    Destinations ++ Acc;
-add_matched(DestinationsArgs, true, Acc) ->
-    lists:foldl(
-      fun({DestQ = #resource{kind = queue}, BindingArgs}, L) ->
-              case rabbit_misc:table_lookup(BindingArgs, <<"x-binding-key">>) of
-                  {longstr, BKey} ->
-                      [{DestQ, BKey} | L];
-                  _ ->
-                      [DestQ | L]
-              end;
-         ({DestX, _BindingArgs}, L) ->
-              [DestX | L]
-      end, Acc, DestinationsArgs).
-
-%% Khepri topic graph
-
-trie_match_in_khepri(X, Words, BKeys) ->
-    try
-        trie_match_in_khepri(X, root, Words, BKeys, [])
-    catch
-        error:badarg ->
-            []
-    end.
+-spec split_topic_key(RoutingKey) -> Words when
+      RoutingKey :: binary(),
+      Words :: [[byte()]].
+split_topic_key(Key) ->
+    split_topic_key(Key, [], []).
 
-trie_match_in_khepri(X, Node, [], BKeys, ResAcc0) ->
-    Destinations = trie_bindings_in_khepri(X, Node, BKeys),
-    ResAcc = add_matched(Destinations, BKeys, ResAcc0),
-    trie_match_part_in_khepri(
-      X, Node, <<"#">>,
-      fun trie_match_skip_any_in_khepri/5, [], BKeys, ResAcc);
-trie_match_in_khepri(X, Node, [W | RestW] = Words, BKeys, ResAcc) ->
-    lists:foldl(fun ({WArg, MatchFun, RestWArg}, Acc) ->
-                        trie_match_part_in_khepri(
-                          X, Node, WArg, MatchFun, RestWArg, BKeys, Acc)
-                end, ResAcc, [{W, fun trie_match_in_khepri/5, RestW},
-                              {<<"*">>, fun trie_match_in_khepri/5, RestW},
-                              {<<"#">>,
-                               fun trie_match_skip_any_in_khepri/5, Words}]).
-
-trie_match_part_in_khepri(X, Node, Search, MatchFun, RestW, BKeys, ResAcc) ->
-    case trie_child_in_khepri(X, Node, Search) of
-        {ok, NextNode} -> MatchFun(X, NextNode, RestW, BKeys, ResAcc);
-        error          -> ResAcc
-    end.
+split_topic_key(<<>>, [], []) ->
+    [];
+split_topic_key(<<>>, RevWordAcc, RevResAcc) ->
+    lists:reverse([lists:reverse(RevWordAcc) | RevResAcc]);
+split_topic_key(<<$., Rest/binary>>, RevWordAcc, RevResAcc) ->
+    split_topic_key(Rest, [], [lists:reverse(RevWordAcc) | RevResAcc]);
+split_topic_key(<<C:8, Rest/binary>>, RevWordAcc, RevResAcc) ->
+    split_topic_key(Rest, [C | RevWordAcc], RevResAcc).
 
-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(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
+%% ==============================================================
+%% Trie-based routing
+%%
+%% Uses two ETS tables:
+%%
+%% 1. Trie edges table (set): {Key, ChildNodeId, ChildCount}
+%%    Key = {XSrc, ParentNodeId, Word}
+%%    Navigation: ets:lookup_element/4 for O(1) edge traversal.
+%%
+%% 2. Leaf bindings table (ordered_set): {{NodeId, BindingKey, Dest}}
+%%    Collection: ets:next/2 probes for fanout 0-2 (fast path), then
+%%    ets:select/2 with a partially bound key for fanout > 2 (does an
+%%    O(log N) seek followed by O(F) range scan).
+%%
+%% Routing walks the trie (branching on literal word, <<"*">>, <<"#">>)
+%% then collects destinations from the bindings table at each matching
+%% leaf. This is O(depth * 3) for the trie walk, plus O(log N) per
+%% leaf for fanout 0-2, or O(log N + F) per leaf for fanout F > 2.
+%% ==============================================================
+
+trie_match(XSrc, Node, [], BKeys, Acc0) ->
+    Acc1 = trie_bindings(Node, BKeys, Acc0),
+    trie_match_try(XSrc, Node, <<"#">>,
+                   fun trie_match_skip_any/5,
+                   [], BKeys, Acc1);
+trie_match(XSrc, Node, [W | RestW] = Words, BKeys, Acc0) ->
+    Acc1 = trie_match_try(XSrc, Node, W,
+                          fun trie_match/5,
+                          RestW, BKeys, Acc0),
+    Acc2 = trie_match_try(XSrc, Node, <<"*">>,
+                          fun trie_match/5,
+                          RestW, BKeys, Acc1),
+    trie_match_try(XSrc, Node, <<"#">>,
+                   fun trie_match_skip_any/5,
+                   Words, BKeys, Acc2).
+
+trie_match_try(XSrc, Node, Word, MatchFun, RestW, BKeys, Acc) ->
+    case ets:lookup_element(?TOPIC_TRIE_PROJECTION,
+                            {XSrc, Node, Word}, 2, undefined) of
+        undefined ->
+            Acc;
+        NextNode ->
+            MatchFun(XSrc, NextNode, RestW, BKeys, Acc)
     end.
 
-trie_bindings_in_khepri(X, Node, BKeys) ->
-    case ets:lookup(
-           ?KHEPRI_PROJECTION,
-           #trie_edge{exchange_name = X,
-                      node_id       = Node,
-                      word          = bindings}) of
-        [#topic_trie_edge_v2{node_id = {bindings, Bindings}}] ->
-            [case BKeys of
-                 true ->
-                     {Dest, Args};
-                 false ->
-                     Dest
-             end || #binding{destination = Dest,
-                             args        = Args} <- sets:to_list(Bindings)];
-        [] ->
-            []
+trie_match_skip_any(XSrc, Node, [], BKeys, Acc) ->
+    trie_match(XSrc, Node, [], BKeys, Acc);
+trie_match_skip_any(XSrc, Node, [_ | RestW] = Words, BKeys, Acc) ->
+    trie_match_skip_any(
+      XSrc, Node, RestW, BKeys,
+      trie_match(XSrc, Node, Words, BKeys, Acc)).
+
+%% Collect all destinations bound at the given trie node.
+%%
+%% Uses ets:next/2 for up to two elements (fast path for the common
+%% fanout 0-2 cases), then switches to ets:select/2 when fanout > 2.
+%%
+%% ets:select/2 occurs the expensive match spec compilation overhead.
+%% For larger fanouts, the cost for compiling the match spec amortises.
+%% ets:select/2 occurs an O(log N) seek followed by an O(F) range scan,
+%% which is cheaper than F individual ets:next/2 calls
+%% (each O(log N) due to CATree fresh-stack allocation).
+trie_bindings(NodeId, BKeys, Acc) ->
+    StartKey = {NodeId, <<>>, {}},
+    case ets:next(?TOPIC_BINDING_PROJECTION, StartKey) of
+        {NodeId, BKey1, Dest1} = Key1 ->
+            case ets:next(?TOPIC_BINDING_PROJECTION, Key1) of
+                {NodeId, BKey2, Dest2} = Key2 ->
+                    case ets:next(?TOPIC_BINDING_PROJECTION, Key2) of
+                        {NodeId, _, _} ->
+                            collect_select(NodeId, BKeys, Acc);
+                        _ ->
+                            Acc1 = collect_binding(Dest1, BKey1, BKeys, Acc),
+                            collect_binding(Dest2, BKey2, BKeys, Acc1)
+                    end;
+                _ ->
+                    collect_binding(Dest1, BKey1, BKeys, Acc)
+            end;
+        _ ->
+            Acc
     end.
+
+collect_binding(#resource{kind = queue} = Dest, BindingKey, true, Acc) ->
+    [{Dest, BindingKey} | Acc];
+collect_binding(Dest, _BindingKey, _ReturnBindingKeys, Acc) ->
+    [Dest | Acc].
+
+collect_select(NodeId, false, Acc) ->
+    Dests = ets:select(?TOPIC_BINDING_PROJECTION,
+                       [{{{NodeId, '_', '$1'}}, [], ['$1']}]),
+    Dests ++ Acc;
+collect_select(NodeId, true, Acc) ->
+    DestsAndBKeys = ets:select(?TOPIC_BINDING_PROJECTION,
+                               [{{{NodeId, '$1', '$2'}}, [], [{{'$2', '$1'}}]}]),
+    format_dest_bkeys(DestsAndBKeys, Acc).
+
+format_dest_bkeys([], Acc) ->
+    Acc;
+format_dest_bkeys([{#resource{kind = queue} = Dest, BKey} | Rest], Acc) ->
+    format_dest_bkeys(Rest, [{Dest, BKey} | Acc]);
+format_dest_bkeys([{Dest, _BKey} | Rest], Acc) ->
+    format_dest_bkeys(Rest, [Dest | Acc]).