Fix at-least-once dead lettering when the target include the source.

If the target for at least once dead lettering included the source queue
the dead letter outbound queue in the quorum queue would never be cleared.

This changes the queue -> dead letter worker message format to better distinguish
between those and queue events for "normal" queue type interactions.

(cherry picked from commit 5d563d0)
(cherry picked from commit 3f6de3e)

Author: kjnilsson

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 ~p from ~p", [Evt, From]),

deps/rabbit/src/rabbit_fifo_dlx_worker.erl

@@ -135,20 +135,21 @@ handle_call(Request, From, State) ->
     rabbit_log:info("~s received unhandled call from ~p: ~p", [?MODULE, From, Request]),
     {noreply, State}.
 
-handle_cast({queue_event, QRef, {_From, {machine, lookup_topology}}},
-            #state{queue_ref = QRef} = State0) ->
+handle_cast({dlx_event, _LeaderPid, {machine, 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").
@@ -1040,6 +1039,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}),
@@ -1050,6 +1050,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}),