enable disc-less nodes

The main trouble with this is
mnesia:change_config(extra_db_nodes,...).  If the current node doesn't
have a schema with all the tables created, it will get it from one of
the nodes it connects to.  This is a problem if we're a disc node and
they're a ram node or vice versa.  So, after we get the tables from
the other node and wait_for_remote_tables, we assert that their
storage location is correct and change it if not.  If the other node
is disc and we are ram, this unfortunately means that the tables are
first written to disc and then converted to ram, but since this is a
one-off operation during startup, it shouldn't be a problem.

3 files changed, 62 insertions, 41 deletions

diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index ff6b0b114f..1a6825eeaa 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -23,7 +23,8 @@
          empty_ram_only_tables/0, copy_db/1, wait_for_tables/1,
          create_cluster_nodes_config/1, read_cluster_nodes_config/0,
          record_running_nodes/0, read_previously_running_nodes/0,
-         delete_previously_running_nodes/0, running_nodes_filename/0]).
+         delete_previously_running_nodes/0, running_nodes_filename/0,
+         is_disc_node/0]).
 
 -export([table_names/0]).
 
@@ -65,6 +66,7 @@
 -spec(read_previously_running_nodes/0 :: () ->  [node()]).
 -spec(delete_previously_running_nodes/0 :: () ->  'ok').
 -spec(running_nodes_filename/0 :: () -> file:filename()).
+-spec(is_disc_node/0 :: () -> boolean()).
 
 -endif.
 
@@ -429,8 +431,8 @@ delete_previously_running_nodes() ->
 init_db(ClusterNodes, Force, SecondaryPostMnesiaFun) ->
     UClusterNodes = lists:usort(ClusterNodes),
     ProperClusterNodes = UClusterNodes -- [node()],
-    IsDiskNode = ClusterNodes == [] orelse lists:member(node(), ClusterNodes),
-    WasDiskNode = filelib:is_regular(filename:join(dir(), "rabbit_durable_exchange.DCD")),
+    IsDiskNode = lists:member(node(), ClusterNodes),
+    WasDiskNode = is_disc_node(),
     case mnesia:change_config(extra_db_nodes, ProperClusterNodes) of
         {ok, Nodes} ->
             case Force of
@@ -448,9 +450,12 @@ init_db(ClusterNodes, Force, SecondaryPostMnesiaFun) ->
             %% three cases and attempt to upgrade the in the other two
             case {Nodes, WasDiskNode, IsDiskNode} of
                 {_, true, false} ->
-                    throw({error, {cannot_convert_disc_to_ram,
-                                   "Cannot convert a disc node to a ram node."
-                                   " Reset first."}});
+                    %% Converting disc node to ram
+                    mnesia:stop(),
+                    move_db(),
+                    rabbit_misc:ensure_ok(mnesia:start(),
+                                          cannot_start_mnesia),
+                    ok = create_schema(false);
                 {[], false, false} ->
                     %% New ram node; start from scratch
                     ok = create_schema(false);
@@ -469,12 +474,13 @@ init_db(ClusterNodes, Force, SecondaryPostMnesiaFun) ->
                     %% Subsequent node in cluster, catch up
                     ensure_version_ok(
                       rpc:call(AnotherNode, rabbit_version, recorded, [])),
-                    ok = wait_for_replicated_tables(),
                     {CopyType, CopyTypeAlt} =
                         case IsDiskNode of
                             true  -> {disc, disc_copies};
                             false -> {ram, ram_copies}
                         end,
+                    ok = wait_for_replicated_tables(),
+                    assert_tables_copy_type(CopyTypeAlt),
                     ok = create_local_table_copy(schema, CopyTypeAlt),
                     ok = create_local_table_copies(CopyType),
                     ok = SecondaryPostMnesiaFun(),
@@ -524,22 +530,25 @@ create_schema() ->
     create_schema(true).
 
 create_schema(OnDisk) ->
-    Nodes = if OnDisk -> [node()];
-               true   -> []
-            end,
-    mnesia:stop(),
-    rabbit_misc:ensure_ok(mnesia:create_schema(Nodes),
-                          cannot_create_schema),
-    rabbit_misc:ensure_ok(mnesia:start(),
-                          cannot_start_mnesia),
-    if not OnDisk ->
-            mnesia:change_table_copy_type(schema, node(), ram_copies);
-       true -> ok
+    if OnDisk ->
+            mnesia:stop(),
+            rabbit_misc:ensure_ok(mnesia:create_schema([node()]),
+                                  cannot_create_schema),
+            rabbit_misc:ensure_ok(mnesia:start(),
+                                  cannot_start_mnesia);
+       true ->
+            ok
     end,
     ok = create_tables(OnDisk),
     ensure_schema_integrity(),
     ok = rabbit_version:record_desired().
 
+is_disc_node() ->
+    %% This is pretty ugly but we can't start Mnesia and ask it (will hang),
+    %% we can't look at the config file (may not include us even if we're a
+    %% disc node).
+    filelib:is_regular(filename:join(dir(), "rabbit_durable_exchange.DCD")).
+
 move_db() ->
     mnesia:stop(),
     MnesiaDir = filename:dirname(dir() ++ "/"),
@@ -582,15 +591,8 @@ create_tables(OnDisk) ->
     lists:foreach(fun ({Tab, TabDef}) ->
                           TabDef1 = proplists:delete(match, TabDef),
                           TabDef2 = case OnDisk of
-                                        true ->
-                                            TabDef1;
-                                        false ->
-                                            [{disc_copies, []},
-                                             {ram_copies, [node()]} |
-                                             proplists:delete(
-                                               ram_copies,
-                                               proplists:delete(disc_copies,
-                                                                TabDef1))]
+                                        true ->  TabDef1;
+                                        false -> copy_type_to_ram(TabDef1)
                                     end,
                           case mnesia:create_table(Tab, TabDef2) of
                               {atomic, ok} -> ok;
@@ -603,9 +605,36 @@ create_tables(OnDisk) ->
                   table_definitions()),
     ok.
 
+copy_type_to_ram(TabDef) ->
+    [{disc_copies, []}, {ram_copies, [node()]}
+     | proplists:delete(ram_copies, proplists:delete(disc_copies, TabDef))].
+
 table_has_copy_type(TabDef, DiscType) ->
     lists:member(node(), proplists:get_value(DiscType, TabDef, [])).
 
+assert_tables_copy_type(CopyTypeAlt) ->
+    lists:foreach(
+      fun({Tab, TabDef}) ->
+              HasDiscCopies     = table_has_copy_type(TabDef, disc_copies),
+              HasDiscOnlyCopies = table_has_copy_type(TabDef, disc_only_copies),
+              StorageType = if HasDiscCopies     -> disc_copies;
+                               HasDiscOnlyCopies -> disc_only_copies;
+                               true              -> ram_copies
+                            end,
+              StorageType1 = if CopyTypeAlt =:= disc_copies -> StorageType;
+                                true                        -> ram_copies
+                             end,
+              case mnesia:table_info(Tab, storage_type) of
+                  StorageType1 -> ok;
+                  unknown      -> ok;
+                  _            -> io:format("~p to ~p: ~p~n", [Tab, StorageType1, mnesia:change_table_copy_type(Tab, node(), StorageType1)])
+              end
+      end, table_definitions()),
+    case mnesia:table_info(schema, storage_type) of
+        CopyTypeAlt -> ok;
+        _           -> io:format("~p to ~p: ~p~n", [schema, CopyTypeAlt, mnesia:change_table_copy_type(schema, node(), CopyTypeAlt)])
+    end.
+
 create_local_table_copies(Type) ->
     lists:foreach(
       fun ({Tab, TabDef}) ->
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index b2d41c64f9..766cad8326 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -951,9 +951,8 @@ test_cluster_management() ->
     ok = control_action(reset, []),
     ok = control_action(start_app, []),
     ok = control_action(stop_app, []),
-    {error, {cannot_convert_disc_to_ram, _}} =
-        control_action(force_cluster, ["invalid1@invalid",
-                                       "invalid2@invalid"]),
+    ok = control_action(force_cluster, ["invalid1@invalid",
+                                        "invalid2@invalid"]),
 
     %% join a non-existing cluster as a ram node
     ok = control_action(reset, []),
@@ -1012,9 +1011,8 @@ test_cluster_management2(SecondaryNode) ->
     ok = control_action(stop_app, []),
 
     %% convert a disk node into a ram node
-    {error, {cannot_convert_disc_to_ram, _}} =
-        control_action(force_cluster, ["invalid1@invalid",
-                                       "invalid2@invalid"]),
+    ok = control_action(force_cluster, ["invalid1@invalid",
+                                        "invalid2@invalid"]),
 
     %% turn a disk node into a ram node
     ok = control_action(reset, []),
diff --git a/src/rabbit_upgrade.erl b/src/rabbit_upgrade.erl
index a2abb1e58b..e6b52c61e3 100644
--- a/src/rabbit_upgrade.erl
+++ b/src/rabbit_upgrade.erl
@@ -144,7 +144,7 @@ upgrade_mode(AllNodes) ->
     case nodes_running(AllNodes) of
         [] ->
             AfterUs = rabbit_mnesia:read_previously_running_nodes(),
-            case {is_disc_node(), AfterUs} of
+            case {rabbit_mnesia:is_disc_node(), AfterUs} of
                 {true, []}  ->
                     primary;
                 {true, _}  ->
@@ -182,12 +182,6 @@ upgrade_mode(AllNodes) ->
             end
     end.
 
-is_disc_node() ->
-    %% This is pretty ugly but we can't start Mnesia and ask it (will hang),
-    %% we can't look at the config file (may not include us even if we're a
-    %% disc node).
-    filelib:is_regular(filename:join(dir(), "rabbit_durable_exchange.DCD")).
-
 die(Msg, Args) ->
     %% We don't throw or exit here since that gets thrown
     %% straight out into do_boot, generating an erl_crash.dump
@@ -218,7 +212,7 @@ force_tables() ->
 
 secondary_upgrade(AllNodes) ->
     %% must do this before we wipe out schema
-    IsDiscNode = is_disc_node(),
+    IsDiscNode = rabbit_mnesia:is_disc_node(),
     rabbit_misc:ensure_ok(mnesia:delete_schema([node()]),
                           cannot_delete_schema),
     %% Note that we cluster with all nodes, rather than all disc nodes