Replace rabbit_fifo_dlx_sup with two-level supervisor hierarchy

deps/rabbit/src/rabbit_fifo.erl

@@ -1113,8 +1113,7 @@ state_enter(leader,
                     enqueuers = Enqs,
                     waiting_consumers = WaitingConsumers,
                     cfg = #cfg{resource = QRes,
-                               dead_letter_handler = DLH},
-                    dlx = DlxState}) ->
+                               dead_letter_handler = DLH}}) ->
     % return effects to monitor all current consumers and enqueuers
     Pids = lists:usort(maps:keys(Enqs)
                        ++ [P || ?CONSUMER_PID(P) <- maps:values(Cons)]
@@ -1124,17 +1123,11 @@ state_enter(leader,
     NodeMons = lists:usort([{monitor, node, node(P)} || P <- Pids]),
     NotifyDecs = notify_decorators_startup(QRes),
     Effects = Mons ++ Nots ++ NodeMons ++ [NotifyDecs],
-
-    case DLH of
-        at_least_once ->
-            ensure_worker_started(QRes, DlxState);
-        _ ->
-            ok
-    end,
-    Effects;
+    maybe_add_dlx_effect(DLH, Effects);
 state_enter(eol, #?STATE{enqueuers = Enqs,
                          consumers = Cons0,
-                         waiting_consumers = WaitingConsumers0}) ->
+                         waiting_consumers = WaitingConsumers0} = State) ->
+    ok = maybe_terminate_worker(State),
     Custs = maps:fold(fun(_K, ?CONSUMER_PID(P) = V, S) ->
                               S#{P => V}
                       end, #{}, Cons0),
@@ -1145,15 +1138,8 @@ state_enter(eol, #?STATE{enqueuers = Enqs,
     [{send_msg, P, eol, ra_event}
      || P <- maps:keys(maps:merge(Enqs, AllConsumers))] ++
     [{aux, eol}];
-state_enter(_, #?STATE{cfg = #cfg{dead_letter_handler = DLH,
-                                  resource = _QRes},
-                       dlx = DlxState}) ->
-    case DLH of
-        at_least_once ->
-            ensure_worker_terminated(DlxState);
-        _ ->
-            ok
-    end,
+state_enter(_, State) ->
+    ok = maybe_terminate_worker(State),
     %% catch all as not handling all states
     [].
 
@@ -1506,26 +1492,17 @@ handle_aux(_RaState, _, force_checkpoint,
                                    ReclaimableBytes, true),
     {no_reply, Aux#?AUX{last_checkpoint = Check}, RaAux, Effects};
 handle_aux(leader, _, {dlx, setup}, Aux, RaAux) ->
-    #?STATE{dlx = DlxState,
-            cfg = #cfg{dead_letter_handler = DLH,
-                       resource = QRes}} = ra_aux:machine_state(RaAux),
-    case DLH of
-        at_least_once ->
-            ensure_worker_started(QRes, DlxState);
-        _ ->
-            ok
-    end,
+    ok = maybe_start_dlx_worker(RaAux),
     {no_reply, Aux, RaAux};
 handle_aux(_, _, {dlx, teardown, Pid}, Aux, RaAux) ->
-    terminate_dlx_worker(Pid),
+    ok = rabbit_fifo_dlx_worker:terminate_worker(Pid),
     {no_reply, Aux, RaAux};
 handle_aux(_, _, Unhandled, Aux, RaAux) ->
     #?STATE{cfg = #cfg{resource = QR}} = ra_aux:machine_state(RaAux),
     ?LOG_DEBUG("~ts: rabbit_fifo: unhandled aux command ~P",
                [rabbit_misc:rs(QR), Unhandled, 10]),
     {no_reply, Aux, RaAux}.
 
-
 eval_gc(RaAux, MacState,
         #?AUX{gc = #aux_gc{last_raft_idx = LastGcIdx} = Gc} = AuxState) ->
     {Idx, _} = ra_aux:log_last_index_term(RaAux),
@@ -3989,16 +3966,7 @@ dlx_apply(_, {dlx, {checkout, ConsumerPid, Prefetch}},
                 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;
-        _ ->
-            ?LOG_DEBUG("Terminating ~p since ~p becomes active rabbit_fifo_dlx_worker",
-                       [OldConsumerPid, ConsumerPid]),
-            %% turn into aux command
-            ensure_worker_terminated(State0)
-    end,
+    ok = ensure_worker_terminated(ConsumerPid, OldConsumerPid, State0),
     %% 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
@@ -4024,58 +3992,71 @@ dlx_apply(_, Cmd, DLH, State) ->
 %% down: 90 bytes
 %% enqueue overhead 210
 
-ensure_worker_started(QRef, #?DLX{consumer = undefined}) ->
-    start_worker(QRef);
-ensure_worker_started(QRef, #?DLX{consumer = #dlx_consumer{pid = Pid}}) ->
-    case is_local_and_alive(Pid) of
+maybe_add_dlx_effect(at_least_once, Effects) ->
+    [{aux, {dlx, setup}} | Effects];
+maybe_add_dlx_effect(_, Effects) ->
+    Effects.
+
+maybe_start_dlx_worker(RaAux) ->
+    #?STATE{dlx = DlxState,
+            cfg = #cfg{dead_letter_handler = DLH,
+                       resource = QRes}} = ra_aux:machine_state(RaAux),
+    maybe_start_dlx_worker(DLH, QRes, DlxState).
+
+maybe_start_dlx_worker(at_least_once, QRes, DlxState) ->
+    ensure_dlx_worker_started(QRes, DlxState);
+maybe_start_dlx_worker(_, _QRes, _DlxState) ->
+    ok.
+
+ensure_dlx_worker_started(QRef, #?DLX{consumer = undefined}) ->
+    start_dlx_worker(QRef);
+ensure_dlx_worker_started(QRef, #?DLX{consumer = #dlx_consumer{pid = Pid}}) ->
+    case rabbit_misc:is_local_process_alive(Pid) of
         true ->
             ?LOG_DEBUG("rabbit_fifo_dlx_worker ~tp already started for ~ts",
                              [Pid, rabbit_misc:rs(QRef)]);
         false ->
-            start_worker(QRef)
+            start_dlx_worker(QRef)
     end.
 
 %% Ensure that starting the rabbit_fifo_dlx_worker succeeds.
 %% Therefore, do not use an effect.
 %% Also therefore, if starting the rabbit_fifo_dlx_worker fails, let the
 %% Ra server process crash in which case another Ra node will become leader.
-start_worker(QRef) ->
-    {ok, Pid} = supervisor:start_child(rabbit_fifo_dlx_sup, [QRef]),
+start_dlx_worker(QRef) ->
+    {ok, Pid} = rabbit_fifo_dlx_sup_sup:start_worker(QRef),
     ?LOG_DEBUG("started rabbit_fifo_dlx_worker ~tp for ~ts",
                      [Pid, rabbit_misc:rs(QRef)]).
 
+%% Since we allow only a single consumer, the new consumer replaces the old consumer.
+ensure_worker_terminated(ConsumerPid, ConsumerPid, _State) ->
+    ok;
+ensure_worker_terminated(ConsumerPid, OldConsumerPid, State) ->
+    ?LOG_DEBUG("Terminating ~p since ~p becomes active rabbit_fifo_dlx_worker",
+                [OldConsumerPid, ConsumerPid]),
+    %% turn into aux command
+    ensure_worker_terminated(State).
+
 ensure_worker_terminated(#?DLX{consumer = undefined}) ->
     ok;
 ensure_worker_terminated(#?DLX{consumer = #dlx_consumer{pid = Pid}}) ->
-    terminate_dlx_worker(Pid).
+    ok = rabbit_fifo_dlx_worker:terminate_worker(Pid).
 
-terminate_dlx_worker(Pid) ->
-    case is_local_and_alive(Pid) of
-        true ->
-            %% Note that we can't return a mod_call effect here
-            %% because mod_call is executed on the leader only.
-            ok = supervisor:terminate_child(rabbit_fifo_dlx_sup, Pid),
-            ?LOG_DEBUG("terminated rabbit_fifo_dlx_worker ~tp", [Pid]);
-        false ->
-            ok
-    end.
+maybe_terminate_worker(#?STATE{cfg = #cfg{dead_letter_handler = at_least_once}, dlx = DlxState}) ->
+    ok = ensure_worker_terminated(DlxState);
+maybe_terminate_worker(_State) ->
+    ok.
 
 local_alive_consumer_pid(#?DLX{consumer = undefined}) ->
     undefined;
 local_alive_consumer_pid(#?DLX{consumer = #dlx_consumer{pid = Pid}}) ->
-    case is_local_and_alive(Pid) of
+    case rabbit_misc:is_local_process_alive(Pid) of
         true ->
             Pid;
         false ->
             undefined
     end.
 
-is_local_and_alive(Pid)
-  when node(Pid) =:= node() ->
-    is_process_alive(Pid);
-is_local_and_alive(_) ->
-    false.
-
 update_config(at_least_once, at_least_once, _, State) ->
     case local_alive_consumer_pid(State) of
         undefined ->

deps/rabbit/src/rabbit_fifo_dlx.erl

@@ -245,9 +245,8 @@ delivery_effects(CPid, Msgs0) ->
 
 -spec state_enter(ra_server:ra_state() | eol, rabbit_types:r('queue'), dead_letter_handler(), state()) ->
     ra_machine:effects().
-state_enter(leader, QRes, at_least_once, State) ->
-    ensure_worker_started(QRes, State),
-    [];
+state_enter(leader, _QRes, at_least_once, _State) ->
+    [{aux, {dlx, setup}}];
 state_enter(_, _, at_least_once, State) ->
     ensure_worker_terminated(State),
     [];
@@ -257,7 +256,7 @@ state_enter(_, _, _, _) ->
 ensure_worker_started(QRef, #?MODULE{consumer = undefined}) ->
     start_worker(QRef);
 ensure_worker_started(QRef, #?MODULE{consumer = #dlx_consumer{pid = Pid}}) ->
-    case is_local_and_alive(Pid) of
+    case rabbit_misc:is_local_process_alive(Pid) of
         true ->
             ?LOG_DEBUG("rabbit_fifo_dlx_worker ~tp already started for ~ts",
                              [Pid, rabbit_misc:rs(QRef)]);
@@ -270,39 +269,25 @@ ensure_worker_started(QRef, #?MODULE{consumer = #dlx_consumer{pid = Pid}}) ->
 %% Also therefore, if starting the rabbit_fifo_dlx_worker fails, let the
 %% Ra server process crash in which case another Ra node will become leader.
 start_worker(QRef) ->
-    {ok, Pid} = supervisor:start_child(rabbit_fifo_dlx_sup, [QRef]),
+    {ok, Pid} = rabbit_fifo_dlx_sup_sup:start_worker(QRef),
     ?LOG_DEBUG("started rabbit_fifo_dlx_worker ~tp for ~ts",
                      [Pid, rabbit_misc:rs(QRef)]).
 
 ensure_worker_terminated(#?MODULE{consumer = undefined}) ->
     ok;
 ensure_worker_terminated(#?MODULE{consumer = #dlx_consumer{pid = Pid}}) ->
-    case is_local_and_alive(Pid) of
-        true ->
-            %% Note that we can't return a mod_call effect here
-            %% because mod_call is executed on the leader only.
-            ok = supervisor:terminate_child(rabbit_fifo_dlx_sup, Pid),
-            ?LOG_DEBUG("terminated rabbit_fifo_dlx_worker ~tp", [Pid]);
-        false ->
-            ok
-    end.
+    ok = rabbit_fifo_dlx_worker:terminate_worker(Pid).
 
 local_alive_consumer_pid(#?MODULE{consumer = undefined}) ->
     undefined;
 local_alive_consumer_pid(#?MODULE{consumer = #dlx_consumer{pid = Pid}}) ->
-    case is_local_and_alive(Pid) of
+    case rabbit_misc:is_local_process_alive(Pid) of
         true ->
             Pid;
         false ->
             undefined
     end.
 
-is_local_and_alive(Pid)
-  when node(Pid) =:= node() ->
-    is_process_alive(Pid);
-is_local_and_alive(_) ->
-    false.
-
 -spec update_config(Old :: dead_letter_handler(), New :: dead_letter_handler(),
                     rabbit_types:r('queue'), state()) ->
     {state(), ra_machine:effects()}.

deps/rabbit/src/rabbit_fifo_dlx_sup.erl → deps/rabbit/src/rabbit_fifo_dlx_sup_sup.erl

@@ -2,9 +2,9 @@
 %% License, v. 2.0. If a copy of the MPL was not distributed with this
 %% file, You can obtain one at https://mozilla.org/MPL/2.0/.
 %%
-%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
 
--module(rabbit_fifo_dlx_sup).
+-module(rabbit_fifo_dlx_sup_sup).
 
 -behaviour(supervisor).
 
@@ -18,19 +18,21 @@
 %% supervisor callback
 -export([init/1]).
 %% client API
--export([start_link/0]).
+-export([start_link/0, start_worker/1]).
 
 start_link() ->
     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
+-spec start_worker(rabbit_amqqueue:name()) -> {ok, pid()}.
+start_worker(QRef) ->
+    {ok, SupPid} = supervisor:start_child(?MODULE, [QRef]),
+    [{_, Pid, worker, _}] = supervisor:which_children(SupPid),
+    {ok, Pid}.
+
 init([]) ->
-    SupFlags = #{strategy => simple_one_for_one,
-                 intensity => 100,
-                 period => 1},
-    Worker = rabbit_fifo_dlx_worker,
-    ChildSpec = #{id => Worker,
-                  start => {Worker, start_link, []},
-                  type => worker,
-                  restart => transient,
-                  modules => [Worker]},
+    SupFlags = #{strategy => simple_one_for_one},
+    ChildSpec = #{id => rabbit_fifo_dlx_worker_sup,
+                  start => {rabbit_fifo_dlx_worker_sup, start_link, []},
+                  type => supervisor,
+                  restart => temporary},
     {ok, {SupFlags, [ChildSpec]}}.

deps/rabbit/src/rabbit_fifo_dlx_worker.erl

@@ -29,13 +29,16 @@
 
 -behaviour(gen_server).
 
--export([start_link/1]).
+-export([start_link/2]).
 %% gen_server callbacks
 -export([init/1, terminate/2, handle_continue/2,
          handle_cast/2, handle_call/3, handle_info/2,
          code_change/3, format_status/1]).
 
+-export([terminate_worker/1]).
+
 -define(HIBERNATE_AFTER, 4*60*1000).
+-define(DLX_WORKER_SUP_PID_KEY, dlx_worker_sup_pid).
 
 -record(pending, {
           %% consumed_msg_id is not to be confused with consumer delivery tag.
@@ -100,14 +103,38 @@
 
 -type state() :: #state{}.
 
-start_link(QRef) ->
-    gen_server:start_link(?MODULE, QRef, [{hibernate_after, ?HIBERNATE_AFTER}]).
+start_link(QRef, SupPid) ->
+    gen_server:start_link(?MODULE, {QRef, SupPid}, [{hibernate_after, ?HIBERNATE_AFTER}]).
 
--spec init(rabbit_amqqueue:name()) ->
+-spec init({rabbit_amqqueue:name(), pid()}) ->
     {ok, undefined, {continue, rabbit_amqqueue:name()}}.
-init(QRef) ->
+init({QRef, SupPid}) ->
+    ok = put_sup_pid(SupPid),
     {ok, undefined, {continue, QRef}}.
 
+-spec put_sup_pid(DlxWorkerSupPid :: pid()) -> ok.
+put_sup_pid(DlxWorkerSupPid) ->
+    %% Stored in the process dictionary so that terminate_dlx_worker/2 can
+    %% retrieve it via process_info(WorkerPid, dictionary) without blocking
+    %% on a gen_server call. Same pattern as rabbit_misc:store_proc_name/1.
+    put(?DLX_WORKER_SUP_PID_KEY, DlxWorkerSupPid),
+    ok.
+
+-spec terminate_worker(DlxWorkerPid :: pid()) -> ok.
+terminate_worker(DlxWorkerPid) ->
+    terminate_worker(DlxWorkerPid, rabbit_misc:is_local_process_alive(DlxWorkerPid)).
+
+-spec terminate_worker(DlxWorkerPid :: pid(), boolean()) -> ok.
+terminate_worker(DlxWorkerPid, true) ->
+    %% Terminate the per-worker supervisor (which also stops the worker).
+    {dictionary, Dict} = erlang:process_info(DlxWorkerPid, dictionary),
+    {?DLX_WORKER_SUP_PID_KEY, SupPid} = lists:keyfind(?DLX_WORKER_SUP_PID_KEY, 1, Dict),
+    ok = supervisor:terminate_child(rabbit_fifo_dlx_sup_sup, SupPid),
+    ?LOG_DEBUG("terminated rabbit_fifo_dlx_worker ~tp", [DlxWorkerPid]),
+    ok;
+terminate_worker(_DlxWorkerPid, _) ->
+    ok.
+
 -spec handle_continue(rabbit_amqqueue:name(), undefined) ->
     {noreply, state()} | {stop, term(), undefined}.
 handle_continue(QRef, undefined) ->

deps/rabbit/src/rabbit_fifo_dlx_worker_sup.erl

@@ -0,0 +1,32 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+
+%% Per-queue supervisor for a single rabbit_fifo_dlx_worker process.
+%% Started as a temporary child of rabbit_fifo_dlx_sup_sup (simple_one_for_one).
+%% This isolation prevents a burst of DLX worker crashes from taking down
+%% the top-level supervisor and stranding unrelated queues.
+-module(rabbit_fifo_dlx_worker_sup).
+
+-behaviour(supervisor).
+
+-export([start_link/1, init/1]).
+
+start_link(QRef) ->
+    supervisor:start_link(?MODULE, [QRef]).
+
+init([QRef]) ->
+    DlxWorkerSupPid = self(),
+    SupFlags = #{strategy => one_for_one,
+                 intensity => 0,
+                 period => 1,
+                 auto_shutdown => any_significant},
+    ChildSpec = #{id => rabbit_fifo_dlx_worker,
+                  start => {rabbit_fifo_dlx_worker, start_link, [QRef, DlxWorkerSupPid]},
+                  type => worker,
+                  restart => transient,
+                  significant => true,
+                  shutdown => 5000},
+    {ok, {SupFlags, [ChildSpec]}}.