Decide the node to do mnesia upgrades based on which was the last disc node to shut down. Blow up with a hopefully helpful error message if the "wrong" disc node is started first. This works; you can now upgrade a disc-only cluster.

2 files changed, 91 insertions, 42 deletions

diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 82e2a30e8b..49d0411642 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -43,6 +43,8 @@
 %% other mnesia-using Erlang applications, such as ejabberd
 -export([create_tables/0]).
 
+-define(EXAMPLE_RABBIT_TABLE, rabbit_durable_exchange).
+
 -include("rabbit.hrl").
 
 %%----------------------------------------------------------------------------
@@ -164,7 +166,7 @@ nodes_of_type(Type) ->
     %% Specifically, we check whether a certain table, which we know
     %% will be written to disk on a disc node, is stored on disk or in
     %% RAM.
-    mnesia:table_info(rabbit_durable_exchange, Type).
+    mnesia:table_info(?EXAMPLE_RABBIT_TABLE, Type).
 
 table_definitions() ->
     [{rabbit_user,
@@ -387,40 +389,50 @@ init_db(ClusterNodes, Force) ->
                 true  -> ok
             end,
             case {Nodes, mnesia:system_info(use_dir)} of
-                {[], true} ->
-                    %% 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], fun forget_other_nodes/0) of
-                        ok                    -> ensure_schema_ok();
-                        version_not_available -> schema_ok_or_move()
-                    end;
                 {[], false} ->
                     %% Nothing there at all, start from scratch
                     ok = create_schema();
-                {[AnotherNode|_], _} ->
-                    %% Subsequent node in cluster, catch up
-                    ensure_version_ok(
-                      rpc:call(AnotherNode, rabbit_upgrade, read_version, [])),
-                    IsDiskNode = ClusterNodes == [] orelse
-                        lists:member(node(), ClusterNodes),
-                    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(case IsDiskNode of
-                                                       true  -> disc;
-                                                       false -> ram
-                                                   end),
-                    ensure_schema_ok()
+                {_, _} ->
+                    DiscNodes = mnesia:table_info(schema, disc_copies),
+                    case are_we_upgrader(DiscNodes) of
+                        true ->
+                            %% True single disc node, or last disc
+                            %% node in cluster to shut down, attempt
+                            %% upgrade
+                            ok = wait_for_tables(),
+                            case rabbit_upgrade:maybe_upgrade(
+                                   [mnesia, local],
+                                   fun () -> ok end,
+                                   fun forget_other_nodes/0) of
+                                ok                    -> ensure_schema_ok();
+                                version_not_available -> schema_ok_or_move()
+                            end;
+                        false ->
+                            %% Subsequent node in cluster, catch up
+                            %% TODO how to do this?
+                            %% ensure_version_ok(
+                            %%   rpc:call(AnotherNode, rabbit_upgrade, read_version, [])),
+                            IsDiskNode = ClusterNodes == [] orelse
+                                lists:member(node(), ClusterNodes),
+                            case rabbit_upgrade:maybe_upgrade(
+                                   [local],
+                                   ensure_nodes_running_fun(DiscNodes),
+                                   reset_fun(DiscNodes -- [node()])) 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(case IsDiskNode of
+                                                               true  -> disc;
+                                                               false -> ram
+                                                           end),
+                            ensure_schema_ok()
+                    end
             end;
         {error, Reason} ->
             %% one reason we may end up here is if we try to join
@@ -459,17 +471,52 @@ ensure_schema_ok() ->
         {error, Reason} -> throw({error, {schema_invalid, Reason}})
     end.
 
-reset_fun(ProperClusterNodes) ->
+ensure_nodes_running_fun(Nodes) ->
+    fun() ->
+            case nodes_running(Nodes) of
+                [] ->
+                    exit("Cluster upgrade needed. The first node you start "
+                         "should be the last node to be shut down.");
+                _ ->
+                    ok
+            end
+    end.
+
+reset_fun(Nodes) ->
     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),
+            {ok, _} = mnesia:change_config(extra_db_nodes, Nodes),
             ok
     end.
 
+%% Were we the last node in the cluster to shut down or is there no cluster?
+%% The answer to this is yes if:
+%% * We are our canonical source for reading a table
+%%    - If the canonical source is "nowhere" or another node, we are out of date
+%% * No other nodes are running Mnesia and have finished booting Rabbit.
+%%    - Since any node will be its own canonical source once the cluster is up.
+
+are_we_upgrader(Nodes) ->
+    Where = mnesia:table_info(?EXAMPLE_RABBIT_TABLE, where_to_read),
+    Node = node(),
+    case {Where, nodes_running(Nodes)} of
+        {Node, []} -> true;
+        {_,    _}  -> false
+    end.
+
+nodes_running(Nodes) ->
+    [N || N <- Nodes, node_running(N)].
+
+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()]),
diff --git a/src/rabbit_upgrade.erl b/src/rabbit_upgrade.erl
index 48c00d69ce..c852a0f953 100644
--- a/src/rabbit_upgrade.erl
+++ b/src/rabbit_upgrade.erl
@@ -21,7 +21,7 @@
 
 -module(rabbit_upgrade).
 
--export([maybe_upgrade/2, read_version/0, write_version/0, desired_version/0]).
+-export([maybe_upgrade/3, read_version/0, write_version/0, desired_version/0]).
 
 -include("rabbit.hrl").
 
@@ -36,7 +36,7 @@
 -type(scope() :: 'mnesia' | 'local').
 -type(version() :: [step()]).
 
--spec(maybe_upgrade/2 :: ([scope()], fun (() -> 'ok'))
+-spec(maybe_upgrade/3 :: ([scope()], fun (() -> 'ok'), fun (() -> 'ok'))
          -> 'ok' | 'version_not_available').
 -spec(read_version/0 :: () -> rabbit_types:ok_or_error2(version(), any())).
 -spec(write_version/0 :: () -> 'ok').
@@ -49,21 +49,22 @@
 %% 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, Fun) ->
+maybe_upgrade(Scopes, GuardFun, UpgradeFun) ->
     case read_version() of
         {ok, CurrentHeads} ->
             with_upgrade_graph(
-              fun (G) -> maybe_upgrade_graph(CurrentHeads, Scopes, Fun, G) end);
+              fun (G) -> maybe_upgrade_graph(CurrentHeads, Scopes,
+                                             GuardFun, UpgradeFun, G) end);
         {error, enoent} ->
             version_not_available
     end.
 
-maybe_upgrade_graph(CurrentHeads, Scopes, Fun, G) ->
+maybe_upgrade_graph(CurrentHeads, Scopes, GuardFun, UpgradeFun, G) ->
     case unknown_heads(CurrentHeads, G) of
         [] ->
             case upgrades_to_apply(CurrentHeads, Scopes, G) of
                 []       -> ok;
-                Upgrades -> apply_upgrades(Upgrades, Fun)
+                Upgrades -> apply_upgrades(Upgrades, GuardFun, UpgradeFun)
             end;
         Unknown ->
             throw({error, {future_upgrades_found, Unknown}})
@@ -132,7 +133,8 @@ heads(G) ->
 
 %% -------------------------------------------------------------------
 
-apply_upgrades(Upgrades, Fun) ->
+apply_upgrades(Upgrades, GuardFun, UpgradeFun) ->
+    GuardFun(),
     LockFile = lock_filename(dir()),
     case rabbit_misc:lock_file(LockFile) of
         ok ->
@@ -147,7 +149,7 @@ apply_upgrades(Upgrades, Fun) ->
                     %% is not intuitive. Remove it.
                     ok = file:delete(lock_filename(BackupDir)),
                     info("Upgrades: Mnesia dir backed up to ~p~n", [BackupDir]),
-                    ok = Fun(),
+                    ok = UpgradeFun(),
                     [apply_upgrade(Upgrade) || Upgrade <- Upgrades],
                     info("Upgrades: All upgrades applied successfully~n", []),
                     ok = write_version(),