WIP projectionskhepri-projections

6 files changed, 361 insertions, 116 deletions

diff --git a/deps/rabbit/src/rabbit_auth_backend_internal.erl b/deps/rabbit/src/rabbit_auth_backend_internal.erl
index 85c6e429d5..e31341ba35 100644
--- a/deps/rabbit/src/rabbit_auth_backend_internal.erl
+++ b/deps/rabbit/src/rabbit_auth_backend_internal.erl
@@ -251,12 +251,16 @@ check_resource_access_in_mnesia(Username, VHostPath, Name, Permission) ->
     end.
 
 check_resource_access_in_khepri(Username, VHostPath, Name, Permission) ->
-    Path = khepri_user_permission_path(Username, VHostPath),
-    case rabbit_khepri:get(Path) of
-        {ok, #user_permission{permission = P}} ->
-            do_check_resource_access(Name, Permission, P);
-        _ ->
-            false
+    UserVHost = #user_vhost{username     = Username,
+                            virtual_host = VHostPath},
+    try
+        P = ets:lookup_element(
+              rabbit_khepri_user_permissions,
+              UserVHost,
+              #user_permission.permission),
+        do_check_resource_access(Name, Permission, P)
+    catch
+        _:_:_ -> false
     end.
 
 do_check_resource_access(Name, Permission, P) ->
@@ -521,10 +525,9 @@ lookup_user_in_mnesia(Username) ->
     rabbit_misc:dirty_read({rabbit_user, Username}).
 
 lookup_user_in_khepri(Username) ->
-    Path = khepri_user_path(Username),
-    case rabbit_khepri:get(Path) of
-        {ok, User} -> {ok, User};
-        _          -> {error, not_found}
+    case ets:lookup(rabbit_khepri_users, Username) of
+        [User] -> {ok, User};
+        _      -> {error, not_found}
     end.
 
 -spec exists(rabbit_types:username()) -> boolean().
diff --git a/deps/rabbit/src/rabbit_khepri.erl b/deps/rabbit/src/rabbit_khepri.erl
index a2872522d5..e1372c3168 100644
--- a/deps/rabbit/src/rabbit_khepri.erl
+++ b/deps/rabbit/src/rabbit_khepri.erl
@@ -12,6 +12,7 @@
 
 -include_lib("khepri/include/khepri.hrl").
 -include_lib("rabbit_common/include/logging.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
 
 -export([setup/0,
          setup/1,
@@ -99,6 +100,7 @@ setup(_) ->
                        friendly_name => ?RA_FRIENDLY_NAME},
     case khepri:start(?RA_SYSTEM, RaServerConfig) of
         {ok, ?STORE_ID} ->
+            register_projections(),
             ?LOG_DEBUG(
                "Khepri-based " ?RA_FRIENDLY_NAME " ready",
                #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
@@ -583,3 +585,212 @@ is_mnesia_table_covered_by_feature_flag(rabbit_user)             -> true;
 is_mnesia_table_covered_by_feature_flag(rabbit_user_permission)  -> true;
 is_mnesia_table_covered_by_feature_flag(rabbit_topic_permission) -> true;
 is_mnesia_table_covered_by_feature_flag(_)                       -> false.
+
+register_projections() ->
+    RegisterFuns = [fun register_rabbit_exchange_projection/0,
+                    fun register_rabbit_queue_projection/0,
+                    fun register_rabbit_vhost_projection/0,
+                    fun register_rabbit_users_projection/0,
+                    fun register_rabbit_user_permissions_projection/0,
+                    fun register_rabbit_bindings_projection/0,
+                    fun register_rabbit_index_route_projection/0,
+                    fun register_rabbit_topic_graph_projection/0],
+    [case RegisterFun() of
+         ok              -> ok;
+         {error, exists} -> ok;
+         {error, Error}  -> throw(Error)
+     end || RegisterFun <- RegisterFuns],
+    ok.
+
+register_rabbit_exchange_projection() ->
+    Name = rabbit_khepri_exchange,
+    PathPattern = [rabbit_store,
+                   exchanges,
+                   _VHost = ?KHEPRI_WILDCARD_STAR,
+                   _Name = ?KHEPRI_WILDCARD_STAR],
+    KeyPos = #exchange.name,
+    register_simple_projection(Name, PathPattern, KeyPos).
+
+register_rabbit_queue_projection() ->
+    Name = rabbit_khepri_queue,
+    PathPattern = [rabbit_store,
+                   queues,
+                   _VHost = ?KHEPRI_WILDCARD_STAR,
+                   _Name = ?KHEPRI_WILDCARD_STAR],
+    KeyPos = 2, %% #amqqueue.name
+    register_simple_projection(Name, PathPattern, KeyPos).
+
+register_rabbit_vhost_projection() ->
+    Name = rabbit_khepri_vhost,
+    PathPattern = [rabbit_vhost, _VHost = ?KHEPRI_WILDCARD_STAR],
+    KeyPos = 2, %% #vhost.virtual_host
+    register_simple_projection(Name, PathPattern, KeyPos).
+
+register_rabbit_users_projection() ->
+    Name = rabbit_khepri_users,
+    PathPattern = [rabbit_auth_backend_internal,
+                   users,
+                   _UserName = ?KHEPRI_WILDCARD_STAR],
+    KeyPos = 2, %% #internal_user.username
+    register_simple_projection(Name, PathPattern, KeyPos).
+
+register_rabbit_user_permissions_projection() ->
+    Name = rabbit_khepri_user_permissions,
+    PathPattern = [rabbit_auth_backend_internal,
+                   users,
+                   _UserName = ?KHEPRI_WILDCARD_STAR,
+                   user_permissions,
+                   _VHost = ?KHEPRI_WILDCARD_STAR],
+    KeyPos = #user_permission.user_vhost,
+    register_simple_projection(Name, PathPattern, KeyPos).
+
+register_simple_projection(Name, PathPattern, KeyPos) ->
+    Options = #{keypos => KeyPos},
+    Fun = fun(_Path, Resource) -> Resource end,
+    Projection = khepri_projection:new(Name, Fun, Options),
+    khepri:register_projection(?RA_CLUSTER_NAME, PathPattern, Projection).
+
+register_rabbit_bindings_projection() ->
+    MapFun = fun(Path, Destination) ->
+                     [rabbit_store, routing, VHost, ExchangeName, RoutingKey] =
+                       Path,
+                     Exchange = rabbit_misc:r(VHost, exchange, ExchangeName),
+                     #binding{source = Exchange,
+                              key = RoutingKey,
+                              destination = Destination}
+             end,
+    ProjectionFun = projection_fun_for_sets(MapFun),
+    Options = #{type => bag, keypos => #binding.source},
+    Projection = khepri_projection:new(
+                   rabbit_khepri_bindings, ProjectionFun, Options),
+    PathPattern = [rabbit_store,
+                   routing,
+                   _VHost = ?KHEPRI_WILDCARD_STAR,
+                   _ExchangeName = ?KHEPRI_WILDCARD_STAR,
+                   _RoutingKey = ?KHEPRI_WILDCARD_STAR],
+    khepri:register_projection(?RA_CLUSTER_NAME, PathPattern, Projection).
+
+register_rabbit_index_route_projection() ->
+    MapFun = fun(Path, Destination) ->
+                     [rabbit_store, routing, VHost, ExchangeName, RoutingKey] =
+                       Path,
+                     Exchange = rabbit_misc:r(VHost, exchange, ExchangeName),
+                     SourceKey = {Exchange, RoutingKey},
+                     #index_route{source_key = SourceKey,
+                                  destination = Destination}
+             end,
+    ProjectionFun = projection_fun_for_sets(MapFun),
+    Options = #{type => bag, keypos => #index_route.source_key},
+    Projection = khepri_projection:new(
+                   rabbit_khepri_index_route, ProjectionFun, Options),
+    PathPattern = [rabbit_store,
+                   routing,
+                   ?KHEPRI_WILDCARD_STAR,
+                   ?KHEPRI_WILDCARD_STAR,
+                   ?KHEPRI_WILDCARD_STAR],
+    khepri:register_projection(?RA_CLUSTER_NAME, PathPattern, Projection).
+
+%% Routing information is stored in the Khepri store as a `set'.
+%% In order to turn these bindings into records in an ETS `bag', we use a
+%% `khepri_projection:extended_projection_fun()' to determine the changes
+%% `khepri_projection' should apply to the ETS table using set algebra.
+projection_fun_for_sets(MapFun) ->
+    fun (_Table, _Path, undefined, undefined) ->
+            ok;
+        (Table, Path, undefined, NewPayload) ->
+            ets:insert(Table, [MapFun(Path, Element) ||
+                               Element <- sets:to_list(NewPayload)]);
+
+        (Table, Path, OldPayload, undefined) ->
+            sets:fold(
+              fun(Element, _Acc) ->
+                      ets:delete_object(Table, MapFun(Path, Element))
+              end, [], OldPayload);
+        (Table, Path, OldPayload, NewPayload) ->
+            Deletions = sets:subtract(OldPayload, NewPayload),
+            Creations = sets:subtract(NewPayload, OldPayload),
+            sets:fold(
+              fun(Element, _Acc) ->
+                      ets:delete_object(Table, MapFun(Path, Element))
+              end, [], Deletions),
+            ets:insert(Table, [MapFun(Path, Element) ||
+                               Element <- sets:to_list(Creations)])
+    end.
+
+register_rabbit_topic_graph_projection() ->
+    Name = rabbit_khepri_topic_trie,
+    Options = #{keypos => #topic_trie_edge.trie_edge},
+    Fun = fun project_topic_trie_binding/4,
+    Projection = khepri_projection:new(Name, Fun, Options),
+    PathPattern = [rabbit_store,
+                   topic_trie_binding,
+                   _VHost = ?KHEPRI_WILDCARD_STAR,
+                   _ExchangeName = ?KHEPRI_WILDCARD_STAR,
+                   _Routes = ?KHEPRI_WILDCARD_STAR_STAR],
+    khepri:register_projection(?RA_CLUSTER_NAME, PathPattern, Projection),
+    ok.
+
+project_topic_trie_binding(_Table, _Path, undefined, undefined) ->
+    ok;
+project_topic_trie_binding(Table, Path, undefined, NewBindings) ->
+    Edges = edges_for_path(Path, NewBindings),
+    ets:insert(Table, Edges);
+project_topic_trie_binding(Table, Path, OldBindings, undefined) ->
+    %% Delete the edge to the bindings and all edges for which the `ToNode'
+    %% has no outgoing edges.
+    [BindingEdge | RestEdges] = edges_for_path(Path, OldBindings),
+    ets:delete_object(Table, BindingEdge),
+    trim_while_out_degree_is_zero(RestEdges);
+project_topic_trie_binding(Table, Path, _OldBindings, NewBindings) ->
+    [BindingEdge | _RestEdges] = edges_for_path(Path, NewBindings),
+    ets:insert(Table, BindingEdge).
+
+edges_for_path(
+  [rabbit_store, topic_trie_binding, VHost, ExchangeName | Components],
+  Bindings) ->
+    Exchange = rabbit_misc:r(VHost, exchange, ExchangeName),
+    edges_for_path([root | Components], Bindings, Exchange, []).
+
+edges_for_path([FromNodeId, To | Rest], Bindings, Exchange, Edges) ->
+    ToNodeId = [To | FromNodeId],
+    Edge = #topic_trie_edge{trie_edge = #trie_edge{exchange_name = Exchange,
+                                                   node_id =       FromNodeId,
+                                                   word =          To},
+                            node_id = ToNodeId},
+    edges_for_path([ToNodeId | Rest], Bindings, Exchange, [Edge | Edges]);
+edges_for_path([LeafId], Bindings, Exchange, Edges) ->
+    ToNodeId = {bindings, Bindings},
+    Edge = #topic_trie_edge{trie_edge = #trie_edge{exchange_name = Exchange,
+                                                   node_id =       LeafId,
+                                                   word =          bindings},
+                            node_id = ToNodeId},
+    [Edge | Edges].
+
+-spec trim_while_out_degree_is_zero(Edges) -> ok
+    when
+      Edges :: [Edge],
+      Edge :: #topic_trie_edge{}.
+
+trim_while_out_degree_is_zero([]) ->
+    ok;
+trim_while_out_degree_is_zero([Edge | Rest]) ->
+    #topic_trie_edge{trie_edge = #trie_edge{exchange_name = Exchange,
+                                            node_id       = _FromNodeId},
+                     node_id = ToNodeId} = Edge,
+    OutEdgePattern = #topic_trie_edge{trie_edge =
+                                      #trie_edge{exchange_name = Exchange,
+                                                 node_id       = ToNodeId,
+                                                 word          = '_'},
+                                      node_id = '_'},
+    case ets:match(rabbit_khepri_topic_trie, OutEdgePattern, 1) of
+        '$end_of_table' ->
+            %% If the ToNode has an out degree of zero, trim the edge to
+            %% the node, effectively erasing ToNode.
+            ets:delete_object(rabbit_khepri_topic_trie, Edge),
+            trim_while_out_degree_is_zero(Rest);
+        {_Match, _Continuation} ->
+            %% Return after finding the first node with a non-zero out-degree.
+            %% If a node has a non-zero out-degree then all of its ancestors
+            %% must as well.
+            ok
+    end.
diff --git a/deps/rabbit/src/rabbit_store.erl b/deps/rabbit/src/rabbit_store.erl
index 974671a0f9..a3f0bc0eff 100644
--- a/deps/rabbit/src/rabbit_store.erl
+++ b/deps/rabbit/src/rabbit_store.erl
@@ -241,12 +241,12 @@ match_exchanges(Pattern0) ->
 lookup_exchange(Name) ->
     rabbit_khepri:try_mnesia_or_khepri(
       fun() -> lookup({rabbit_exchange, Name}, mnesia) end,
-      fun() -> lookup(khepri_exchange_path(Name), khepri) end).
+      fun() -> lookup_resource_in_khepri_projection(rabbit_khepri_exchange, Name) end).
 
 lookup_many_exchanges(Names) ->
     rabbit_khepri:try_mnesia_or_khepri(
       fun() -> lookup_many(rabbit_exchange, Names, mnesia) end,
-      fun() -> lookup_many(fun khepri_exchange_path/1, Names, khepri) end).
+      fun() -> lookup_resources_in_khepri_projection(rabbit_khepri_exchange, Names) end).
 
 peek_exchange_serial(XName, LockType) ->
     rabbit_khepri:try_mnesia_or_khepri(
@@ -489,39 +489,34 @@ list_bindings(VHost) ->
               [B || #route{binding = B} <- list_in_mnesia(rabbit_route, Match)]
       end,
       fun() ->
-              Path = khepri_routes_path() ++ [VHost, if_has_data_wildcard()],
-              lists:foldl(fun(SetOfBindings, Acc) ->
-                                  sets:to_list(SetOfBindings) ++ Acc
-                          end, [], list_in_khepri(Path))
+              VHostResource = rabbit_misc:r(VHost, '_'),
+              Match = #binding{source      = VHostResource,
+                               destination = VHostResource,
+                               _           = '_'},
+              ets:match(rabbit_khepri_bindings, Match)
       end).
 
-list_bindings_for_source(#resource{virtual_host = VHost, name = Name} = Resource) ->
+list_bindings_for_source(Source) ->
     rabbit_khepri:try_mnesia_or_khepri(
       fun() ->
-              Route = #route{binding = #binding{source = Resource, _ = '_'}},
+              Route = #route{binding = #binding{source = Source, _ = '_'}},
               [B || #route{binding = B} <- list_in_mnesia(rabbit_route, Route)]
       end,
-      fun() ->
-              Path = khepri_routes_path() ++ [VHost, Name, if_has_data_wildcard()],
-              lists:foldl(fun(SetOfBindings, Acc) ->
-                                  sets:to_list(SetOfBindings) ++ Acc
-                          end, [], list_in_khepri(Path))
-      end).
+      fun() -> ets:lookup(rabbit_khepri_bindings, Source) end).
 
-list_bindings_for_destination(#resource{virtual_host = VHost, name = Name,
-                                        kind = Kind} = Resource) ->
+list_bindings_for_destination(Destination) ->
     rabbit_khepri:try_mnesia_or_khepri(
       fun() ->
-              Route = rabbit_binding:reverse_route(#route{binding = #binding{destination = Resource,
+              Route = rabbit_binding:reverse_route(#route{binding = #binding{destination = Destination,
                                                                              _ = '_'}}),
               [rabbit_binding:reverse_binding(B) ||
                   #reverse_route{reverse_binding = B} <- list_in_mnesia(rabbit_reverse_route, Route)]
       end,
       fun() ->
-              Path = khepri_routes_path() ++ [VHost, ?KHEPRI_WILDCARD_STAR, Kind, Name, if_has_data_wildcard()],
-              lists:foldl(fun(SetOfBindings, Acc) ->
-                                  sets:to_list(SetOfBindings) ++ Acc
-                          end, [], list_in_khepri(Path))
+              %% TODO: projection for bindings indexed by destination?
+              Match = #binding{destination = Destination,
+                               _           = '_'},
+              ets:match(rabbit_khepri_bindings, Match)
       end).
 
 list_bindings_for_source_and_destination(SrcName, DstName) ->
@@ -533,10 +528,10 @@ list_bindings_for_source_and_destination(SrcName, DstName) ->
               [B || #route{binding = B} <- list_in_mnesia(rabbit_route, Route)]
       end,
       fun() ->
-              Values = match_source_and_destination_in_khepri(SrcName, DstName),
-              lists:foldl(fun(SetOfBindings, Acc) ->
-                                  sets:to_list(SetOfBindings) ++ Acc
-                          end, [], Values)
+              Match = #binding{source      = SrcName,
+                               destination = DstName,
+                               _           = '_'},
+              ets:match(rabbit_khepri_bindings, Match)
       end).
 
 list_explicit_bindings() ->
@@ -728,12 +723,12 @@ store_durable_queue(Q) ->
 lookup_queue(Name) ->
     rabbit_khepri:try_mnesia_or_khepri(
       fun() -> lookup({rabbit_queue, Name}, mnesia) end,
-      fun() -> lookup(khepri_queue_path(Name), khepri) end).
+      fun() -> lookup_resource_in_khepri_projection(rabbit_khepri_queue, Name) end).
 
 lookup_durable_queue(Name) ->
     rabbit_khepri:try_mnesia_or_khepri(
       fun() -> lookup({rabbit_durable_queue, Name}, mnesia) end,
-      fun() -> lookup(khepri_queue_path(Name), khepri) end).
+      fun() -> lookup_resource_in_khepri_projection(rabbit_khepri_queue, Name) end).
 
 %% TODO this should be internal, it's here because of mirrored queues
 lookup_queue_in_khepri_tx(Name) ->
@@ -742,14 +737,14 @@ lookup_queue_in_khepri_tx(Name) ->
 lookup_queues(Names) when is_list(Names) ->
     rabbit_khepri:try_mnesia_or_khepri(
       fun() -> lookup_many(rabbit_queue, Names, mnesia) end,
-      fun() -> lookup_many(fun khepri_queue_path/1, Names, khepri) end);
+      fun() -> lookup_resources_in_khepri_projection(rabbit_khepri_queue, Names) end);
 lookup_queues(Name) ->
     lookup_queue(Name).
 
 lookup_durable_queues(Names) when is_list(Names) ->
     rabbit_khepri:try_mnesia_or_khepri(
       fun() -> lookup_many(rabbit_durable_queue, Names, mnesia) end,
-      fun() -> lookup_many(fun khepri_queue_path/1, Names, khepri) end);
+      fun() -> lookup_resources_in_khepri_projection(rabbit_khepri_queue, Names) end);
 lookup_durable_queues(Name) ->
     lookup_durable_queue(Name).
 
@@ -759,7 +754,7 @@ exists_queue(Name) ->
               ets:member(rabbit_queue, Name)
       end,
       fun() ->
-              rabbit_khepri:exists(khepri_queue_path(Name))
+              ets:member(rabbit_khepri_queue, Name)
       end).
 
 update_queue(#resource{virtual_host = VHost, name = Name} = QName, Fun) ->
@@ -894,11 +889,9 @@ match_bindings(SrcName, Match) ->
                            Routes, Match(Binding)]
       end,
       fun() ->
-              Data = match_source_in_khepri(SrcName),
-              Bindings = lists:foldl(fun(SetOfBindings, Acc) ->
-                                             sets:to_list(SetOfBindings) ++ Acc
-                                     end, [], maps:values(Data)),
-              [Dest || Binding = #binding{destination = Dest} <- Bindings, Match(Binding)]
+              Routes = ets:lookup(rabbit_khepri_bindings, SrcName),
+              [Dest || Binding = #binding{destination = Dest} <-
+                           Routes, Match(Binding)]
       end).
 
 match_routing_key(SrcName, RoutingKeys, UseIndex) ->
@@ -906,13 +899,18 @@ match_routing_key(SrcName, RoutingKeys, UseIndex) ->
       fun() ->
               case UseIndex of
                   true ->
-                      route_in_mnesia_v2(SrcName, RoutingKeys);
+                      route_v2(rabbit_index_route, SrcName, RoutingKeys);
                   _ ->
                       route_in_mnesia_v1(SrcName, RoutingKeys)
               end
       end,
       fun() ->
-              match_source_and_key_in_khepri(SrcName, RoutingKeys)
+              case UseIndex of
+                  true ->
+                      route_v2(rabbit_khepri_index_route, SrcName, RoutingKeys);
+                  _ ->
+                      match_source_and_key_in_khepri(SrcName, RoutingKeys)
+              end
       end).
 
 route_in_mnesia_v1(SrcName, [RoutingKey]) ->
@@ -945,27 +943,27 @@ route_in_mnesia_v1(SrcName, [_|_] = RoutingKeys) ->
 %% ets:select/2 is expensive because it needs to compile the match spec every
 %% time and lookup does not happen by a hash key.
 %%
-%% In contrast, route_v2/2 increases end-to-end message sending throughput
+%% In contrast, route_v2/3 increases end-to-end message sending throughput
 %% (i.e. from RabbitMQ client to the queue process) by up to 35% by using ets:lookup_element/3.
 %% Only the direct exchange type uses the rabbit_index_route table to store its
 %% bindings by table key tuple {SourceExchange, RoutingKey}.
--spec route_in_mnesia_v2(rabbit_types:binding_source(), [rabbit_router:routing_key(), ...]) ->
+-spec route_v2(ets:table(), rabbit_types:binding_source(), [rabbit_router:routing_key(), ...]) ->
     rabbit_router:match_result().
-route_in_mnesia_v2(SrcName, [RoutingKey]) ->
+route_v2(Table, SrcName, [RoutingKey]) ->
     %% optimization
-    destinations(SrcName, RoutingKey);
-route_in_mnesia_v2(SrcName, [_|_] = RoutingKeys) ->
+    destinations(Table, SrcName, RoutingKey);
+route_v2(Table, SrcName, [_|_] = RoutingKeys) ->
     lists:flatmap(fun(Key) ->
-                          destinations(SrcName, Key)
+                          destinations(Table, SrcName, Key)
                   end, RoutingKeys).
 
-destinations(SrcName, RoutingKey) ->
+destinations(Table, SrcName, RoutingKey) ->
     %% Prefer try-catch block over checking Key existence with ets:member/2.
     %% The latter reduces throughput by a few thousand messages per second because
     %% of function db_member_hash in file erl_db_hash.c.
     %% We optimise for the happy path, that is the binding / table key is present.
     try
-        ets:lookup_element(rabbit_index_route,
+        ets:lookup_element(Table,
                            {SrcName, RoutingKey},
                            #index_route.destination)
     catch
@@ -1003,26 +1001,8 @@ add_topic_trie_binding_tx(XName, RoutingKey, Destination, Args) ->
     ok = khepri_tx:put(Path, Set).
 
 route_delivery_for_exchange_type_topic(XName, RoutingKey) ->
-    Words = lists:map(fun(W) -> #if_any{conditions = [W, <<"*">>]} end,
-                      split_topic_trie_key(RoutingKey)),
-    Root = khepri_exchange_type_topic_path(XName),
-    Path0 = Root ++ Words,
-    {Hd, [Tl]} = lists:split(length(Path0) - 1, Path0),
-    Path = Hd ++ [if_has_data([Tl])],
-    Fanout = Root ++ [<<"#">>],
-    Map = case rabbit_khepri:get(Fanout) of
-              {ok, Data} when Data =/= undefined ->
-                  #{Fanout => Data};
-              _ ->
-                  case rabbit_khepri:match(Path) of
-                      {ok, Map0} -> Map0;
-                      _ -> #{}
-                  end
-          end,
-    maps:fold(fun(_, Data, Acc) ->
-                      Bindings = sets:to_list(Data),
-                      [maps:get(destination, B) || B <- Bindings] ++ Acc
-              end, [], Map).
+    Words = split_topic_trie_key(RoutingKey),
+    trie_match(XName, Words).
 
 delete_topic_trie_bindings_for_exchange(XName) ->
     ok = rabbit_khepri:delete(khepri_exchange_type_topic_path(XName)).
@@ -1932,22 +1912,15 @@ binding_has_for_source_in_khepri(#resource{virtual_host = VHost, name = Name}) -
             Error
     end.
 
-match_source_in_khepri(#resource{virtual_host = VHost, name = Name}) ->
-    Path = khepri_routes_path() ++ [VHost, Name, if_has_data_wildcard()],
-    {ok, Map} = rabbit_khepri:match(Path),
-    Map.
-
 match_source_and_key_in_khepri(Src, ['_']) ->
-    Path = khepri_routing_path(Src, if_has_data_wildcard()),
-    case rabbit_khepri:match(Path) of
-        {ok, Map} ->
-            maps:fold(fun(_, Dsts, Acc) ->
-                              sets:to_list(Dsts) ++ Acc
-                      end, [], Map);
-        {error, {khepri, node_not_found, _}} ->
+    try
+        ets:lookup_element(rabbit_khepri_bindings, Src, #binding.destination)
+    catch
+        error:badarg ->
             []
     end;
 match_source_and_key_in_khepri(Src, RoutingKeys) ->
+    %% TODO: use {@link ets:select/2}
     lists:foldl(
       fun(RK, Acc) ->
               Path = khepri_routing_path(Src, RK),
@@ -2302,3 +2275,63 @@ recover_mnesia_tables() ->
         ++ [Table || {Table, _} <- rabbit_table:definitions()],
     [mnesia:change_table_access_mode(Table, read_write) || Table <- Tables],
     ok.
+
+lookup_resource_in_khepri_projection(Projection, Key) ->
+    case ets:lookup(Projection, Key) of
+        [Resource] -> {ok, Resource};
+        []         -> {error, not_found}
+    end.
+
+lookup_resources_in_khepri_projection(Projection, Keys) ->
+    lists:append([ets:lookup(Projection, Name) || Name <- Keys]).
+
+%%--->8--- Khepri Topic Exchange Routing --->8---
+%% This is slightly different than the mnesia version. We use only one
+%% table. Bindings are the leaf nodes of the trie.
+
+%% Also, lists are rewritten as binaries. Why does this work before?
+
+trie_match(X, Words) ->
+    trie_match(X, root, Words, []).
+
+trie_match(X, Node, [], ResAcc) ->
+    trie_match_part(X, Node, <<"#">>, fun trie_match_skip_any/4, [],
+                    trie_bindings(X, Node) ++ ResAcc);
+trie_match(X, Node, [W | RestW] = Words, ResAcc) ->
+    lists:foldl(fun ({WArg, MatchFun, RestWArg}, Acc) ->
+                        trie_match_part(X, Node, WArg, MatchFun, RestWArg, Acc)
+                end, ResAcc, [{W, fun trie_match/4, RestW},
+                              {<<"*">>, fun trie_match/4, RestW},
+                              {<<"#">>, fun trie_match_skip_any/4, Words}]).
+
+trie_match_part(X, Node, Search, MatchFun, RestW, ResAcc) ->
+    case trie_child(X, Node, Search) of
+        {ok, NextNode} -> MatchFun(X, NextNode, RestW, ResAcc);
+        error          -> ResAcc
+    end.
+
+trie_match_skip_any(X, Node, [], ResAcc) ->
+    trie_match(X, Node, [], ResAcc);
+trie_match_skip_any(X, Node, [_ | RestW] = Words, ResAcc) ->
+    trie_match_skip_any(X, Node, RestW,
+                        trie_match(X, Node, Words, ResAcc)).
+
+trie_child(X, Node, Word) ->
+    case ets:lookup(rabbit_khepri_topic_trie,
+                    #trie_edge{exchange_name = X,
+                               node_id       = Node,
+                               word          = Word}) of
+        [#topic_trie_edge{node_id = NextNode}] -> {ok, NextNode};
+        []                                     -> error
+    end.
+
+trie_bindings(X, Node) ->
+    case ets:lookup(rabbit_khepri_topic_trie,
+                    #trie_edge{exchange_name = X,
+                               node_id       = Node,
+                               word          = bindings}) of
+        [#topic_trie_edge{node_id = {bindings, Bindings}}] ->
+            [Dest || #{destination := Dest} <- sets:to_list(Bindings)];
+        [] ->
+            []
+    end.
diff --git a/deps/rabbit/src/rabbit_vhost.erl b/deps/rabbit/src/rabbit_vhost.erl
index 4fd7a76286..3d464f22be 100644
--- a/deps/rabbit/src/rabbit_vhost.erl
+++ b/deps/rabbit/src/rabbit_vhost.erl
@@ -580,8 +580,7 @@ exists_in_mnesia(VHost) ->
     mnesia:dirty_read({rabbit_vhost, VHost}) /= [].
 
 exists_in_khepri(VHost) ->
-    Path = khepri_vhost_path(VHost),
-    rabbit_khepri:exists(Path).
+    ets:member(rabbit_khepri_vhost, VHost).
 
 -spec list_names() -> [vhost:name()].
 list_names() ->
@@ -593,11 +592,7 @@ list_names_in_mnesia() ->
     mnesia:dirty_all_keys(rabbit_vhost).
 
 list_names_in_khepri() ->
-    Path = khepri_vhosts_path(),
-    case rabbit_khepri:list_child_nodes(Path) of
-        {ok, Result} -> Result;
-        _            -> []
-    end.
+    ets:select(rabbit_khepri_vhost, [{vhost:pattern_match_names(), [], ['$1']}]).
 
 %% Exists for backwards compatibility, prefer list_names/0.
 -spec list() -> [vhost:name()].
@@ -613,11 +608,7 @@ all_in_mnesia() ->
     mnesia:dirty_match_object(rabbit_vhost, vhost:pattern_match_all()).
 
 all_in_khepri() ->
-    Path = khepri_vhosts_path(),
-    case rabbit_khepri:list(Path) of
-        {ok, VHosts} -> maps:values(VHosts);
-        _            -> []
-    end.
+    ets:tab2list(rabbit_khepri_vhost).
 
 -spec all_tagged_with(atom()) -> [vhost:vhost()].
 all_tagged_with(TagName) ->
@@ -655,10 +646,9 @@ lookup_in_mnesia(VHostName) ->
     end.
 
 lookup_in_khepri(VHostName) ->
-    Path = khepri_vhost_path(VHostName),
-    case rabbit_khepri:get(Path) of
-        {ok, Record} -> Record;
-        _            -> {error, {no_such_vhost, VHostName}}
+    case ets:lookup(rabbit_khepri_vhost, VHostName) of
+        [Record] -> Record;
+        _        -> {error, {no_such_vhost, VHostName}}
     end.
 
 with_in_mnesia(VHostName, Thunk) ->
@@ -869,10 +859,9 @@ info_in_mnesia(Key) ->
     end.
 
 info_in_khepri(Key) ->
-    Path = khepri_vhost_path(Key),
-    case rabbit_khepri:get(Path) of
-        {ok, VHost} -> infos(?INFO_KEYS, VHost);
-        _           -> []
+    case ets:lookup(rabbit_khepri_vhost, Key) of
+        [VHost] -> infos(?INFO_KEYS, VHost);
+        _       -> []
     end.
 
 -spec info(vhost:vhost(), rabbit_types:info_keys()) -> rabbit_types:infos().
diff --git a/deps/rabbit/src/vhost.erl b/deps/rabbit/src/vhost.erl
index cdd3ba3ce5..fb0cb99c6d 100644
--- a/deps/rabbit/src/vhost.erl
+++ b/deps/rabbit/src/vhost.erl
@@ -20,6 +20,7 @@
   upgrade/1,
   upgrade_to/2,
   pattern_match_all/0,
+  pattern_match_names/0,
   get_name/1,
   get_limits/1,
   get_metadata/1,
@@ -46,7 +47,7 @@
 
 -record(vhost, {
     %% name as a binary
-    virtual_host :: name() | '_',
+    virtual_host :: name() | '_' | '$1',
     %% proplist of limits configured, if any
     limits :: list() | '_',
     metadata :: metadata() | '_'
@@ -60,7 +61,7 @@
 
 -type vhost_pattern() :: vhost_v2_pattern().
 -type vhost_v2_pattern() :: #vhost{
-                                  virtual_host :: name() | '_',
+                                  virtual_host :: name() | '_' | '$1',
                                   limits :: '_',
                                   metadata :: '_'
                                  }.
@@ -112,10 +113,13 @@ info_keys() ->
      cluster_state].
 
 -spec pattern_match_all() -> vhost_pattern().
-
 pattern_match_all() ->
     #vhost{_ = '_'}.
 
+-spec pattern_match_names() -> vhost_pattern().
+pattern_match_names() ->
+    #vhost{virtual_host = '$1', _ = '_'}.
+
 -spec get_name(vhost()) -> name().
 get_name(#vhost{virtual_host = Value}) -> Value.
 
diff --git a/workspace_helpers.bzl b/workspace_helpers.bzl
index 625b10ed8e..45b6e38932 100644
--- a/workspace_helpers.bzl
+++ b/workspace_helpers.bzl
@@ -125,8 +125,8 @@ def rabbitmq_external_deps(rabbitmq_workspace = "@rabbitmq-server"):
     github_erlang_app(
         name = "khepri",
         org = "rabbitmq",
-        ref = "main",
-        version = "main",
+        ref = "b7f9bac94857078a0b0584aa72a920f7596a1e49",
+        version = "b7f9bac94857078a0b0584aa72a920f7596a1e49",
         build_file = rabbitmq_workspace + "//:BUILD.khepri",
     )
 
@@ -180,11 +180,16 @@ sed -i"_orig" -E '/VERSION/ s/[0-9]+\\.[0-9]+\\.[0-9]+/'${VERSION}'/' BUILD.baze
         sha256 = "282a8a323ca2a845c9e6f787d166348f776c1d4a41ede63046d72d422e3da946",
     )
 
-    git_repository(
+    hex_pm_erlang_app(
         name = "ra",
-        branch = "main",
-        remote = "https://github.com/rabbitmq/ra.git",
-        patch_cmds = [RA_INJECT_GIT_VERSION],
+        version = "2.4.0",
+        sha256 = "f716146ee1755823fbff741669053efc76e241caa95465782fc4c0a4389eb40c",
+        deps = [
+            "@gen_batch_server//:erlang_app",
+        ],
+        runtime_deps = [
+            "@aten//:erlang_app",
+        ]
     )
 
     hex_archive(