Merge branch 'stable' into rabbitmq-management-236

10 files changed, 357 insertions, 472 deletions

diff --git a/src/gm.erl b/src/gm.erl
index 74e19ee6fd..41aa01f04d 100644
--- a/src/gm.erl
+++ b/src/gm.erl
@@ -1175,11 +1175,20 @@ record_new_member_in_group(NewMember, Left, GroupName, TxnFun) ->
                 try
                     Group = #gm_group { members = Members, version = Ver } =
                         check_membership(Left, read_group(GroupName)),
-                    {Prefix, [Left | Suffix]} =
-                        lists:splitwith(fun (M) -> M =/= Left end, Members),
-                    write_group(Group #gm_group {
-                                  members = Prefix ++ [Left, NewMember | Suffix],
-                                  version = Ver + 1 })
+                    case lists:member(NewMember, Members) of
+                        true ->
+                            %% This avois duplicates during partial partitions,
+                            %% as inconsistent views might happen during them
+                            rabbit_log:warning("(~p) GM avoiding duplicate of ~p",
+                                               [self(), NewMember]),
+                            Group;
+                        false ->
+                            {Prefix, [Left | Suffix]} =
+                                lists:splitwith(fun (M) -> M =/= Left end, Members),
+                            write_group(Group #gm_group {
+                                          members = Prefix ++ [Left, NewMember | Suffix],
+                                          version = Ver + 1 })
+                    end
                 catch
                     lost_membership ->
                         %% The transaction must not be abruptly crashed, but
diff --git a/src/pg2_fixed.erl b/src/pg2_fixed.erl
deleted file mode 100644
index 222a0bc849..0000000000
--- a/src/pg2_fixed.erl
+++ /dev/null
@@ -1,399 +0,0 @@
-%% This is the version of pg2 from R14B02, which contains the fix
-%% described at
-%% http://erlang.2086793.n4.nabble.com/pg2-still-busted-in-R13B04-td2230601.html.
-%% The changes are a search-and-replace to rename the module and avoid
-%% clashes with other versions of pg2, and also a simple rewrite of
-%% "andalso" and "orelse" expressions to case statements where the second
-%% operand is not a boolean since R12B does not allow this.
-
-%%
-%% %CopyrightBegin%
-%%
-%% Copyright Ericsson AB 1997-2010. All Rights Reserved.
-%%
-%% The contents of this file are subject to the Erlang Public License,
-%% Version 1.1, (the "License"); you may not use this file except in
-%% compliance with the License. You should have received a copy of the
-%% Erlang Public License along with this software. If not, it can be
-%% retrieved online at http://www.erlang.org/.
-%%
-%% Software distributed under the License is distributed on an "AS IS"
-%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
-%% the License for the specific language governing rights and limitations
-%% under the License.
-%%
-%% %CopyrightEnd%
-%%
--module(pg2_fixed).
-
--export([create/1, delete/1, join/2, leave/2]).
--export([get_members/1, get_local_members/1]).
--export([get_closest_pid/1, which_groups/0]).
--export([start/0,start_link/0,init/1,handle_call/3,handle_cast/2,handle_info/2,
-         terminate/2]).
-
-%%% As of R13B03 monitors are used instead of links.
-
-%%%
-%%% Exported functions
-%%%
-
--spec start_link() -> {'ok', pid()} | {'error', term()}.
-
-start_link() ->
-    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
-
--spec start() -> {'ok', pid()} | {'error', term()}.
-
-start() ->
-    ensure_started().
-
--spec create(term()) -> 'ok'.
-
-create(Name) ->
-    _ = ensure_started(),
-    case ets:member(pg2_fixed_table, {group, Name}) of
-        false ->
-            global:trans({{?MODULE, Name}, self()},
-                         fun() ->
-                                 gen_server:multi_call(?MODULE, {create, Name})
-                         end),
-            ok;
-        true ->
-            ok
-    end.
-
--type name() :: term().
-
--spec delete(name()) -> 'ok'.
-
-delete(Name) ->
-    _ = ensure_started(),
-    global:trans({{?MODULE, Name}, self()},
-                 fun() ->
-                         gen_server:multi_call(?MODULE, {delete, Name})
-                 end),
-    ok.
-
--spec join(name(), pid()) -> 'ok' | {'error', {'no_such_group', term()}}.
-
-join(Name, Pid) when is_pid(Pid) ->
-    _ = ensure_started(),
-    case ets:member(pg2_fixed_table, {group, Name}) of
-        false ->
-            {error, {no_such_group, Name}};
-        true ->
-            global:trans({{?MODULE, Name}, self()},
-                         fun() ->
-                                 gen_server:multi_call(?MODULE,
-                                                       {join, Name, Pid})
-                         end),
-            ok
-    end.
-
--spec leave(name(), pid()) -> 'ok' | {'error', {'no_such_group', name()}}.
-
-leave(Name, Pid) when is_pid(Pid) ->
-    _ = ensure_started(),
-    case ets:member(pg2_fixed_table, {group, Name}) of
-        false ->
-            {error, {no_such_group, Name}};
-        true ->
-            global:trans({{?MODULE, Name}, self()},
-                         fun() ->
-                                 gen_server:multi_call(?MODULE,
-                                                       {leave, Name, Pid})
-                         end),
-            ok
-    end.
-
--type get_members_ret() :: [pid()] | {'error', {'no_such_group', name()}}.
-
--spec get_members(name()) -> get_members_ret().
-
-get_members(Name) ->
-    _ = ensure_started(),
-    case ets:member(pg2_fixed_table, {group, Name}) of
-        true ->
-            group_members(Name);
-        false ->
-            {error, {no_such_group, Name}}
-    end.
-
--spec get_local_members(name()) -> get_members_ret().
-
-get_local_members(Name) ->
-    _ = ensure_started(),
-    case ets:member(pg2_fixed_table, {group, Name}) of
-        true ->
-            local_group_members(Name);
-        false ->
-            {error, {no_such_group, Name}}
-    end.
-
--spec which_groups() -> [name()].
-
-which_groups() ->
-    _ = ensure_started(),
-    all_groups().
-
--type gcp_error_reason() :: {'no_process', term()} | {'no_such_group', term()}.
-
--spec get_closest_pid(term()) -> pid() | {'error', gcp_error_reason()}.
-
-get_closest_pid(Name) ->
-    case get_local_members(Name) of
-        [Pid] ->
-            Pid;
-        [] ->
-            case get_members(Name) of
-                [] -> {error, {no_process, Name}};
-                Members ->
-                    X = time_compat:erlang_system_time(micro_seconds),
-                    lists:nth((X rem length(Members))+1, Members)
-            end;
-        Members when is_list(Members) ->
-            X = time_compat:erlang_system_time(micro_seconds),
-            lists:nth((X rem length(Members))+1, Members);
-        Else ->
-            Else
-    end.
-
-%%%
-%%% Callback functions from gen_server
-%%%
-
--record(state, {}).
-
--spec init([]) -> {'ok', #state{}}.
-
-init([]) ->
-    Ns = nodes(),
-    _ = net_kernel:monitor_nodes(true),
-    lists:foreach(fun(N) ->
-                          {?MODULE, N} ! {new_pg2_fixed, node()},
-                          self() ! {nodeup, N}
-                  end, Ns),
-    pg2_fixed_table = ets:new(pg2_fixed_table, [ordered_set, protected, named_table]),
-    {ok, #state{}}.
-
--type call() :: {'create', name()}
-              | {'delete', name()}
-              | {'join', name(), pid()}
-              | {'leave', name(), pid()}.
-
--spec handle_call(call(), _, #state{}) ->
-        {'reply', 'ok', #state{}}.
-
-handle_call({create, Name}, _From, S) ->
-    assure_group(Name),
-    {reply, ok, S};
-handle_call({join, Name, Pid}, _From, S) ->
-    case ets:member(pg2_fixed_table, {group, Name}) of
-        true -> _ = join_group(Name, Pid),
-                ok;
-        _    -> ok
-    end,
-    {reply, ok, S};
-handle_call({leave, Name, Pid}, _From, S) ->
-    case ets:member(pg2_fixed_table, {group, Name}) of
-        true -> leave_group(Name, Pid);
-        _    -> ok
-    end,
-    {reply, ok, S};
-handle_call({delete, Name}, _From, S) ->
-    delete_group(Name),
-    {reply, ok, S};
-handle_call(Request, From, S) ->
-    error_logger:warning_msg("The pg2_fixed server received an unexpected message:\n"
-                             "handle_call(~p, ~p, _)\n",
-                             [Request, From]),
-    {noreply, S}.
-
--type all_members() :: [[name(),...]].
--type cast() :: {'exchange', node(), all_members()}
-              | {'del_member', name(), pid()}.
-
--spec handle_cast(cast(), #state{}) -> {'noreply', #state{}}.
-
-handle_cast({exchange, _Node, List}, S) ->
-    store(List),
-    {noreply, S};
-handle_cast(_, S) ->
-    %% Ignore {del_member, Name, Pid}.
-    {noreply, S}.
-
--spec handle_info(tuple(), #state{}) -> {'noreply', #state{}}.
-
-handle_info({'DOWN', MonitorRef, process, _Pid, _Info}, S) ->
-    member_died(MonitorRef),
-    {noreply, S};
-handle_info({nodeup, Node}, S) ->
-    gen_server:cast({?MODULE, Node}, {exchange, node(), all_members()}),
-    {noreply, S};
-handle_info({new_pg2_fixed, Node}, S) ->
-    gen_server:cast({?MODULE, Node}, {exchange, node(), all_members()}),
-    {noreply, S};
-handle_info(_, S) ->
-    {noreply, S}.
-
--spec terminate(term(), #state{}) -> 'ok'.
-
-terminate(_Reason, _S) ->
-    true = ets:delete(pg2_fixed_table),
-    ok.
-
-%%%
-%%% Local functions
-%%%
-
-%%% One ETS table, pg2_fixed_table, is used for bookkeeping. The type of the
-%%% table is ordered_set, and the fast matching of partially
-%%% instantiated keys is used extensively.
-%%%
-%%% {{group, Name}}
-%%%    Process group Name.
-%%% {{ref, Pid}, RPid, MonitorRef, Counter}
-%%% {{ref, MonitorRef}, Pid}
-%%%    Each process has one monitor. Sometimes a process is spawned to
-%%%    monitor the pid (RPid). Counter is incremented when the Pid joins
-%%%    some group.
-%%% {{member, Name, Pid}, GroupCounter}
-%%% {{local_member, Name, Pid}}
-%%%    Pid is a member of group Name, GroupCounter is incremented when the
-%%%    Pid joins the group Name.
-%%% {{pid, Pid, Name}}
-%%%    Pid is a member of group Name.
-
-store(List) ->
-    _ = [assure_group(Name)
-         andalso
-         [join_group(Name, P) || P <- Members -- group_members(Name)] ||
-            [Name, Members] <- List],
-    ok.
-
-assure_group(Name) ->
-    Key = {group, Name},
-    ets:member(pg2_fixed_table, Key) orelse true =:= ets:insert(pg2_fixed_table, {Key}).
-
-delete_group(Name) ->
-    _ = [leave_group(Name, Pid) || Pid <- group_members(Name)],
-    true = ets:delete(pg2_fixed_table, {group, Name}),
-    ok.
-
-member_died(Ref) ->
-    [{{ref, Ref}, Pid}] = ets:lookup(pg2_fixed_table, {ref, Ref}),
-    Names = member_groups(Pid),
-    _ = [leave_group(Name, P) ||
-            Name <- Names,
-            P <- member_in_group(Pid, Name)],
-    %% Kept for backward compatibility with links. Can be removed, eventually.
-    _ = [gen_server:abcast(nodes(), ?MODULE, {del_member, Name, Pid}) ||
-            Name <- Names],
-    ok.
-
-join_group(Name, Pid) ->
-    Ref_Pid = {ref, Pid},
-    try _ = ets:update_counter(pg2_fixed_table, Ref_Pid, {4, +1})
-    catch _:_ ->
-            {RPid, Ref} = do_monitor(Pid),
-            true = ets:insert(pg2_fixed_table, {Ref_Pid, RPid, Ref, 1}),
-            true = ets:insert(pg2_fixed_table, {{ref, Ref}, Pid})
-    end,
-    Member_Name_Pid = {member, Name, Pid},
-    try _ = ets:update_counter(pg2_fixed_table, Member_Name_Pid, {2, +1, 1, 1})
-    catch _:_ ->
-            true = ets:insert(pg2_fixed_table, {Member_Name_Pid, 1}),
-            _ = [ets:insert(pg2_fixed_table, {{local_member, Name, Pid}}) ||
-                    node(Pid) =:= node()],
-            true = ets:insert(pg2_fixed_table, {{pid, Pid, Name}})
-    end.
-
-leave_group(Name, Pid) ->
-    Member_Name_Pid = {member, Name, Pid},
-    try ets:update_counter(pg2_fixed_table, Member_Name_Pid, {2, -1, 0, 0}) of
-        N ->
-            if
-                N =:= 0 ->
-                    true = ets:delete(pg2_fixed_table, {pid, Pid, Name}),
-                    _ = [ets:delete(pg2_fixed_table, {local_member, Name, Pid}) ||
-                            node(Pid) =:= node()],
-                    true = ets:delete(pg2_fixed_table, Member_Name_Pid);
-                true ->
-                    ok
-            end,
-            Ref_Pid = {ref, Pid},
-            case ets:update_counter(pg2_fixed_table, Ref_Pid, {4, -1}) of
-                0 ->
-                    [{Ref_Pid,RPid,Ref,0}] = ets:lookup(pg2_fixed_table, Ref_Pid),
-                    true = ets:delete(pg2_fixed_table, {ref, Ref}),
-                    true = ets:delete(pg2_fixed_table, Ref_Pid),
-                    true = erlang:demonitor(Ref, [flush]),
-                    kill_monitor_proc(RPid, Pid);
-                _ ->
-                    ok
-            end
-    catch _:_ ->
-            ok
-    end.
-
-all_members() ->
-    [[G, group_members(G)] || G <- all_groups()].
-
-group_members(Name) ->
-    [P ||
-        [P, N] <- ets:match(pg2_fixed_table, {{member, Name, '$1'},'$2'}),
-        _ <- lists:seq(1, N)].
-
-local_group_members(Name) ->
-    [P ||
-        [Pid] <- ets:match(pg2_fixed_table, {{local_member, Name, '$1'}}),
-        P <- member_in_group(Pid, Name)].
-
-member_in_group(Pid, Name) ->
-    case ets:lookup(pg2_fixed_table, {member, Name, Pid}) of
-        [] -> [];
-        [{{member, Name, Pid}, N}] ->
-            lists:duplicate(N, Pid)
-    end.
-
-member_groups(Pid) ->
-    [Name || [Name] <- ets:match(pg2_fixed_table, {{pid, Pid, '$1'}})].
-
-all_groups() ->
-    [N || [N] <- ets:match(pg2_fixed_table, {{group,'$1'}})].
-
-ensure_started() ->
-    case whereis(?MODULE) of
-        undefined ->
-            C = {pg2_fixed, {?MODULE, start_link, []}, permanent,
-                 1000, worker, [?MODULE]},
-            supervisor:start_child(kernel_safe_sup, C);
-        Pg2_FixedPid ->
-            {ok, Pg2_FixedPid}
-    end.
-
-
-kill_monitor_proc(RPid, Pid) ->
-    case RPid of
-        Pid -> ok;
-        _   -> exit(RPid, kill)
-    end.
-
-%% When/if erlang:monitor() returns before trying to connect to the
-%% other node this function can be removed.
-do_monitor(Pid) ->
-    case (node(Pid) =:= node()) orelse lists:member(node(Pid), nodes()) of
-        true ->
-            %% Assume the node is still up
-            {Pid, erlang:monitor(process, Pid)};
-        false ->
-            F = fun() ->
-                        Ref = erlang:monitor(process, Pid),
-                        receive
-                            {'DOWN', Ref, process, Pid, _Info} ->
-                                exit(normal)
-                        end
-                end,
-            erlang:spawn_monitor(F)
-    end.
diff --git a/src/rabbit.app.src b/src/rabbit.app.src
index 872336bd8e..dd38ad4072 100644
--- a/src/rabbit.app.src
+++ b/src/rabbit.app.src
@@ -99,5 +99,13 @@
          {credit_flow_default_credit, {200, 100}},
          %% see rabbitmq-server#248
          %% and rabbitmq-server#667
-         {channel_operation_timeout, 15000}
+         {channel_operation_timeout, 15000},
+         {config_entry_decoder, [
+             {cipher, aes_cbc256},
+             {hash, sha512},
+             {iterations, 1000},
+             {passphrase, undefined}
+         ]},
+         %% rabbitmq-server-973
+         {lazy_queue_explicit_gc_run_operation_threshold, 250}
         ]}]}.
diff --git a/src/rabbit.erl b/src/rabbit.erl
index c9a008c446..f0d66f4110 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -24,7 +24,7 @@
          start_fhc/0]).
 -export([start/2, stop/1, prep_stop/1]).
 -export([start_apps/1, stop_apps/1]).
--export([log_location/1, config_files/0]). %% for testing and mgmt-agent
+-export([log_location/1, config_files/0, decrypt_config/2]). %% for testing and mgmt-agent
 
 %%---------------------------------------------------------------------------
 %% Boot steps.
@@ -449,6 +449,38 @@ stop_and_halt() ->
 
 start_apps(Apps) ->
     app_utils:load_applications(Apps),
+
+    ConfigEntryDecoder = case application:get_env(rabbit, config_entry_decoder) of
+        undefined ->
+            [];
+        {ok, Val} ->
+            Val
+    end,
+    PassPhrase = case proplists:get_value(passphrase, ConfigEntryDecoder) of
+        prompt ->
+            IoDevice = get_input_iodevice(),
+            io:setopts(IoDevice, [{echo, false}]),
+            PP = lists:droplast(io:get_line(IoDevice,
+                "\nPlease enter the passphrase to unlock encrypted "
+                "configuration entries.\n\nPassphrase: ")),
+            io:setopts(IoDevice, [{echo, true}]),
+            io:format(IoDevice, "~n", []),
+            PP;
+        {file, Filename} ->
+            {ok, File} = file:read_file(Filename),
+            [PP|_] = binary:split(File, [<<"\r\n">>, <<"\n">>]),
+            PP;
+        PP ->
+            PP
+    end,
+    Algo = {
+        proplists:get_value(cipher, ConfigEntryDecoder, rabbit_pbe:default_cipher()),
+        proplists:get_value(hash, ConfigEntryDecoder, rabbit_pbe:default_hash()),
+        proplists:get_value(iterations, ConfigEntryDecoder, rabbit_pbe:default_iterations()),
+        PassPhrase
+    },
+    decrypt_config(Apps, Algo),
+
     OrderedApps = app_utils:app_dependency_order(Apps, false),
     case lists:member(rabbit, Apps) of
         false -> rabbit_boot_steps:run_boot_steps(Apps); %% plugin activation
@@ -457,6 +489,78 @@ start_apps(Apps) ->
     ok = app_utils:start_applications(OrderedApps,
                                       handle_app_error(could_not_start)).
 
+%% This function retrieves the correct IoDevice for requesting
+%% input. The problem with using the default IoDevice is that
+%% the Erlang shell prevents us from getting the input.
+%%
+%% Instead we therefore look for the io process used by the
+%% shell and if it can't be found (because the shell is not
+%% started e.g with -noshell) we use the 'user' process.
+%%
+%% This function will not work when either -oldshell or -noinput
+%% options are passed to erl.
+get_input_iodevice() ->
+    case whereis(user) of
+        undefined -> user;
+        User ->
+            case group:interfaces(User) of
+                [] ->
+                    user;
+                [{user_drv, Drv}] ->
+                    case user_drv:interfaces(Drv) of
+                        [] ->
+                            user;
+                        [{current_group, IoDevice}] ->
+                            IoDevice
+                    end
+            end
+    end.
+
+decrypt_config([], _) ->
+    ok;
+decrypt_config([App|Apps], Algo) ->
+    decrypt_app(App, application:get_all_env(App), Algo),
+    decrypt_config(Apps, Algo).
+
+decrypt_app(_, [], _) ->
+    ok;
+decrypt_app(App, [{Key, Value}|Tail], Algo) ->
+    try begin
+            case decrypt(Value, Algo) of
+                Value ->
+                    ok;
+                NewValue ->
+                    application:set_env(App, Key, NewValue)
+            end
+        end
+    catch
+        exit:{bad_configuration, config_entry_decoder} ->
+            exit({bad_configuration, config_entry_decoder});
+        _:Msg ->
+            rabbit_log:info("Error while decrypting key '~p'. Please check encrypted value, passphrase, and encryption configuration~n", [Key]),
+            exit({decryption_error, {key, Key}, Msg})
+    end,
+    decrypt_app(App, Tail, Algo).
+
+decrypt({encrypted, _}, {_, _, _, undefined}) ->
+    exit({bad_configuration, config_entry_decoder});
+decrypt({encrypted, EncValue}, {Cipher, Hash, Iterations, Password}) ->
+    rabbit_pbe:decrypt_term(Cipher, Hash, Iterations, Password, EncValue);
+decrypt(List, Algo) when is_list(List) ->
+    decrypt_list(List, Algo, []);
+decrypt(Value, _) ->
+    Value.
+
+%% We make no distinction between strings and other lists.
+%% When we receive a string, we loop through each element
+%% and ultimately return the string unmodified, as intended.
+decrypt_list([], _, Acc) ->
+    lists:reverse(Acc);
+decrypt_list([{Key, Value}|Tail], Algo, Acc) when Key =/= encrypted ->
+    decrypt_list(Tail, Algo, [{Key, decrypt(Value, Algo)}|Acc]);
+decrypt_list([Value|Tail], Algo, Acc) ->
+    decrypt_list(Tail, Algo, [decrypt(Value, Algo)|Acc]).
+
 stop_apps(Apps) ->
     ok = app_utils:stop_applications(
            Apps, handle_app_error(error_during_shutdown)),
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 4f8581f78a..b7df5a9a19 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -1310,7 +1310,13 @@ handle_cast(policy_changed, State = #q{q = #amqqueue{name = Name}}) ->
     %% This also has the side effect of waking us up so we emit a
     %% stats event - so event consumers see the changed policy.
     {ok, Q} = rabbit_amqqueue:lookup(Name),
-    noreply(process_args_policy(State#q{q = Q})).
+    noreply(process_args_policy(State#q{q = Q}));
+
+handle_cast({sync_start, _, _}, State = #q{q = #amqqueue{name = Name}}) ->
+    %% Only a slave should receive this, it means we are a duplicated master
+    rabbit_mirror_queue_misc:log_warning(
+      Name, "Stopping after receiving sync_start from another master", []),
+    stop(State).
 
 handle_info({maybe_expire, Vsn}, State = #q{args_policy_version = Vsn}) ->
     case is_unused(State) of
diff --git a/src/rabbit_control_main.erl b/src/rabbit_control_main.erl
index 92898c2a2c..8c245892b7 100644
--- a/src/rabbit_control_main.erl
+++ b/src/rabbit_control_main.erl
@@ -92,7 +92,8 @@
          {trace_off, [?VHOST_DEF]},
          set_vm_memory_high_watermark,
          set_disk_free_limit,
-         help
+         help,
+         {encode, [?DECODE_DEF, ?CIPHER_DEF, ?HASH_DEF, ?ITERATIONS_DEF, ?LIST_CIPHERS_DEF, ?LIST_HASHES_DEF]}
         ]).
 
 -define(GLOBAL_QUERIES,
@@ -114,7 +115,7 @@
         [stop, stop_app, start_app, wait, reset, force_reset, rotate_logs,
          join_cluster, change_cluster_node_type, update_cluster_nodes,
          forget_cluster_node, rename_cluster_node, cluster_status, status,
-         environment, eval, force_boot, help, hipe_compile]).
+         environment, eval, force_boot, help, hipe_compile, encode]).
 
 %% [Command | {Command, DefaultTimeoutInMilliSeconds}]
 -define(COMMANDS_WITH_TIMEOUT,
@@ -579,6 +580,17 @@ action(eval, Node, [Expr], _Opts, _Inform) ->
 action(help, _Node, _Args, _Opts, _Inform) ->
     io:format("~s", [rabbit_ctl_usage:usage()]);
 
+action(encode, _Node, Args, Opts, _Inform) ->
+    ListCiphers = lists:member({?LIST_CIPHERS_OPT, true}, Opts),
+    ListHashes = lists:member({?LIST_HASHES_OPT, true}, Opts),
+    Decode = lists:member({?DECODE_OPT, true}, Opts),
+    Cipher = list_to_atom(proplists:get_value(?CIPHER_OPT, Opts)),
+    Hash = list_to_atom(proplists:get_value(?HASH_OPT, Opts)),
+    Iterations = list_to_integer(proplists:get_value(?ITERATIONS_OPT, Opts)),
+
+    {_, Msg} = rabbit_control_pbe:encode(ListCiphers, ListHashes, Decode, Cipher, Hash, Iterations, Args),
+    io:format(Msg ++ "~n");
+
 action(Command, Node, Args, Opts, Inform) ->
     %% For backward compatibility, run commands accepting a timeout with
     %% the default timeout.
diff --git a/src/rabbit_control_pbe.erl b/src/rabbit_control_pbe.erl
new file mode 100644
index 0000000000..2fa2c90a6e
--- /dev/null
+++ b/src/rabbit_control_pbe.erl
@@ -0,0 +1,79 @@
+% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2016 Pivotal Software, Inc.  All rights reserved.
+%%
+
+-module(rabbit_control_pbe).
+
+-export([encode/7]).
+
+% for testing purposes
+-export([evaluate_input_as_term/1]).
+
+encode(ListCiphers, _ListHashes, _Decode, _Cipher, _Hash, _Iterations, _Args) when ListCiphers ->
+    {ok, io_lib:format("~p", [rabbit_pbe:supported_ciphers()])};
+
+encode(_ListCiphers, ListHashes, _Decode, _Cipher, _Hash, _Iterations, _Args) when ListHashes ->
+    {ok, io_lib:format("~p", [rabbit_pbe:supported_hashes()])};
+
+encode(_ListCiphers, _ListHashes, Decode, Cipher, Hash, Iterations, Args) ->
+    CipherExists = lists:member(Cipher, rabbit_pbe:supported_ciphers()),
+    HashExists = lists:member(Hash, rabbit_pbe:supported_hashes()),
+    encode_encrypt_decrypt(CipherExists, HashExists, Decode, Cipher, Hash, Iterations, Args).
+
+encode_encrypt_decrypt(CipherExists, _HashExists, _Decode, _Cipher, _Hash, _Iterations, _Args) when CipherExists =:= false ->
+    {error, io_lib:format("The requested cipher is not supported", [])};
+
+encode_encrypt_decrypt(_CipherExists, HashExists, _Decode, _Cipher, _Hash, _Iterations, _Args) when HashExists =:= false ->
+    {error, io_lib:format("The requested hash is not supported", [])};
+
+encode_encrypt_decrypt(_CipherExists, _HashExists, _Decode, _Cipher, _Hash, Iterations, _Args) when Iterations =< 0 ->
+    {error, io_lib:format("The requested number of iterations is incorrect", [])};
+
+encode_encrypt_decrypt(_CipherExists, _HashExists, Decode, Cipher, Hash, Iterations, Args) when length(Args) == 2, Decode =:= false ->
+    [Value, PassPhrase] = Args,
+    try begin
+            TermValue = evaluate_input_as_term(Value),
+            Result = rabbit_pbe:encrypt_term(Cipher, Hash, Iterations, list_to_binary(PassPhrase), TermValue),
+            {ok, io_lib:format("~p", [{encrypted, Result}])}
+        end
+    catch
+        _:Msg -> {error, io_lib:format("Error during cipher operation: ~p", [Msg])}
+    end;
+
+encode_encrypt_decrypt(_CipherExists, _HashExists, Decode, Cipher, Hash, Iterations, Args) when length(Args) == 2, Decode ->
+    [Value, PassPhrase] = Args,
+    try begin
+            TermValue = evaluate_input_as_term(Value),
+            TermToDecrypt = case TermValue of
+                {encrypted, EncryptedTerm} ->
+                    EncryptedTerm;
+                _ ->
+                    TermValue
+            end,
+            Result = rabbit_pbe:decrypt_term(Cipher, Hash, Iterations, list_to_binary(PassPhrase), TermToDecrypt),
+            {ok, io_lib:format("~p", [Result])}
+        end
+    catch
+        _:Msg -> {error, io_lib:format("Error during cipher operation: ~p", [Msg])}
+    end;
+
+encode_encrypt_decrypt(_CipherExists, _HashExists, _Decode, _Cipher, _Hash, _Iterations, _Args) ->
+    {error, io_lib:format("Please provide a value to encode/decode and a passphrase", [])}.
+
+evaluate_input_as_term(Input) ->
+    {ok,Tokens,_EndLine} = erl_scan:string(Input ++ "."),
+    {ok,AbsForm} = erl_parse:parse_exprs(Tokens),
+    {value,TermValue,_Bs} = erl_eval:exprs(AbsForm, erl_eval:new_bindings()),
+    TermValue.
diff --git a/src/rabbit_mirror_queue_mode_nodes.erl b/src/rabbit_mirror_queue_mode_nodes.erl
index e63f340373..31c55722a5 100644
--- a/src/rabbit_mirror_queue_mode_nodes.erl
+++ b/src/rabbit_mirror_queue_mode_nodes.erl
@@ -32,29 +32,37 @@
 description() ->
     [{description, <<"Mirror queue to specified nodes">>}].
 
-suggested_queue_nodes(Nodes0, MNode, _SNodes, SSNodes, Poss) ->
-    Nodes1 = [list_to_atom(binary_to_list(Node)) || Node <- Nodes0],
+suggested_queue_nodes(PolicyNodes0, CurrentMaster, _SNodes, SSNodes, NodesRunningRabbitMQ) ->
+    PolicyNodes1 = [list_to_atom(binary_to_list(Node)) || Node <- PolicyNodes0],
     %% If the current master is not in the nodes specified, then what we want
     %% to do depends on whether there are any synchronised slaves. If there
     %% are then we can just kill the current master - the admin has asked for
     %% a migration and we should give it to them. If there are not however
     %% then we must keep the master around so as not to lose messages.
-    Nodes = case SSNodes of
-                [] -> lists:usort([MNode | Nodes1]);
-                _  -> Nodes1
-            end,
-    Unavailable = Nodes -- Poss,
-    Available = Nodes -- Unavailable,
-    case Available of
+
+    PolicyNodes = case SSNodes of
+                      [] -> lists:usort([CurrentMaster | PolicyNodes1]);
+                      _  -> PolicyNodes1
+                  end,
+    Unavailable = PolicyNodes -- NodesRunningRabbitMQ,
+    AvailablePolicyNodes = PolicyNodes -- Unavailable,
+    case AvailablePolicyNodes of
         [] -> %% We have never heard of anything? Not much we can do but
               %% keep the master alive.
-              {MNode, []};
-        _  -> case lists:member(MNode, Available) of
-                  true  -> {MNode, Available -- [MNode]};
+              {CurrentMaster, []};
+        _  -> case lists:member(CurrentMaster, AvailablePolicyNodes) of
+                  true  -> {CurrentMaster,
+                            AvailablePolicyNodes -- [CurrentMaster]};
                   false -> %% Make sure the new master is synced! In order to
                            %% get here SSNodes must not be empty.
-                           [NewMNode | _] = SSNodes,
-                           {NewMNode, Available -- [NewMNode]}
+                           SyncPolicyNodes = [Node ||
+                                              Node <- AvailablePolicyNodes,
+                                              lists:member(Node, SSNodes)],
+                           NewMaster = case SyncPolicyNodes of
+                                          [Node | _] -> Node;
+                                          []         -> erlang:hd(SSNodes)
+                                      end,
+                           {NewMaster, AvailablePolicyNodes -- [NewMaster]}
               end
     end.
 
diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl
index 4770018f9e..6017e5a028 100644
--- a/src/rabbit_mirror_queue_slave.erl
+++ b/src/rabbit_mirror_queue_slave.erl
@@ -314,7 +314,12 @@ handle_cast({set_ram_duration_target, Duration},
             State = #state { backing_queue       = BQ,
                              backing_queue_state = BQS }) ->
     BQS1 = BQ:set_ram_duration_target(Duration, BQS),
-    noreply(State #state { backing_queue_state = BQS1 }).
+    noreply(State #state { backing_queue_state = BQS1 });
+
+handle_cast(policy_changed, State) ->
+    %% During partial partitions, we might end up receiving messages expected by a master
+    %% Ignore them
+    noreply(State).
 
 handle_info(update_ram_duration, State = #state{backing_queue       = BQ,
                                                 backing_queue_state = BQS}) ->
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 297df086ad..dd92256146 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -306,7 +306,11 @@
           io_batch_size,
 
           %% default queue or lazy queue
-          mode
+          mode,
+          %% number of reduce_memory_usage executions, once it
+          %% reaches a threshold the queue will manually trigger a runtime GC
+	        %% see: maybe_execute_gc/1
+          memory_reduction_run_count
         }).
 
 -record(rates, { in, out, ack_in, ack_out, timestamp }).
@@ -402,7 +406,8 @@
              disk_write_count      :: non_neg_integer(),
 
              io_batch_size         :: pos_integer(),
-             mode                  :: 'default' | 'lazy' }.
+             mode                  :: 'default' | 'lazy',
+             memory_reduction_run_count :: non_neg_integer()}.
 %% Duplicated from rabbit_backing_queue
 -spec ack([ack()], state()) -> {[rabbit_guid:guid()], state()}.
 
@@ -427,6 +432,21 @@
 %% rabbit_amqqueue_process need fairly fresh rates.
 -define(MSGS_PER_RATE_CALC, 100).
 
+
+%% we define the garbage collector threshold
+%% it needs to tune the GC calls inside `reduce_memory_use`
+%% see: rabbitmq-server-973 and `maybe_execute_gc` function
+-define(DEFAULT_EXPLICIT_GC_RUN_OP_THRESHOLD, 250).
+-define(EXPLICIT_GC_RUN_OP_THRESHOLD,
+    case get(explicit_gc_run_operation_threshold) of
+        undefined ->
+            Val = rabbit_misc:get_env(rabbit, lazy_queue_explicit_gc_run_operation_threshold,
+                ?DEFAULT_EXPLICIT_GC_RUN_OP_THRESHOLD),
+            put(explicit_gc_run_operation_threshold, Val),
+            Val;
+        Val       -> Val
+    end).
+
 %%----------------------------------------------------------------------------
 %% Public API
 %%----------------------------------------------------------------------------
@@ -633,25 +653,28 @@ ack([], State) ->
 %% optimisation: this head is essentially a partial evaluation of the
 %% general case below, for the single-ack case.
 ack([SeqId], State) ->
-    {#msg_status { msg_id        = MsgId,
-                   is_persistent = IsPersistent,
-                   msg_in_store  = MsgInStore,
-                   index_on_disk = IndexOnDisk },
-     State1 = #vqstate { index_state       = IndexState,
-                         msg_store_clients = MSCState,
-                         ack_out_counter   = AckOutCount }} =
-        remove_pending_ack(true, SeqId, State),
-    IndexState1 = case IndexOnDisk of
-                      true  -> rabbit_queue_index:ack([SeqId], IndexState);
-                      false -> IndexState
-                  end,
-    case MsgInStore of
-        true  -> ok = msg_store_remove(MSCState, IsPersistent, [MsgId]);
-        false -> ok
-    end,
-    {[MsgId],
-     a(State1 #vqstate { index_state      = IndexState1,
-                         ack_out_counter  = AckOutCount + 1 })};
+    case remove_pending_ack(true, SeqId, State) of
+        {none, _} ->
+            State;
+        {#msg_status { msg_id        = MsgId,
+                       is_persistent = IsPersistent,
+                       msg_in_store  = MsgInStore,
+                       index_on_disk = IndexOnDisk },
+         State1 = #vqstate { index_state       = IndexState,
+                             msg_store_clients = MSCState,
+                             ack_out_counter   = AckOutCount }} ->
+            IndexState1 = case IndexOnDisk of
+                              true  -> rabbit_queue_index:ack([SeqId], IndexState);
+                              false -> IndexState
+                          end,
+            case MsgInStore of
+                true  -> ok = msg_store_remove(MSCState, IsPersistent, [MsgId]);
+                false -> ok
+            end,
+            {[MsgId],
+             a(State1 #vqstate { index_state      = IndexState1,
+                                 ack_out_counter  = AckOutCount + 1 })}
+    end;
 ack(AckTags, State) ->
     {{IndexOnDiskSeqIds, MsgIdsByStore, AllMsgIds},
      State1 = #vqstate { index_state       = IndexState,
@@ -659,8 +682,12 @@ ack(AckTags, State) ->
                          ack_out_counter   = AckOutCount }} =
         lists:foldl(
           fun (SeqId, {Acc, State2}) ->
-                  {MsgStatus, State3} = remove_pending_ack(true, SeqId, State2),
-                  {accumulate_ack(MsgStatus, Acc), State3}
+                  case remove_pending_ack(true, SeqId, State2) of
+                      {none, _} ->
+                          {Acc, State2};
+                      {MsgStatus, State3} ->
+                          {accumulate_ack(MsgStatus, Acc), State3}
+                  end
           end, {accumulate_ack_init(), State}, AckTags),
     IndexState1 = rabbit_queue_index:ack(IndexOnDiskSeqIds, IndexState),
     remove_msgs_by_id(MsgIdsByStore, MSCState),
@@ -1330,7 +1357,8 @@ init(IsDurable, IndexState, DeltaCount, DeltaBytes, Terms,
 
       io_batch_size       = IoBatchSize,
 
-      mode                = default },
+      mode                = default,
+      memory_reduction_run_count = 0},
     a(maybe_deltas_to_betas(State)).
 
 blank_rates(Now) ->
@@ -1977,8 +2005,12 @@ lookup_pending_ack(SeqId, #vqstate { ram_pending_ack  = RPA,
 
 %% First parameter = UpdateStats
 remove_pending_ack(true, SeqId, State) ->
-    {MsgStatus, State1} = remove_pending_ack(false, SeqId, State),
-    {MsgStatus, stats({0, -1}, {MsgStatus, none}, State1)};
+    case remove_pending_ack(false, SeqId, State) of
+        {none, _} ->
+            {none, State};
+        {MsgStatus, State1} ->
+            {MsgStatus, stats({0, -1}, {MsgStatus, none}, State1)}
+    end;
 remove_pending_ack(false, SeqId, State = #vqstate{ram_pending_ack  = RPA,
                                                   disk_pending_ack = DPA,
                                                   qi_pending_ack   = QPA}) ->
@@ -1990,9 +2022,13 @@ remove_pending_ack(false, SeqId, State = #vqstate{ram_pending_ack  = RPA,
                               DPA1 = gb_trees:delete(SeqId, DPA),
                               {V, State#vqstate{disk_pending_ack = DPA1}};
                           none ->
-                              QPA1 = gb_trees:delete(SeqId, QPA),
-                              {gb_trees:get(SeqId, QPA),
-                               State#vqstate{qi_pending_ack = QPA1}}
+                              case gb_trees:lookup(SeqId, QPA) of
+                                  {value, V} ->
+                                      QPA1 = gb_trees:delete(SeqId, QPA),
+                                      {V, State#vqstate{qi_pending_ack = QPA1}};
+                                  none ->
+                                      {none, State}
+                              end
                       end
     end.
 
@@ -2143,11 +2179,15 @@ queue_merge([SeqId | Rest] = SeqIds, Q, Front, MsgIds,
                         Limit, PubFun, State);
         {_, _Q1} ->
             %% enqueue from the remaining list of sequence ids
-            {MsgStatus, State1} = msg_from_pending_ack(SeqId, State),
-            {#msg_status { msg_id = MsgId } = MsgStatus1, State2} =
-                PubFun(MsgStatus, State1),
-            queue_merge(Rest, Q, ?QUEUE:in(MsgStatus1, Front), [MsgId | MsgIds],
-                        Limit, PubFun, State2)
+            case msg_from_pending_ack(SeqId, State) of
+                {none, _} ->
+                    queue_merge(Rest, Q, Front, MsgIds, Limit, PubFun, State);
+                {MsgStatus, State1} ->
+                    {#msg_status { msg_id = MsgId } = MsgStatus1, State2} =
+                        PubFun(MsgStatus, State1),
+                    queue_merge(Rest, Q, ?QUEUE:in(MsgStatus1, Front), [MsgId | MsgIds],
+                                Limit, PubFun, State2)
+            end
     end;
 queue_merge(SeqIds, Q, Front, MsgIds,
             _Limit, _PubFun, State) ->
@@ -2156,22 +2196,28 @@ queue_merge(SeqIds, Q, Front, MsgIds,
 delta_merge([], Delta, MsgIds, State) ->
     {Delta, MsgIds, State};
 delta_merge(SeqIds, Delta, MsgIds, State) ->
-    lists:foldl(fun (SeqId, {Delta0, MsgIds0, State0}) ->
-                        {#msg_status { msg_id = MsgId } = MsgStatus, State1} =
-                            msg_from_pending_ack(SeqId, State0),
-                        {_MsgStatus, State2} =
-                            maybe_prepare_write_to_disk(true, true, MsgStatus, State1),
-                        {expand_delta(SeqId, Delta0), [MsgId | MsgIds0],
-                         stats({1, -1}, {MsgStatus, none}, State2)}
+    lists:foldl(fun (SeqId, {Delta0, MsgIds0, State0} = Acc) ->
+                        case msg_from_pending_ack(SeqId, State0) of
+                            {none, _} ->
+                                Acc;
+                        {#msg_status { msg_id = MsgId } = MsgStatus, State1} ->
+                                {_MsgStatus, State2} =
+                                    maybe_prepare_write_to_disk(true, true, MsgStatus, State1),
+                                {expand_delta(SeqId, Delta0), [MsgId | MsgIds0],
+                                 stats({1, -1}, {MsgStatus, none}, State2)}
+                        end
                 end, {Delta, MsgIds, State}, SeqIds).
 
 %% Mostly opposite of record_pending_ack/2
 msg_from_pending_ack(SeqId, State) ->
-    {#msg_status { msg_props = MsgProps } = MsgStatus, State1} =
-        remove_pending_ack(false, SeqId, State),
-    {MsgStatus #msg_status {
-       msg_props = MsgProps #message_properties { needs_confirming = false } },
-     State1}.
+    case remove_pending_ack(false, SeqId, State) of
+        {none, _} ->
+            {none, State};
+        {#msg_status { msg_props = MsgProps } = MsgStatus, State1} ->
+            {MsgStatus #msg_status {
+               msg_props = MsgProps #message_properties { needs_confirming = false } },
+             State1}
+    end.
 
 beta_limit(Q) ->
     case ?QUEUE:peek(Q) of
@@ -2264,6 +2310,14 @@ ifold(Fun, Acc, Its, State) ->
 %% Phase changes
 %%----------------------------------------------------------------------------
 
+maybe_execute_gc(State = #vqstate {memory_reduction_run_count = MRedRunCount}) ->
+    case MRedRunCount >= ?EXPLICIT_GC_RUN_OP_THRESHOLD of
+	true -> garbage_collect(),
+		 State#vqstate{memory_reduction_run_count =  0};
+        false ->    State#vqstate{memory_reduction_run_count =  MRedRunCount + 1}
+
+    end.
+
 reduce_memory_use(State = #vqstate { target_ram_count = infinity }) ->
     State;
 reduce_memory_use(State = #vqstate {
@@ -2336,8 +2390,7 @@ reduce_memory_use(State = #vqstate {
             S2 ->
                 push_betas_to_deltas(S2, State1)
         end,
-    garbage_collect(),
-    State3.
+    maybe_execute_gc(State3).
 
 limit_ram_acks(0, State) ->
     {0, ui(State)};