Merge pull request #8821 from rabbitmq/mergify/bp/v3.12.x/pr-8799

Fix at-least-once dead lettering when the target include the source (backport #8799)

Author: michaelklishin

deps/rabbit/src/rabbit_fifo_dlx.erl

@@ -236,7 +236,7 @@ delivery_effects(CPid, Msgs0) ->
               Msgs = lists:zipwith(fun (Cmd, {Reason, MsgId}) ->
                                            {MsgId, {Reason, rabbit_fifo:get_msg(Cmd)}}
                                    end, Log, RsnIds),
-              [{send_msg, CPid, {dlx_delivery, Msgs}, [ra_event]}]
+              [{send_msg, CPid, {dlx_event, self(), {dlx_delivery, Msgs}}, [cast]}]
       end}].
 
 -spec state_enter(ra_server:ra_state() | eol, rabbit_types:r('queue'), dead_letter_handler(), state()) ->
@@ -308,7 +308,7 @@ update_config(at_least_once, at_least_once, _, State) ->
             {State, []};
         Pid ->
             %% Notify rabbit_fifo_dlx_worker about potentially updated policies.
-            {State, [{send_msg, Pid, lookup_topology, ra_event}]}
+            {State, [{send_msg, Pid, {dlx_event, self(), lookup_topology}, cast}]}
     end;
 update_config(SameDLH, SameDLH, _, State) ->
     {State, []};

deps/rabbit/src/rabbit_fifo_dlx_client.erl

@@ -57,9 +57,10 @@ process_command(Cmd, #state{leader = Leader} = State, Tries) ->
             process_command(Cmd, State, Tries - 1)
     end.
 
--spec handle_ra_event(ra:server_id(), term(), state()) ->
+-spec handle_ra_event(pid(), term(), state()) ->
     {ok, state(), actions()}.
-handle_ra_event(Leader, {machine, {dlx_delivery, _} = Del}, #state{leader = Leader} = State) ->
+handle_ra_event(Leader, {dlx_delivery, _} = Del,
+                #state{leader = _Leader} = State) when node(Leader) == node() ->
     handle_delivery(Del, State);
 handle_ra_event(From, Evt, State) ->
     rabbit_log:debug("Ignoring ra event ~tp from ~tp", [Evt, From]),

deps/rabbit/src/rabbit_fifo_dlx_worker.erl

@@ -135,20 +135,21 @@ handle_call(Request, From, State) ->
     rabbit_log:info("~ts received unhandled call from ~tp: ~tp", [?MODULE, From, Request]),
     {noreply, State}.
 
-handle_cast({queue_event, QRef, {_From, {machine, lookup_topology}}},
-            #state{queue_ref = QRef} = State0) ->
+handle_cast({dlx_event, _LeaderPid, lookup_topology},
+            #state{queue_ref = _} = State0) ->
     State = lookup_topology(State0),
     redeliver_and_ack(State);
-handle_cast({queue_event, QRef, {From, Evt}},
-            #state{queue_ref = QRef,
+handle_cast({dlx_event, LeaderPid, Evt},
+            #state{queue_ref = _QRef,
                    dlx_client_state = DlxState0} = State0) ->
     %% received dead-letter message from source queue
-    {ok, DlxState, Actions} = rabbit_fifo_dlx_client:handle_ra_event(From, Evt, DlxState0),
+    {ok, DlxState, Actions} = rabbit_fifo_dlx_client:handle_ra_event(LeaderPid, Evt, DlxState0),
     State1 = State0#state{dlx_client_state = DlxState},
     State = handle_queue_actions(Actions, State1),
     {noreply, State};
 handle_cast({queue_event, QRef, Evt},
             #state{queue_type_state = QTypeState0} = State0) ->
+
     case rabbit_queue_type:handle_event(QRef, Evt, QTypeState0) of
         {ok, QTypeState1, Actions} ->
             %% received e.g. confirm from target queue

deps/rabbit/test/dead_lettering_SUITE.erl

@@ -9,7 +9,6 @@
 -module(dead_lettering_SUITE).
 
 -include_lib("common_test/include/ct.hrl").
--include_lib("kernel/include/file.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
@@ -1043,6 +1042,7 @@ dead_letter_headers_cycle(Config) ->
     publish(Ch, QName, [P]),
     wait_for_messages(Config, [[QName, <<"1">>, <<"1">>, <<"0">>]]),
     [DTag] = consume(Ch, QName, [P]),
+    wait_for_messages(Config, [[QName, <<"1">>, <<"0">>, <<"1">>]]),
     amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DTag,
                                         multiple     = false,
                                         requeue      = false}),
@@ -1053,6 +1053,7 @@ dead_letter_headers_cycle(Config) ->
     {array, [{table, Death1}]} = rabbit_misc:table_lookup(Headers1, <<"x-death">>),
     ?assertEqual({long, 1}, rabbit_misc:table_lookup(Death1, <<"count">>)),
 
+    wait_for_messages(Config, [[QName, <<"1">>, <<"0">>, <<"1">>]]),
     amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DTag1,
                                         multiple     = false,
                                         requeue      = false}),

deps/rabbit/test/rabbit_fifo_dlx_integration_SUITE.erl

@@ -589,7 +589,8 @@ reject_publish(Config, QArg) when is_tuple(QArg) ->
     ok = publish_confirm(Ch, SourceQ),
     RaName = ra_name(SourceQ),
     eventually(?_assertMatch([{2, 2}], %% 2 messages with 1 byte each
-                             dirty_query([Server], RaName, fun rabbit_fifo:query_stat_dlx/1))),
+                             dirty_query([Server], RaName,
+                                         fun rabbit_fifo:query_stat_dlx/1))),
     %% Now, we have 2 expired messages in the source quorum queue's discards queue.
     %% Now that we are over the limit we expect publishes to be rejected.
     ?assertEqual(fail, publish_confirm(Ch, SourceQ)),