Replace rabbit_fifo_dlx_sup with a two-level supervisor hierarchy

lukebakken · lukebakken · commit 9594857a0e81 · 2026-06-12T19:24:46.000Z
The flat simple_one_for_one rabbit_fifo_dlx_sup allowed a burst of DLX worker crashes to trip the supervisor's restart intensity and render it permanently unavailable, stranding all at_least_once DLX quorum queues cluster-wide (GitHub #16652). Replace it with: - rabbit_fifo_dlx_sup_sup: simple_one_for_one top-level supervisor whose children are per-queue worker supervisors (temporary). - rabbit_fifo_dlx_worker_sup: one_for_one per-queue supervisor with intensity 0. A worker crash terminates only its own supervisor, isolating failures to a single queue. Additional fixes: - ensure_worker_started now checks live supervisor state via find_worker_sup/1 before starting a new worker, preventing duplicate worker_sups from concurrent state_enter and {dlx, setup} aux calls. - state_enter(eol, ...) now terminates the DLX worker on queue deletion, preventing orphaned worker_sups after queue cleanup.
diff --git a/deps/rabbit/src/rabbit_fifo.erl b/deps/rabbit/src/rabbit_fifo.erl
@@ -1134,7 +1134,13 @@ state_enter(leader,
     Effects;
 state_enter(eol, #?STATE{enqueuers = Enqs,
                          consumers = Cons0,
-                         waiting_consumers = WaitingConsumers0}) ->
+                         waiting_consumers = WaitingConsumers0,
+                         cfg = #cfg{dead_letter_handler = DLH},
+                         dlx = DlxState}) ->
+    case DLH of
+        at_least_once -> ensure_worker_terminated(DlxState);
+        _ -> ok
+    end,
     Custs = maps:fold(fun(_K, ?CONSUMER_PID(P) = V, S) ->
                               S#{P => V}
                       end, #{}, Cons0),
@@ -4025,22 +4031,39 @@ dlx_apply(_, Cmd, DLH, State) ->
 %% enqueue overhead 210
 
 ensure_worker_started(QRef, #?DLX{consumer = undefined}) ->
-    start_worker(QRef);
+    case find_worker_sup(QRef) of
+        undefined -> start_worker(QRef);
+        _SupPid -> ok
+    end;
 ensure_worker_started(QRef, #?DLX{consumer = #dlx_consumer{pid = Pid}}) ->
     case is_local_and_alive(Pid) of
         true ->
             ?LOG_DEBUG("rabbit_fifo_dlx_worker ~tp already started for ~ts",
                              [Pid, rabbit_misc:rs(QRef)]);
         false ->
-            start_worker(QRef)
+            case find_worker_sup(QRef) of
+                undefined -> start_worker(QRef);
+                _SupPid -> ok
+            end
     end.
 
+find_worker_sup(QRef) ->
+    lists:foldl(
+      fun({_, SupPid, _, _}, undefined) ->
+              case rabbit_fifo_dlx_worker_sup:get_qref(SupPid) of
+                  QRef -> SupPid;
+                  _ -> undefined
+              end;
+         (_, Found) -> Found
+      end, undefined, supervisor:which_children(rabbit_fifo_dlx_sup_sup)).
+
 %% 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]),
+    {ok, SupPid} = supervisor:start_child(rabbit_fifo_dlx_sup_sup, [QRef]),
+    [{_, Pid, worker, _}] = supervisor:which_children(SupPid),
     ?LOG_DEBUG("started rabbit_fifo_dlx_worker ~tp for ~ts",
                      [Pid, rabbit_misc:rs(QRef)]).
 
@@ -4054,7 +4077,10 @@ terminate_dlx_worker(Pid) ->
         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),
+            %% Terminate the per-worker supervisor (which also stops the worker).
+            {dictionary, Dict} = erlang:process_info(Pid, dictionary),
+            {_, [SupPid | _]} = lists:keyfind('$ancestors', 1, Dict),
+            ok = supervisor:terminate_child(rabbit_fifo_dlx_sup_sup, SupPid),
             ?LOG_DEBUG("terminated rabbit_fifo_dlx_worker ~tp", [Pid]);
         false ->
             ok
diff --git a/deps/rabbit/src/rabbit_fifo_dlx.erl b/deps/rabbit/src/rabbit_fifo_dlx.erl
@@ -255,22 +255,39 @@ state_enter(_, _, _, _) ->
     [].
 
 ensure_worker_started(QRef, #?MODULE{consumer = undefined}) ->
-    start_worker(QRef);
+    case find_worker_sup(QRef) of
+        undefined -> start_worker(QRef);
+        _SupPid -> ok
+    end;
 ensure_worker_started(QRef, #?MODULE{consumer = #dlx_consumer{pid = Pid}}) ->
     case is_local_and_alive(Pid) of
         true ->
             ?LOG_DEBUG("rabbit_fifo_dlx_worker ~tp already started for ~ts",
                              [Pid, rabbit_misc:rs(QRef)]);
         false ->
-            start_worker(QRef)
+            case find_worker_sup(QRef) of
+                undefined -> start_worker(QRef);
+                _SupPid -> ok
+            end
     end.
 
+find_worker_sup(QRef) ->
+    lists:foldl(
+      fun({_, SupPid, _, _}, undefined) ->
+              case rabbit_fifo_dlx_worker_sup:get_qref(SupPid) of
+                  QRef -> SupPid;
+                  _ -> undefined
+              end;
+         (_, Found) -> Found
+      end, undefined, supervisor:which_children(rabbit_fifo_dlx_sup_sup)).
+
 %% 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]),
+    {ok, SupPid} = supervisor:start_child(rabbit_fifo_dlx_sup_sup, [QRef]),
+    [{_, Pid, worker, _}] = supervisor:which_children(SupPid),
     ?LOG_DEBUG("started rabbit_fifo_dlx_worker ~tp for ~ts",
                      [Pid, rabbit_misc:rs(QRef)]).
 
@@ -281,7 +298,10 @@ ensure_worker_terminated(#?MODULE{consumer = #dlx_consumer{pid = Pid}}) ->
         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),
+            %% Terminate the per-worker supervisor (which also stops the worker).
+            {dictionary, Dict} = erlang:process_info(Pid, dictionary),
+            {_, [SupPid | _]} = lists:keyfind('$ancestors', 1, Dict),
+            ok = supervisor:terminate_child(rabbit_fifo_dlx_sup_sup, SupPid),
             ?LOG_DEBUG("terminated rabbit_fifo_dlx_worker ~tp", [Pid]);
         false ->
             ok
diff --git a/deps/rabbit/src/rabbit_fifo_dlx_sup_sup.erl b/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).
 
@@ -24,13 +24,9 @@ start_link() ->
     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
 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]}}.
diff --git a/deps/rabbit/src/rabbit_fifo_dlx_worker_sup.erl b/deps/rabbit/src/rabbit_fifo_dlx_worker_sup.erl
@@ -0,0 +1,39 @@
+%% 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, get_qref/1]).
+
+start_link(QRef) ->
+    supervisor:start_link(?MODULE, [QRef]).
+
+init([QRef]) ->
+    erlang:put(rabbit_fifo_dlx_qref, QRef),
+    SupFlags = #{strategy => one_for_one,
+                 intensity => 0,
+                 period => 1},
+    ChildSpec = #{id => rabbit_fifo_dlx_worker,
+                  start => {rabbit_fifo_dlx_worker, start_link, [QRef]},
+                  type => worker,
+                  restart => transient,
+                  shutdown => 5000},
+    {ok, {SupFlags, [ChildSpec]}}.
+
+-spec get_qref(pid()) -> rabbit_fifo:rabbit_fifo_queue_ref() | undefined.
+get_qref(SupPid) ->
+    case erlang:process_info(SupPid, dictionary) of
+        {dictionary, Dict} ->
+            proplists:get_value(rabbit_fifo_dlx_qref, Dict);
+        undefined ->
+            undefined
+    end.
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -2842,7 +2842,7 @@ queue_vm_stats_sups() ->
     {[quorum_queue_procs,
       quorum_queue_dlx_procs],
      [[ra_server_sup_sup],
-      [rabbit_fifo_dlx_sup]]}.
+      [rabbit_fifo_dlx_sup_sup]]}.
 
 queue_vm_ets() ->
     {[quorum_ets],
diff --git a/deps/rabbit/src/rabbit_vm.erl b/deps/rabbit/src/rabbit_vm.erl
@@ -22,7 +22,7 @@ memory() ->
     %% example for existing info keys:
     %% [{queue_procs,          queue_sups()},
     %%  {quorum_queue_procs,   [ra_server_sup_sup]},
-    %%  {quorum_queue_dlx_procs, [rabbit_fifo_dlx_sup]},
+    %%  {quorum_queue_dlx_procs, [rabbit_fifo_dlx_sup_sup]},
     %%  {stream_queue_procs,   [osiris_server_sup]},
     %%  {stream_queue_replica_reader_procs,  [osiris_replica_reader_sup]},
     %%  {stream_queue_coordinator_procs, [rabbit_stream_coordinator]}]
diff --git a/deps/rabbit/test/rabbit_fifo_dlx_SUITE.erl b/deps/rabbit/test/rabbit_fifo_dlx_SUITE.erl
@@ -128,7 +128,7 @@ switch_strategies(_Config) ->
                      name = <<"blah">>},
     application:set_env(rabbit, dead_letter_worker_consumer_prefetch, 1),
     application:set_env(rabbit, dead_letter_worker_publisher_confirm_timeout, 1000),
-    {ok, _} = rabbit_fifo_dlx_sup:start_link(),
+    {ok, _} = rabbit_fifo_dlx_sup_sup:start_link(),
     S0 = rabbit_fifo_dlx:init(),
 
     Handler0 = undefined,
@@ -141,7 +141,8 @@ switch_strategies(_Config) ->
                   {aux, {dlx, setup}}],
                  Effects0),
     rabbit_fifo_dlx:handle_aux(leader, {dlx, setup}, fake_aux, QRes, Handler1, S1),
-    [{_, WorkerPid, worker, _}] = supervisor:which_children(rabbit_fifo_dlx_sup),
+    [{_, WorkerSupPid, supervisor, _}] = supervisor:which_children(rabbit_fifo_dlx_sup_sup),
+    [{_, WorkerPid, worker, _}] = supervisor:which_children(WorkerSupPid),
     {S2, _} = rabbit_fifo_dlx:discard([make_msg(1)], because, Handler1, S1),
     Checkout = rabbit_fifo_dlx:make_checkout(WorkerPid, 1),
     {S3, _} = rabbit_fifo_dlx:apply(meta(2), Checkout, Handler1, S2),
@@ -158,7 +159,7 @@ switch_strategies(_Config) ->
                     [1, 1, "queue 'blah' in vhost '/'"]]}],
                  Effects),
     ?assertMatch([_, {active, 0}, _, _],
-                 supervisor:count_children(rabbit_fifo_dlx_sup)),
+                 supervisor:count_children(rabbit_fifo_dlx_sup_sup)),
     ?assertMatch(#{num_discarded := 0}, rabbit_fifo_dlx:overview(S5)),
     ok.
 
diff --git a/deps/rabbit/test/rabbit_fifo_dlx_integration_SUITE.erl b/deps/rabbit/test/rabbit_fifo_dlx_integration_SUITE.erl
@@ -154,7 +154,7 @@ end_per_testcase(Testcase, Config) ->
     ?awaitMatch(
        true,
        begin
-           DlxWorkers = rabbit_ct_broker_helpers:rpc_all(Config, supervisor, which_children, [rabbit_fifo_dlx_sup]),
+           DlxWorkers = rabbit_ct_broker_helpers:rpc_all(Config, supervisor, which_children, [rabbit_fifo_dlx_sup_sup]),
            lists:all(fun(L) -> L =:= [] end, DlxWorkers)
        end, 60000),
 
@@ -513,7 +513,7 @@ drop_head_falls_back_to_at_most_once(Config) ->
     consistently(
       ?_assertMatch(
          [_, {active, 0}, _, _],
-         rpc(Config, Server, supervisor, count_children, [rabbit_fifo_dlx_sup]))).
+         rpc(Config, Server, supervisor, count_children, [rabbit_fifo_dlx_sup_sup]))).
 
 %% Test that dynamically switching dead-letter-strategy works.
 switch_strategy(Config) ->
@@ -935,7 +935,7 @@ single_dlx_worker(Config) ->
        [[_, {active, 1}, _, _],
         [_, {active, 0}, _, _],
         [_, {active, 0}, _, _]],
-       rabbit_ct_broker_helpers:rpc_all(Config, supervisor, count_children, [rabbit_fifo_dlx_sup])),
+       rabbit_ct_broker_helpers:rpc_all(Config, supervisor, count_children, [rabbit_fifo_dlx_sup_sup])),
 
     ok = rabbit_ct_broker_helpers:stop_broker(Config, Server1),
     RaName = ra_name(SourceQ),
@@ -948,7 +948,7 @@ single_dlx_worker(Config) ->
     consistently(
       ?_assertEqual(
          0,
-         length(rpc(Config, Server1, supervisor, which_children, [rabbit_fifo_dlx_sup], 1000)))),
+         length(rpc(Config, Server1, supervisor, which_children, [rabbit_fifo_dlx_sup_sup], 1000)))),
 
     %% Kill the leader node to verify the DLX worker follows leadership.
     %% We kill the node rather than just the Ra process: killing only the process
@@ -965,7 +965,7 @@ single_dlx_worker(Config) ->
 
 assert_active_dlx_workers(N, Config, Server) ->
     ?awaitMatch(N,
-                length(rpc(Config, Server, supervisor, which_children, [rabbit_fifo_dlx_sup], 5000)),
+                length(rpc(Config, Server, supervisor, which_children, [rabbit_fifo_dlx_sup_sup], 5000)),
                 60_000).
 
 declare_queue(Channel, Queue, Args) ->
