reset

Author: dumbbell

deps/rabbit/src/rabbit_db.erl

@@ -129,6 +129,7 @@ clear_init_finished() ->
 %% @doc Resets the database and the node.
 
 reset() ->
+    ok = rabbit:stop(),
     ok = case rabbit_khepri:is_enabled() of
              true  -> reset_using_khepri();
              false -> reset_using_mnesia()
@@ -198,6 +199,8 @@ force_load_on_next_boot_using_mnesia() ->
 post_reset() ->
     rabbit_feature_flags:reset(),
 
+    wipe_data_dir(),
+
     %% The cluster status files that RabbitMQ uses when Mnesia is the database
     %% are initially created from rabbit_prelaunch_cluster. However, it will
     %% only be done once the `rabbit` app is restarted. Meanwhile, they are
@@ -210,6 +213,17 @@ post_reset() ->
 
     ok.
 
+wipe_data_dir() ->
+    DataDir = dir(),
+    Glob = filename:join(DataDir, "*"),
+    FilesToRemove = filelib:wildcard(Glob),
+    ?LOG_DEBUG(
+       "DB: wipe files in data directory `~ts`:~p",
+       [DataDir, FilesToRemove],
+       #{domain => ?RMQLOG_DOMAIN_DB}),
+    ok = rabbit_file:recursive_delete(FilesToRemove),
+    ok.
+
 %% -------------------------------------------------------------------
 %% is_virgin_node().
 %% -------------------------------------------------------------------

deps/rabbit/src/rabbit_db_cluster.erl

@@ -379,19 +379,42 @@ forget_member_locked(Node, RemoveWhenOffline)
         ok -> post_forget_member_locked(Node, RemoveWhenOffline);
         _  -> ok
     end,
-    Ret.
+    Ret;
+forget_member_locked(Node, RemoveWhenOffline)
+  when is_atom(Node) andalso Node =:= node() ->
+    OtherNodes = members() -- [Node],
+    forget_member_locked_remotely(OtherNodes, Node, RemoveWhenOffline).
 
-post_forget_member_locked(Node, false = _RemoveWhenOffline) ->
-    ?LOG_DEBUG(
-       "DB: removing node `~s` from various Ra clusters", [Node],
-       #{domain => ?RMQLOG_DOMAIN_DB}),
-    _ = rabbit_amqqueue:forget_all(Node),
-    _ = rabbit_quorum_queue:shrink_all(Node),
-    _ = rabbit_stream_queue:delete_all_replicas(Node),
-    _ = rabbit_stream_coordinator:forget_node(Node),
-    rabbit_node_monitor:notify_left_cluster(Node),
-    ok;
-post_forget_member_locked(_Node, true = _RemoveWhenOffline) ->
+forget_member_locked_remotely([OtherNode | Rest], Node, RemoveWhenOffline) ->
+    try
+        ?LOG_DEBUG(
+           "DB: removing cluster member `~ts` (this node); doing it from "
+           "remote node `~s`",
+           [Node, OtherNode],
+           #{domain => ?RMQLOG_DOMAIN_DB}),
+        Ret = erpc:call(
+                OtherNode,
+                ?MODULE, forget_member_locked, [Node, RemoveWhenOffline]),
+        case Ret of
+            ok  ->
+                ok;
+            Error ->
+                ?LOG_DEBUG(
+                   "DB: failed to remove cluster member `~ts` from node "
+                   "`~s`: ~0p",
+                   [Node, OtherNode, Error],
+                   #{domain => ?RMQLOG_DOMAIN_DB}),
+                forget_member_locked_remotely(Rest, Node, RemoveWhenOffline)
+        end
+    catch
+        _:Reason ->
+            ?LOG_DEBUG(
+               "DB: failed to remove cluster member `~ts` from node `~s`: ~0p",
+               [Node, OtherNode, Reason],
+               #{domain => ?RMQLOG_DOMAIN_DB}),
+            forget_member_locked_remotely(Rest, Node, RemoveWhenOffline)
+    end;
+forget_member_locked_remotely([], _Node, _RemoveWhenOffline) ->
     ok.
 
 forget_member_using_mnesia(Node, RemoveWhenOffline) ->
@@ -406,6 +429,19 @@ forget_member_using_khepri(_Node, true) ->
 forget_member_using_khepri(Node, false = _RemoveWhenOffline) ->
     rabbit_khepri:remove_member(Node).
 
+post_forget_member_locked(Node, false = _RemoveWhenOffline) ->
+    ?LOG_DEBUG(
+       "DB: removing node `~s` from various Ra clusters", [Node],
+       #{domain => ?RMQLOG_DOMAIN_DB}),
+    _ = rabbit_amqqueue:forget_all(Node),
+    _ = rabbit_quorum_queue:shrink_all(Node),
+    _ = rabbit_stream_queue:delete_all_replicas(Node),
+    _ = rabbit_stream_coordinator:forget_node(Node),
+    rabbit_node_monitor:notify_left_cluster(Node),
+    ok;
+post_forget_member_locked(_Node, true = _RemoveWhenOffline) ->
+    ok.
+
 lock_cluster_changes(ChangingNode) ->
     RabbitWasRunning = stop_rabbit_if_running(ChangingNode),
     InitialState = #{changing_node => ChangingNode,

deps/rabbit/src/rabbit_khepri.erl

@@ -302,6 +302,9 @@ 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;
@@ -372,19 +375,14 @@ await_replication() ->
 %% @private
 
 reset() ->
-    case rabbit:is_running() of
-        false ->
-            %% Rabbit should be stopped, but Khepri needs to be running.
-            %% Restart it.
-            ok = setup(),
-            ok = khepri_cluster:reset(?RA_CLUSTER_NAME),
-            ok = khepri:stop(?RA_CLUSTER_NAME),
+    ?assertNot(rabbit:is_running()),
 
-            _ = file:delete(rabbit_guid:filename()),
-            ok;
-        true ->
-            throw({error, rabbitmq_unexpectedly_running})
-    end.
+    ok = setup(),
+    ThisNode = node(),
+    rabbit_db_cluster:forget_member(ThisNode),
+    ok = khepri:stop(?RA_CLUSTER_NAME),
+    ok = ensure_ra_system_stopped(),
+    ok.
 
 -spec dir() -> Dir when
       Dir :: file:filename_all().