Interpret x-death header from AMQP 0.9.1 client

Fixes #10709
Fixes #11331

This commit fixes the following regression which worked in 3.12.x, but
stopped working in 3.13.0 - 3.13.3:

```
AMQP 0.9.1 client    --publish-->
Q                    --dead-letter-->
DLQ                  --consume-->
AMQP 0.9.1 client (death count is now 1) --republish-same-message-with-headers-as-just-received-->
Q                    --dead-letter-->
DLQ                  --consume -->
AMQP 0.9.1 (death count is now 1, but should be 2)
```

The reason this behaviour stopped to work in 3.13.0 is that the broker
won't specially interpret x-headers in general, and the x-death header
specifically in this case anymore.

In other words, the new desired 3.13 behaviour with message containers
is that "x-headers belong to the broker".

While this is correct, it does break client applications which depended
on the previous use case.
One simple fix is that the client application does not re-publish with
the x-death header, but instead sets its own custom count header to
determine the number of times it retries.

This commit will only be packported to v3.13.x branch.
In other words, 4.0 won't interpret x-headers as done in 3.13.0 - 3.13.3.

The reason we backport this commit to v3.13.x is that the Spring
documentation expliclity recommends re-publishing the message with
x-death header being set:
https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-rabbit/3.0.6.RELEASE/reference/html/spring-cloud-stream-binder-rabbit.html#_retry_with_the_rabbitmq_binder

deps/rabbit/BUILD.bazel

@@ -815,6 +815,9 @@ rabbitmq_suite(
 rabbitmq_suite(
     name = "mc_unit_SUITE",
     size = "small",
+    runtime_deps = [
+        "@meck//:erlang_app",
+    ],
     deps = [
         "//deps/amqp10_common:erlang_app",
         "//deps/rabbit_common:erlang_app",

deps/rabbit/include/mc.hrl

@@ -14,6 +14,11 @@
 -define(ANN_PRIORITY, p).
 
 -define(FF_MC_DEATHS_V2, message_containers_deaths_v2).
+-define(MC_ENV,
+        case rabbit_feature_flags:is_enabled(?FF_MC_DEATHS_V2) of
+            true -> #{};
+            false -> #{?FF_MC_DEATHS_V2 => false}
+        end).
 
 -type death_key() :: {SourceQueue :: rabbit_misc:resource_name(), rabbit_dead_letter:reason()}.
 -type death_anns() :: #{%% timestamp of the first time this message

deps/rabbit/src/mc.erl

@@ -93,7 +93,7 @@
 %% protocol specific init function
 %% returns a map of additional annotations to merge into the
 %% protocol generic annotations map, e.g. ttl, priority and durable
--callback init(term()) ->
+-callback init(term(), environment()) ->
     {proto_state(), annotations()}.
 
 %% the size of the payload and other meta data respectively
@@ -147,7 +147,7 @@ init(Proto, Data, Anns0, Env)
   when is_atom(Proto)
        andalso is_map(Anns0)
        andalso is_map(Env) ->
-    {ProtoData, ProtoAnns} = Proto:init(Data),
+    {ProtoData, ProtoAnns} = Proto:init(Data, Env),
     Anns = case maps:size(Env) == 0 of
                true ->
                    Anns0;
@@ -389,9 +389,9 @@ record_death(Reason, SourceQueue,
                                             [{Key, NewDeath} | Deaths0]
                                     end
                             end,
-                   Anns0#{<<"x-last-death-reason">> := atom_to_binary(Reason),
-                          <<"x-last-death-queue">> := SourceQueue,
-                          <<"x-last-death-exchange">> := Exchange,
+                   Anns0#{<<"x-last-death-reason">> => atom_to_binary(Reason),
+                          <<"x-last-death-queue">> => SourceQueue,
+                          <<"x-last-death-exchange">> => Exchange,
                           deaths := Deaths};
                _ ->
                    Deaths = case Env of

deps/rabbit/src/mc_amqp.erl

@@ -5,7 +5,7 @@
 -include("mc.hrl").
 
 -export([
-         init/1,
+         init/2,
          size/1,
          x_header/2,
          property/2,
@@ -59,18 +59,18 @@
              ]).
 
 %% mc implementation
-init(Sections) when is_list(Sections) ->
+init(Sections, Env) when is_list(Sections) ->
     Msg = decode(Sections, #msg{}),
-    init(Msg);
-init(#msg{} = Msg) ->
+    init(Msg, Env);
+init(#msg{} = Msg, _Env) ->
     %% TODO: as the essential annotations, durable, priority, ttl and delivery_count
     %% is all we are interested in it isn't necessary to keep hold of the
     %% incoming AMQP header inside the state
     Anns = essential_properties(Msg),
     {Msg, Anns}.
 
-convert_from(?MODULE, Sections, _Env) ->
-    element(1, init(Sections));
+convert_from(?MODULE, Sections, Env) ->
+    element(1, init(Sections, Env));
 convert_from(_SourceProto, _, _Env) ->
     not_implemented.
 

deps/rabbit/src/mc_amqpl.erl

@@ -8,7 +8,7 @@
 
 %% mc
 -export([
-         init/1,
+         init/2,
          size/1,
          x_header/2,
          routing_headers/2,
@@ -48,10 +48,10 @@
              ]).
 
 %% mc implementation
-init(#content{} = Content0) ->
+init(#content{} = Content0, Env) ->
     Content1 = rabbit_binary_parser:ensure_content_decoded(Content0),
     %% project essential properties into annotations
-    Anns = essential_properties(Content1),
+    Anns = essential_properties(Content1, Env),
     Content = strip_header(Content1, ?DELETED_HEADER),
     {Content, Anns}.
 
@@ -483,7 +483,8 @@ message(#resource{name = ExchangeNameBin}, RoutingKey,
             {ok, mc:init(?MODULE,
                          Content,
                          Anns#{?ANN_ROUTING_KEYS => [RoutingKey | HeaderRoutes],
-                               ?ANN_EXCHANGE => ExchangeNameBin})}
+                               ?ANN_EXCHANGE => ExchangeNameBin},
+                         ?MC_ENV)}
     end;
 message(#resource{} = XName, RoutingKey,
         #content{} = Content, Anns, false) ->
@@ -514,75 +515,6 @@ from_basic_message(#basic_message{content = Content,
 
 %% Internal
 
-deaths_to_headers(Deaths, Headers0) ->
-    Infos = case Deaths of
-                #deaths{records = Records} ->
-                    %% sort records by the last timestamp
-                    List = lists:sort(
-                             fun({_, #death{anns = #{last_time := L1}}},
-                                 {_, #death{anns = #{last_time := L2}}}) ->
-                                     L1 =< L2
-                             end, maps:to_list(Records)),
-                    lists:foldl(fun(Record, Acc) ->
-                                        Table = death_table(Record),
-                                        [Table | Acc]
-                                end, [], List);
-                _ ->
-                    lists:map(fun death_table/1, Deaths)
-            end,
-    rabbit_misc:set_table_value(Headers0, <<"x-death">>, array, Infos).
-
-convert_from_amqp_deaths({array, map, Maps}) ->
-    L = lists:map(
-          fun({map, KvList}) ->
-                  {Ttl, KvList1} = case KvList of
-                                       [{{symbol, <<"ttl">>}, {uint, Ttl0}} | Tail] ->
-                                           {Ttl0, Tail};
-                                       _ ->
-                                           {undefined, KvList}
-                                   end,
-                  [
-                   {{symbol, <<"queue">>}, {utf8, Queue}},
-                   {{symbol, <<"reason">>}, {symbol, Reason}},
-                   {{symbol, <<"count">>}, {ulong, Count}},
-                   {{symbol, <<"first-time">>}, {timestamp, FirstTime}},
-                   {{symbol, <<"last-time">>}, {timestamp, _LastTime}},
-                   {{symbol, <<"exchange">>}, {utf8, Exchange}},
-                   {{symbol, <<"routing-keys">>}, {array, utf8, RKeys0}}
-                  ] = KvList1,
-                  RKeys = [Key || {utf8, Key} <- RKeys0],
-                  death_table(Queue, Reason, Exchange, RKeys, Count, FirstTime, Ttl)
-          end, Maps),
-    {true, {<<"x-death">>, array, L}};
-convert_from_amqp_deaths(_IgnoreUnknownValue) ->
-    false.
-
-death_table({{QName, Reason},
-             #death{exchange = Exchange,
-                    routing_keys = RoutingKeys,
-                    count = Count,
-                    anns = DeathAnns = #{first_time := FirstTime}}}) ->
-    death_table(QName, Reason, Exchange, RoutingKeys, Count, FirstTime,
-                maps:get(ttl, DeathAnns, undefined)).
-
-death_table(QName, Reason, Exchange, RoutingKeys, Count, FirstTime, Ttl) ->
-    L0 = [
-          {<<"count">>, long, Count},
-          {<<"reason">>, longstr, rabbit_data_coercion:to_binary(Reason)},
-          {<<"queue">>, longstr, QName},
-          {<<"time">>, timestamp, FirstTime div 1000},
-          {<<"exchange">>, longstr, Exchange},
-          {<<"routing-keys">>, array, [{longstr, Key} || Key <- RoutingKeys]}
-         ],
-    L = case Ttl of
-            undefined ->
-                L0;
-            _ ->
-                Expiration = integer_to_binary(Ttl),
-                [{<<"original-expiration">>, longstr, Expiration} | L0]
-        end,
-    {table, L}.
-
 strip_header(#content{properties = #'P_basic'{headers = undefined}}
              = DecodedContent, _Key) ->
     DecodedContent;
@@ -732,11 +664,11 @@ message_id({utf8, S}, HKey, H0) ->
 message_id(undefined, _HKey, H) ->
     {H, undefined}.
 
-essential_properties(#content{} = C) ->
+essential_properties(#content{properties = Props}, Env) ->
     #'P_basic'{delivery_mode = Mode,
                priority = Priority,
                timestamp = TimestampRaw,
-               headers = Headers} = Props = C#content.properties,
+               headers = Headers} = Props,
     {ok, MsgTTL} = rabbit_basic:parse_expiration(Props),
     Timestamp = case TimestampRaw of
                     undefined ->
@@ -752,6 +684,8 @@ essential_properties(#content{} = C) ->
                   _ ->
                       undefined
               end,
+    Deaths = headers_to_deaths(Headers, Env),
+
     maps_put_truthy(
       ?ANN_PRIORITY, Priority,
       maps_put_truthy(
@@ -762,7 +696,125 @@ essential_properties(#content{} = C) ->
             ?ANN_DURABLE, Durable,
             maps_put_truthy(
               bcc, BccKeys,
-              #{}))))).
+              maps_put_truthy(
+                deaths, Deaths,
+                #{})))))).
+
+headers_to_deaths(_, #{?FF_MC_DEATHS_V2 := false}) ->
+    undefined;
+headers_to_deaths(undefined, _) ->
+    undefined;
+headers_to_deaths(Headers, _) ->
+    case lists:keymember(<<"x-death">>, 1, Headers) of
+        true ->
+            case rabbit_misc:amqp_table(Headers) of
+                #{<<"x-death">> := XDeathList}
+                  when is_list(XDeathList) ->
+                    recover_deaths(XDeathList, []);
+                _ ->
+                    undefined
+            end;
+        false ->
+            undefined
+    end.
+
+recover_deaths([], Acc) ->
+    lists:reverse(Acc);
+recover_deaths([Map = #{<<"exchange">> := Exchange,
+                        <<"queue">> := Queue,
+                        <<"routing-keys">> := RKeys,
+                        <<"reason">> := ReasonBin,
+                        <<"count">> := Count,
+                        <<"time">> := Ts} | Rem], Acc0) ->
+    Reason = binary_to_existing_atom(ReasonBin),
+    DeathAnns0 = #{first_time => Ts,
+                   %% Given that this timestamp is absent in the AMQP 0.9.1
+                   %% x-death header, the last_time we set here is incorrect
+                   %% if the message was dead lettered more than one time.
+                   last_time => Ts},
+    DeathAnns = case Map of
+                    #{<<"original-expiration">> := Exp} ->
+                        DeathAnns0#{ttl => binary_to_integer(Exp)};
+                    _ ->
+                        DeathAnns0
+                end,
+    Acc = [{{Queue, Reason},
+            #death{anns = DeathAnns,
+                   exchange = Exchange,
+                   count = Count,
+                   routing_keys = RKeys}} | Acc0],
+    recover_deaths(Rem, Acc);
+recover_deaths([_IgnoreInvalid | Rem], Acc) ->
+    recover_deaths(Rem, Acc).
+
+deaths_to_headers(Deaths, Headers0) ->
+    Infos = case Deaths of
+                #deaths{records = Records} ->
+                    %% sort records by the last timestamp
+                    List = lists:sort(
+                             fun({_, #death{anns = #{last_time := L1}}},
+                                 {_, #death{anns = #{last_time := L2}}}) ->
+                                     L1 =< L2
+                             end, maps:to_list(Records)),
+                    lists:foldl(fun(Record, Acc) ->
+                                        Table = death_table(Record),
+                                        [Table | Acc]
+                                end, [], List);
+                _ ->
+                    lists:map(fun death_table/1, Deaths)
+            end,
+    rabbit_misc:set_table_value(Headers0, <<"x-death">>, array, Infos).
+
+convert_from_amqp_deaths({array, map, Maps}) ->
+    L = lists:map(
+          fun({map, KvList}) ->
+                  {Ttl, KvList1} = case KvList of
+                                       [{{symbol, <<"ttl">>}, {uint, Ttl0}} | Tail] ->
+                                           {Ttl0, Tail};
+                                       _ ->
+                                           {undefined, KvList}
+                                   end,
+                  [
+                   {{symbol, <<"queue">>}, {utf8, Queue}},
+                   {{symbol, <<"reason">>}, {symbol, Reason}},
+                   {{symbol, <<"count">>}, {ulong, Count}},
+                   {{symbol, <<"first-time">>}, {timestamp, FirstTime}},
+                   {{symbol, <<"last-time">>}, {timestamp, _LastTime}},
+                   {{symbol, <<"exchange">>}, {utf8, Exchange}},
+                   {{symbol, <<"routing-keys">>}, {array, utf8, RKeys0}}
+                  ] = KvList1,
+                  RKeys = [Key || {utf8, Key} <- RKeys0],
+                  death_table(Queue, Reason, Exchange, RKeys, Count, FirstTime, Ttl)
+          end, Maps),
+    {true, {<<"x-death">>, array, L}};
+convert_from_amqp_deaths(_IgnoreUnknownValue) ->
+    false.
+
+death_table({{QName, Reason},
+             #death{exchange = Exchange,
+                    routing_keys = RoutingKeys,
+                    count = Count,
+                    anns = DeathAnns = #{first_time := FirstTime}}}) ->
+    death_table(QName, Reason, Exchange, RoutingKeys, Count, FirstTime,
+                maps:get(ttl, DeathAnns, undefined)).
+
+death_table(QName, Reason, Exchange, RoutingKeys, Count, FirstTime, Ttl) ->
+    L0 = [
+          {<<"count">>, long, Count},
+          {<<"reason">>, longstr, rabbit_data_coercion:to_binary(Reason)},
+          {<<"queue">>, longstr, QName},
+          {<<"time">>, timestamp, FirstTime div 1000},
+          {<<"exchange">>, longstr, Exchange},
+          {<<"routing-keys">>, array, [{longstr, Key} || Key <- RoutingKeys]}
+         ],
+    L = case Ttl of
+            undefined ->
+                L0;
+            _ ->
+                Expiration = integer_to_binary(Ttl),
+                [{<<"original-expiration">>, longstr, Expiration} | L0]
+        end,
+    {table, L}.
 
 %% headers that are added as annotations during conversions
 is_internal_header(<<"x-basic-", _/binary>>) ->

deps/rabbit/src/rabbit_dead_letter.erl

@@ -31,11 +31,7 @@ publish(Msg0, Reason, #exchange{name = XName} = DLX, RK,
                   _ ->
                       [RK]
               end,
-    Env = case rabbit_feature_flags:is_enabled(?FF_MC_DEATHS_V2) of
-              true -> #{};
-              false -> #{?FF_MC_DEATHS_V2 => false}
-          end,
-    Msg1 = mc:record_death(Reason, SourceQName, Msg0, Env),
+    Msg1 = mc:record_death(Reason, SourceQName, Msg0, ?MC_ENV),
     {Ttl, Msg2} = mc:take_annotation(dead_letter_ttl, Msg1),
     Msg3 = mc:set_ttl(Ttl, Msg2),
     Msg4 = mc:set_annotation(?ANN_ROUTING_KEYS, DLRKeys, Msg3),

deps/rabbit/src/rabbit_fifo_dlx_worker.erl

@@ -324,11 +324,7 @@ forward(ConsumedMsg, ConsumedMsgId, ConsumedQRef, DLX, Reason,
                   _ ->
                       [RKey]
               end,
-    Env = case rabbit_feature_flags:is_enabled(?FF_MC_DEATHS_V2) of
-              true -> #{};
-              false -> #{?FF_MC_DEATHS_V2 => false}
-          end,
-    Msg0 = mc:record_death(Reason, SourceQName, ConsumedMsg, Env),
+    Msg0 = mc:record_death(Reason, SourceQName, ConsumedMsg, ?MC_ENV),
     Msg1 = mc:set_ttl(undefined, Msg0),
     Msg2 = mc:set_annotation(?ANN_ROUTING_KEYS, DLRKeys, Msg1),
     Msg = mc:set_annotation(?ANN_EXCHANGE, DLXName, Msg2),