Skip to content

Commit a64f5f5

Browse files
ansdansd
committed
Stop using deaths{} record for deaths v2
given that first and last death keys are already stored as mc annotations and that last death is the 1st entry in the list
1 parent 0a9afb9 commit a64f5f5

File tree

4 files changed

+59
-62
lines changed

4 files changed

+59
-62
lines changed

‎deps/rabbit/include/mc.hrl

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -32,8 +32,5 @@
3232

3333
-record(deaths, {first :: death_key(), % redundant to mc annotations x-first-death-*
3434
last :: death_key(), % redundant to mc annotations x-last-death-*
35-
records :: #{death_key() := #death{}} |
36-
%% Records are ordered by death recency when feature flag
37-
%% message_containers_deaths_v2 is enabled
38-
[{death_key(), #death{}}]
35+
records :: #{death_key() := #death{}}
3936
}).

‎deps/rabbit/src/mc.erl

Lines changed: 40 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -376,46 +376,49 @@ record_death(Reason, SourceQueue,
376376
count = 1,
377377
anns = DeathAnns},
378378
Anns = case Anns0 of
379-
#{deaths := Deaths = #deaths{records = Rs0}} ->
380-
Rs = case is_list(Rs0) of
381-
true ->
382-
%% records are ordered by recency
383-
case lists:keytake(Key, 1, Rs0) of
384-
{value, {Key, D0}, Rs1} ->
385-
D = update_death(D0, Timestamp),
386-
[{Key, D} | Rs1];
387-
false ->
388-
[{Key, NewDeath} | Rs0]
389-
end;
390-
false ->
391-
maps:update_with(
392-
Key,
393-
fun(Death) -> update_death(Death, Timestamp) end,
394-
NewDeath,
395-
Rs0)
396-
end,
379+
#{deaths := Deaths0} ->
380+
Deaths = case Deaths0 of
381+
#deaths{records = Rs0} ->
382+
Rs = maps:update_with(
383+
Key,
384+
fun(Death) ->
385+
update_death(Death, Timestamp)
386+
end,
387+
NewDeath,
388+
Rs0),
389+
Deaths0#deaths{last = Key,
390+
records = Rs};
391+
_ ->
392+
%% Deaths are ordered by recency
393+
case lists:keytake(Key, 1, Deaths0) of
394+
{value, {Key, D0}, Deaths1} ->
395+
D = update_death(D0, Timestamp),
396+
[{Key, D} | Deaths1];
397+
false ->
398+
[{Key, NewDeath} | Deaths0]
399+
end
400+
end,
397401
Anns0#{<<"x-last-death-reason">> := atom_to_binary(Reason),
398402
<<"x-last-death-queue">> := SourceQueue,
399403
<<"x-last-death-exchange">> := Exchange,
400-
deaths := Deaths#deaths{last = Key,
401-
records = Rs}};
404+
deaths := Deaths};
402405
_ ->
403-
Rs = case Env of
404-
#{?FF_MC_DEATHS_V2 := false} ->
405-
#{Key => NewDeath};
406-
_ ->
407-
[{Key, NewDeath}]
408-
end,
406+
Deaths = case Env of
407+
#{?FF_MC_DEATHS_V2 := false} ->
408+
#deaths{last = Key,
409+
first = Key,
410+
records = #{Key => NewDeath}};
411+
_ ->
412+
[{Key, NewDeath}]
413+
end,
409414
ReasonBin = atom_to_binary(Reason),
410415
Anns0#{<<"x-first-death-reason">> => ReasonBin,
411416
<<"x-first-death-queue">> => SourceQueue,
412417
<<"x-first-death-exchange">> => Exchange,
413418
<<"x-last-death-reason">> => ReasonBin,
414419
<<"x-last-death-queue">> => SourceQueue,
415420
<<"x-last-death-exchange">> => Exchange,
416-
deaths => #deaths{last = Key,
417-
first = Key,
418-
records = Rs}}
421+
deaths => Deaths}
419422
end,
420423
State#?MODULE{annotations = Anns};
421424
record_death(Reason, SourceQueue, BasicMsg, Env) ->
@@ -428,11 +431,9 @@ update_death(#death{count = Count,
428431

429432
-spec is_death_cycle(rabbit_misc:resource_name(), state()) -> boolean().
430433
is_death_cycle(TargetQueue, #?MODULE{annotations = #{deaths := #deaths{records = Rs}}}) ->
431-
if is_list(Rs) ->
432-
is_cycle_v2(TargetQueue, Rs);
433-
is_map(Rs) ->
434-
is_cycle_v1(TargetQueue, maps:keys(Rs))
435-
end;
434+
is_cycle_v1(TargetQueue, maps:keys(Rs));
435+
is_death_cycle(TargetQueue, #?MODULE{annotations = #{deaths := Deaths}}) ->
436+
is_cycle_v2(TargetQueue, Deaths);
436437
is_death_cycle(_TargetQueue, #?MODULE{}) ->
437438
false;
438439
is_death_cycle(TargetQueue, BasicMsg) ->
@@ -441,13 +442,11 @@ is_death_cycle(TargetQueue, BasicMsg) ->
441442
%% Returns death queue names ordered by recency.
442443
-spec death_queue_names(state()) -> [rabbit_misc:resource_name()].
443444
death_queue_names(#?MODULE{annotations = #{deaths := #deaths{records = Rs}}}) ->
444-
if is_list(Rs) ->
445-
lists:map(fun({{Queue, _Reason}, _Death}) ->
446-
Queue
447-
end, Rs);
448-
is_map(Rs) ->
449-
proplists:get_keys(maps:keys(Rs))
450-
end;
445+
proplists:get_keys(maps:keys(Rs));
446+
death_queue_names(#?MODULE{annotations = #{deaths := Deaths}}) ->
447+
lists:map(fun({{Queue, _Reason}, _Death}) ->
448+
Queue
449+
end, Deaths);
451450
death_queue_names(#?MODULE{}) ->
452451
[];
453452
death_queue_names(BasicMsg) ->

‎deps/rabbit/src/mc_amqp.erl

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -402,9 +402,9 @@ protocol_state_message_annotations(MA, Anns) ->
402402
maps_upsert(K, mc_util:infer_type(V), L);
403403
(<<"timestamp_in_ms">>, V, L) ->
404404
maps_upsert(<<"x-opt-rabbitmq-received-time">>, {timestamp, V}, L);
405-
(deaths, #deaths{records = Records}, L)
406-
when is_list(Records) ->
407-
Maps = encode_deaths(Records),
405+
(deaths, Deaths, L)
406+
when is_list(Deaths) ->
407+
Maps = encode_deaths(Deaths),
408408
maps_upsert(<<"x-opt-deaths">>, {array, map, Maps}, L);
409409
(_, _, Acc) ->
410410
Acc

‎deps/rabbit/src/mc_amqpl.erl

Lines changed: 15 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -574,20 +574,21 @@ from_basic_message(#basic_message{content = Content,
574574

575575
%% Internal
576576

577-
deaths_to_headers(#deaths{records = Records}, Headers0) ->
578-
Infos = if is_list(Records) ->
579-
lists:map(fun death_table/1, Records);
580-
is_map(Records) ->
581-
%% sort records by the last timestamp
582-
List = lists:sort(
583-
fun({_, #death{anns = #{last_time := L1}}},
584-
{_, #death{anns = #{last_time := L2}}}) ->
585-
L1 =< L2
586-
end, maps:to_list(Records)),
587-
lists:foldl(fun(Record, Acc) ->
588-
Table = death_table(Record),
589-
[Table | Acc]
590-
end, [], List)
577+
deaths_to_headers(Deaths, Headers0) ->
578+
Infos = case Deaths of
579+
#deaths{records = Records} ->
580+
%% sort records by the last timestamp
581+
List = lists:sort(
582+
fun({_, #death{anns = #{last_time := L1}}},
583+
{_, #death{anns = #{last_time := L2}}}) ->
584+
L1 =< L2
585+
end, maps:to_list(Records)),
586+
lists:foldl(fun(Record, Acc) ->
587+
Table = death_table(Record),
588+
[Table | Acc]
589+
end, [], List);
590+
_ ->
591+
lists:map(fun death_table/1, Deaths)
591592
end,
592593
rabbit_misc:set_table_value(Headers0, <<"x-death">>, array, Infos).
593594

0 commit comments

Comments
 (0)