Store first and last death time in a stream

Given that we store first and last death time for a given {Queue,
Reason} tuple, let's store these infos in a stream and return these
infos to the AMQP client.

Last death time, similar to first death time can be useful to AMQP
clients when there are multiple deaths for the same {Queue, Reason}
tuple.

AMQP 0.9.1 only exposes a time header referring on the first time.

Author: ansd

deps/rabbit/include/mc.hrl

@@ -30,6 +30,8 @@
 
 -record(deaths, {first :: death_key(), % redundant to mc annotations x-first-death-*
                  last :: death_key(), % redundant to mc annotations x-last-death-*
-                 records :: #{death_key() := #death{}} | % feature flag message_containers_deaths_v2 disabled
-                            [{death_key(), #death{}}] % feature flag message_containers_deaths_v2 enabled
+                 records :: #{death_key() := #death{}} |
+                             %% Records are ordered by death recency when feature flag
+                             %% message_containers_deaths_v2 is enabled
+                             [{death_key(), #death{}}]
                 }).

deps/rabbit/src/mc_amqp.erl

@@ -587,13 +587,15 @@ encode_deaths(Deaths) ->
            #death{exchange = Exchange,
                   routing_keys = RoutingKeys,
                   count = Count,
-                  anns = Anns = #{first_time := FirstTime}}}) ->
+                  anns = Anns = #{first_time := FirstTime,
+                                  last_time := LastTime}}}) ->
               RKeys = [{utf8, Rk} || Rk <- RoutingKeys],
               Map0 = [
                       {{symbol, <<"queue">>}, {utf8, Queue}},
                       {{symbol, <<"reason">>}, {symbol, atom_to_binary(Reason)}},
                       {{symbol, <<"count">>}, {ulong, Count}},
-                      {{symbol, <<"time">>}, {timestamp, FirstTime}},
+                      {{symbol, <<"first-time">>}, {timestamp, FirstTime}},
+                      {{symbol, <<"last-time">>}, {timestamp, LastTime}},
                       {{symbol, <<"exchange">>}, {utf8, Exchange}},
                       {{symbol, <<"routing-keys">>}, {array, utf8, RKeys}}
                      ],

deps/rabbit/src/mc_amqpl.erl

@@ -604,12 +604,13 @@ convert_from_amqp_deaths({array, map, Maps}) ->
                    {{symbol, <<"queue">>}, {utf8, Queue}},
                    {{symbol, <<"reason">>}, {symbol, Reason}},
                    {{symbol, <<"count">>}, {ulong, Count}},
-                   {{symbol, <<"time">>}, {timestamp, Timestamp}},
+                   {{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, Timestamp, Ttl)
+                  death_table(Queue, Reason, Exchange, RKeys, Count, FirstTime, Ttl)
           end, Maps),
     {true, {<<"x-death">>, array, L}};
 convert_from_amqp_deaths(_IgnoreUnknownValue) ->

deps/rabbit/test/amqp_client_SUITE.erl

@@ -3542,7 +3542,8 @@ dead_letter_headers_exchange(Config) ->
                                            {{symbol, <<"queue">>}, {utf8, QName1}},
                                            {{symbol, <<"reason">>}, {symbol, <<"expired">>}},
                                            {{symbol, <<"count">>}, {ulong, 1}},
-                                           {{symbol, <<"time">>}, {timestamp, Timestamp}},
+                                           {{symbol, <<"first-time">>}, {timestamp, Timestamp}},
+                                           {{symbol, <<"last-time">>}, {timestamp, Timestamp}},
                                            {{symbol, <<"exchange">>},{utf8, <<>>}},
                                            {{symbol, <<"routing-keys">>}, {array, utf8, [{utf8, QName1}]}}
                                           ]}]}
@@ -3628,28 +3629,31 @@ dead_letter_reject(Config) ->
            {{symbol, <<"queue">>}, {utf8, QName1}},
            {{symbol, <<"reason">>}, {symbol, <<"expired">>}},
            {{symbol, <<"count">>}, {ulong, 3}},
-           {{symbol, <<"time">>}, {timestamp, Timestamp1}},
+           {{symbol, <<"first-time">>}, {timestamp, Ts1}},
+           {{symbol, <<"last-time">>}, {timestamp, Ts2}},
            {{symbol, <<"exchange">>},{utf8, <<>>}},
            {{symbol, <<"routing-keys">>}, {array, utf8, [{utf8, QName1}]}}
           ]} = D1,
     {map, [
            {{symbol, <<"queue">>}, {utf8, QName2}},
            {{symbol, <<"reason">>}, {symbol, <<"rejected">>}},
            {{symbol, <<"count">>}, {ulong, 2}},
-           {{symbol, <<"time">>}, {timestamp, Timestamp2}},
+           {{symbol, <<"first-time">>}, {timestamp, Ts3}},
+           {{symbol, <<"last-time">>}, {timestamp, Ts4}},
            {{symbol, <<"exchange">>},{utf8, <<>>}},
            {{symbol, <<"routing-keys">>}, {array, utf8, [{utf8, QName2}]}}
           ]} = D2,
     {map, [
            {{symbol, <<"queue">>}, {utf8, QName3}},
            {{symbol, <<"reason">>}, {symbol, <<"expired">>}},
            {{symbol, <<"count">>}, {ulong, 2}},
-           {{symbol, <<"time">>}, {timestamp, Timestamp3}},
+           {{symbol, <<"first-time">>}, {timestamp, Ts5}},
+           {{symbol, <<"last-time">>}, {timestamp, Ts6}},
            {{symbol, <<"exchange">>},{utf8, <<>>}},
            {{symbol, <<"routing-keys">>}, {array, utf8, [{utf8, QName3}]}}
           ]} = D3,
-    ?assert(Timestamp1 < Timestamp2),
-    ?assert(Timestamp2 < Timestamp3),
+    ?assertEqual([Ts1, Ts3, Ts5, Ts4, Ts6, Ts2],
+                 lists:sort([Ts1, Ts2, Ts3, Ts4, Ts5, Ts6])),
 
     ok = amqp10_client:detach_link(Receiver),
     ok = amqp10_client:detach_link(Sender),
@@ -3710,7 +3714,8 @@ dead_letter_into_stream(Config) ->
                                    {{symbol, <<"queue">>}, {utf8, QName0}},
                                    {{symbol, <<"reason">>}, {symbol, <<"expired">>}},
                                    {{symbol, <<"count">>}, {ulong, 1}},
-                                   {{symbol, <<"time">>}, {timestamp, Timestamp}},
+                                   {{symbol, <<"first-time">>}, {timestamp, Timestamp}},
+                                   {{symbol, <<"last-time">>}, {timestamp, Timestamp}},
                                    {{symbol, <<"exchange">>},{utf8, <<>>}},
                                    {{symbol, <<"routing-keys">>}, {array, utf8, [{utf8, QName0}]}}
                                   ]}]}