Holy %$*! it works. Still needs tidying up somewhat...

3 files changed, 82 insertions, 98 deletions

diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index a11347ffb0..345ca82afe 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,
-         empty_ram_only_tables/0, copy_db/1]).
+         empty_ram_only_tables/0, copy_db/1, create_cluster_nodes_config/1]).
 
 -export([table_names/0]).
 
@@ -94,7 +94,6 @@ status() ->
      {running_nodes, running_clustered_nodes()}].
 
 init() ->
-    ok = maybe_reset_for_upgrades(),
     ok = ensure_mnesia_running(),
     ok = ensure_mnesia_dir(),
     ok = init_db(read_cluster_nodes_config(), true),
@@ -399,35 +398,19 @@ init_db(ClusterNodes, Force) ->
     end.
 
 setup_existing_node(ClusterNodes, Nodes) ->
-    DiscNodes = mnesia:table_info(schema, disc_copies),
-    Node = node(),
-    case upgrader(DiscNodes) of
-        Node ->
-            %% True single disc node, or upgrader node - attempt
-            %% upgrade if necessary
+    case Nodes of
+        [] ->
+            %% We're the first node up
             ok = wait_for_tables(),
-            case rabbit_upgrade:maybe_upgrade([mnesia, local]) of
+            case rabbit_upgrade:maybe_upgrade([local]) of
                 ok                    -> ensure_schema_ok();
                 version_not_available -> schema_ok_or_move()
             end;
-        _ ->
+        [AnotherNode|_] ->
             %% Subsequent node in cluster, catch up
-            case Nodes of
-                [AnotherNode|_] ->
-                    ensure_version_ok(
-                      rpc:call(AnotherNode, rabbit_upgrade, read_version, []));
-                [] ->
-                    ok
-            end,
+            ensure_version_ok(
+              rpc:call(AnotherNode, rabbit_upgrade, read_version, [])),
             ok = wait_for_tables(),
-            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,
             IsDiskNode = ClusterNodes == [] orelse
                 lists:member(node(), ClusterNodes),
             ok = wait_for_replicated_tables(),
@@ -436,13 +419,13 @@ setup_existing_node(ClusterNodes, Nodes) ->
                                                true  -> disc;
                                                false -> ram
                                            end),
-            ensure_schema_ok(),
-            %% If we're just starting up a new node we won't have
-            %% a version
-            case rabbit_upgrade:read_version() of
-                {error, _} -> rabbit_upgrade:write_version();
-                _          -> ok
-            end
+            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,
+            ensure_schema_ok()
     end.
 
 schema_ok_or_move() ->
@@ -475,55 +458,6 @@ ensure_schema_ok() ->
         {error, Reason} -> throw({error, {schema_invalid, Reason}})
     end.
 
-maybe_reset_for_upgrades() ->
-    case rabbit_upgrade:upgrade_required([mnesia]) of
-        true ->
-            DiscNodes = all_clustered_nodes(),
-            Upgrader = upgrader(DiscNodes),
-            case node() of
-                Upgrader ->
-                    reset_for_primary_upgrade(DiscNodes);
-                _ ->
-                    reset_for_non_primary_upgrade(Upgrader, DiscNodes)
-            end;
-        false ->
-            ok
-    end.
-
-reset_for_primary_upgrade(DiscNodes) ->
-    Others = DiscNodes -- [node()],
-    ensure_mnesia_running(),
-    force_tables(),
-    [{atomic, ok} = mnesia:del_table_copy(schema, Node) || Node <- Others],
-    ok.
-
-reset_for_non_primary_upgrade(Upgrader, DiscNodes) ->
-    case node_running(Upgrader) of
-        false ->
-            exit(lists:flatten(
-                   io_lib:format(
-                     "Cluster upgrade needed. Please start node ~s first",
-                     [Upgrader])));
-        true ->
-            OtherNodes = DiscNodes -- [node()],
-            mnesia:stop(),
-            rabbit_misc:ensure_ok(mnesia:delete_schema([node()]),
-                                  cannot_delete_schema),
-            mnesia:start(),
-            {ok, _} = mnesia:change_config(extra_db_nodes, OtherNodes),
-            ok
-    end.
-
-upgrader(Nodes) ->
-    [Upgrader|_] = lists:usort(Nodes),
-    Upgrader.
-
-node_running(Node) ->
-    case rpc:call(Node, application, which_applications, []) of
-        {badrpc, _} -> false;
-        Apps        -> lists:keysearch(rabbit, 1, Apps) =/= false
-    end.
-
 create_schema() ->
     mnesia:stop(),
     rabbit_misc:ensure_ok(mnesia:create_schema([node()]),
@@ -637,9 +571,6 @@ wait_for_tables(TableNames) ->
             throw({error, {failed_waiting_for_tables, Reason}})
     end.
 
-force_tables() ->
-    [mnesia:force_load_table(T) || T <- table_names()].
-
 reset(Force) ->
     ok = ensure_mnesia_not_running(),
     Node = node(),
diff --git a/src/rabbit_prelaunch.erl b/src/rabbit_prelaunch.erl
index 8ae45abda3..c5ee63bad7 100644
--- a/src/rabbit_prelaunch.erl
+++ b/src/rabbit_prelaunch.erl
@@ -250,8 +250,8 @@ post_process_script(ScriptFile) ->
             {error, {failed_to_load_script, Reason}}
     end.
 
-process_entry(Entry = {apply,{application,start_boot,[rabbit,permanent]}}) ->
-    [{apply,{rabbit,prepare,[]}}, Entry];
+process_entry(Entry = {apply,{application,start_boot,[mnesia,permanent]}}) ->
+    [{apply,{rabbit_upgrade,maybe_upgrade_mnesia,[]}}, Entry];
 process_entry(Entry) ->
     [Entry].
 
diff --git a/src/rabbit_upgrade.erl b/src/rabbit_upgrade.erl
index 260f85a1c6..9f9e8806d3 100644
--- a/src/rabbit_upgrade.erl
+++ b/src/rabbit_upgrade.erl
@@ -21,7 +21,7 @@
 
 -module(rabbit_upgrade).
 
--export([maybe_upgrade/1, upgrade_required/1]).
+-export([maybe_upgrade_mnesia/0, maybe_upgrade/1]).
 -export([read_version/0, write_version/0, desired_version/0]).
 
 -include("rabbit.hrl").
@@ -37,8 +37,8 @@
 -type(scope() :: 'mnesia' | 'local').
 -type(version() :: [step()]).
 
+-spec(maybe_upgrade_mnesia/0 :: () -> 'ok').
 -spec(maybe_upgrade/1 :: ([scope()]) -> 'ok' | 'version_not_available').
--spec(upgrade_required/1 :: ([scope()]) -> boolean()).
 -spec(read_version/0 :: () -> rabbit_types:ok_or_error2(version(), any())).
 -spec(write_version/0 :: () -> 'ok').
 -spec(desired_version/0 :: () -> version()).
@@ -47,9 +47,69 @@
 
 %% -------------------------------------------------------------------
 
-%% 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_mnesia() ->
+    rabbit:prepare(),
+    case upgrades_required([mnesia]) of
+        Upgrades = [_|_] ->
+            DiscNodes = rabbit_mnesia:all_clustered_nodes(),
+            Upgrader = upgrader(DiscNodes),
+            case node() of
+                Upgrader ->
+                    primary_upgrade(Upgrades, DiscNodes);
+                _ ->
+                    non_primary_upgrade(Upgrader, DiscNodes)
+            end;
+        [] ->
+            ok;
+        version_not_available ->
+            ok
+    end.
+
+upgrader(Nodes) ->
+    [Upgrader|_] = lists:usort(Nodes),
+    Upgrader.
+
+primary_upgrade(Upgrades, DiscNodes) ->
+    Others = DiscNodes -- [node()],
+    %% TODO this should happen after backing up!
+    rabbit_misc:ensure_ok(mnesia:start(),
+                          cannot_start_mnesia),
+    force_tables(),
+    [{atomic, ok} = mnesia:del_table_copy(schema, Node) || Node <- Others],
+    apply_upgrades(Upgrades),
+    ok.
+
+force_tables() ->
+    [mnesia:force_load_table(T) || T <- rabbit_mnesia:table_names()].
+
+non_primary_upgrade(Upgrader, DiscNodes) ->
+    case node_running(Upgrader) of
+        false ->
+            Msg = "~n~n * Cluster upgrade needed. Please start node ~s "
+                "first. * ~n~n~n",
+            Args = [Upgrader],
+            %% We don't throw or exit here since that gets thrown
+            %% straight out into do_boot, generating an erl_crash.dump
+            %% and displaying any error message in a confusing way.
+            error_logger:error_msg(Msg, Args),
+            io:format(Msg, Args),
+            error_logger:logfile(close),
+            halt(1);
+        true ->
+            rabbit_misc:ensure_ok(mnesia:delete_schema([node()]),
+                                  cannot_delete_schema),
+            ok = rabbit_mnesia:create_cluster_nodes_config(DiscNodes),
+            ok
+    end.
+
+node_running(Node) ->
+    case rpc:call(Node, application, which_applications, []) of
+        {badrpc, _} -> false;
+        Apps        -> lists:keysearch(rabbit, 1, Apps) =/= false
+    end.
+
+%% -------------------------------------------------------------------
+
 maybe_upgrade(Scopes) ->
     case upgrades_required(Scopes) of
         version_not_available -> version_not_available;
@@ -57,13 +117,6 @@ maybe_upgrade(Scopes) ->
         Upgrades              -> apply_upgrades(Upgrades)
     end.
 
-upgrade_required(Scopes) ->
-    case upgrades_required(Scopes) of
-        version_not_available -> false;
-        []                    -> false;
-        _                     -> true
-    end.
-
 read_version() ->
     case rabbit_misc:read_term_file(schema_filename()) of
         {ok, [Heads]}    -> {ok, Heads};