Eager mirror synchronisation.

4 files changed, 81 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..68ae481674 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, 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 39060c09cb..bc2d21acd8 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/2]).
 
 -behaviour(rabbit_backing_queue).
 
@@ -127,6 +127,33 @@ stop_mirroring(State = #state { coordinator         = CPid,
     stop_all_slaves(shutdown, State),
     {BQ, BQS}.
 
+sync_mirrors(SPids, #state { backing_queue       = BQ,
+                             backing_queue_state = BQS }) ->
+    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],
+    BQ:fold(fun (M = #basic_message{}, none) ->
+                    [SPid ! {sync_message, Ref, M} || SPid <- SPids1],
+                    none
+            end, none, BQS),
+    [SPid ! {sync_complete, Ref} || SPid <- SPids1],
+    ok.
+
 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..cd2b205c56 100644
--- a/src/rabbit_mirror_queue_slave.erl
+++ b/src/rabbit_mirror_queue_slave.erl
@@ -264,6 +264,14 @@ 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, State#state{backing_queue_state = BQS1}));
+
 handle_info(Msg, State) ->
     {stop, {unexpected_info, Msg}, State}.
 
@@ -830,3 +838,27 @@ record_synchronised(#amqqueue { name = QName }) ->
                       ok
               end
       end).
+
+sync_loop(Ref, MRef, 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.)
+            State#state{backing_queue_state = BQ:purge(BQS)};
+        {sync_complete, Ref} ->
+            erlang:demonitor(MRef),
+            set_delta(0, State);
+        {sync_message, Ref, M} ->
+            %% TODO expiry / delivered need fixing
+            Props = #message_properties{expiry           = undefined,
+                                        needs_confirming = false,
+                                        delivered        = false},
+            BQS1 = BQ:publish(M, Props, none, BQS),
+            sync_loop(Ref, MRef, State#state{backing_queue_state = BQS1})
+    end.
+