rabbit_khepri: Register projections during virgin init/1

This covers a specific case where we need to register projections not
covered by the enable callback of the `khepri_db` feature flag. The
feature flag may be enabled if a node has been part of a cluster which
enabled the flag, but the metadata store might be reset. Upon init the
feature flag will be enabled but the store will be empty and the
projections will not exist, so operations like inserting default data
will fail when asserting that a vhost exists for example.

This fixes the `cluster_management_SUITE:forget_cluster_node_in_khepri/1`
case when running the suite with `RABBITMQ_METADATA_STORE=khepri`, which
fails as mentioned above.

We could run projection registration always when using Khepri but once
projections are registered the command is idempotent so there's no need
to, and the commands are somewhat large.

Author: the-mikedavis

deps/rabbit/src/rabbit_db.erl

@@ -67,8 +67,8 @@ init() ->
     end,
 
     Ret = case rabbit_khepri:is_enabled() of
-              true  -> init_using_khepri();
-              false -> init_using_mnesia()
+              true  -> init_using_khepri(IsVirgin);
+              false -> init_using_mnesia(IsVirgin)
           end,
     case Ret of
         ok ->
@@ -91,19 +91,19 @@ pre_init(IsVirgin) ->
     OtherMembers = rabbit_nodes:nodes_excl_me(Members),
     rabbit_db_cluster:ensure_feature_flags_are_in_sync(OtherMembers, IsVirgin).
 
-init_using_mnesia() ->
+init_using_mnesia(_IsVirgin) ->
     ?LOG_DEBUG(
       "DB: initialize Mnesia",
       #{domain => ?RMQLOG_DOMAIN_DB}),
     ok = rabbit_mnesia:init(),
     ?assertEqual(rabbit:data_dir(), mnesia_dir()),
     rabbit_sup:start_child(mnesia_sync).
 
-init_using_khepri() ->
+init_using_khepri(IsVirgin) ->
     ?LOG_DEBUG(
       "DB: initialize Khepri",
       #{domain => ?RMQLOG_DOMAIN_DB}),
-    rabbit_khepri:init().
+    rabbit_khepri:init(IsVirgin).
 
 init_finished() ->
     %% Used during initialisation by rabbit_logger_exchange_h.erl

deps/rabbit/src/rabbit_khepri.erl

@@ -99,7 +99,7 @@
 -export([setup/0,
          setup/1,
          register_projections/0,
-         init/0,
+         init/1,
          can_join_cluster/1,
          add_member/2,
          remove_member/1,
@@ -288,26 +288,38 @@ retry_timeout() ->
 
 %% @private
 
--spec init() -> Ret when
+-spec init(IsVirgin) -> Ret when
+      IsVirgin :: boolean(),
       Ret :: ok | timeout_error().
 
-init() ->
+init(IsVirgin) ->
     case members() of
         [] ->
             timer:sleep(1000),
-            init();
+            init(IsVirgin);
         Members ->
             ?LOG_NOTICE(
                "Found the following metadata store members: ~p", [Members],
                #{domain => ?RMQLOG_DOMAIN_DB}),
-            %% Delete transient queues on init.
-            %% Note that we also do this in the
-            %% `rabbit_amqqueue:on_node_down/1' callback. We must try this
-            %% deletion during init because the cluster may have been in a
-            %% minority when this node went down. We wait for a majority while
-            %% booting (via `rabbit_khepri:setup/0') though so this deletion is
-            %% likely to succeed.
-            rabbit_amqqueue:delete_transient_queues_on_node(node())
+            Ret = case IsVirgin of
+                      true ->
+                          register_projections();
+                      false ->
+                          ok
+                  end,
+            case Ret of
+                ok ->
+                    %% Delete transient queues on init.
+                    %% Note that we also do this in the
+                    %% `rabbit_amqqueue:on_node_down/1' callback. We must try
+                    %% this deletion during init because the cluster may have
+                    %% been in a minority when this node went down. We wait for
+                    %% a majority while registering projections above
+                    %% though so this deletion is likely to succeed.
+                    rabbit_amqqueue:delete_transient_queues_on_node(node());
+                {error, _} = Error ->
+                    Error
+            end
     end.
 
 %% @private