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

* 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.

* Improve function names

Improve function names and add function specs.

Author: ansd

deps/rabbit/src/rabbit_core_ff.erl

@@ -244,3 +244,14 @@
                          {rabbit_db_queue,
                           tie_binding_to_dest_with_keep_while_cond_enable}}
      }}).
+
+-rabbit_feature_flag(
+   {topic_binding_projection_v4,
+    #{desc          => "Enable the topic binding Khepri projection v4",
+      stability     => stable,
+      depends_on    => ['rabbitmq_4.3.0'],
+      callbacks     => #{enable =>
+                         {rabbit_khepri, topic_binding_projection_enable},
+                         post_enable =>
+                         {rabbit_khepri, topic_binding_projection_post_enable}}
+     }}).

deps/rabbit/src/rabbit_db_topic_exchange.erl

@@ -14,7 +14,10 @@
 %% Used by Khepri projections and the Mnesia-to-Khepri migration.
 -export([split_topic_key_binary/1]).
 
--define(KHEPRI_PROJECTION, rabbit_khepri_topic_trie_v3).
+-define(KHEPRI_PROJECTION_V3, 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()}].
@@ -31,14 +34,20 @@
 %%
 %% @private
 
-match(XName, RoutingKey, Opts) ->
+match(#resource{virtual_host = VHost, name = XName} = X, 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().
-%% --------------------------------------------------------------
+    case rabbit_khepri:get_effective_topic_binding_projection_version() of
+        V when V >= 4 ->
+            try
+                trie_match({VHost, XName}, root, Words, BKeys, [])
+            catch
+                error:badarg ->
+                    []
+            end;
+        _ ->
+            trie_match_v3(X, Words, BKeys)
+    end.
 
 -spec split_topic_key_binary(RoutingKey) -> Words when
       RoutingKey :: binary(),
@@ -58,81 +67,164 @@ split_topic_key_binary(RoutingKey) ->
     end,
     binary:split(RoutingKey, Pattern, [global]).
 
-%% --------------------------------------------------------------
-%% Internal
-%% --------------------------------------------------------------
+%% ==============================================================
+%% 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.
+%% ==============================================================
 
--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).
+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.
 
-%% Khepri topic graph
+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)).
 
-trie_match_in_khepri(X, Words, BKeys) ->
+%% 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]).
+
+%% ==============================================================
+%% Old v3 Khepri topic graph.
+%% Delete these *_v3 functions when feature flag
+%% topic_binding_projection_v4 becomes required.
+%% ==============================================================
+
+trie_match_v3(X, Words, BKeys) ->
     try
-        trie_match_in_khepri(X, root, Words, BKeys, [])
+        trie_match_v3(X, root, Words, BKeys, [])
     catch
         error:badarg ->
             []
     end.
 
-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(
+trie_match_v3(X, Node, [], BKeys, ResAcc0) ->
+    Destinations = trie_bindings_v3(X, Node, BKeys),
+    ResAcc = add_matched_v3(Destinations, BKeys, ResAcc0),
+    trie_match_part_v3(
       X, Node, <<"#">>,
-      fun trie_match_skip_any_in_khepri/5, [], BKeys, ResAcc);
-trie_match_in_khepri(X, Node, [W | RestW] = Words, BKeys, ResAcc) ->
+      fun trie_match_skip_any_v3/5, [], BKeys, ResAcc);
+trie_match_v3(X, Node, [W | RestW] = Words, BKeys, ResAcc) ->
     lists:foldl(fun ({WArg, MatchFun, RestWArg}, Acc) ->
-                        trie_match_part_in_khepri(
+                        trie_match_part_v3(
                           X, Node, WArg, MatchFun, RestWArg, BKeys, Acc)
-                end, ResAcc, [{W, fun trie_match_in_khepri/5, RestW},
-                              {<<"*">>, fun trie_match_in_khepri/5, RestW},
+                end, ResAcc, [{W, fun trie_match_v3/5, RestW},
+                              {<<"*">>, fun trie_match_v3/5, RestW},
                               {<<"#">>,
-                               fun trie_match_skip_any_in_khepri/5, Words}]).
+                               fun trie_match_skip_any_v3/5, Words}]).
 
-trie_match_part_in_khepri(X, Node, Search, MatchFun, RestW, BKeys, ResAcc) ->
-    case trie_child_in_khepri(X, Node, Search) of
+trie_match_part_v3(X, Node, Search, MatchFun, RestW, BKeys, ResAcc) ->
+    case trie_child_v3(X, Node, Search) of
         {ok, NextNode} -> MatchFun(X, NextNode, RestW, BKeys, ResAcc);
         error          -> ResAcc
     end.
 
-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(
+trie_match_skip_any_v3(X, Node, [], BKeys, ResAcc) ->
+    trie_match_v3(X, Node, [], BKeys, ResAcc);
+trie_match_skip_any_v3(X, Node, [_ | RestW] = Words, BKeys, ResAcc) ->
+    trie_match_skip_any_v3(
       X, Node, RestW, BKeys,
-      trie_match_in_khepri(X, Node, Words, BKeys, ResAcc)).
+      trie_match_v3(X, Node, Words, BKeys, ResAcc)).
 
-trie_child_in_khepri(X, Node, Word) ->
+trie_child_v3(X, Node, Word) ->
     case ets:lookup(
-           ?KHEPRI_PROJECTION,
+           ?KHEPRI_PROJECTION_V3,
            #trie_edge{exchange_name = X,
                       node_id       = Node,
                       word          = Word}) of
         [#topic_trie_edge_v2{node_id = NextNode}] -> {ok, NextNode};
         []                                        -> error
     end.
 
-trie_bindings_in_khepri(X, Node, BKeys) ->
+trie_bindings_v3(X, Node, BKeys) ->
     case ets:lookup(
-           ?KHEPRI_PROJECTION,
+           ?KHEPRI_PROJECTION_V3,
            #trie_edge{exchange_name = X,
                       node_id       = Node,
                       word          = bindings}) of
@@ -147,3 +239,18 @@ trie_bindings_in_khepri(X, Node, BKeys) ->
         [] ->
             []
     end.
+
+add_matched_v3(Destinations, false, Acc) ->
+    Destinations ++ Acc;
+add_matched_v3(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).