Merge pull request #11608 from rabbitmq/v3.13.x-dead-letter-ts-fix

3.13.x only: fix timestamp when recovering x-death header

Author: michaelklishin

deps/rabbit/src/mc_amqpl.erl

@@ -725,13 +725,15 @@ recover_deaths([Map = #{<<"exchange">> := Exchange,
                         <<"routing-keys">> := RKeys,
                         <<"reason">> := ReasonBin,
                         <<"count">> := Count,
-                        <<"time">> := Ts} | Rem], Acc0) ->
+                        <<"time">> := TsSeconds} | Rem], Acc0)
+  when is_integer(TsSeconds) ->
     Reason = binary_to_existing_atom(ReasonBin),
-    DeathAnns0 = #{first_time => Ts,
+    TsMillis = TsSeconds * 1000,
+    DeathAnns0 = #{first_time => TsMillis,
                    %% 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},
+                   last_time => TsMillis},
     DeathAnns = case Map of
                     #{<<"original-expiration">> := Exp} ->
                         DeathAnns0#{ttl => binary_to_integer(Exp)};

deps/rabbit/test/dead_lettering_SUITE.erl

@@ -1598,6 +1598,7 @@ x_death_header_from_amqpl_client(Config) ->
     DLXQName = ?config(queue_name_dlx, Config),
     declare_dead_letter_queues(Ch, Config, QName, DLXQName, [{<<"x-message-ttl">>, long, 0}]),
 
+    Ts = os:system_time(second),
     Payload = <<"my payload">>,
     ok = amqp_channel:call(Ch,
                            #'basic.publish'{routing_key = QName},
@@ -1610,6 +1611,10 @@ x_death_header_from_amqpl_client(Config) ->
                                            no_ack = true}),
     {array, [{table, Death1}]} = rabbit_misc:table_lookup(Headers1, <<"x-death">>),
     ?assertEqual({long, 1}, rabbit_misc:table_lookup(Death1, <<"count">>)),
+    %% AMQP 0.9.1 timestamp should be seconds since epoch.
+    {timestamp, Ts1} = rabbit_misc:table_lookup(Death1, <<"time">>),
+    ?assert(Ts1 > Ts - 10),
+    ?assert(Ts1 < Ts + 10),
 
     ok = amqp_channel:call(Ch,
                            #'basic.publish'{routing_key = QName},
@@ -1623,7 +1628,10 @@ x_death_header_from_amqpl_client(Config) ->
     } = amqp_channel:call(Ch, #'basic.get'{queue = DLXQName,
                                            no_ack = true}),
     {array, [{table, Death2}]} = rabbit_misc:table_lookup(Headers2, <<"x-death">>),
-    ?assertEqual({long, 2}, rabbit_misc:table_lookup(Death2, <<"count">>)).
+    ?assertEqual({long, 2}, rabbit_misc:table_lookup(Death2, <<"count">>)),
+    %% Timestamp should be exactly the same as the 1st timestamp because it denotes
+    %% when this message was dead lettered the **first** time from this queue for this reason.
+    ?assertEqual({timestamp, Ts1}, rabbit_misc:table_lookup(Death2, <<"time">>)).
 
 set_queue_options(QName, Options) ->
     rabbit_amqqueue:update(rabbit_misc:r(<<"/">>, queue, QName),

deps/rabbit/test/mc_unit_SUITE.erl

@@ -245,6 +245,7 @@ amqpl_death_records(Env) ->
 amqpl_parse_x_death(_Config) ->
     Q = <<"my queue">>,
     DLQ = <<"my dead letter queue">>,
+    Ts = os:system_time(second),
 
     Content0 = #content{class_id = 60,
                         properties = #'P_basic'{headers = [],
@@ -273,7 +274,9 @@ amqpl_parse_x_death(_Config) ->
     ?assertMatch({_, longstr, <<"rejected">>}, header(<<"reason">>, T1)),
     ?assertMatch({_, longstr, Q}, header(<<"queue">>, T1)),
     ?assertMatch({_, longstr, <<"exch">>}, header(<<"exchange">>, T1)),
-    ?assertMatch({_, timestamp, _}, header(<<"time">>, T1)),
+    {_, timestamp, Ts1} = header(<<"time">>, T1),
+    ?assert(Ts1 > Ts - 10),
+    ?assert(Ts1 < Ts + 10),
     ?assertMatch({_, array, [{longstr, <<"apple">>}]}, header(<<"routing-keys">>, T1)),
     ?assertMatch({_, longstr, <<"9999">>}, header(<<"original-expiration">>, T1)),