Merge pull request #11339 from rabbitmq/v3.13.x-amqpl-x-death

Interpret x-death header from AMQP 0.9.1 client

Author: michaelklishin

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/app.bzl

@@ -1721,6 +1721,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         testonly = True,
         srcs = ["test/unit_classic_mirrored_queue_sync_throttling_SUITE.erl"],
         outs = ["test/unit_classic_mirrored_queue_sync_throttling_SUITE.beam"],
+        hdrs = ["include/mc.hrl"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
         deps = ["//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),