Merge bug25848 again.

12 files changed, 125 insertions, 42 deletions

diff --git a/docs/rabbitmq.config.example b/docs/rabbitmq.config.example
index 4fad154255..e8b5666098 100644
--- a/docs/rabbitmq.config.example
+++ b/docs/rabbitmq.config.example
@@ -27,6 +27,11 @@
    %%
    %% {ssl_listeners, [5671]},
 
+   %% Maximum time for AMQP 0-8/0-9/0-9-1 handshake (after socket connection
+   %% and SSL handshake), in milliseconds.
+   %%
+   %% {handshake_timeout, 10000},
+
    %% Log levels (currently just used for connection logging).
    %% One of 'info', 'warning', 'error' or 'none', in decreasing order
    %% of verbosity. Defaults to 'info'.
@@ -103,6 +108,10 @@
    %%
    %% {ssl_cert_login_from, common_name},
 
+   %% SSL handshake timeout, in milliseconds.
+   %%
+   %% {ssl_handshake_timeout, 5000},
+
    %%
    %% Default User / VHost
    %% ====================
diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in
index 3647c04ade..f26e0f7709 100644
--- a/ebin/rabbit_app.in
+++ b/ebin/rabbit_app.in
@@ -46,6 +46,8 @@
          {trace_vhosts, []},
          {log_levels, [{connection, info}]},
          {ssl_cert_login_from, distinguished_name},
+         {ssl_handshake_timeout, 5000},
+         {handshake_timeout, 10000},
          {reverse_dns_lookups, false},
          {cluster_partition_handling, ignore},
          {tcp_listen_options, [binary,
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index c13868030a..7a40f9ebf0 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -52,6 +52,7 @@
           arguments,                   %% immutable
           pid,                         %% durable (just so we know home node)
           slave_pids, sync_slave_pids, %% transient
+          down_slave_nodes,            %% durable
           policy,                      %% durable, implicit update as above
           gm_pids,                     %% transient
           decorators}).                %% transient, recalculated as above
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 8a1d162a7a..4e23dbd242 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -29,7 +29,7 @@
 -export([basic_get/4, basic_consume/10, basic_cancel/4, notify_decorators/1]).
 -export([notify_sent/2, notify_sent_queue_down/1, resume/2]).
 -export([notify_down_all/2, activate_limit_all/2, credit/5]).
--export([on_node_down/1]).
+-export([on_node_up/1, on_node_down/1]).
 -export([update/2, store_queue/1, update_decorators/1, policy_changed/2]).
 -export([start_mirroring/1, stop_mirroring/1, sync_mirrors/1,
          cancel_sync_mirrors/1]).
@@ -174,6 +174,7 @@
          (fun ((atom(), A) -> {[rabbit_types:msg_id()], A}))) -> 'ok').
 -spec(set_ram_duration_target/2 :: (pid(), number() | 'infinity') -> 'ok').
 -spec(set_maximum_since_use/2 :: (pid(), non_neg_integer()) -> 'ok').
+-spec(on_node_up/1 :: (node()) -> 'ok').
 -spec(on_node_down/1 :: (node()) -> 'ok').
 -spec(pseudo_queue/2 :: (name(), pid()) -> rabbit_types:amqqueue()).
 -spec(immutable/1 :: (rabbit_types:amqqueue()) -> rabbit_types:amqqueue()).
@@ -257,15 +258,16 @@ declare(QueueName, Durable, AutoDelete, Args, Owner) ->
 declare(QueueName, Durable, AutoDelete, Args, Owner, Node) ->
     ok = check_declare_arguments(QueueName, Args),
     Q = rabbit_queue_decorator:set(
-          rabbit_policy:set(#amqqueue{name            = QueueName,
-                                      durable         = Durable,
-                                      auto_delete     = AutoDelete,
-                                      arguments       = Args,
-                                      exclusive_owner = Owner,
-                                      pid             = none,
-                                      slave_pids      = [],
-                                      sync_slave_pids = [],
-                                      gm_pids         = []})),
+          rabbit_policy:set(#amqqueue{name             = QueueName,
+                                      durable          = Durable,
+                                      auto_delete      = AutoDelete,
+                                      arguments        = Args,
+                                      exclusive_owner  = Owner,
+                                      pid              = none,
+                                      slave_pids       = [],
+                                      sync_slave_pids  = [],
+                                      down_slave_nodes = [],
+                                      gm_pids          = []})),
     Node = rabbit_mirror_queue_misc:initial_queue_node(Q, Node),
     gen_server2:call(start_queue_process(Node, Q), {init, new}, infinity).
 
@@ -665,15 +667,23 @@ forget_all_durable(Node) ->
           fun () ->
                   Qs = mnesia:match_object(rabbit_durable_queue,
                                            #amqqueue{_ = '_'}, write),
-                  [rabbit_binding:process_deletions(
-                     internal_delete1(Name)) ||
-                      #amqqueue{name = Name, pid = Pid} = Q <- Qs,
-                      node(Pid) =:= Node,
-                      rabbit_policy:get(<<"ha-mode">>, Q) =:= undefined],
+                  [forget_node_for_queue(Q) || #amqqueue{pid = Pid} = Q <- Qs,
+                      node(Pid) =:= Node],
                   ok
           end),
     ok.
 
+forget_node_for_queue(#amqqueue{name             = Name,
+                                down_slave_nodes = []}) ->
+    %% No slaves to recover from, queue is gone
+    rabbit_binding:process_deletions(internal_delete1(Name));
+
+forget_node_for_queue(Q = #amqqueue{down_slave_nodes = [H|T]}) ->
+    %% Promote a slave while down - it'll happily recover as a master
+    Q1 = Q#amqqueue{pid              = rabbit_misc:node_to_fake_pid(H),
+                    down_slave_nodes = T},
+    ok = mnesia:write(rabbit_durable_queue, Q1, write).
+
 run_backing_queue(QPid, Mod, Fun) ->
     gen_server2:cast(QPid, {run_backing_queue, Mod, Fun}).
 
@@ -689,6 +699,20 @@ stop_mirroring(QPid)  -> ok = delegate:cast(QPid, stop_mirroring).
 sync_mirrors(QPid)        -> delegate:call(QPid, sync_mirrors).
 cancel_sync_mirrors(QPid) -> delegate:call(QPid, cancel_sync_mirrors).
 
+on_node_up(Node) ->
+    ok = rabbit_misc:execute_mnesia_transaction(
+           fun () ->
+                   Qs = mnesia:match_object(rabbit_queue,
+                                            #amqqueue{_ = '_'}, write),
+                   [case lists:member(Node, DSNs) of
+                        true  -> DSNs1 = DSNs -- [Node],
+                                 store_queue(
+                                   Q#amqqueue{down_slave_nodes = DSNs1});
+                        false -> ok
+                    end || #amqqueue{down_slave_nodes = DSNs} = Q <- Qs],
+                   ok
+           end).
+
 on_node_down(Node) ->
     rabbit_misc:execute_mnesia_tx_with_tail(
       fun () -> QsDels =
@@ -724,12 +748,13 @@ pseudo_queue(QueueName, Pid) ->
               pid          = Pid,
               slave_pids   = []}.
 
-immutable(Q) -> Q#amqqueue{pid             = none,
-                           slave_pids      = none,
-                           sync_slave_pids = none,
-                           gm_pids         = none,
-                           policy          = none,
-                           decorators      = none}.
+immutable(Q) -> Q#amqqueue{pid              = none,
+                           slave_pids       = none,
+                           sync_slave_pids  = none,
+                           down_slave_nodes = none,
+                           gm_pids          = none,
+                           policy           = none,
+                           decorators       = none}.
 
 deliver([], _Delivery, _Flow) ->
     %% /dev/null optimisation
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 97206df350..4082c53d33 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -84,6 +84,7 @@
          memory,
          slave_pids,
          synchronised_slave_pids,
+         down_slave_nodes,
          backing_queue_status,
          state
         ]).
@@ -810,6 +811,14 @@ i(synchronised_slave_pids, #q{q = #amqqueue{name = Name}}) ->
         false -> '';
         true  -> SSPids
     end;
+i(down_slave_nodes, #q{q = #amqqueue{name    = Name,
+                                     durable = Durable}}) ->
+    {ok, Q = #amqqueue{down_slave_nodes = Nodes}} =
+        rabbit_amqqueue:lookup(Name),
+    case Durable andalso rabbit_mirror_queue_misc:is_mirrored(Q) of
+        false -> '';
+        true  -> Nodes
+    end;
 i(state, #q{status = running}) -> credit_flow:state();
 i(state, #q{status = State})   -> State;
 i(backing_queue_status, #q{backing_queue_state = BQS, backing_queue = BQ}) ->
diff --git a/src/rabbit_mirror_queue_misc.erl b/src/rabbit_mirror_queue_misc.erl
index 7aec1ac81f..9e8c4a1891 100644
--- a/src/rabbit_mirror_queue_misc.erl
+++ b/src/rabbit_mirror_queue_misc.erl
@@ -78,9 +78,10 @@ remove_from_queue(QueueName, Self, DeadGMPids) ->
               %% get here.
               case mnesia:read({rabbit_queue, QueueName}) of
                   [] -> {error, not_found};
-                  [Q = #amqqueue { pid        = QPid,
-                                   slave_pids = SPids,
-                                   gm_pids    = GMPids }] ->
+                  [Q = #amqqueue { pid              = QPid,
+                                   slave_pids       = SPids,
+                                   gm_pids          = GMPids,
+                                   down_slave_nodes = DSNs}] ->
                       {DeadGM, AliveGM} = lists:partition(
                                             fun ({GM, _}) ->
                                                     lists:member(GM, DeadGMPids)
@@ -89,6 +90,9 @@ remove_from_queue(QueueName, Self, DeadGMPids) ->
                       AlivePids = [Pid || {_GM, Pid} <- AliveGM],
                       Alive     = [Pid || Pid <- [QPid | SPids],
                                           lists:member(Pid, AlivePids)],
+                      DSNs1     = [node(Pid) ||
+                                      Pid <- SPids,
+                                      not lists:member(Pid, AlivePids)] ++ DSNs,
                       {QPid1, SPids1} = promote_slave(Alive),
                       case {{QPid, SPids}, {QPid1, SPids1}} of
                           {Same, Same} ->
@@ -97,9 +101,10 @@ remove_from_queue(QueueName, Self, DeadGMPids) ->
                               %% Either master hasn't changed, so
                               %% we're ok to update mnesia; or we have
                               %% become the master.
-                              Q1 = Q#amqqueue{pid        = QPid1,
-                                              slave_pids = SPids1,
-                                              gm_pids    = AliveGM},
+                              Q1 = Q#amqqueue{pid              = QPid1,
+                                              slave_pids       = SPids1,
+                                              gm_pids          = AliveGM,
+                                              down_slave_nodes = DSNs1},
                               store_updated_slaves(Q1),
                               %% If we add and remove nodes at the same time we
                               %% might tell the old master we need to sync and
@@ -109,8 +114,9 @@ remove_from_queue(QueueName, Self, DeadGMPids) ->
                           _ ->
                               %% Master has changed, and we're not it.
                               %% [1].
-                              Q1 = Q#amqqueue{slave_pids = Alive,
-                                              gm_pids    = AliveGM},
+                              Q1 = Q#amqqueue{slave_pids       = Alive,
+                                              gm_pids          = AliveGM,
+                                              down_slave_nodes = DSNs1},
                               store_updated_slaves(Q1)
                       end,
                       {ok, QPid1, DeadPids}
@@ -239,12 +245,16 @@ log(Level, QName, Fmt, Args) ->
     rabbit_log:log(mirroring, Level, "Mirrored ~s: " ++ Fmt,
                    [rabbit_misc:rs(QName) | Args]).
 
-store_updated_slaves(Q = #amqqueue{slave_pids      = SPids,
-                                   sync_slave_pids = SSPids}) ->
+store_updated_slaves(Q = #amqqueue{pid              = MPid,
+                                   slave_pids       = SPids,
+                                   sync_slave_pids  = SSPids,
+                                   down_slave_nodes = DSNs}) ->
     %% TODO now that we clear sync_slave_pids in rabbit_durable_queue,
     %% do we still need this filtering?
     SSPids1 = [SSPid || SSPid <- SSPids, lists:member(SSPid, SPids)],
-    Q1 = Q#amqqueue{sync_slave_pids = SSPids1},
+    DSNs1 = DSNs -- [node(P) || P <- [MPid | SPids]],
+    Q1 = Q#amqqueue{sync_slave_pids  = SSPids1,
+                    down_slave_nodes = DSNs1},
     ok = rabbit_amqqueue:store_queue(Q1),
     %% Wake it up so that we emit a stats event
     rabbit_amqqueue:notify_policy_changed(Q1),
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 6f353da574..fd4b7b1137 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -45,7 +45,7 @@
 -export([with_local_io/1, local_info_msg/2]).
 -export([unfold/2, ceil/1, queue_fold/3]).
 -export([sort_field_table/1]).
--export([pid_to_string/1, string_to_pid/1]).
+-export([pid_to_string/1, string_to_pid/1, node_to_fake_pid/1]).
 -export([version_compare/2, version_compare/3]).
 -export([version_minor_equivalent/2]).
 -export([dict_cons/3, orddict_cons/3, gb_trees_cons/3]).
@@ -193,6 +193,7 @@
         (rabbit_framing:amqp_table()) -> rabbit_framing:amqp_table()).
 -spec(pid_to_string/1 :: (pid()) -> string()).
 -spec(string_to_pid/1 :: (string()) -> pid()).
+-spec(node_to_fake_pid/1 :: (atom()) -> pid()).
 -spec(version_compare/2 :: (string(), string()) -> 'lt' | 'eq' | 'gt').
 -spec(version_compare/3 ::
         (string(), string(), ('lt' | 'lte' | 'eq' | 'gte' | 'gt'))
@@ -709,6 +710,10 @@ string_to_pid(Str) ->
             throw(Err)
     end.
 
+%% node(node_to_fake_pid(Node)) =:= Node.
+node_to_fake_pid(Node) ->
+    string_to_pid(format("<~s.0.0.0>", [Node])).
+
 version_compare(A, B, lte) ->
     case version_compare(A, B) of
         eq -> true;
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index 9082dbd353..96448f3227 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -37,8 +37,6 @@
 -include("rabbit.hrl").
 -include_lib("kernel/include/inet.hrl").
 
--define(SSL_TIMEOUT, 5). %% seconds
-
 -define(FIRST_TEST_BIND_PORT, 10000).
 
 %%----------------------------------------------------------------------------
@@ -168,9 +166,14 @@ ensure_ssl() ->
             end
     end.
 
+ssl_timeout() ->
+    {ok, Val} = application:get_env(rabbit, ssl_handshake_timeout),
+    Val.
+
 ssl_transform_fun(SslOpts) ->
     fun (Sock) ->
-            case catch ssl:ssl_accept(Sock, SslOpts, ?SSL_TIMEOUT * 1000) of
+            Timeout = ssl_timeout(),
+            case catch ssl:ssl_accept(Sock, SslOpts, Timeout) of
                 {ok, SslSock} ->
                     {ok, #ssl_socket{tcp = Sock, ssl = SslSock}};
                 {error, timeout} ->
@@ -185,7 +188,7 @@ ssl_transform_fun(SslOpts) ->
                     %% form, according to the TLS spec). So we give
                     %% the ssl_connection a little bit of time to send
                     %% such alerts.
-                    timer:sleep(?SSL_TIMEOUT * 1000),
+                    timer:sleep(Timeout),
                     {error, {ssl_upgrade_error, Reason}};
                 {'EXIT', Reason} ->
                     {error, {ssl_upgrade_failure, Reason}}
diff --git a/src/rabbit_node_monitor.erl b/src/rabbit_node_monitor.erl
index 1496147848..1c971c1da8 100644
--- a/src/rabbit_node_monitor.erl
+++ b/src/rabbit_node_monitor.erl
@@ -415,6 +415,7 @@ ensure_ping_timer(State) ->
       State, #state.down_ping_timer, ?RABBIT_DOWN_PING_INTERVAL, ping_nodes).
 
 handle_live_rabbit(Node) ->
+    ok = rabbit_amqqueue:on_node_up(Node),
     ok = rabbit_alarm:on_node_up(Node),
     ok = rabbit_mnesia:on_node_up(Node).
 
diff --git a/src/rabbit_prelaunch.erl b/src/rabbit_prelaunch.erl
index 4cc9cd12f1..6a6a4ee680 100644
--- a/src/rabbit_prelaunch.erl
+++ b/src/rabbit_prelaunch.erl
@@ -90,9 +90,9 @@ dist_port_set_check() ->
                         {none, none} -> ok;
                         _            -> rabbit_misc:quit(?DIST_PORT_CONFIGURED)
                     end;
+                {ok, _} ->
+                    ok;
                 {error, _} ->
-                    %% TODO can we present errors more nicely here
-                    %% than after -config has failed?
                     ok
             end
     end.
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 906c4b6e2f..2d46dd54b4 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -27,7 +27,6 @@
 
 -export([conserve_resources/3, server_properties/1]).
 
--define(HANDSHAKE_TIMEOUT, 10).
 -define(NORMAL_TIMEOUT, 3).
 -define(CLOSING_TIMEOUT, 30).
 -define(CHANNEL_TERMINATION_TIMEOUT, 3).
@@ -216,8 +215,9 @@ start_connection(Parent, HelperSup, Deb, Sock, SockTransform) ->
                                     exit(normal)
            end,
     log(info, "accepting AMQP connection ~p (~s)~n", [self(), Name]),
+    {ok, HandshakeTimeout} = application:get_env(rabbit, handshake_timeout),
     ClientSock = socket_op(Sock, SockTransform),
-    erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(), handshake_timeout),
+    erlang:send_after(HandshakeTimeout, self(), handshake_timeout),
     {PeerHost, PeerPort, Host, Port} =
         socket_op(Sock, fun (S) -> rabbit_net:socket_ends(S, inbound) end),
     ?store_proc_name(list_to_binary(Name)),
@@ -231,7 +231,7 @@ start_connection(Parent, HelperSup, Deb, Sock, SockTransform) ->
                   peer_port          = PeerPort,
                   protocol           = none,
                   user               = none,
-                  timeout_sec        = ?HANDSHAKE_TIMEOUT,
+                  timeout_sec        = (HandshakeTimeout / 1000),
                   frame_max          = ?FRAME_MIN_SIZE,
                   vhost              = none,
                   client_properties  = none,
diff --git a/src/rabbit_upgrade_functions.erl b/src/rabbit_upgrade_functions.erl
index b6d378525e..1104f3731a 100644
--- a/src/rabbit_upgrade_functions.erl
+++ b/src/rabbit_upgrade_functions.erl
@@ -48,6 +48,7 @@
 -rabbit_upgrade({queue_decorators,      mnesia, [gm_pids]}).
 -rabbit_upgrade({internal_system_x,     mnesia, [exchange_decorators]}).
 -rabbit_upgrade({cluster_name,          mnesia, [runtime_parameters]}).
+-rabbit_upgrade({down_slave_nodes,      mnesia, [queue_decorators]}).
 
 %% -------------------------------------------------------------------
 
@@ -77,6 +78,8 @@
 -spec(policy_apply_to/0       :: () -> 'ok').
 -spec(queue_decorators/0      :: () -> 'ok').
 -spec(internal_system_x/0     :: () -> 'ok').
+-spec(cluster_name/0          :: () -> 'ok').
+-spec(down_slave_nodes/0      :: () -> 'ok').
 
 -endif.
 
@@ -382,6 +385,21 @@ cluster_name_tx() ->
     [mnesia:delete(T, K, write) || K <- Ks],
     ok.
 
+down_slave_nodes() ->
+    ok = down_slave_nodes(rabbit_queue),
+    ok = down_slave_nodes(rabbit_durable_queue).
+
+down_slave_nodes(Table) ->
+    transform(
+      Table,
+      fun ({amqqueue, Name, Durable, AutoDelete, ExclusiveOwner, Arguments,
+            Pid, SlavePids, SyncSlavePids, Policy, GmPids, Decorators}) ->
+              {amqqueue, Name, Durable, AutoDelete, ExclusiveOwner, Arguments,
+               Pid, SlavePids, SyncSlavePids, [], Policy, GmPids, Decorators}
+      end,
+      [name, durable, auto_delete, exclusive_owner, arguments, pid, slave_pids,
+       sync_slave_pids, down_slave_nodes, policy, gm_pids, decorators]).
+
 %%--------------------------------------------------------------------
 
 transform(TableName, Fun, FieldList) ->