Merge pull request #3286 from rabbitmq/mergify/bp/v3.8.x/pr-3284

Revert "CQ: don't deliver right before acking in the index" (backport #3284)

deps/rabbit/src/rabbit_queue_index.erl

@@ -841,9 +841,9 @@ action_to_entry(RelSeq, Action, JEntries) ->
              end};
         ({Pub,    no_del, no_ack}) when Action == del ->
             {set, {Pub,    del, no_ack}};
-        ({no_pub,    _del, no_ack}) when Action == ack ->
+        ({no_pub,    del, no_ack}) when Action == ack ->
             {set, {no_pub, del,    ack}};
-        ({?PUB,      _del, no_ack}) when Action == ack ->
+        ({?PUB,      del, no_ack}) when Action == ack ->
             {reset, none}
     end.
 

deps/rabbit/src/rabbit_variable_queue.erl

@@ -1774,21 +1774,22 @@ purge_betas_and_deltas(DelsAndAcksFun, State = #vqstate { mode = Mode }) ->
 
 remove_queue_entries(Q, DelsAndAcksFun,
                      State = #vqstate{msg_store_clients = MSCState}) ->
-    {MsgIdsByStore, Acks, State1} =
+    {MsgIdsByStore, Delivers, Acks, State1} =
         ?QUEUE:foldl(fun remove_queue_entries1/2,
-                     {maps:new(), [], State}, Q),
+                     {maps:new(), [], [], State}, Q),
     remove_msgs_by_id(MsgIdsByStore, MSCState),
-    DelsAndAcksFun([], Acks, State1).
+    DelsAndAcksFun(Delivers, Acks, State1).
 
 remove_queue_entries1(
-  #msg_status { msg_id = MsgId, seq_id = SeqId,
+  #msg_status { msg_id = MsgId, seq_id = SeqId, is_delivered = IsDelivered,
                 msg_in_store = MsgInStore, index_on_disk = IndexOnDisk,
                 is_persistent = IsPersistent} = MsgStatus,
-  {MsgIdsByStore, Acks, State}) ->
+  {MsgIdsByStore, Delivers, Acks, State}) ->
     {case MsgInStore of
          true  -> rabbit_misc:maps_cons(IsPersistent, MsgId, MsgIdsByStore);
          false -> MsgIdsByStore
      end,
+     cons_if(IndexOnDisk andalso not IsDelivered, SeqId, Delivers),
      cons_if(IndexOnDisk, SeqId, Acks),
      stats({-1, 0}, {MsgStatus, none}, 0, State)}.