Improve rabbit_fifo_dlx_worker resiliency

deps/rabbit/src/rabbit_fifo_dlx.erl

@@ -114,11 +114,20 @@ apply(_, {dlx, #checkout{consumer = Pid,
 apply(_, {dlx, #checkout{consumer = ConsumerPid,
                          prefetch = Prefetch}},
       at_least_once,
-      #?MODULE{consumer = #dlx_consumer{checked_out = CheckedOutOldConsumer},
+      #?MODULE{consumer = #dlx_consumer{pid = OldConsumerPid,
+                                        checked_out = CheckedOutOldConsumer},
                discards = Discards0,
                msg_bytes = Bytes,
                msg_bytes_checkout = BytesCheckout} = State0) ->
     %% Since we allow only a single consumer, the new consumer replaces the old consumer.
+    case ConsumerPid of
+        OldConsumerPid ->
+            ok;
+        _ ->
+            rabbit_log:debug("Terminating ~p since ~p becomes active rabbit_fifo_dlx_worker",
+                             [OldConsumerPid, ConsumerPid]),
+            ensure_worker_terminated(State0)
+    end,
     %% All checked out messages to the old consumer need to be returned to the discards queue
     %% such that these messages will be re-delivered to the new consumer.
     %% When inserting back into the discards queue, we respect the original order in which messages

deps/rabbit/src/rabbit_fifo_dlx_client.erl

@@ -32,7 +32,7 @@ settle(MsgIds, #state{leader = Leader} = State)
     {ok, State}.
 
 -spec checkout(rabbit_amqqueue:name(), ra:server_id(), non_neg_integer()) ->
-    {ok, state()} | {error, ra_command_failed}.
+    {ok, state()} | {error, non_local_leader | ra_command_failed}.
 checkout(QResource, Leader, NumUnsettled) ->
     Cmd = rabbit_fifo_dlx:make_checkout(self(), NumUnsettled),
     State = #state{queue_resource = QResource,
@@ -46,10 +46,10 @@ process_command(Cmd, #state{leader = Leader} = State, Tries) ->
     case ra:process_command(Leader, Cmd, 60_000) of
         {ok, ok, Leader} ->
             {ok, State#state{leader = Leader}};
-        {ok, ok, L} ->
+        {ok, ok, NonLocalLeader} ->
             rabbit_log:warning("Failed to process command ~tp on quorum queue leader ~tp because actual leader is ~tp.",
-                               [Cmd, Leader, L]),
-            {error, ra_command_failed};
+                               [Cmd, Leader, NonLocalLeader]),
+            {error, non_local_leader};
         Err ->
             rabbit_log:warning("Failed to process command ~tp on quorum queue leader ~tp: ~tp~n"
                                "Trying ~b more time(s)...",

deps/rabbit/src/rabbit_fifo_dlx_sup.erl

@@ -24,11 +24,12 @@ start_link() ->
 
 init([]) ->
     SupFlags = #{strategy => simple_one_for_one,
-                 intensity => 1,
-                 period => 5},
+                 intensity => 100,
+                 period => 1},
     Worker = rabbit_fifo_dlx_worker,
     ChildSpec = #{id => Worker,
                   start => {Worker, start_link, []},
                   type => worker,
+                  restart => transient,
                   modules => [Worker]},
     {ok, {SupFlags, [ChildSpec]}}.

deps/rabbit/src/rabbit_fifo_dlx_worker.erl

@@ -106,23 +106,27 @@ init(QRef) ->
     {ok, undefined, {continue, QRef}}.
 
 -spec handle_continue(rabbit_amqqueue:name(), undefined) ->
-    {noreply, state()}.
+    {noreply, state()} | {stop, term(), undefined}.
 handle_continue(QRef, undefined) ->
     {ok, Prefetch} = application:get_env(rabbit,
                                          dead_letter_worker_consumer_prefetch),
     {ok, SettleTimeout} = application:get_env(rabbit,
                                               dead_letter_worker_publisher_confirm_timeout),
     {ok, Q} = rabbit_amqqueue:lookup(QRef),
     {ClusterName, _MaybeOldLeaderNode} = amqqueue:get_pid(Q),
-    {ok, ConsumerState} = rabbit_fifo_dlx_client:checkout(QRef,
-                                                          {ClusterName, node()},
-                                                          Prefetch),
-    {noreply, lookup_topology(#state{queue_ref = QRef,
-                                     queue_type_state = rabbit_queue_type:init(),
-                                     settle_timeout = SettleTimeout,
-                                     dlx_client_state = ConsumerState,
-                                     monitor_ref = erlang:monitor(process, ClusterName)
-                                    })}.
+    case rabbit_fifo_dlx_client:checkout(QRef, {ClusterName, node()}, Prefetch) of
+        {ok, ConsumerState} ->
+            {noreply, lookup_topology(#state{queue_ref = QRef,
+                                             queue_type_state = rabbit_queue_type:init(),
+                                             settle_timeout = SettleTimeout,
+                                             dlx_client_state = ConsumerState,
+                                             monitor_ref = erlang:monitor(process, ClusterName)
+                                            })};
+        {error, non_local_leader = Reason} ->
+            {stop, {shutdown, Reason}, undefined};
+        Error ->
+            {stop, Error, undefined}
+    end.
 
 terminate(_Reason, State) ->
     cancel_timer(State).