rabbit_db_*: Wrap ets calls to projections in whereis/1 checks

Projections might not be available in a mixed-version scenario where a
cluster has nodes which are all blank/uninitialized and the majority
of nodes run a version of Khepri with a new machine version while the
minority does not have the new machine version's code.

In this case, the cluster's effective machine version will be set to
the newer version as the majority of members have access to the new
code. The older version members will be unable to apply commands
including the `register_projection` commands that set up these ETS
tables. When these ETS tables don't exist, calls like `ets:tab2list/1`
or `ets:lookup/2` cause `badarg` errors.

We use default empty values when `ets:whereis/1` returns `undefined` for
a projection table name. Instead we could use local queries or leader
queries. Writing equivalent queries is a fair amount more work and the
code would be hard to test. `ets:whereis/1` should only return
`undefined` in the above scenario which should only be a problem in
our mixed-version testing - not in practice.

Author: the-mikedavis

deps/rabbit/src/rabbit_db_binding.erl

@@ -53,6 +53,8 @@
 -define(MNESIA_SEMI_DURABLE_TABLE, rabbit_semi_durable_route).
 -define(MNESIA_REVERSE_TABLE, rabbit_reverse_route).
 -define(MNESIA_INDEX_TABLE, rabbit_index_route).
+-define(KHEPRI_BINDINGS_PROJECTION, rabbit_khepri_bindings).
+-define(KHEPRI_INDEX_ROUTE_PROJECTION, rabbit_khepri_index_route).
 
 %% -------------------------------------------------------------------
 %% exists().
@@ -411,7 +413,12 @@ get_all_in_mnesia() ->
       end).
 
 get_all_in_khepri() ->
-     [B || #route{binding = B} <- ets:tab2list(rabbit_khepri_bindings)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            [B || #route{binding = B} <- ets:tab2list(Table)]
+    end.
 
 -spec get_all(VHostName) -> [Binding] when
       VHostName :: vhost:name(),
@@ -437,11 +444,16 @@ get_all_in_mnesia(VHost) ->
     [B || #route{binding = B} <- rabbit_db:list_in_mnesia(?MNESIA_TABLE, Match)].
 
 get_all_in_khepri(VHost) ->
-    VHostResource = rabbit_misc:r(VHost, '_'),
-    Match = #route{binding = #binding{source      = VHostResource,
-                                      destination = VHostResource,
-                                      _           = '_'}},
-    [B || #route{binding = B} <- ets:match_object(rabbit_khepri_bindings, Match)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            VHostResource = rabbit_misc:r(VHost, '_'),
+            Match = #route{binding = #binding{source      = VHostResource,
+                                              destination = VHostResource,
+                                              _           = '_'}},
+            [B || #route{binding = B} <- ets:match_object(Table, Match)]
+    end.
 
 -spec get_all(Src, Dst, Reverse) -> [Binding] when
       Src :: rabbit_types:binding_source(),
@@ -469,10 +481,15 @@ get_all_in_mnesia(SrcName, DstName, Reverse) ->
     mnesia:async_dirty(Fun).
 
 get_all_in_khepri(SrcName, DstName) ->
-    MatchHead = #route{binding = #binding{source      = SrcName,
-                                          destination = DstName,
-                                          _           = '_'}},
-    [B || #route{binding = B} <- ets:match_object(rabbit_khepri_bindings, MatchHead)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            MatchHead = #route{binding = #binding{source      = SrcName,
+                                                  destination = DstName,
+                                                  _           = '_'}},
+            [B || #route{binding = B} <- ets:match_object(Table, MatchHead)]
+    end.
 
 %% -------------------------------------------------------------------
 %% get_all_for_source().
@@ -511,8 +528,13 @@ list_for_route(Route, true) ->
     end.
 
 get_all_for_source_in_khepri(Resource) ->
-    Route = #route{binding = #binding{source = Resource, _ = '_'}},
-    [B || #route{binding = B} <- ets:match_object(rabbit_khepri_bindings, Route)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            Route = #route{binding = #binding{source = Resource, _ = '_'}},
+            [B || #route{binding = B} <- ets:match_object(Table, Route)]
+    end.
 
 %% -------------------------------------------------------------------
 %% get_all_for_destination().
@@ -541,9 +563,14 @@ get_all_for_destination_in_mnesia(Dst) ->
     mnesia:async_dirty(Fun).
 
 get_all_for_destination_in_khepri(Destination) ->
-    Match = #route{binding = #binding{destination = Destination,
-                                      _           = '_'}},
-    [B || #route{binding = B} <- ets:match_object(rabbit_khepri_bindings, Match)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            Match = #route{binding = #binding{destination = Destination,
+                                              _           = '_'}},
+            [B || #route{binding = B} <- ets:match_object(Table, Match)]
+    end.
 
 %% -------------------------------------------------------------------
 %% fold().
@@ -617,11 +644,16 @@ match_in_mnesia(SrcName, Match) ->
                  Routes, Match(Binding)].
 
 match_in_khepri(SrcName, Match) ->
-    MatchHead = #route{binding = #binding{source      = SrcName,
-                                          _           = '_'}},
-    Routes = ets:select(rabbit_khepri_bindings, [{MatchHead, [], [['$_']]}]),
-    [Dest || [#route{binding = Binding = #binding{destination = Dest}}] <-
-                 Routes, Match(Binding)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            MatchHead = #route{binding = #binding{source      = SrcName,
+                                                  _           = '_'}},
+            Routes = ets:select(Table, [{MatchHead, [], [['$_']]}]),
+            [Dest || [#route{binding = Binding = #binding{destination = Dest}}] <-
+                         Routes, Match(Binding)]
+    end.
 
 %% Routing - HOT CODE PATH
 %% -------------------------------------------------------------------
@@ -654,18 +686,26 @@ match_routing_key_in_mnesia(SrcName, RoutingKeys, UseIndex) ->
             route_in_mnesia_v1(SrcName, RoutingKeys)
     end.
 
-match_routing_key_in_khepri(Src, ['_']) ->
+match_routing_key_in_khepri(Src, RoutingKeys) ->
+    case ets:whereis(?KHEPRI_INDEX_ROUTE_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            do_match_routing_key_in_khepri(Table, Src, RoutingKeys)
+    end.
+
+do_match_routing_key_in_khepri(Table, Src, ['_']) ->
     MatchHead = #index_route{source_key  = {Src, '_'},
                              destination = '$1',
                              _           = '_'},
-    ets:select(rabbit_khepri_index_route, [{MatchHead, [], ['$1']}]);
+    ets:select(Table, [{MatchHead, [], ['$1']}]);
 
-match_routing_key_in_khepri(Src, RoutingKeys) ->
+do_match_routing_key_in_khepri(Table, Src, RoutingKeys) ->
     lists:foldl(
       fun(RK, Acc) ->
               try
                   Dst = ets:lookup_element(
-                          rabbit_khepri_index_route,
+                          Table,
                           {Src, RK},
                           #index_route.destination),
                   Dst ++ Acc

deps/rabbit/src/rabbit_db_exchange.erl

@@ -57,6 +57,7 @@
 -define(MNESIA_TABLE, rabbit_exchange).
 -define(MNESIA_DURABLE_TABLE, rabbit_durable_exchange).
 -define(MNESIA_SERIAL_TABLE, rabbit_exchange_serial).
+-define(KHEPRI_PROJECTION, rabbit_khepri_exchange).
 
 %% -------------------------------------------------------------------
 %% get_all().
@@ -182,9 +183,14 @@ get_in_mnesia(Name) ->
     rabbit_mnesia:dirty_read({?MNESIA_TABLE, Name}).
 
 get_in_khepri(Name) ->
-    case ets:lookup(rabbit_khepri_exchange, Name) of
-        [X] -> {ok, X};
-        []  -> {error, not_found}
+    case ets:whereis(?KHEPRI_PROJECTION) of
+        undefined ->
+            {error, not_found};
+        Table ->
+            case ets:lookup(Table, Name) of
+                [X] -> {ok, X};
+                []  -> {error, not_found}
+            end
     end.
 
 %% -------------------------------------------------------------------
@@ -227,7 +233,12 @@ 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:append([ets:lookup(rabbit_khepri_exchange, Name) || Name <- Names]).
+    case ets:whereis(?KHEPRI_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            lists:append([ets:lookup(Table, Name) || Name <- Names])
+    end.
 
 %% -------------------------------------------------------------------
 %% count().