QQ: track number of requeues for message.

To be able to calculate the correct value for the AMQP delivery_count
header we need to be able to distinguish between messages that were
"released" or returned in QQ speak and those that were returned
due to errors such as channel termination.

This commit implement such tracking as well as the calculation
of a new mc annotations `delivery_count` that AMQP makes use
of to set the header value accordingly.

Author: kjnilsson

deps/rabbit/BUILD.bazel

@@ -698,6 +698,9 @@ rabbitmq_suite(
 rabbitmq_suite(
     name = "rabbit_fifo_int_SUITE",
     size = "medium",
+    additional_beam = [
+        ":test_test_util_beam",
+    ],
     deps = [
         "//deps/rabbit_common:erlang_app",
         "@aten//:erlang_app",

deps/rabbit/app.bzl

@@ -1341,7 +1341,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/rabbit_fifo_int_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/rabbit_common:erlang_app"],
+        deps = ["//deps/rabbitmq_ct_helpers:erlang_app"],
     )
     erlang_bytecode(
         name = "rabbit_fifo_prop_SUITE_beam_files",

deps/rabbit/src/mc_amqp.erl

@@ -222,14 +222,7 @@ get_property(priority, Msg) ->
 -spec protocol_state(state(), mc:annotations()) -> iolist().
 protocol_state(Msg0 = #msg_body_decoded{header = Header0,
                                         message_annotations = MA0}, Anns) ->
-    FirstAcquirer = first_acquirer(Anns),
-    Header = case Header0 of
-                 undefined ->
-                     #'v1_0.header'{durable = true,
-                                    first_acquirer = FirstAcquirer};
-                 #'v1_0.header'{} ->
-                     Header0#'v1_0.header'{first_acquirer = FirstAcquirer}
-             end,
+    Header = update_header_from_anns(Header0, Anns),
     MA = protocol_state_message_annotations(MA0, Anns),
     Msg = Msg0#msg_body_decoded{header = Header,
                                 message_annotations = MA},
@@ -238,14 +231,7 @@ protocol_state(Msg0 = #msg_body_decoded{header = Header0,
 protocol_state(#msg_body_encoded{header = Header0,
                                  message_annotations = MA0,
                                  bare_and_footer = BareAndFooter}, Anns) ->
-    FirstAcquirer = first_acquirer(Anns),
-    Header = case Header0 of
-                 undefined ->
-                     #'v1_0.header'{durable = true,
-                                    first_acquirer = FirstAcquirer};
-                 #'v1_0.header'{} ->
-                     Header0#'v1_0.header'{first_acquirer = FirstAcquirer}
-             end,
+    Header = update_header_from_anns(Header0, Anns),
     MA = protocol_state_message_annotations(MA0, Anns),
     Sections = to_sections(Header, MA, []),
     [encode(Sections), BareAndFooter];
@@ -269,10 +255,9 @@ protocol_state(#v1{message_annotations = MA0,
               _ ->
                   undefined
           end,
-    Header = #'v1_0.header'{durable = Durable,
-                            priority = Priority,
-                            ttl = Ttl,
-                            first_acquirer = first_acquirer(Anns)},
+    Header = update_header_from_anns(#'v1_0.header'{durable = Durable,
+                                                    priority = Priority,
+                                                    ttl = Ttl}, Anns),
     MA = protocol_state_message_annotations(MA0, Anns),
     Sections = to_sections(Header, MA, []),
     [encode(Sections), BareAndFooter].
@@ -573,13 +558,13 @@ msg_body_encoded([{{pos, Pos}, {body, Code}}], BarePos, Msg)
 binary_part_bare_and_footer(Payload, Start) ->
     binary_part(Payload, Start, byte_size(Payload) - Start).
 
--spec first_acquirer(mc:annotations()) -> boolean().
-first_acquirer(Anns) ->
-    Redelivered = case Anns of
-                      #{redelivered := R} -> R;
-                      _ -> false
-                  end,
-    not Redelivered.
+update_header_from_anns(undefined, Anns) ->
+    update_header_from_anns(#'v1_0.header'{durable = true}, Anns);
+update_header_from_anns(Header, Anns) ->
+    FirstAcq = not maps:get(redelivered, Anns, false),
+    DeliveryCount = {uint, maps:get(delivery_count, Anns, 0)},
+    Header#'v1_0.header'{first_acquirer = FirstAcq,
+                         delivery_count = DeliveryCount}.
 
 encode_deaths(Deaths) ->
     lists:map(

deps/rabbit/src/rabbit_fifo.erl

@@ -96,7 +96,8 @@
 -ifdef(TEST).
 -export([update_header/4,
          chunk_disk_msgs/3,
-         smallest_raft_index/1]).
+         smallest_raft_index/1,
+         make_requeue/4]).
 -endif.
 
 -import(serial_number, [add/2, diff/2]).
@@ -308,7 +309,8 @@ apply(#{index := Idx} = Meta,
           when is_map_key(MsgId, Checked0) ->
             %% construct a message with the current raft index
             %% and update delivery count before adding it to the message queue
-            Header = update_header(delivery_count, fun incr/1, 1, Header0),
+            Header1 = update_header(delivery_count, fun incr/1, 1, Header0),
+            Header = update_header(return_count, fun incr/1, 1, Header1),
             State0 = add_bytes_return(Header, State00),
             Con = Con0#consumer{checked_out = maps:remove(MsgId, Checked0),
                                 credit = increase_credit(Con0, 1)},
@@ -1745,17 +1747,19 @@ maybe_enqueue(RaftIdx, Ts, From, MsgSeqNo, RawMsg,
 return(#{index := IncomingRaftIdx} = Meta,
        ConsumerKey, MsgIds, Checked, Effects0, State0) ->
     %% We requeue in the same order as messages got returned by the client.
-    {State1, Effects1} = lists:foldl(
-                           fun(MsgId, Acc = {S0, E0}) ->
-                                   case Checked of
-                                       #{MsgId := Val} ->
-                                           {_At, Msg} = Val,
-                                           return_one(Meta, MsgId, Msg,
-                                                      S0, E0, ConsumerKey);
-                                       #{} ->
-                                           Acc
-                                   end
-                           end, {State0, Effects0}, MsgIds),
+    {State1, Effects1} =
+        lists:foldl(
+          fun(MsgId, Acc = {S0, E0}) ->
+                  case Checked of
+                      #{MsgId := {_At, Msg0}} ->
+                          Msg = update_msg_header(return_count, fun incr/1, 1,
+                                                  Msg0),
+                          return_one(Meta, MsgId, Msg,
+                                     S0, E0, ConsumerKey);
+                      #{} ->
+                          Acc
+                  end
+          end, {State0, Effects0}, MsgIds),
     State2 = case State1#?STATE.consumers of
                  #{ConsumerKey := Con} ->
                      update_or_remove_con(Meta, ConsumerKey, Con, State1);
@@ -1904,11 +1908,13 @@ update_header(Key, UpdateFun, Default, Size)
   when is_integer(Size) ->
     update_header(Key, UpdateFun, Default, #{size => Size});
 update_header(Key, UpdateFun, Default, ?TUPLE(Size, Expiry))
-  when is_integer(Size), is_integer(Expiry) ->
+  when is_integer(Size) andalso
+       is_integer(Expiry) ->
     update_header(Key, UpdateFun, Default, #{size => Size,
                                              expiry => Expiry});
 update_header(Key, UpdateFun, Default, Header)
-  when is_map(Header), is_map_key(size, Header) ->
+  when is_map(Header) andalso
+       is_map_key(size, Header) ->
     maps:update_with(Key, UpdateFun, Default, Header).
 
 get_msg_header(?MSG(_Idx, Header)) ->

deps/rabbit/src/rabbit_fifo.hrl

@@ -42,11 +42,13 @@
                       optimised_tuple(msg_size(), Expiry :: milliseconds()) |
                       #{size := msg_size(),
                         delivery_count => non_neg_integer(),
+                        return_count => non_neg_integer(),
                         expiry => milliseconds()}.
 %% The message header:
 %% size: The size of the message payload in bytes.
-%% delivery_count: the number of unsuccessful delivery attempts.
+%% delivery_count: The number of unsuccessful delivery attempts.
 %%                 A non-zero value indicates a previous attempt.
+%% return_count: The number of explicit returns.
 %% expiry: Epoch time in ms when a message expires. Set during enqueue.
 %%         Value is determined by per-queue or per-message message TTL.
 %% If it contains only the size it can be condensed to an integer.

deps/rabbit/src/rabbit_fifo_client.erl

@@ -208,12 +208,7 @@ dequeue(QueueName, ConsumerTag, Settlement,
         {ok, {dequeue, empty}, Leader} ->
             {empty, State0#state{leader = Leader}};
         {ok, {dequeue, {MsgId, {MsgHeader, Msg0}}, MsgsReady}, Leader} ->
-            Count = case MsgHeader of
-                        #{delivery_count := C} -> C;
-                       _ -> 0
-                    end,
-            IsDelivered = Count > 0,
-            Msg = add_delivery_count_header(Msg0, Count),
+            {Msg, IsDelivered} = add_delivery_count_header(Msg0, MsgHeader),
             {ok, MsgsReady,
              {QueueName, qref(Leader), MsgId, IsDelivered, Msg},
              State0#state{leader = Leader}};
@@ -223,14 +218,21 @@ dequeue(QueueName, ConsumerTag, Settlement,
             Err
     end.
 
-add_delivery_count_header(Msg, Count) ->
-    case mc:is(Msg) of
-        true when is_integer(Count) andalso
-                  Count > 0 ->
-            mc:set_annotation(<<"x-delivery-count">>, Count, Msg);
+add_delivery_count_header(Msg, #{delivery_count := DelCount} = Header)
+  when is_integer(DelCount) ->
+    {case mc:is(Msg) of
+        true ->
+             %% the "delivery-count" header in the AMQP spec does not include
+             %% returns (released outcomes)
+             AmqpDelCount = DelCount - maps:get(return_count, Header, 0),
+             mc:set_annotation(delivery_count, AmqpDelCount,
+                               mc:set_annotation(<<"x-delivery-count">>,
+                                                 DelCount, Msg));
         _ ->
             Msg
-    end.
+    end, DelCount > 0};
+add_delivery_count_header(Msg, _Header) ->
+    {Msg, false}.
 
 
 %% @doc Settle a message. Permanently removes message from the queue.
@@ -840,13 +842,7 @@ handle_delivery(_QName, _Leader, {delivery, Tag, [_ | _] = IdMsgs},
 transform_msgs(QName, QRef, Msgs) ->
     lists:map(
       fun({MsgId, {MsgHeader, Msg0}}) ->
-              {Msg, Redelivered} = case MsgHeader of
-                                       #{delivery_count := C} ->
-                                           {add_delivery_count_header(Msg0, C), true};
-                                       _ ->
-                                           {Msg0, false}
-                                   end,
-
+              {Msg, Redelivered} = add_delivery_count_header(Msg0, MsgHeader),
               {QName, QRef, MsgId, Redelivered, Msg}
       end, Msgs).
 

deps/rabbit/test/amqp_system_SUITE.erl

@@ -34,6 +34,7 @@ groups() ->
           %% TODO at_most_once,
           reject,
           redelivery,
+          released,
           routing,
           invalid_routes,
           auth_failure,
@@ -80,6 +81,13 @@ end_per_group(_, Config) ->
       rabbit_ct_broker_helpers:teardown_steps()).
 
 init_per_testcase(Testcase, Config) ->
+    enable_feature_flags(Config,
+                         [
+                          message_containers_store_amqp_v1,
+                          credit_api_v2,
+                          quorum_queues_v4
+                          % amqp_address_v1
+                         ]),
     rabbit_ct_helpers:testcase_started(Config, Testcase).
 
 end_per_testcase(Testcase, Config) ->
@@ -115,22 +123,20 @@ build_maven_test_project(Config) ->
 %% -------------------------------------------------------------------
 
 roundtrip(Config) ->
+    declare_queue(Config, ?FUNCTION_NAME, "quorum"),
     run(Config, [{dotnet, "roundtrip"},
                  {java, "RoundTripTest"}]).
 
 streams(Config) ->
-    _ = rabbit_ct_broker_helpers:enable_feature_flag(Config,
-                                                     message_containers_store_amqp_v1),
-    Ch = rabbit_ct_client_helpers:open_channel(Config),
-    amqp_channel:call(Ch, #'queue.declare'{queue = <<"stream_q2">>,
-                                           durable = true,
-                                           arguments = [{<<"x-queue-type">>, longstr, "stream"}]}),
+    declare_queue(Config, ?FUNCTION_NAME, "stream"),
     run(Config, [{dotnet, "streams"}]).
 
 roundtrip_to_amqp_091(Config) ->
+    declare_queue(Config, ?FUNCTION_NAME, "classic"),
     run(Config, [{dotnet, "roundtrip_to_amqp_091"}]).
 
 default_outcome(Config) ->
+    declare_queue(Config, ?FUNCTION_NAME, "classic"),
     run(Config, [{dotnet, "default_outcome"}]).
 
 no_routes_is_released(Config) ->
@@ -140,28 +146,41 @@ no_routes_is_released(Config) ->
     run(Config, [{dotnet, "no_routes_is_released"}]).
 
 outcomes(Config) ->
+    declare_queue(Config, ?FUNCTION_NAME, "classic"),
     run(Config, [{dotnet, "outcomes"}]).
 
 fragmentation(Config) ->
+    declare_queue(Config, ?FUNCTION_NAME, "classic"),
     run(Config, [{dotnet, "fragmentation"}]).
 
 message_annotations(Config) ->
+    declare_queue(Config, ?FUNCTION_NAME, "classic"),
     run(Config, [{dotnet, "message_annotations"}]).
 
 footer(Config) ->
+    declare_queue(Config, ?FUNCTION_NAME, "classic"),
     run(Config, [{dotnet, "footer"}]).
 
 data_types(Config) ->
+    declare_queue(Config, ?FUNCTION_NAME, "classic"),
     run(Config, [{dotnet, "data_types"}]).
 
 reject(Config) ->
+    declare_queue(Config, ?FUNCTION_NAME, "classic"),
     run(Config, [{dotnet, "reject"}]).
 
 redelivery(Config) ->
+    declare_queue(Config, ?FUNCTION_NAME, "quorum"),
     run(Config, [{dotnet, "redelivery"}]).
 
+released(Config) ->
+    declare_queue(Config, ?FUNCTION_NAME, "quorum"),
+    run(Config, [{dotnet, "released"}]).
+
 routing(Config) ->
     Ch = rabbit_ct_client_helpers:open_channel(Config),
+    amqp_channel:call(Ch, #'queue.declare'{queue = <<"test">>,
+                                           durable = true}),
     amqp_channel:call(Ch, #'queue.declare'{queue = <<"transient_q">>,
                                            durable = false}),
     amqp_channel:call(Ch, #'queue.declare'{queue = <<"durable_q">>,
@@ -174,6 +193,18 @@ routing(Config) ->
                                            arguments = [{<<"x-queue-type">>, longstr, <<"stream">>}]}),
     amqp_channel:call(Ch, #'queue.declare'{queue = <<"autodel_q">>,
                                            auto_delete = true}),
+    amqp_channel:call(Ch, #'queue.declare'{queue = <<"fanout_q">>,
+                                           durable = false}),
+    amqp_channel:call(Ch, #'queue.bind'{queue = <<"fanout_q">>,
+                                        exchange = <<"amq.fanout">>
+                                       }),
+    amqp_channel:call(Ch, #'queue.declare'{queue = <<"direct_q">>,
+                                           durable = false}),
+    amqp_channel:call(Ch, #'queue.bind'{queue = <<"direct_q">>,
+                                        exchange = <<"amq.direct">>,
+                                        routing_key = <<"direct_q">>
+                                       }),
+
     run(Config, [
         {dotnet, "routing"}
       ]).
@@ -227,6 +258,7 @@ run_dotnet_test(Config, Method) ->
       [
         {cd, TestProjectDir}
       ]),
+    ct:pal("~s: result ~p", [?FUNCTION_NAME, Ret]),
     {ok, _} = Ret.
 
 run_java_test(Config, Class) ->
@@ -239,3 +271,23 @@ run_java_test(Config, Class) ->
       ],
       [{cd, TestProjectDir}]),
     {ok, _} = Ret.
+
+
+enable_feature_flags(Config, Flags) ->
+    [begin
+         case rabbit_ct_broker_helpers:enable_feature_flag(Config, Flag) of
+             ok -> ok;
+             _ ->
+                 throw({skip, "feature flag ~s could not be enabled"})
+         end
+     end || Flag <- Flags].
+
+declare_queue(Config, Name, Type) ->
+    Ch = rabbit_ct_client_helpers:open_channel(Config),
+    #'queue.declare_ok'{} =
+        amqp_channel:call(Ch, #'queue.declare'{queue = atom_to_binary(Name, utf8),
+                                               durable = true,
+                                               arguments = [{<<"x-queue-type">>,
+                                                             longstr, Type}]}),
+    rabbit_ct_client_helpers:close_channel(Ch),
+    ok.