Fix bugs in previous commit, and be rather more thorough about monitoring and cleaning up credit flow.

4 files changed, 51 insertions, 25 deletions

diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index ad81ba032c..e5f1cb908b 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -174,7 +174,8 @@
 -spec(start_mirroring/1 :: (pid()) -> 'ok').
 -spec(stop_mirroring/1 :: (pid()) -> 'ok').
 -spec(sync_mirrors/1 :: (rabbit_types:amqqueue()) ->
-    'ok' | error('queue_has_pending_acks') | error('queue_not_mirrored')).
+    'ok' | rabbit_types:error('queue_has_pending_acks')
+         | rabbit_types:error('queue_not_mirrored')).
 
 -endif.
 
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 7b167a9a66..8c2fafa653 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -1179,8 +1179,9 @@ handle_call(sync_mirrors, From,
         0 -> {ok, #amqqueue{slave_pids = SPids, sync_slave_pids = SSPids}} =
                  rabbit_amqqueue:lookup(Name),
              gen_server2:reply(From, ok),
-             noreply(rabbit_mirror_queue_master:sync_mirrors(
-                       SPids -- SSPids, Name, BQS));
+             noreply(State#q{backing_queue_state =
+                                 rabbit_mirror_queue_master:sync_mirrors(
+                                   SPids -- SSPids, Name, BQS)});
         _ -> reply({error, queue_has_pending_acks}, State)
     end;
 
diff --git a/src/rabbit_mirror_queue_master.erl b/src/rabbit_mirror_queue_master.erl
index ea9e04fedf..542d724af1 100644
--- a/src/rabbit_mirror_queue_master.erl
+++ b/src/rabbit_mirror_queue_master.erl
@@ -144,44 +144,65 @@ sync_mirrors(SPids, Name, State = #state { backing_queue       = BQ,
     %% 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, BQS1} =
-        BQ:fold(fun ({Msg, MsgProps}, I) ->
-                        wait_for_credit(),
+    SPidsMRefs1 = sync_foreach(SPidsMRefs, Ref, fun sync_receive_ready/3),
+    {{Total, SPidsMRefs2}, BQS1} =
+        BQ:fold(fun ({Msg, MsgProps}, {I, SPMR}) ->
+                        SPMR1 = wait_for_credit(SPMR, Ref),
                         [begin
                              credit_flow:send(SPid, ?CREDIT_DISC_BOUND),
                              SPid ! {sync_message, Ref, Msg, MsgProps}
-                         end || SPid <- SPids1],
+                         end || {SPid, _} <- SPMR1],
                         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],
+                        {I + 1, SPMR1}
+                end, {0, SPidsMRefs1}, BQS),
+    sync_foreach(SPidsMRefs2, Ref, fun sync_receive_complete/3),
     rabbit_log:info("Synchronising ~s: ~p messages; complete~n",
                     [rabbit_misc:rs(Name), Total]),
     State#state{backing_queue_state = BQS1}.
 
-wait_for_credit() ->
+wait_for_credit(SPidsMRefs, Ref) ->
     case credit_flow:blocked() of
-        true  -> receive
-                     {bump_credit, Msg} -> credit_flow:handle_bump_msg(Msg),
-                                           wait_for_credit()
-                 end;
-        false -> ok
+        true  -> wait_for_credit(sync_foreach(SPidsMRefs, Ref,
+                                              fun sync_receive_credit/3), Ref);
+        false -> SPidsMRefs
     end.
 
+sync_foreach(SPidsMRefs, Ref, Fun) ->
+    [{SPid, MRef} || {SPid, MRef} <- SPidsMRefs,
+                     SPid1 <- [Fun(SPid, MRef, Ref)],
+                     SPid1 =/= dead].
+
+sync_receive_ready(SPid, MRef, Ref) ->
+    receive
+        {sync_ready, Ref, SPid}    -> SPid;
+        {'DOWN', MRef, _, SPid, _} -> dead
+    end.
+
+sync_receive_credit(SPid, MRef, Ref) ->
+    receive
+        {bump_credit, {SPid, _} = Msg} -> credit_flow:handle_bump_msg(Msg),
+                                          sync_receive_credit(SPid, MRef, Ref);
+        {'DOWN', MRef, _, SPid, _}     -> credit_flow:peer_down(SPid),
+                                          dead
+    after 0 ->
+            SPid
+    end.
+
+sync_receive_complete(SPid, MRef, Ref) ->
+    SPid ! {sync_complete, Ref},
+    receive
+        {sync_complete_ok, Ref, SPid} -> ok;
+        {'DOWN', MRef, _, SPid, _}    -> ok
+    end,
+    erlang:demonitor(MRef, [flush]),
+    credit_flow:peer_down(SPid).
+
+
 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 f4130e0250..311c6ca658 100644
--- a/src/rabbit_mirror_queue_slave.erl
+++ b/src/rabbit_mirror_queue_slave.erl
@@ -851,12 +851,15 @@ sync_loop(Ref, MRef, MPid, State = #state{backing_queue       = BQ,
             %% messages from it, we have a hole in the middle. So the
             %% only thing to do here is purge.)
             {_MsgCount, BQS1} = BQ:purge(BQS),
+            credit_flow:peer_down(MPid),
             State#state{backing_queue_state = BQS1};
         {bump_credit, Msg} ->
             credit_flow:handle_bump_msg(Msg),
             sync_loop(Ref, MRef, MPid, State);
         {sync_complete, Ref} ->
+            MPid ! {sync_complete_ok, Ref, self()},
             erlang:demonitor(MRef),
+            credit_flow:peer_down(MPid),
             %% We can only sync when there are no pending acks
             set_delta(0, State);
         {sync_message, Ref, Msg, Props0} ->