rabbit_khepri: Implement proper is_virgin_node/0 detection

[Why]
The previous implementation checked if the store was empty. This part is
unchanged. However if `is_empty/0` returned an error, typically because
the store was stopped, it would consider that the node is also virgin.

This is a wrong approximation because an admin could have executed
`rabbitmqctl stop_app` before something checks the node is virgin.

[How]
With this patch, we introduce `rabbit_khepri:is_virgin_node/0`. It will
start the store if it is stopped, then it will query if it is empty. It
will stop the store again if it was stopped initially.

This way, we have a more accurate answer to the question.

Author: dumbbell

deps/rabbit/src/rabbit_db.erl

@@ -231,10 +231,7 @@ is_virgin_node_using_mnesia() ->
     rabbit_mnesia:is_virgin_node().
 
 is_virgin_node_using_khepri() ->
-    case rabbit_khepri:is_empty() of
-        {error, _} -> true;
-        IsEmpty    -> IsEmpty
-    end.
+    rabbit_khepri:is_virgin_node().
 
 -spec is_virgin_node(Node) -> IsVirgin | undefined when
       Node :: node(),

deps/rabbit/src/rabbit_khepri.erl

@@ -104,6 +104,7 @@
          setup/1,
          init/1,
          reset/0,
+         is_virgin_node/0,
 
          dir/0,
          get_ra_cluster_name/0,
@@ -297,10 +298,16 @@ setup(_Context) ->
             exit(Error)
     end.
 
+is_ra_system_running() ->
+    rabbit_ra_systems:is_running(?RA_SYSTEM).
+
 ensure_ra_system_started() ->
     {ok, _} = application:ensure_all_started(khepri),
     ok = rabbit_ra_systems:ensure_ra_system_started(?RA_SYSTEM).
 
+ensure_ra_system_stopped() ->
+    ok = rabbit_ra_systems:ensure_ra_system_stopped(?RA_SYSTEM).
+
 retry_timeout() ->
     case application:get_env(rabbit, khepri_leader_wait_retry_timeout) of
         {ok, T} when is_integer(T) andalso T >= 0 -> T;
@@ -385,6 +392,30 @@ reset() ->
             throw({error, rabbitmq_unexpectedly_running})
     end.
 
+is_virgin_node() ->
+    IsSystemRunning = is_ra_system_running(),
+    IsStoreRunning = khepri_cluster:is_store_running(?STORE_ID),
+    case IsSystemRunning of
+        true  -> ok;
+        false -> ensure_ra_system_started()
+    end,
+    case IsStoreRunning of
+        true  -> ok;
+        false -> setup()
+    end,
+
+    IsEmpty = is_empty() =:= true,
+
+    case IsStoreRunning of
+        true  -> ok;
+        false -> _ = khepri:stop(?RA_CLUSTER_NAME)
+    end,
+    case IsSystemRunning of
+        true  -> ok;
+        false -> ensure_ra_system_stopped()
+    end,
+    IsEmpty.
+
 -spec dir() -> Dir when
       Dir :: file:filename_all().
 %% @doc Returns the Khepri store directory.

deps/rabbit/src/rabbit_ra_systems.erl

@@ -14,6 +14,7 @@
 -export([setup/0,
          setup/1,
          all_ra_systems/0,
+         is_running/1,
          are_running/0,
          ensure_ra_system_started/1,
          ensure_ra_system_stopped/1,
@@ -46,6 +47,20 @@ all_ra_systems() ->
     [coordination,
      quorum_queues].
 
+-spec is_running(RaSystem) -> IsRunning when
+      RaSystem :: ra_system_name(),
+      IsRunning :: boolean().
+
+is_running(RaSystem) ->
+    try
+        %% FIXME: We hard-code the name of an internal Ra process here.
+        Children = supervisor:which_children(ra_systems_sup),
+        is_ra_system_running(Children, RaSystem)
+    catch
+        exit:{noproc, _} ->
+            false
+    end.
+
 -spec are_running() -> AreRunning when
       AreRunning :: boolean().