Merge bug25303

4 files changed, 114 insertions, 2 deletions

diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 8ce1160ca7..4d95778947 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -31,7 +31,7 @@
 -export([notify_down_all/2, limit_all/3]).
 -export([on_node_down/1]).
 -export([update/2, store_queue/1, policy_changed/2]).
--export([start_mirroring/1, stop_mirroring/1]).
+-export([start_mirroring/1, stop_mirroring/1, sync_mirrors/1]).
 
 %% internal
 -export([internal_declare/2, internal_delete/2, run_backing_queue/3,
@@ -591,6 +591,12 @@ set_maximum_since_use(QPid, Age) ->
 start_mirroring(QPid) -> ok = delegate_call(QPid, start_mirroring).
 stop_mirroring(QPid)  -> ok = delegate_call(QPid, stop_mirroring).
 
+sync_mirrors(Name) ->
+    case lookup(Name) of
+        {ok, #amqqueue{pid = QPid}} -> delegate_cast(QPid, sync_mirrors);
+        _                           -> ok
+    end.
+
 on_node_down(Node) ->
     rabbit_misc:execute_mnesia_tx_with_tail(
       fun () -> QsDels =
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index b3c44a50a1..07f4c3b190 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -1300,6 +1300,20 @@ handle_cast({dead_letter, Msgs, Reason}, State = #q{dlx = XName}) ->
             cleanup_after_confirm([AckTag || {_, AckTag} <- Msgs], State)
     end;
 
+handle_cast(sync_mirrors,
+            State = #q{q                   = #amqqueue{name = Name},
+                       backing_queue       = BQ,
+                       backing_queue_state = BQS}) ->
+    case BQ of
+        rabbit_mirror_queue_master ->
+            {ok, #amqqueue{slave_pids = SPids, sync_slave_pids = SSPids}} =
+                rabbit_amqqueue:lookup(Name),
+            rabbit_mirror_queue_master:sync_mirrors(SPids -- SSPids, Name, BQS);
+        _ ->
+            ok
+    end,
+    noreply(State);
+
 handle_cast(wake_up, State) ->
     noreply(State).
 
diff --git a/src/rabbit_mirror_queue_master.erl b/src/rabbit_mirror_queue_master.erl
index 15aeea011b..a9f5e5ac34 100644
--- a/src/rabbit_mirror_queue_master.erl
+++ b/src/rabbit_mirror_queue_master.erl
@@ -28,7 +28,7 @@
 
 -export([promote_backing_queue_state/7, sender_death_fun/0, depth_fun/0]).
 
--export([init_with_existing_bq/3, stop_mirroring/1]).
+-export([init_with_existing_bq/3, stop_mirroring/1, sync_mirrors/3]).
 
 -behaviour(rabbit_backing_queue).
 
@@ -127,6 +127,61 @@ stop_mirroring(State = #state { coordinator         = CPid,
     stop_all_slaves(shutdown, State),
     {BQ, BQS}.
 
+sync_mirrors([], Name, _State) ->
+    rabbit_log:info("Synchronising ~s: nothing to do~n",
+                    [rabbit_misc:rs(Name)]),
+    ok;
+sync_mirrors(SPids, Name, #state { backing_queue       = BQ,
+                                   backing_queue_state = BQS }) ->
+    rabbit_log:info("Synchronising ~s with slaves ~p~n",
+                    [rabbit_misc:rs(Name), SPids]),
+    Ref = make_ref(),
+    SPidsMRefs = [begin
+                      SPid ! {sync_start, Ref, self()},
+                      MRef = erlang:monitor(process, SPid),
+                      {SPid, MRef}
+                  end || SPid <- SPids],
+    %% We wait for a reply from the slaves so that we know they are in
+    %% a receive block and will thus receive messages we send to them
+    %% *without* those messages ending up in their gen_server2 pqueue.
+    SPids1 = [SPid1 || {SPid, MRef} <- SPidsMRefs,
+                       SPid1 <- [receive
+                                     {'DOWN', MRef, process, SPid, _Reason} ->
+                                         dead;
+                                     {sync_ready, Ref, SPid} ->
+                                         SPid
+                                 end],
+                       SPid1 =/= dead],
+    [erlang:demonitor(MRef) || {_, MRef} <- SPidsMRefs],
+    {Total, _BQS} =
+        BQ:fold(fun (M = #basic_message{}, I) ->
+                        wait_for_credit(),
+                        [begin
+                             credit_flow:send(SPid, ?CREDIT_DISC_BOUND),
+                             SPid ! {sync_message, Ref, M}
+                         end || SPid <- SPids1],
+                        case I rem 1000 of
+                            0 -> rabbit_log:info(
+                                   "Synchronising ~s: ~p messages~n",
+                                   [rabbit_misc:rs(Name), I]);
+                            _ -> ok
+                        end,
+                        I + 1
+                end, 0, BQS),
+    [SPid ! {sync_complete, Ref} || SPid <- SPids1],
+    rabbit_log:info("Synchronising ~s: ~p messages; complete~n",
+                    [rabbit_misc:rs(Name), Total]),
+    ok.
+
+wait_for_credit() ->
+    case credit_flow:blocked() of
+        true  -> receive
+                     {bump_credit, Msg} -> credit_flow:handle_bump_msg(Msg),
+                                           wait_for_credit()
+                 end;
+        false -> ok
+    end.
+
 terminate({shutdown, dropped} = Reason,
           State = #state { backing_queue       = BQ,
                            backing_queue_state = BQS }) ->
diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl
index 3ad8eb7785..a8615cee4b 100644
--- a/src/rabbit_mirror_queue_slave.erl
+++ b/src/rabbit_mirror_queue_slave.erl
@@ -264,6 +264,15 @@ handle_info({bump_credit, Msg}, State) ->
     credit_flow:handle_bump_msg(Msg),
     noreply(State);
 
+handle_info({sync_start, Ref, MPid},
+            State = #state { backing_queue       = BQ,
+                             backing_queue_state = BQS }) ->
+    MRef = erlang:monitor(process, MPid),
+    MPid ! {sync_ready, Ref, self()},
+    {_MsgCount, BQS1} = BQ:purge(BQS),
+    noreply(
+      sync_loop(Ref, MRef, MPid, State#state{backing_queue_state = BQS1}));
+
 handle_info(Msg, State) ->
     {stop, {unexpected_info, Msg}, State}.
 
@@ -830,3 +839,31 @@ record_synchronised(#amqqueue { name = QName }) ->
                       ok
               end
       end).
+
+sync_loop(Ref, MRef, MPid, State = #state{backing_queue       = BQ,
+                                    backing_queue_state = BQS}) ->
+    receive
+        {'DOWN', MRef, process, MPid, _Reason} ->
+            %% If the master dies half way we are not in the usual
+            %% half-synced state (with messages nearer the tail of the
+            %% queue; instead we have ones nearer the head. If we then
+            %% sync with a newly promoted master, or even just receive
+            %% messages from it, we have a hole in the middle. So the
+            %% only thing to do here is purge.)
+            {_MsgCount, BQS1} = BQ:purge(BQS),
+            State#state{backing_queue_state = BQS1};
+        {bump_credit, Msg} ->
+            credit_flow:handle_bump_msg(Msg),
+            sync_loop(Ref, MRef, MPid, State);
+        {sync_complete, Ref} ->
+            erlang:demonitor(MRef),
+            set_delta(0, State);
+        {sync_message, Ref, M} ->
+            credit_flow:ack(MPid, ?CREDIT_DISC_BOUND),
+            %% TODO expiry needs fixing
+            Props = #message_properties{expiry           = undefined,
+                                        needs_confirming = false,
+                                        delivered        = true},
+            BQS1 = BQ:publish(M, Props, none, BQS),
+            sync_loop(Ref, MRef, MPid, State#state{backing_queue_state = BQS1})
+    end.