Fix quorum queue drop-head dead letter order

Prior to this commit, quorum queue at-most-once dead lettering
for the overflow behaviour `drop-head` was dead lettering in the wrong order.

(cherry picked from commit 960918e)
(cherry picked from commit 117652b)

Author: ansd

deps/rabbit/src/rabbit_fifo.erl

@@ -1639,30 +1639,32 @@ drop_head(#?STATE{ra_indexes = Indexes0} = State0, Effects) ->
             #?STATE{cfg = #cfg{dead_letter_handler = DLH},
                     dlx = DlxState} = State = State3,
             {_, DlxEffects} = rabbit_fifo_dlx:discard([Msg], maxlen, DLH, DlxState),
-            {State, combine_effects(DlxEffects, Effects)};
+            {State, add_drop_head_effects(DlxEffects, Effects)};
         empty ->
             {State0, Effects}
     end.
 
-%% combine global counter update effects to avoid bulding a huge list of
-%% effects if many messages are dropped at the same time as could happen
-%% when the `max_length' is changed via a configuration update.
-combine_effects([{mod_call,
-                  rabbit_global_counters,
-                  messages_dead_lettered,
-                  [Reason, rabbit_quorum_queue, Type, NewLen]}],
-                [{mod_call,
-                  rabbit_global_counters,
-                  messages_dead_lettered,
-                  [Reason, rabbit_quorum_queue, Type, PrevLen]} | Rem]) ->
+add_drop_head_effects([{mod_call,
+                        rabbit_global_counters,
+                        messages_dead_lettered,
+                        [Reason, rabbit_quorum_queue, Type, NewLen]}],
+                      [{mod_call,
+                        rabbit_global_counters,
+                        messages_dead_lettered,
+                        [Reason, rabbit_quorum_queue, Type, PrevLen]} | Rem]) ->
+    %% combine global counter update effects to avoid bulding a huge list of
+    %% effects if many messages are dropped at the same time as could happen
+    %% when the `max_length' is changed via a configuration update.
     [{mod_call,
       rabbit_global_counters,
       messages_dead_lettered,
       [Reason, rabbit_quorum_queue, Type, PrevLen + NewLen]} | Rem];
-combine_effects(New, Old) ->
+add_drop_head_effects([{log, _, _}] = DlxEffs, Effs) ->
+    %% dead letter in the correct order
+    Effs ++ DlxEffs;
+add_drop_head_effects(New, Old) ->
     New ++ Old.
 
-
 maybe_set_msg_ttl(Msg, RaCmdTs, Header,
                   #?STATE{cfg = #cfg{msg_ttl = MsgTTL}}) ->
     case mc:is(Msg) of

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -167,6 +167,7 @@ all_tests() ->
      subscribe_redelivery_count,
      message_bytes_metrics,
      queue_length_limit_drop_head,
+     queue_length_bytes_limit_drop_head,
      queue_length_limit_reject_publish,
      queue_length_limit_policy_cleared,
      subscribe_redelivery_limit,
@@ -3677,6 +3678,50 @@ queue_length_limit_drop_head(Config) ->
                  amqp_channel:call(Ch, #'basic.get'{queue = QQ,
                                                     no_ack = true})).
 
+queue_length_bytes_limit_drop_head(Config) ->
+    [Server | _] = Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+    QQ = ?config(queue_name, Config),
+    DLQ = <<"dead letter queue">>,
+
+    ?assertEqual({'queue.declare_ok', DLQ, 0, 0},
+                 declare(Ch, DLQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>},
+                                  {<<"x-overflow">>, longstr, <<"drop-head">>},
+                                  {<<"x-max-length-bytes">>, long, 1000},
+                                  {<<"x-dead-letter-exchange">>, longstr, <<>>},
+                                  {<<"x-dead-letter-routing-key">>, longstr, DLQ}])),
+
+    LargePayload = binary:copy(<<"x">>, 1500),
+    ok = amqp_channel:cast(Ch,
+                           #'basic.publish'{routing_key = QQ},
+                           #amqp_msg{payload = <<"m1">>}),
+    ok = amqp_channel:cast(Ch,
+                           #'basic.publish'{routing_key = QQ},
+                           #amqp_msg{payload = <<"m2">>}),
+    ok = amqp_channel:cast(Ch,
+                           #'basic.publish'{routing_key = QQ},
+                           #amqp_msg{payload = LargePayload}),
+    wait_for_consensus(QQ, Config),
+    wait_for_consensus(DLQ, Config),
+    RaName = ra_name(DLQ),
+    wait_for_messages_ready(Servers, RaName, 3),
+    ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"m1">>}},
+                 amqp_channel:call(Ch, #'basic.get'{queue = DLQ,
+                                                    no_ack = true})),
+    ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"m2">>}},
+                 amqp_channel:call(Ch, #'basic.get'{queue = DLQ,
+                                                    no_ack = true})),
+    ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = LargePayload}},
+                 amqp_channel:call(Ch, #'basic.get'{queue = DLQ,
+                                                    no_ack = true})),
+
+    [?assertEqual(#'queue.delete_ok'{message_count = 0},
+                  amqp_channel:call(Ch, #'queue.delete'{queue = Q}))
+     || Q <- [QQ, DLQ]].
+
 queue_length_limit_reject_publish(Config) ->
     [Server | _] = Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),