Merge pull request #1809 from rabbitmq/qq-purge-bug-fix

Quorum queue: purge should not remove checkout out messages.

Author: dcorbacho

src/rabbit_fifo.erl

@@ -413,26 +413,20 @@ apply(_, #checkout{spec = Spec, meta = Meta,
     State1 = update_consumer(ConsumerId, Meta, Spec, State0),
     checkout(State1, [{monitor, process, Pid}]);
 apply(#{index := RaftIdx}, #purge{},
-      #state{consumers = Cons0, ra_indexes = Indexes } = State0) ->
-    Total = rabbit_fifo_index:size(Indexes),
-    {State1, Effects1} =
-        maps:fold(
-          fun(ConsumerId, C = #consumer{checked_out = Checked0},
-              {StateAcc0, EffectsAcc0}) ->
-                  MsgRaftIdxs = [RIdx || {_MsgInId, {RIdx, _}}
-                                             <- maps:values(Checked0)],
-                  complete(ConsumerId, MsgRaftIdxs, maps:size(Checked0), C,
-                           #{}, EffectsAcc0, StateAcc0)
-          end, {State0, []}, Cons0),
-        {State, _, Effects} =
-            update_smallest_raft_index(
-              RaftIdx, Indexes,
-              State1#state{ra_indexes = rabbit_fifo_index:empty(),
-                           messages = #{},
-                           returns = lqueue:new(),
-                           msg_bytes_enqueue = 0,
-                           msg_bytes_checkout = 0,
-                           low_msg_num = undefined}, Effects1),
+      #state{ra_indexes = Indexes0,
+             messages = Messages} = State0) ->
+    Total = maps:size(Messages),
+    Indexes = lists:foldl(fun rabbit_fifo_index:delete/2,
+                          Indexes0,
+                          [I || {I, _} <- lists:sort(maps:values(Messages))]),
+    {State, _, Effects} =
+        update_smallest_raft_index(RaftIdx, Indexes0,
+                                   State0#state{ra_indexes = Indexes,
+                                                messages = #{},
+                                                returns = lqueue:new(),
+                                                msg_bytes_enqueue = 0,
+                                                low_msg_num = undefined},
+                                   []),
     %% as we're not checking out after a purge (no point) we have to
     %% reverse the effects ourselves
     {State, {purge, Total},
@@ -1876,11 +1870,12 @@ purge_with_checkout_test() ->
     %% assert message bytes are non zero
     ?assert(State2#state.msg_bytes_checkout > 0),
     ?assert(State2#state.msg_bytes_enqueue > 0),
-    {State3, {purge, 2}, _} = apply(meta(2), make_purge(), State2),
-    ?assertEqual(0, State3#state.msg_bytes_checkout),
+    {State3, {purge, 1}, _} = apply(meta(2), make_purge(), State2),
+    ?assert(State2#state.msg_bytes_checkout > 0),
     ?assertEqual(0, State3#state.msg_bytes_enqueue),
+    ?assertEqual(1, rabbit_fifo_index:size(State3#state.ra_indexes)),
     #consumer{checked_out = Checked} = maps:get(Cid, State3#state.consumers),
-    ?assertEqual(0, maps:size(Checked)),
+    ?assertEqual(1, maps:size(Checked)),
     ok.
 
 down_returns_checked_out_in_order_test() ->

src/rabbit_fifo_index.erl

@@ -56,10 +56,10 @@ return(Key, Value, #?MODULE{data = Data} = State)
   when is_integer(Key) ->
     State#?MODULE{data = maps:put(Key, Value, Data)}.
 
--spec delete(integer(), state()) -> state().
+-spec delete(Index :: integer(), state()) -> state().
 delete(Smallest, #?MODULE{data = Data0,
-                        largest = Largest,
-                        smallest = Smallest} = State) ->
+                          largest = Largest,
+                          smallest = Smallest} = State) ->
     Data = maps:remove(Smallest, Data0),
     case find_next(Smallest + 1, Largest, Data) of
         undefined ->

test/quorum_queue_SUITE.erl

@@ -1539,8 +1539,8 @@ purge(Config) ->
     _DeliveryTag = consume(Ch, QQ, false),
     wait_for_messages_ready(Servers, RaName, 1),
     wait_for_messages_pending_ack(Servers, RaName, 1),
-    {'queue.purge_ok', 2} = amqp_channel:call(Ch, #'queue.purge'{queue = QQ}),
-    wait_for_messages_pending_ack(Servers, RaName, 0),
+    {'queue.purge_ok', 1} = amqp_channel:call(Ch, #'queue.purge'{queue = QQ}),
+    wait_for_messages_pending_ack(Servers, RaName, 1),
     wait_for_messages_ready(Servers, RaName, 0).
 
 sync_queue(Config) ->