Prevent federation links from restarting during node shutdown

michaelklishin · michaelklishin · commit adc25bbb3578 · 2026-01-13T16:40:53.000-08:00
or plugin shutdown, for that matter. With this guardrail in place, nodes with hundreds or thousands of federation links will avoid potentially significant shutdown delays that have to do with links being restarted while the node as a whole is preparing to shut down. Per discussion with @dcorbacho @ansd.
diff --git a/deps/rabbitmq_exchange_federation/src/rabbit_exchange_federation_app.erl b/deps/rabbitmq_exchange_federation/src/rabbit_exchange_federation_app.erl
@@ -13,7 +13,7 @@
 -include_lib("kernel/include/logger.hrl").
 
 -behaviour(application).
--export([start/2, stop/1]).
+-export([start/2, prep_stop/1, stop/1]).
 
 %% Dummy supervisor - see Ulf Wiger's comment at
 %% http://erlang.org/pipermail/erlang-questions/2010-April/050508.html
@@ -37,6 +37,10 @@ start(_Type, _StartArgs) ->
                 #{link_module => rabbit_federation_exchange_link_sup_sup}}),
     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
+prep_stop(State) ->
+    rabbit_federation_app_state:mark_as_shutting_down(),
+    State.
+
 stop(_State) ->
     ets:delete(?FEDERATION_ETS, rabbitmq_exchange_federation),
     rabbit_federation_pg:stop_scope(?FEDERATION_PG_SCOPE),
diff --git a/deps/rabbitmq_exchange_federation/src/rabbit_federation_exchange_link.erl b/deps/rabbitmq_exchange_federation/src/rabbit_federation_exchange_link.erl
@@ -66,6 +66,14 @@ start_link(Args) ->
     gen_server2:start_link(?MODULE, Args, [{timeout, infinity}]).
 
 init({Upstream, XName}) ->
+    case rabbit_federation_app_state:is_shutting_down() of
+        true ->
+            ignore;
+        false ->
+            init_link({Upstream, XName})
+    end.
+
+init_link({Upstream, XName}) ->
     logger:set_process_metadata(#{domain => ?RMQLOG_DOMAIN_FEDERATION,
                                   exchange => XName}),
     %% If we are starting up due to a policy change then it's possible
diff --git a/deps/rabbitmq_exchange_federation/src/rabbit_federation_exchange_link_sup_sup.erl b/deps/rabbitmq_exchange_federation/src/rabbit_federation_exchange_link_sup_sup.erl
@@ -31,6 +31,7 @@ start_link() ->
     %% and other places, so we have to start it very early outside of the supervision tree.
     %% The scope is stopped in stop/1.
     _ = rabbit_federation_pg:start_scope(?FEDERATION_PG_SCOPE),
+    rabbit_federation_app_state:reset_shutting_down_marker(),
     mirrored_supervisor:start_link({local, ?SUPERVISOR}, ?SUPERVISOR,
                                    ?MODULE, []).
 
diff --git a/deps/rabbitmq_federation_common/src/rabbit_federation_app_state.erl b/deps/rabbitmq_federation_common/src/rabbit_federation_app_state.erl
@@ -0,0 +1,26 @@
+%% 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.
+%%
+
+%% Tracks transient application state for federation plugins.
+%% Used to prevent link restarts during node shutdown.
+-module(rabbit_federation_app_state).
+
+-export([is_shutting_down/0, mark_as_shutting_down/0, reset_shutting_down_marker/0]).
+
+-define(APP, rabbitmq_federation_common).
+
+-spec is_shutting_down() -> boolean().
+is_shutting_down() ->
+    application:get_env(?APP, shutting_down, false).
+
+-spec mark_as_shutting_down() -> ok.
+mark_as_shutting_down() ->
+    application:set_env(?APP, shutting_down, true).
+
+-spec reset_shutting_down_marker() -> ok.
+reset_shutting_down_marker() ->
+    application:unset_env(?APP, shutting_down).
diff --git a/deps/rabbitmq_federation_common/test/unit_SUITE.erl b/deps/rabbitmq_federation_common/test/unit_SUITE.erl
@@ -17,7 +17,10 @@ all() -> [
     obfuscate_upstream,
     obfuscate_upstream_params_network,
     obfuscate_upstream_params_network_with_char_list_password_value,
-    obfuscate_upstream_params_direct
+    obfuscate_upstream_params_direct,
+    shutdown_flag_defaults_to_false,
+    shutdown_flag_can_be_set,
+    shutdown_flag_can_be_cleared
 ].
 
 init_per_suite(Config) ->
@@ -63,3 +66,23 @@ obfuscate_upstream_params_network_with_char_list_password_value(_Config) ->
     ObfuscatedUpstreamParams = rabbit_federation_util:obfuscate_upstream_params(UpstreamParams),
     ?assertEqual(UpstreamParams, rabbit_federation_util:deobfuscate_upstream_params(ObfuscatedUpstreamParams)),
     ok.
+
+shutdown_flag_defaults_to_false(_Config) ->
+    application:unset_env(rabbitmq_federation_common, shutting_down),
+    ?assertEqual(false, rabbit_federation_app_state:is_shutting_down()),
+    ok.
+
+shutdown_flag_can_be_set(_Config) ->
+    application:unset_env(rabbitmq_federation_common, shutting_down),
+    ?assertEqual(false, rabbit_federation_app_state:is_shutting_down()),
+    ok = rabbit_federation_app_state:mark_as_shutting_down(),
+    ?assertEqual(true, rabbit_federation_app_state:is_shutting_down()),
+    application:unset_env(rabbitmq_federation_common, shutting_down),
+    ok.
+
+shutdown_flag_can_be_cleared(_Config) ->
+    ok = rabbit_federation_app_state:mark_as_shutting_down(),
+    ?assertEqual(true, rabbit_federation_app_state:is_shutting_down()),
+    ok = rabbit_federation_app_state:reset_shutting_down_marker(),
+    ?assertEqual(false, rabbit_federation_app_state:is_shutting_down()),
+    ok.
diff --git a/deps/rabbitmq_queue_federation/src/rabbit_federation_queue_link.erl b/deps/rabbitmq_queue_federation/src/rabbit_federation_queue_link.erl
@@ -54,6 +54,14 @@ q(QName) ->
 %%----------------------------------------------------------------------------
 
 init({Upstream, Queue}) when ?is_amqqueue(Queue) ->
+    case rabbit_federation_app_state:is_shutting_down() of
+        true ->
+            ignore;
+        false ->
+            init_link({Upstream, Queue})
+    end.
+
+init_link({Upstream, Queue}) when ?is_amqqueue(Queue) ->
     QName = amqqueue:get_name(Queue),
     logger:set_process_metadata(#{domain => ?RMQLOG_DOMAIN_FEDERATION,
                                   queue => QName}),
diff --git a/deps/rabbitmq_queue_federation/src/rabbit_federation_queue_link_sup_sup.erl b/deps/rabbitmq_queue_federation/src/rabbit_federation_queue_link_sup_sup.erl
@@ -30,6 +30,7 @@ start_link() ->
     %% and other places, so we have to start it very early outside of the supervision tree.
     %% The scope is stopped in stop/1.
     _ = rabbit_federation_pg:start_scope(?FEDERATION_PG_SCOPE),
+    rabbit_federation_app_state:reset_shutting_down_marker(),
     mirrored_supervisor:start_link({local, ?SUPERVISOR}, ?SUPERVISOR,
                                    ?MODULE, []).
 
diff --git a/deps/rabbitmq_queue_federation/src/rabbit_queue_federation_app.erl b/deps/rabbitmq_queue_federation/src/rabbit_queue_federation_app.erl
@@ -12,7 +12,7 @@
 -include("rabbit_queue_federation.hrl").
 
 -behaviour(application).
--export([start/2, stop/1]).
+-export([start/2, prep_stop/1, stop/1]).
 
 %% Dummy supervisor - see Ulf Wiger's comment at
 %% http://erlang.org/pipermail/erlang-questions/2010-April/050508.html
@@ -36,6 +36,10 @@ start(_Type, _StartArgs) ->
                 #{link_module => rabbit_federation_queue_link_sup_sup}}),
     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
+prep_stop(State) ->
+    rabbit_federation_app_state:mark_as_shutting_down(),
+    State.
+
 stop(_State) ->
     ets:delete(?FEDERATION_ETS, rabbitmq_queue_federation),
     rabbit_federation_pg:stop_scope(?FEDERATION_PG_SCOPE),
diff --git a/deps/rabbitmq_queue_federation/test/queue_SUITE.erl b/deps/rabbitmq_queue_federation/test/queue_SUITE.erl
@@ -146,6 +146,9 @@ end_per_group(_, Config) ->
       rabbit_ct_client_helpers:teardown_steps() ++
       rabbit_ct_broker_helpers:teardown_steps()).
 
+init_per_testcase(dynamic_plugin_stop_start = Testcase, Config) ->
+    ct:timetrap({seconds, 90}),
+    rabbit_ct_helpers:testcase_started(Config, Testcase);
 init_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_started(Config, Testcase).
 
