Fix message IDs settlement order

ansd · ansd · commit b5a0b17be51c · 2024-06-25T12:08:07.000+02:00
## What? This commit fixes issues that were present only on `main` branch and were introduced by #9022. 1. Classic queues (specifically `rabbit_queue_consumers:subtract_acks/3`) expect message IDs to be (n)acked in the order as they were delivered to the channel / session proc. Hence, the `lists:usort(MsgIds0)` in `rabbit_classic_queue:settle/5` was wrong causing not all messages to be acked adding a regression to also AMQP 0.9.1. 2. The order in which the session proc requeues or rejects multiple message IDs at once is important. For example, if the client sends a DISPOSITION with first=3 and last=5, the message IDs corresponding to delivery IDs 3,4,5 must be requeued or rejected in exactly that order. For example, quorum queues use this order of message IDs in https://github.com/rabbitmq/rabbitmq-server/blob/34d3f943742bdcf7d34859edff8d45f35e4007d4/deps/rabbit/src/rabbit_fifo.erl#L226-L234 to dead letter in that order. ## How? The session proc will settle (internal queue) message IDs in ascending (AMQP) delivery ID order, i.e. in the order messages were sent to the client and in the order messages were settled by the client. This commit chooses to keep the session's outgoing_unsettled_map map data structure. An alternative would have been to use a queue or lqueue for the outgoing_unsettled_map as done in * https://github.com/rabbitmq/rabbitmq-server/blob/34d3f943742bdcf7d34859edff8d45f35e4007d4/deps/rabbit/src/rabbit_channel.erl#L135 * https://github.com/rabbitmq/rabbitmq-server/blob/34d3f943742bdcf7d34859edff8d45f35e4007d4/deps/rabbit/src/rabbit_queue_consumers.erl#L43 Whether a queue (as done by `rabbit_channel`) or a map (as done by `rabbit_amqp_session`) performs better depends on the pattern how clients ack messages. A queue will likely perform good enough because usually the oldest delivered messages will be acked first. However, given that there can be many different consumers on an AQMP 0.9.1 channel or AMQP 1.0 session, this commit favours a map because it will likely generate less garbage and is very efficient when for example a new single message (or few new messages) gets acked while many (older) messages are still checked out by the session (but by possibly different AMQP 1.0 receivers).
diff --git a/deps/amqp10_common/src/serial_number.erl b/deps/amqp10_common/src/serial_number.erl
@@ -59,10 +59,7 @@ compare(A, B) ->
     [serial_number()].
 usort(L) ->
     lists:usort(fun(A, B) ->
-                        case compare(A, B) of
-                            greater -> false;
-                            _ -> true
-                        end
+                        compare(A, B) =/= greater
                 end, L).
 
 %% Takes a list of serial numbers and returns tuples
diff --git a/deps/rabbit/BUILD.bazel b/deps/rabbit/BUILD.bazel
@@ -402,6 +402,10 @@ rabbitmq_integration_suite(
     shard_count = 6,
 )
 
+rabbitmq_integration_suite(
+    name = "amqpl_consumer_ack_SUITE",
+)
+
 rabbitmq_integration_suite(
     name = "message_containers_deaths_v2_SUITE",
     size = "medium",
diff --git a/deps/rabbit/app.bzl b/deps/rabbit/app.bzl
@@ -2134,3 +2134,12 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = ["//deps/amqp_client:erlang_app"],
     )
+    erlang_bytecode(
+        name = "amqpl_consumer_ack_SUITE_beam_files",
+        testonly = True,
+        srcs = ["test/amqpl_consumer_ack_SUITE.erl"],
+        outs = ["test/amqpl_consumer_ack_SUITE.beam"],
+        app_name = "rabbit",
+        erlc_opts = "//:test_erlc_opts",
+        deps = ["//deps/amqp_client:erlang_app"],
+    )
diff --git a/deps/rabbit/src/rabbit_amqp_session.erl b/deps/rabbit/src/rabbit_amqp_session.erl
@@ -1307,9 +1307,13 @@ handle_control(#'v1_0.disposition'{role = ?AMQP_ROLE_RECEIVER,
             case DispositionRangeSize =< UnsettledMapSize of
                 true ->
                     %% It is cheaper to iterate over the range of settled delivery IDs.
-                    serial_number:foldl(fun settle_delivery_id/2, {#{}, UnsettledMap0}, First, Last);
+                    serial_number:foldl(fun settle_delivery_id/2,
+                                        {#{}, UnsettledMap0},
+                                        First, Last);
                 false ->
                     %% It is cheaper to iterate over the outgoing unsettled map.
+                    Iter = maps:iterator(UnsettledMap0,
+                                         fun(A, B) -> compare(A, B) =/= greater end),
                     {Settled0, UnsettledList} =
                     maps:fold(
                       fun (DeliveryId,
@@ -1329,14 +1333,15 @@ handle_control(#'v1_0.disposition'{role = ?AMQP_ROLE_RECEIVER,
                                       {SettledAcc, [{DeliveryId, Unsettled} | UnsettledAcc]}
                               end
                       end,
-                      {#{}, []}, UnsettledMap0),
+                      {#{}, []}, Iter),
                     {Settled0, maps:from_list(UnsettledList)}
             end,
 
             SettleOp = settle_op_from_outcome(Outcome),
             {QStates, Actions} =
             maps:fold(
-              fun({QName, Ctag}, MsgIds, {QS0, ActionsAcc}) ->
+              fun({QName, Ctag}, MsgIdsRev, {QS0, ActionsAcc}) ->
+                      MsgIds = lists:reverse(MsgIdsRev),
                       case rabbit_queue_type:settle(QName, SettleOp, Ctag, MsgIds, QS0) of
                           {ok, QS, Actions0} ->
                               messages_acknowledged(SettleOp, QName, QS, MsgIds),
diff --git a/deps/rabbit/src/rabbit_channel.erl b/deps/rabbit/src/rabbit_channel.erl
@@ -1972,7 +1972,7 @@ collect_acks(AcknowledgedAcc, RemainingAcc, UAMQ, DeliveryTag, Multiple) ->
     end.
 
 %% Settles (acknowledges) messages at the queue replica process level.
-%% This happens in the youngest-first order (ascending by delivery tag).
+%% This happens in the oldest-first order (ascending by delivery tag).
 settle_acks(Acks, State = #ch{queue_states = QueueStates0}) ->
     {QueueStates, Actions} =
         foreach_per_queue(
diff --git a/deps/rabbit/src/rabbit_classic_queue.erl b/deps/rabbit/src/rabbit_classic_queue.erl
@@ -275,9 +275,7 @@ cancel(Q, ConsumerTag, OkMsg, ActingUser, State) ->
 -spec settle(rabbit_amqqueue:name(), rabbit_queue_type:settle_op(),
              rabbit_types:ctag(), [non_neg_integer()], state()) ->
     {state(), rabbit_queue_type:actions()}.
-settle(_QName, Op, _CTag, MsgIds0, State = #?STATE{pid = Pid}) ->
-    %% Classic queues expect message IDs in sorted order.
-    MsgIds = lists:usort(MsgIds0),
+settle(_QName, Op, _CTag, MsgIds, State = #?STATE{pid = Pid}) ->
     Arg = case Op of
               complete ->
                   {ack, MsgIds, self()};
diff --git a/deps/rabbit/src/rabbit_queue_consumers.erl b/deps/rabbit/src/rabbit_queue_consumers.erl
@@ -40,8 +40,8 @@
 %% channel record
 -record(cr, {ch_pid,
              monitor_ref,
-             acktags,
-             consumer_count,
+             acktags :: ?QUEUE:?QUEUE({ack(), rabbit_types:ctag() | none}),
+             consumer_count :: non_neg_integer(),
              %% Queue of {ChPid, #consumer{}} for consumers which have
              %% been blocked (rate/prefetch limited) for any reason
              blocked_consumers,
diff --git a/deps/rabbit/test/amqp_client_SUITE.erl b/deps/rabbit/test/amqp_client_SUITE.erl
@@ -108,6 +108,12 @@ groups() ->
        classic_priority_queue,
        dead_letter_headers_exchange,
        dead_letter_reject,
+       dead_letter_reject_message_order_classic_queue,
+       dead_letter_reject_message_order_quorum_queue,
+       accept_multiple_message_order_classic_queue,
+       accept_multiple_message_order_quorum_queue,
+       release_multiple_message_order_classic_queue,
+       release_multiple_message_order_quorum_queue,
        immutable_bare_message,
        receive_many_made_available_over_time_classic_queue,
        receive_many_made_available_over_time_quorum_queue,
@@ -158,7 +164,10 @@ groups() ->
 init_per_suite(Config) ->
     {ok, _} = application:ensure_all_started(amqp10_client),
     rabbit_ct_helpers:log_environment(),
-    Config.
+    rabbit_ct_helpers:merge_app_env(
+      Config, {rabbit, [{quorum_tick_interval, 1000},
+                        {stream_tick_interval, 1000}
+                       ]}).
 
 end_per_suite(Config) ->
     Config.
@@ -3747,6 +3756,97 @@ dead_letter_reject(Config) ->
     ok = end_session_sync(Session),
     ok = amqp10_client:close_connection(Connection).
 
+dead_letter_reject_message_order_classic_queue(Config) ->
+    dead_letter_reject_message_order(<<"classic">>, Config).
+
+dead_letter_reject_message_order_quorum_queue(Config) ->
+    dead_letter_reject_message_order(<<"quorum">>, Config).
+
+dead_letter_reject_message_order(QType, Config) ->
+    {Connection, Session, LinkPair} = init(Config),
+    QName1 = <<"q1">>,
+    QName2 = <<"q2">>,
+    {ok, #{type := QType}} = rabbitmq_amqp_client:declare_queue(
+                               LinkPair,
+                               QName1,
+                               #{arguments => #{<<"x-queue-type">> => {utf8, QType},
+                                                <<"x-dead-letter-exchange">> => {utf8, <<>>},
+                                                <<"x-dead-letter-routing-key">> => {utf8, QName2}
+                                               }}),
+    %% We don't care about the target dead letter queue type.
+    {ok, _} = rabbitmq_amqp_client:declare_queue(LinkPair, QName2, #{}),
+
+    {ok, Sender} = amqp10_client:attach_sender_link(
+                     Session, <<"sender">>, <<"/queue/", QName1/binary>>, unsettled),
+    wait_for_credit(Sender),
+    {ok, Receiver1} = amqp10_client:attach_receiver_link(
+                        Session, <<"receiver 1">>, <<"/queue/", QName1/binary>>, unsettled),
+    {ok, Receiver2} = amqp10_client:attach_receiver_link(
+                        Session, <<"receiver 2">>, <<"/queue/", QName2/binary>>, unsettled),
+
+    [begin
+         Bin = integer_to_binary(N),
+         Msg = amqp10_msg:new(Bin, Bin, true),
+         ok = amqp10_client:send_msg(Sender, Msg)
+     end || N <- lists:seq(1, 5)],
+
+    {ok, Msg1} = amqp10_client:get_msg(Receiver1),
+    {ok, Msg2} = amqp10_client:get_msg(Receiver1),
+    {ok, _Msg3} = amqp10_client:get_msg(Receiver1),
+    {ok, Msg4} = amqp10_client:get_msg(Receiver1),
+    {ok, Msg5} = amqp10_client:get_msg(Receiver1),
+    assert_messages(QName1, 5, 5, Config),
+
+    %% Reject messages in the following order: 2, 3, 4, 1, 5
+    ok = amqp10_client_session:disposition(
+           Receiver1,
+           amqp10_msg:delivery_id(Msg2),
+           amqp10_msg:delivery_id(Msg4),
+           true,
+           rejected),
+    ok = amqp10_client_session:disposition(
+           Receiver1,
+           amqp10_msg:delivery_id(Msg1),
+           amqp10_msg:delivery_id(Msg5),
+           true,
+           rejected),
+
+    assert_messages(QName1, 0, 0, Config),
+    %% All 5 messages should be in the dead letter queue.
+    assert_messages(QName2, 5, 0, Config),
+
+    {ok, MsgDead2} = amqp10_client:get_msg(Receiver2),
+    {ok, MsgDead3} = amqp10_client:get_msg(Receiver2),
+    {ok, MsgDead4} = amqp10_client:get_msg(Receiver2),
+    {ok, MsgDead1} = amqp10_client:get_msg(Receiver2),
+    {ok, MsgDead5} = amqp10_client:get_msg(Receiver2),
+    assert_messages(QName2, 5, 5, Config),
+
+    %% Messages should be dead lettered in the order we rejected.
+    ?assertEqual(<<"2">>, amqp10_msg:body_bin(MsgDead2)),
+    ?assertEqual(<<"3">>, amqp10_msg:body_bin(MsgDead3)),
+    ?assertEqual(<<"4">>, amqp10_msg:body_bin(MsgDead4)),
+    ?assertEqual(<<"1">>, amqp10_msg:body_bin(MsgDead1)),
+    ?assertEqual(<<"5">>, amqp10_msg:body_bin(MsgDead5)),
+
+    %% Accept all messages in the dead letter queue.
+    ok = amqp10_client_session:disposition(
+           Receiver2,
+           amqp10_msg:delivery_id(MsgDead2),
+           amqp10_msg:delivery_id(MsgDead5),
+           true,
+           accepted),
+    assert_messages(QName2, 0, 0, Config),
+
+    ok = amqp10_client:detach_link(Receiver1),
+    ok = amqp10_client:detach_link(Receiver2),
+    ok = amqp10_client:detach_link(Sender),
+    {ok, #{message_count := 0}} = rabbitmq_amqp_client:delete_queue(LinkPair, QName1),
+    {ok, #{message_count := 0}} = rabbitmq_amqp_client:delete_queue(LinkPair, QName2),
+    ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPair),
+    ok = end_session_sync(Session),
+    ok = amqp10_client:close_connection(Connection).
+
 %% Dead letter from a quorum queue into a stream.
 dead_letter_into_stream(Config) ->
     {Connection0, Session0, LinkPair0} = init(0, Config),
@@ -3820,6 +3920,124 @@ dead_letter_into_stream(Config) ->
     ok = amqp10_client:close_connection(Connection0),
     ok = amqp10_client:close_connection(Connection1).
 
+accept_multiple_message_order_classic_queue(Config) ->
+    accept_multiple_message_order(<<"classic">>, Config).
+
+accept_multiple_message_order_quorum_queue(Config) ->
+    accept_multiple_message_order(<<"quorum">>, Config).
+
+accept_multiple_message_order(QType, Config) ->
+    QName = atom_to_binary(?FUNCTION_NAME),
+    Address = <<"/queue/", QName/binary>>,
+
+    {Connection, Session, LinkPair} = init(Config),
+    QProps = #{arguments => #{<<"x-queue-type">> => {utf8, QType}}},
+    {ok, #{type := QType}} = rabbitmq_amqp_client:declare_queue(LinkPair, QName, QProps),
+
+    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, Address, settled),
+    ok = wait_for_credit(Sender),
+    [begin
+         Bin = integer_to_binary(N),
+         Msg = amqp10_msg:new(Bin, Bin, true),
+         ok = amqp10_client:send_msg(Sender, Msg)
+     end || N <- lists:seq(1, 5)],
+    ok = amqp10_client:detach_link(Sender),
+    assert_messages(QName, 5, 0, Config),
+
+    {ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"receiver">>, QName, unsettled),
+    {ok, Msg1} = amqp10_client:get_msg(Receiver),
+    {ok, Msg2} = amqp10_client:get_msg(Receiver),
+    {ok, _Msg3} = amqp10_client:get_msg(Receiver),
+    {ok, Msg4} = amqp10_client:get_msg(Receiver),
+    {ok, Msg5} = amqp10_client:get_msg(Receiver),
+    assert_messages(QName, 5, 5, Config),
+
+    %% Accept messages out of order.
+    ok = amqp10_client_session:disposition(
+           Receiver,
+           amqp10_msg:delivery_id(Msg2),
+           amqp10_msg:delivery_id(Msg4),
+           true,
+           accepted),
+    assert_messages(QName, 2, 2, Config),
+
+    ok = amqp10_client:accept_msg(Receiver, Msg5),
+    assert_messages(QName, 1, 1, Config),
+
+    ok = amqp10_client:accept_msg(Receiver, Msg1),
+    assert_messages(QName, 0, 0, Config),
+
+    ok = amqp10_client:detach_link(Receiver),
+    ?assertMatch({ok, #{message_count := 0}}, rabbitmq_amqp_client:delete_queue(LinkPair, QName)),
+    ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPair),
+    ok = end_session_sync(Session),
+    ok = amqp10_client:close_connection(Connection).
+
+release_multiple_message_order_classic_queue(Config) ->
+    release_multiple_message_order(<<"classic">>, Config).
+
+release_multiple_message_order_quorum_queue(Config) ->
+    release_multiple_message_order(<<"quorum">>, Config).
+
+release_multiple_message_order(QType, Config) ->
+    QName = atom_to_binary(?FUNCTION_NAME),
+    Address = <<"/queue/", QName/binary>>,
+
+    {Connection, Session, LinkPair} = init(Config),
+    QProps = #{arguments => #{<<"x-queue-type">> => {utf8, QType}}},
+    {ok, #{type := QType}} = rabbitmq_amqp_client:declare_queue(LinkPair, QName, QProps),
+
+    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, Address, settled),
+    ok = wait_for_credit(Sender),
+    [begin
+         Bin = integer_to_binary(N),
+         Msg = amqp10_msg:new(Bin, Bin, true),
+         ok = amqp10_client:send_msg(Sender, Msg)
+     end || N <- lists:seq(1, 4)],
+    ok = amqp10_client:detach_link(Sender),
+    assert_messages(QName, 4, 0, Config),
+
+    {ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"receiver">>, QName, unsettled),
+    {ok, Msg1} = amqp10_client:get_msg(Receiver),
+    {ok, Msg2} = amqp10_client:get_msg(Receiver),
+    {ok, Msg3} = amqp10_client:get_msg(Receiver),
+    {ok, Msg4} = amqp10_client:get_msg(Receiver),
+    assert_messages(QName, 4, 4, Config),
+
+    %% Release messages out of order.
+    ok = amqp10_client_session:disposition(
+           Receiver,
+           amqp10_msg:delivery_id(Msg2),
+           amqp10_msg:delivery_id(Msg3),
+           true,
+           released),
+    %% Both messages should be requeued and redelivered.
+    assert_messages(QName, 4, 2, Config),
+
+    {ok, Msg2b} = amqp10_client:get_msg(Receiver),
+    {ok, Msg3b} = amqp10_client:get_msg(Receiver),
+    assert_messages(QName, 4, 4, Config),
+    ?assertEqual([<<"2">>], amqp10_msg:body(Msg2b)),
+    ?assertEqual([<<"3">>], amqp10_msg:body(Msg3b)),
+
+    ok = amqp10_client_session:disposition(
+           Receiver,
+           amqp10_msg:delivery_id(Msg4),
+           amqp10_msg:delivery_id(Msg3b),
+           true,
+           accepted),
+    assert_messages(QName, 1, 1, Config),
+
+    ok = amqp10_client:accept_msg(Receiver, Msg1),
+    assert_messages(QName, 0, 0, Config),
+
+    ok = amqp10_client:detach_link(Receiver),
+    ?assertMatch({ok, #{message_count := 0}}, rabbitmq_amqp_client:delete_queue(LinkPair, QName)),
+    ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPair),
+    ok = end_session_sync(Session),
+    ok = amqp10_client:close_connection(Connection).
+
+
 %% This test asserts the following §3.2 requirement:
 %% "The bare message is immutable within the AMQP network. That is, none of the sections can be
 %% changed by any node acting as an AMQP intermediary. If a section of the bare message is
@@ -4359,6 +4577,7 @@ tcp_back_pressure_rabbitmq_internal_flow(QType, Config) ->
     ok = end_session_sync(Session),
     ok = amqp10_client:close_connection(Connection).
 
+
 %% internal
 %%
 
@@ -4592,13 +4811,16 @@ send_messages_with_group_id(Sender, N, GroupId) ->
      end || I <- lists:seq(1, N)].
 
 assert_messages(QNameBin, NumTotalMsgs, NumUnackedMsgs, Config) ->
+    assert_messages(QNameBin, NumTotalMsgs, NumUnackedMsgs, Config, 0).
+
+assert_messages(QNameBin, NumTotalMsgs, NumUnackedMsgs, Config, Node) ->
     Vhost = ?config(rmq_vhost, Config),
     eventually(
       ?_assertEqual(
          lists:sort([{messages, NumTotalMsgs}, {messages_unacknowledged, NumUnackedMsgs}]),
          begin
-             {ok, Q} = rpc(Config, rabbit_amqqueue, lookup, [QNameBin, Vhost]),
-             Infos = rpc(Config, rabbit_amqqueue, info, [Q, [messages, messages_unacknowledged]]),
+             {ok, Q} = rpc(Config, Node, rabbit_amqqueue, lookup, [QNameBin, Vhost]),
+             Infos = rpc(Config, Node, rabbit_amqqueue, info, [Q, [messages, messages_unacknowledged]]),
              lists:sort(Infos)
          end
         ), 500, 5).
diff --git a/deps/rabbit/test/amqpl_consumer_ack_SUITE.erl b/deps/rabbit/test/amqpl_consumer_ack_SUITE.erl
diff --git a/deps/rabbit/test/dead_lettering_SUITE.erl b/deps/rabbit/test/dead_lettering_SUITE.erl
