Khepri: use projections

Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com>

8 files changed, 200 insertions, 140 deletions

diff --git a/deps/rabbit/src/rabbit_db_binding.erl b/deps/rabbit/src/rabbit_db_binding.erl
index fcbbeb56e6..3e3779bb92 100644
--- a/deps/rabbit/src/rabbit_db_binding.erl
+++ b/deps/rabbit/src/rabbit_db_binding.erl
@@ -203,12 +203,10 @@ create_in_khepri(#binding{source = SrcName,
                                                        already_exists;
                                                    false ->
                                                        ok = khepri_tx:put(Path, sets:add_element(Binding, Set)),
-                                                       add_routing(Binding),
                                                        serial_in_khepri(MaybeSerial, Src)
                                                end;
                                            _ ->
                                                ok = khepri_tx:put(Path, sets:add_element(Binding, sets:new())),
-                                               add_routing(Binding),
                                                serial_in_khepri(MaybeSerial, Src)
                                        end
                                end, rw),
@@ -236,17 +234,6 @@ lookup_resource(#resource{kind = exchange} = Name) ->
         _ -> []
     end.
 
-add_routing(#binding{destination = Dst, source = Src} = Binding) ->
-    Path = khepri_routing_path(Binding),
-    KeepWhile = #{rabbit_db_exchange:path(Src) => #if_node_exists{exists = true}},
-    Options = #{keep_while => KeepWhile},
-    case khepri_tx:get(Path) of
-        {ok, Data} ->
-            ok = khepri_tx:put(Path, sets:add_element(Dst, Data), Options);
-        _ ->
-            ok = khepri_tx:put(Path, sets:add_element(Dst, sets:new()), Options)
-    end.
-
 serial_in_khepri(false, _) ->
     none;
 serial_in_khepri(true, X) ->
@@ -339,7 +326,6 @@ delete_in_khepri(#binding{source = SrcName,
                                    case ChecksFun(Src, Dst) of
                                        ok ->
                                            ok = delete_in_khepri(Binding),
-                                           ok = delete_routing(Binding),
                                            maybe_auto_delete_exchange_in_khepri(Binding#binding.source, [Binding], rabbit_binding:new_deletions(), false);
                                        {error, _} = Err ->
                                            Err
@@ -381,21 +367,6 @@ delete_in_khepri(Binding) ->
             ok
     end.
 
-delete_routing(#binding{destination = Dst} = Binding) ->
-    Path = khepri_routing_path(Binding),
-    case khepri_tx:get(Path) of
-        {ok, Data0} ->
-            Data = sets:del_element(Dst, Data0),
-            case sets:is_empty(Data) of
-                true ->
-                    ok = khepri_tx:delete(Path);
-                false ->
-                    ok = khepri_tx:put(Path, Data)
-            end;
-        _ ->
-            ok
-    end.
-
 maybe_auto_delete_exchange_in_khepri(XName, Bindings, Deletions, OnlyDurable) ->
     {Entry, Deletions1} =
         case rabbit_db_exchange:maybe_auto_delete_in_khepri(XName, OnlyDurable) of
@@ -434,10 +405,7 @@ get_all_in_mnesia() ->
       end).
 
 get_all_in_khepri() ->
-    Path = khepri_routes_path() ++ [rabbit_khepri:if_has_data_wildcard()],
-    lists:foldl(fun(SetOfBindings, Acc) ->
-                        sets:to_list(SetOfBindings) ++ Acc
-                end, [], rabbit_db:list_in_khepri(Path)).
+     [B || #route{binding = B} <- ets:tab2list(rabbit_khepri_bindings)].
 
 -spec get_all(VHostName) -> [Binding] when
       VHostName :: vhost:name(),
@@ -493,14 +461,11 @@ get_all_in_mnesia(SrcName, DstName, Reverse) ->
     Fun = list_for_route(Route, Reverse),
     mnesia:async_dirty(Fun).
 
-get_all_in_khepri(#resource{virtual_host = VHost, name = SrcName},
-                  #resource{kind = Kind, name = DstName}) ->
-    Path = khepri_routes_path() ++ [VHost, SrcName, Kind, DstName,
-                                    rabbit_khepri:if_has_data_wildcard()],
-    Values = rabbit_db:list_in_khepri(Path),
-    lists:foldl(fun(SetOfBindings, Acc) ->
-                        sets:to_list(SetOfBindings) ++ Acc
-                end, [], Values).
+get_all_in_khepri(SrcName, DstName) ->
+    Match = #binding{source      = SrcName,
+                     destination = DstName,
+                     _           = '_'},
+    ets:match_object(rabbit_khepri_bindings, Match).
 
 %% -------------------------------------------------------------------
 %% get_all_for_source().
@@ -538,11 +503,8 @@ list_for_route(Route, true) ->
                                  rabbit_binding:reverse_route(Route), read)]
     end.
 
-get_all_for_source_in_khepri(#resource{virtual_host = VHost, name = Name}) ->
-    Path = khepri_routes_path() ++ [VHost, Name, rabbit_khepri:if_has_data_wildcard()],
-    lists:foldl(fun(SetOfBindings, Acc) ->
-                        sets:to_list(SetOfBindings) ++ Acc
-                end, [], rabbit_db:list_in_khepri(Path)).
+get_all_for_source_in_khepri(Resource) ->
+    ets:lookup(rabbit_khepri_bindings, Resource).
 
 %% -------------------------------------------------------------------
 %% get_all_for_destination().
@@ -570,13 +532,11 @@ get_all_for_destination_in_mnesia(Dst) ->
     Fun = list_for_route(Route, true),
     mnesia:async_dirty(Fun).
 
-get_all_for_destination_in_khepri(#resource{virtual_host = VHost, name = Name,
-                                            kind = Kind}) ->
-    Path = khepri_routes_path() ++ [VHost, ?KHEPRI_WILDCARD_STAR, Kind, Name,
-                                    rabbit_khepri:if_has_data_wildcard()],
-    lists:foldl(fun(SetOfBindings, Acc) ->
-                        sets:to_list(SetOfBindings) ++ Acc
-                end, [], rabbit_db:list_in_khepri(Path)).
+get_all_for_destination_in_khepri(Destination) ->
+     %% TODO: projection for bindings indexed by destination?
+    Match = #binding{destination = Destination,
+                     _           = '_'},
+    ets:match_object(rabbit_khepri_bindings, Match).
 
 %% -------------------------------------------------------------------
 %% fold().
@@ -649,16 +609,9 @@ match_in_mnesia(SrcName, Match) ->
                  Routes, Match(Binding)].
 
 match_in_khepri(SrcName, Match) ->
-    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)].
-
-match_source_in_khepri(#resource{virtual_host = VHost, name = Name}) ->
-    Path = khepri_routes_path() ++ [VHost, Name, rabbit_khepri:if_has_data_wildcard()],
-    {ok, Map} = rabbit_khepri:match(Path),
-    Map.
+    Routes = ets:lookup(rabbit_khepri_bindings, SrcName),
+    [Dest || Binding = #binding{destination = Dest} <-
+                 Routes, Match(Binding)].
 
 %% Routing - HOT CODE PATH
 %% -------------------------------------------------------------------
@@ -692,25 +645,22 @@ match_routing_key_in_mnesia(SrcName, RoutingKeys, UseIndex) ->
     end.
 
 match_routing_key_in_khepri(Src, ['_']) ->
-    Path = khepri_routing_path(Src, rabbit_khepri: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, _}} ->
-            []
-    end;
+    MatchHead = #index_route{source_key      = {Src, '_'},
+                             destination = '$1',
+                             _           = '_'},
+    ets:select(rabbit_khepri_index_route, [{MatchHead, [], ['$1']}]);
+
 match_routing_key_in_khepri(Src, RoutingKeys) ->
     lists:foldl(
       fun(RK, Acc) ->
-              Path = khepri_routing_path(Src, RK),
-              %% Don't use transaction if we want to hit the cache
-              case rabbit_khepri:get(Path) of
-                  {ok, Dsts} ->
-                      sets:to_list(Dsts) ++ Acc;
-                  {error, {khepri, node_not_found, _}} ->
-                      Acc
+              try
+                  Dst = ets:lookup_element(
+                          rabbit_khepri_index_route,
+                          {Src, RK},
+                          #index_route.destination),
+                  Dst ++ Acc
+              catch
+                  _:_:_ -> Acc
               end
       end, [], RoutingKeys).
 
@@ -822,9 +772,12 @@ delete_all_for_exchange_in_khepri(X = #exchange{name = XName}, OnlyDurable, Remo
 
 delete_for_source_in_khepri(#resource{virtual_host = VHost, name = Name}) ->
     Path = khepri_routes_path() ++ [VHost, Name],
-    {ok, Bindings} = khepri_tx:get_many(Path ++ [rabbit_khepri:if_has_data_wildcard()]),
-    ok = khepri_tx:delete(Path),
-    maps:fold(fun(_, Set, Acc) ->
+    {ok, Bindings} = khepri_tx:get_many(Path ++ [rabbit_db:if_has_data_wildcard()]),
+    %% ok = khepri_tx:delete(Path),
+    maps:fold(fun(P, Set, Acc) ->
+                      %% TODO projections are not updated if we just use the previous
+                      %% delete of the root path. Let's keep it like this for now...
+                      ok = khepri_tx:delete(P),
                       sets:to_list(Set) ++ Acc
               end, [], Bindings).
 
@@ -874,7 +827,6 @@ delete_for_destination_in_khepri(DstName, OnlyDurable) ->
     Bindings = maps:fold(fun(_, Set, Acc) ->
                                  sets:to_list(Set) ++ Acc
                          end, [], BindingsMap),
-    lists:foreach(fun(Binding) -> delete_routing(Binding) end, Bindings),
     rabbit_binding:group_bindings_fold(fun maybe_auto_delete_exchange_in_khepri/4,
                                        lists:keysort(#binding.source, Bindings), OnlyDurable).
 
@@ -975,9 +927,7 @@ add_binding_tx(Path, Binding) ->
               _ ->
                   sets:new()
           end,
-    ok = khepri_tx:put(Path, sets:add_element(Binding, Set)),
-    add_routing(Binding),
-    ok.
+    ok = khepri_tx:put(Path, sets:add_element(Binding, Set)).
 
 -spec mnesia_delete_to_khepri(Table, ToDelete) -> ok when
       Table :: atom(),
@@ -1000,8 +950,7 @@ mnesia_delete_to_khepri(rabbit_semi_durable_route, Name) ->
       Table :: atom().
 
 clear_data_in_khepri(rabbit_route) ->
-    khepri_delete(khepri_routes_path()),
-    khepri_delete(khepri_routing_path());
+    khepri_delete(khepri_routes_path());
 %% There is a single khepri entry for routes and it should be already deleted
 clear_data_in_khepri(rabbit_durable_route) ->
     ok;
@@ -1040,7 +989,6 @@ clear_in_mnesia() ->
 
 clear_in_khepri() ->
     khepri_delete(khepri_routes_path()),
-    khepri_delete(khepri_routing_path()),
     ok.
 
 %% --------------------------------------------------------------
@@ -1054,19 +1002,6 @@ khepri_route_path(#binding{source = #resource{virtual_host = VHost, name = SrcNa
 khepri_routes_path() ->
     [?MODULE, routes].
 
-%% Routing optimisation, probably the most relevant on the hot code path.
-%% It only needs to store a list of destinations to be used by rabbit_router.
-%% Unless there is a high queue churn, this should barely change. Thus, the small
-%% penalty for updates should be worth it.
-khepri_routing_path() ->
-    [?MODULE, routing].
-
-khepri_routing_path(#binding{source = Src, key = RoutingKey}) ->
-    khepri_routing_path(Src, RoutingKey).
-
-khepri_routing_path(#resource{virtual_host = VHost, name = Name}, RoutingKey) ->
-    [?MODULE, routing, VHost, Name, RoutingKey].
-
 %% --------------------------------------------------------------
 %% Internal
 %% --------------------------------------------------------------
diff --git a/deps/rabbit/src/rabbit_db_exchange.erl b/deps/rabbit/src/rabbit_db_exchange.erl
index b76ceb2236..072a9b4cf4 100644
--- a/deps/rabbit/src/rabbit_db_exchange.erl
+++ b/deps/rabbit/src/rabbit_db_exchange.erl
@@ -179,10 +179,9 @@ get_in_mnesia(Name) ->
     rabbit_mnesia:dirty_read({?MNESIA_TABLE, Name}).
 
 get_in_khepri(Name) ->
-    Path = khepri_exchange_path(Name),
-    case rabbit_khepri:get(Path) of
-        {ok, X} -> {ok, X};
-        _ -> {error, not_found}
+    case ets:lookup(rabbit_khepri_exchange, Name) of
+        [X] -> {ok, X};
+        []  -> {error, not_found}
     end.
 
 %% -------------------------------------------------------------------
@@ -225,12 +224,7 @@ get_many_in_mnesia(Table, Names) when is_list(Names) ->
     lists:append([ets:lookup(Table, Name) || Name <- Names]).
 
 get_many_in_khepri(Names) when is_list(Names) ->
-    lists:foldl(fun(Name, Acc) ->
-                        case get_in_khepri(Name) of
-                            {ok, X} -> [X | Acc];
-                            _ -> Acc
-                        end
-                end, [], Names).
+    lists:append([ets:lookup(rabbit_khepri_exchange, Name) || Name <- Names]).
 
 %% -------------------------------------------------------------------
 %% count().
diff --git a/deps/rabbit/src/rabbit_db_queue.erl b/deps/rabbit/src/rabbit_db_queue.erl
index d2fd09f872..f03a40f9c8 100644
--- a/deps/rabbit/src/rabbit_db_queue.erl
+++ b/deps/rabbit/src/rabbit_db_queue.erl
@@ -660,7 +660,7 @@ exists_in_mnesia(QName) ->
     ets:member(?MNESIA_TABLE, QName).
 
 exists_in_khepri(QName) ->
-    rabbit_khepri:exists(khepri_queue_path(QName)).
+    ets:member(rabbit_khepri_queue, QName).
 
 %% -------------------------------------------------------------------
 %% exists().
diff --git a/deps/rabbit/src/rabbit_db_user.erl b/deps/rabbit/src/rabbit_db_user.erl
index 41ec8df7b6..159a8c590f 100644
--- a/deps/rabbit/src/rabbit_db_user.erl
+++ b/deps/rabbit/src/rabbit_db_user.erl
@@ -157,10 +157,9 @@ get_in_mnesia(Username) ->
     end.
 
 get_in_khepri(Username) ->
-    Path = khepri_user_path(Username),
-    case rabbit_khepri:get(Path) of
-        {ok, User} -> User;
-        _          -> undefined
+    case ets:lookup(rabbit_khepri_users, Username) of
+        [User] -> {ok, User};
+        _      -> {error, not_found}
     end.
 
 %% -------------------------------------------------------------------
@@ -263,10 +262,11 @@ get_user_permissions_in_mnesia(Username, VHostName) ->
     end.
 
 get_user_permissions_in_khepri(Username, VHostName) ->
-    Path = khepri_user_permission_path(Username, VHostName),
-    case rabbit_khepri:get(Path) of
-        {ok, UserPermission} -> UserPermission;
-        _ -> undefined
+    UserVHost = #user_vhost{username     = Username,
+                            virtual_host = VHostName},
+    case ets:lookup(rabbit_khepri_user_permissions, UserVHost) of
+        [UserPermission] -> UserPermission;
+        _      -> undefined
     end.
 
 %% -------------------------------------------------------------------
diff --git a/deps/rabbit/src/rabbit_db_vhost.erl b/deps/rabbit/src/rabbit_db_vhost.erl
index ade8447747..64b84e7d0e 100644
--- a/deps/rabbit/src/rabbit_db_vhost.erl
+++ b/deps/rabbit/src/rabbit_db_vhost.erl
@@ -226,8 +226,7 @@ exists_in_mnesia(VHostName) ->
     mnesia:dirty_read({?MNESIA_TABLE, VHostName}) /= [].
 
 exists_in_khepri(VHostName) ->
-    Path = khepri_vhost_path(VHostName),
-    rabbit_khepri:exists(Path).
+    ets:member(rabbit_khepri_vhost, VHostName).
 
 %% -------------------------------------------------------------------
 %% get().
@@ -255,10 +254,9 @@ get_in_mnesia(VHostName) ->
     end.
 
 get_in_khepri(VHostName) ->
-    Path = khepri_vhost_path(VHostName),
-    case rabbit_khepri:get(Path) of
-        {ok, Record} -> Record;
-        _            -> undefined
+    case ets:lookup(rabbit_khepri_vhost, VHostName) of
+        [Record] -> Record;
+        _        -> undefined
     end.
 
 %% -------------------------------------------------------------------
@@ -282,11 +280,7 @@ get_all_in_mnesia() ->
     mnesia:dirty_match_object(?MNESIA_TABLE, vhost:pattern_match_all()).
 
 get_all_in_khepri() ->
-    Path = khepri_vhosts_path(),
-    case rabbit_khepri:list(Path) of
-        {ok, VHosts} -> maps:values(VHosts);
-        _            -> []
-    end.
+    ets:tab2list(rabbit_khepri_vhost).
 
 %% -------------------------------------------------------------------
 %% list().
@@ -309,11 +303,7 @@ list_in_mnesia() ->
     mnesia:dirty_all_keys(?MNESIA_TABLE).
 
 list_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']}]).
 
 %% -------------------------------------------------------------------
 %% update_in_*tx().
diff --git a/deps/rabbit/src/rabbit_exchange_type_direct.erl b/deps/rabbit/src/rabbit_exchange_type_direct.erl
index 11c5fed22b..684aa16b5d 100644
--- a/deps/rabbit/src/rabbit_exchange_type_direct.erl
+++ b/deps/rabbit/src/rabbit_exchange_type_direct.erl
@@ -38,7 +38,7 @@ route(#exchange{name = Name, type = Type},
             rabbit_db_binding:match_routing_key(Name, Routes, true);
         _ ->
             rabbit_db_binding:match_routing_key(Name, Routes, false)
-    end.
+        end.
 
 validate(_X) -> ok.
 validate_binding(_X, _B) -> ok.
diff --git a/deps/rabbit/src/rabbit_khepri.erl b/deps/rabbit/src/rabbit_khepri.erl
index a7973b05cf..fb1a4ea8cd 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,
@@ -105,6 +106,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}),
@@ -586,3 +588,137 @@ use_khepri() ->
     %rabbit_log:notice("~s: ~p", [?FUNCTION_NAME, Ret]),
     Ret.
 -endif.
+
+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],
+    [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_db_exchange,
+                   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_db_queue,
+                   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_db_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_db_user,
+                   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_db_user,
+                   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, Binding) ->
+                     Binding
+             end,
+    ProjectionFun = projection_fun_for_sets(MapFun),
+    Options = #{type => bag, keypos => #binding.source},
+    Projection = khepri_projection:new(
+                   rabbit_khepri_bindings, ProjectionFun, Options),
+    PathPattern = [rabbit_db_binding,
+                   routes,
+                   _VHost = ?KHEPRI_WILDCARD_STAR,
+                   _ExchangeName = ?KHEPRI_WILDCARD_STAR,
+                   _Kind = ?KHEPRI_WILDCARD_STAR,
+                   _DstName = ?KHEPRI_WILDCARD_STAR,
+                   _RoutingKey = ?KHEPRI_WILDCARD_STAR],
+    khepri:register_projection(?RA_CLUSTER_NAME, PathPattern, Projection).
+
+register_rabbit_index_route_projection() ->
+    MapFun = fun(Path, _) ->
+                     [rabbit_db_binding, routes, VHost, ExchangeName, Kind, DstName,
+                      RoutingKey] = Path,
+                     Exchange = rabbit_misc:r(VHost, exchange, ExchangeName),
+                     Destination = rabbit_misc:r(VHost, Kind, DstName),
+                     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_db_binding,
+                   routes,
+                   _VHost = ?KHEPRI_WILDCARD_STAR,
+                   _ExchangeName = ?KHEPRI_WILDCARD_STAR,
+                   _Kind = ?KHEPRI_WILDCARD_STAR,
+                   _DstName = ?KHEPRI_WILDCARD_STAR,
+                   _RoutingKey = ?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, NewPayload0) ->
+            NewPayload = maps:get(data, NewPayload0, sets:new()),
+            ets:insert(Table, [MapFun(Path, Element) ||
+                               Element <- sets:to_list(NewPayload)]);
+
+        (Table, Path, OldPayload0, undefined) ->
+            OldPayload = maps:get(data, OldPayload0, sets:new()),
+            sets:fold(
+              fun(Element, _Acc) ->
+                      ets:delete_object(Table, MapFun(Path, Element))
+              end, [], OldPayload);
+        (Table, Path, OldPayload0, NewPayload0) ->
+            NewPayload = maps:get(data, NewPayload0, sets:new()),
+            OldPayload = maps:get(data, OldPayload0, sets:new()),
+            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.
diff --git a/deps/rabbit/src/vhost.erl b/deps/rabbit/src/vhost.erl
index 635e0d14e4..d6ce7e44a0 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,
@@ -53,7 +54,7 @@
 
 -record(vhost, {
     %% name as a binary
-    virtual_host :: name() | '_',
+    virtual_host :: name() | '_' | '$1',
     %% proplist of limits configured, if any
     limits :: limits() | '_',
     metadata :: metadata() | '_'
@@ -67,7 +68,7 @@
 
 -type vhost_pattern() :: vhost_v2_pattern().
 -type vhost_v2_pattern() :: #vhost{
-                                  virtual_host :: name() | '_',
+                                  virtual_host :: name() | '_' | '$1',
                                   limits :: '_',
                                   metadata :: '_'
                                  }.
@@ -128,6 +129,10 @@ info_keys() ->
 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.