rabbit_db_cluster: Move forget_node() generic steps from rabbit_mnesia

[Why]

`rabbit_mnesia` indirectly checks that `rabbit` is stopped on the remote
node because `mnesia:del_table_copy()` requires that Mnesia is stopped
to delete the schema. However, this is not specific to Mnesia and we
want `rabbit` to be stopped when we use Khepri in the future.

Then, after a successful `forget_node()`, `rabbit_mnesia` would remove
some queues and emit an event. Again, this is not specific to Mnesia and
we want that to be handled by `rabbit_db_cluster`.

[How]
We use `rabbit:is_running(Node)` to query the status of RabbitMQ on the
remote node to forget. This is not atomic so there is a small chance
that RabbitMQ is restarted between the check and the actual forget.

The rest was simply moved from `rabbit_mnesia` to `rabbit_db_cluster`.

Author: dumbbell

deps/rabbit/src/rabbit_db_cluster.erl

@@ -109,9 +109,27 @@ join_using_mnesia(ClusterNodes, NodeType) when is_list(ClusterNodes) ->
 %% @doc Removes `Node' from the cluster.
 
 forget_member(Node, RemoveWhenOffline) ->
-    rabbit_db:run(
-      #{mnesia =>
-        fun() -> forget_member_using_mnesia(Node, RemoveWhenOffline) end}).
+    IsRunning = rabbit:is_running(Node),
+    case rabbit:is_running(Node) of
+        false ->
+            Ret = rabbit_db:run(
+                    #{mnesia => fun() ->
+                                        forget_member_using_mnesia(
+                                          Node, RemoveWhenOffline)
+                                end
+                     }),
+            case Ret of
+                ok when IsRunning andalso not RemoveWhenOffline ->
+                    rabbit_amqqueue:forget_all_durable(Node),
+                    rabbit_node_monitor:notify_left_cluster(Node),
+                    ok;
+                _ ->
+                    ok
+            end,
+            Ret;
+        true ->
+            {error, {failed_to_remove_node, Node, rabbit_still_running}}
+    end.
 
 forget_member_using_mnesia(Node, RemoveWhenOffline) ->
     rabbit_mnesia:forget_cluster_node(Node, RemoveWhenOffline).

deps/rabbit/src/rabbit_mnesia.erl

@@ -891,8 +891,6 @@ remove_node_if_mnesia_running(Node) ->
             %% change being propagated to all nodes
             case mnesia:del_table_copy(schema, Node) of
                 {atomic, ok} ->
-                    rabbit_amqqueue:forget_all_durable(Node),
-                    rabbit_node_monitor:notify_left_cluster(Node),
                     ok;
                 {aborted, Reason} ->
                     {error, {failed_to_remove_node, Node, Reason}}

deps/rabbitmq_cli/lib/rabbitmq/cli/ctl/commands/forget_cluster_node_command.ex

@@ -58,6 +58,10 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ForgetClusterNodeCommand do
       end
 
     case ret do
+      {:error, {:failed_to_remove_node, ^atom_name, :rabbit_still_running}} ->
+        {:error,
+         "RabbitMQ on node #{node_to_remove} must be stopped with 'rabbitmqctl -n #{node_to_remove} stop_app' before it can be removed"}
+
       {:error, {:failed_to_remove_node, ^atom_name, {:active, _, _}}} ->
         {:error,
          "RabbitMQ on node #{node_to_remove} must be stopped with 'rabbitmqctl -n #{node_to_remove} stop_app' before it can be removed"}