Make disc node reclustering work, various cleanups.

2 files changed, 53 insertions, 65 deletions

diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index f1e007a1d6..e5929f86b6 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -35,7 +35,7 @@
 -export([ensure_mnesia_dir/0, dir/0, status/0, init/0, is_db_empty/0,
          cluster/1, force_cluster/1, reset/0, force_reset/0,
          is_clustered/0, running_clustered_nodes/0, all_clustered_nodes/0,
-         forget_other_nodes/0, empty_ram_only_tables/0, copy_db/1]).
+         empty_ram_only_tables/0, copy_db/1]).
 
 -export([table_names/0]).
 
@@ -66,7 +66,6 @@
 -spec(is_clustered/0 :: () -> boolean()).
 -spec(running_clustered_nodes/0 :: () -> [node()]).
 -spec(all_clustered_nodes/0 :: () -> [node()]).
--spec(forget_other_nodes/0 :: () -> 'ok').
 -spec(empty_ram_only_tables/0 :: () -> 'ok').
 -spec(create_tables/0 :: () -> 'ok').
 -spec(copy_db/1 :: (file:filename()) ->  rabbit_types:ok_or_error(any())).
@@ -127,8 +126,8 @@ cluster(ClusterNodes, Force) ->
 %% return node to its virgin state, where it is not member of any
 %% cluster, has no cluster configuration, no local database, and no
 %% persisted messages
-reset()       -> reset(all).
-force_reset() -> reset(force_all).
+reset()       -> reset(false).
+force_reset() -> reset(true).
 
 is_clustered() ->
     RunningNodes = running_clustered_nodes(),
@@ -388,10 +387,11 @@ init_db(ClusterNodes, Force) ->
             end,
             case {Nodes, mnesia:system_info(use_dir)} of
                 {[], true} ->
-                    %% True single disc node, or master" (i.e. without
+                    %% True single disc node, or "master" (i.e. without
                     %% config) disc node in cluster, attempt upgrade
                     ok = wait_for_tables(),
-                    case rabbit_upgrade:maybe_upgrade([mnesia, local]) of
+                    case rabbit_upgrade:maybe_upgrade(
+                           [mnesia, local], fun forget_other_nodes/0) of
                         ok                    -> ensure_schema_ok();
                         version_not_available -> schema_ok_or_move()
                     end;
@@ -400,37 +400,27 @@ init_db(ClusterNodes, Force) ->
                     ok = create_schema();
                 {[AnotherNode|_], _} ->
                     %% Subsequent node in cluster, catch up
-                    IsDiskNode = ClusterNodes == [] orelse
-                        lists:member(node(), ClusterNodes),
-                    case IsDiskNode of
-                        true ->
-                            %% TODO test this branch ;)
-                            %% TODO don't just reset every time we start up!
-                            mnesia:stop(),
-                            reset(mnesia),
-                            mnesia:start(),
-                            %% TODO what should we ensure?
-                            %% ensure_version_ok(rabbit_upgrade:read_version()),
-                            %% ensure_version_ok(
-                            %%   rpc:call(AnotherNode, rabbit_upgrade, read_version, [])),
-                            %% TODO needed?
-                            ok = wait_for_replicated_tables(),
-                            ok = create_local_table_copy(schema, disc_copies),
-                            ok = create_local_table_copies(disc);
-                        false ->
-                            ok = wait_for_replicated_tables(),
-                            %% TODO can we live without this on disc?
-                            ok = create_local_table_copy(schema, disc_copies),
-                            ok = create_local_table_copies(ram),
-                            case rabbit_upgrade:maybe_upgrade([local]) of
-                                ok ->
-                                    ok;
-                                %% If we're just starting up a new node
-                                %% we won't have a version
-                                version_not_available ->
-                                    ok = rabbit_upgrade:write_version()
-                            end
+                    %% TODO what should we ensure?
+                    %% ensure_version_ok(rabbit_upgrade:read_version()),
+                    %% ensure_version_ok(
+                    %%   rpc:call(AnotherNode, rabbit_upgrade, read_version, [])),
+                    Type = case ClusterNodes == [] orelse
+                               lists:member(node(), ClusterNodes) of
+                               true  -> disc;
+                               false -> ram
+                           end,
+                    case rabbit_upgrade:maybe_upgrade(
+                           [local], reset_fun(ProperClusterNodes)) of
+                        ok ->
+                            ok;
+                        %% If we're just starting up a new node
+                        %% we won't have a version
+                        version_not_available ->
+                            ok = rabbit_upgrade:write_version()
                     end,
+                    ok = wait_for_replicated_tables(),
+                    ok = create_local_table_copy(schema, disc_copies),
+                    ok = create_local_table_copies(Type),
                     ensure_schema_ok()
             end;
         {error, Reason} ->
@@ -470,6 +460,16 @@ ensure_schema_ok() ->
         {error, Reason} -> throw({error, {schema_invalid, Reason}})
     end.
 
+reset_fun(ProperClusterNodes) ->
+    fun() ->
+            mnesia:stop(),
+            rabbit_misc:ensure_ok(mnesia:delete_schema([node()]),
+                                  cannot_delete_schema),
+            rabbit_misc:ensure_ok(mnesia:start(),
+                                  cannot_start_mnesia),
+            {ok, _} = mnesia:change_config(extra_db_nodes, ProperClusterNodes)
+    end.
+
 create_schema() ->
     mnesia:stop(),
     rabbit_misc:ensure_ok(mnesia:create_schema([node()]),
@@ -583,15 +583,12 @@ wait_for_tables(TableNames) ->
             throw({error, {failed_waiting_for_tables, Reason}})
     end.
 
-%% Mode: force_all - get rid of everything unconditionally
-%%       all       - get rid of everything, conditional on Mnesia working
-%%       mnesia    - just get rid of Mnesia, leave everything else
-reset(Mode) ->
+reset(Force) ->
     ok = ensure_mnesia_not_running(),
     Node = node(),
-    case Mode of
-        force_all -> ok;
-        _ ->
+    case Force of
+        true  -> ok;
+        false ->
             ok = ensure_mnesia_dir(),
             rabbit_misc:ensure_ok(mnesia:start(), cannot_start_mnesia),
             {Nodes, RunningNodes} =
@@ -606,14 +603,9 @@ reset(Mode) ->
             rabbit_misc:ensure_ok(mnesia:delete_schema([Node]),
                                   cannot_delete_schema)
     end,
-    case Mode of
-        mnesia ->
-            ok;
-        _ ->
-            ok = delete_cluster_nodes_config(),
-            %% remove persisted messages and any other garbage we find
-            ok = rabbit_misc:recursive_delete(filelib:wildcard(dir() ++ "/*"))
-    end,
+    ok = delete_cluster_nodes_config(),
+    %% remove persisted messages and any other garbage we find
+    ok = rabbit_misc:recursive_delete(filelib:wildcard(dir() ++ "/*")),
     ok.
 
 leave_cluster([], _) -> ok;
diff --git a/src/rabbit_upgrade.erl b/src/rabbit_upgrade.erl
index dee08f48df..7e59faaf43 100644
--- a/src/rabbit_upgrade.erl
+++ b/src/rabbit_upgrade.erl
@@ -21,7 +21,7 @@
 
 -module(rabbit_upgrade).
 
--export([maybe_upgrade/1, read_version/0, write_version/0, desired_version/0]).
+-export([maybe_upgrade/2, read_version/0, write_version/0, desired_version/0]).
 
 -include("rabbit.hrl").
 
@@ -36,7 +36,8 @@
 -type(scope() :: 'mnesia' | 'local').
 -type(version() :: [step()]).
 
--spec(maybe_upgrade/1 :: ([scope()]) -> 'ok' | 'version_not_available').
+%% TODO update
+%%-spec(maybe_upgrade/1 :: ([scope()]) -> 'ok' | 'version_not_available').
 -spec(read_version/0 :: () -> rabbit_types:ok_or_error2(version(), any())).
 -spec(write_version/0 :: () -> 'ok').
 -spec(desired_version/0 :: () -> version()).
@@ -48,23 +49,21 @@
 %% Try to upgrade the schema. If no information on the existing schema
 %% could be found, do nothing. rabbit_mnesia:check_schema_integrity()
 %% will catch the problem.
-maybe_upgrade(Scopes) ->
+maybe_upgrade(Scopes, Fun) ->
     case read_version() of
         {ok, CurrentHeads} ->
             with_upgrade_graph(
-              fun (G) -> maybe_upgrade_graph(CurrentHeads, Scopes, G) end);
+              fun (G) -> maybe_upgrade_graph(CurrentHeads, Scopes, Fun, G) end);
         {error, enoent} ->
             version_not_available
     end.
 
-maybe_upgrade_graph(CurrentHeads, Scopes, G) ->
+maybe_upgrade_graph(CurrentHeads, Scopes, Fun, G) ->
     case unknown_heads(CurrentHeads, G) of
         [] ->
             case upgrades_to_apply(CurrentHeads, Scopes, G) of
-                [] ->
-                    ok;
-                Upgrades ->
-                    apply_upgrades(Upgrades, lists:member(mnesia, Scopes))
+                []       -> ok;
+                Upgrades -> apply_upgrades(Upgrades, Fun)
             end;
         Unknown ->
             throw({error, {future_upgrades_found, Unknown}})
@@ -133,7 +132,7 @@ heads(G) ->
 
 %% -------------------------------------------------------------------
 
-apply_upgrades(Upgrades, ForgetOthers) ->
+apply_upgrades(Upgrades, Fun) ->
     LockFile = lock_filename(dir()),
     case rabbit_misc:lock_file(LockFile) of
         ok ->
@@ -148,10 +147,7 @@ apply_upgrades(Upgrades, ForgetOthers) ->
                     %% is not intuitive. Remove it.
                     ok = file:delete(lock_filename(BackupDir)),
                     info("Upgrades: Mnesia dir backed up to ~p~n", [BackupDir]),
-                    case ForgetOthers of
-                        true -> rabbit_mnesia:forget_other_nodes();
-                        _    -> ok
-                    end,
+                    Fun(),
                     [apply_upgrade(Upgrade) || Upgrade <- Upgrades],
                     info("Upgrades: All upgrades applied successfully~n", []),
                     ok = write_version(),