Merge pull request #16243 from rabbitmq/mergify/bp/v4.2.x/pr-16239

By @tete17: Federation: handle `{error, already_present}` in link supervisors (backport #16234) (backport #16239)

Author: michaelklishin

deps/rabbitmq_exchange_federation/src/rabbit_federation_exchange_link_sup_sup.erl

@@ -50,6 +50,13 @@ start_child(X) ->
           ?LOG_DEBUG("Federation link for exchange ~tp was already started",
                      [rabbit_misc:rs(ExchangeName)]),
           ok;
+        {error, already_present} ->
+          #exchange{name = ExchangeName} = X,
+          ?LOG_WARNING("Federation link for exchange ~tp had a stale child spec; "
+                       "removing it so the link can be restarted",
+                       [rabbit_misc:rs(ExchangeName)]),
+          _ = mirrored_supervisor:delete_child(?SUPERVISOR, id(X)),
+          ok;
         %% A link returned {stop, gone}, the link_sup shut down, that's OK.
         {error, {shutdown, _}} -> ok
     catch

deps/rabbitmq_exchange_federation/test/unit_SUITE.erl

@@ -8,6 +8,7 @@
 -module(unit_SUITE).
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
 
 -include("rabbit_exchange_federation.hrl").
 
@@ -17,7 +18,8 @@ all() -> [
     reconnect_all_empty_scope,
     reconnect_all_broadcasts_to_members,
     adjust_when_supervisor_not_running,
-    adjust_clear_upstream_when_supervisor_not_running
+    adjust_clear_upstream_when_supervisor_not_running,
+    start_child_handles_already_present
 ].
 
 init_per_suite(Config) ->
@@ -77,3 +79,21 @@ adjust_clear_upstream_when_supervisor_not_running(_Config) ->
     %% adjust/1 with clear_upstream should not fail
     ?assertEqual(ok, rabbit_federation_exchange_link_sup_sup:adjust({clear_upstream, <<"/">>, <<"test">>})),
     ?assertEqual(ok, rabbit_federation_exchange_link_sup_sup:adjust({clear_upstream_set, <<"test">>})).
+
+start_child_handles_already_present(_Config) ->
+    XName = #resource{virtual_host = <<"/">>, kind = exchange, name = <<"x">>},
+    X = #exchange{name = XName, type = direct, durable = true,
+                  auto_delete = false, internal = false, arguments = []},
+    ExpectedId = (rabbit_exchange:immutable(X))#exchange{policy = X#exchange.policy},
+    ok = meck:new(mirrored_supervisor, [unstick, passthrough]),
+    ok = meck:expect(mirrored_supervisor, start_child,
+                     fun(_Sup, _ChildSpec) -> {error, already_present} end),
+    ok = meck:expect(mirrored_supervisor, delete_child,
+                     fun(_Sup, _Id) -> ok end),
+    try
+        ?assertEqual(ok, rabbit_federation_exchange_link_sup_sup:start_child(X)),
+        ?assert(meck:called(mirrored_supervisor, delete_child,
+                            [rabbit_federation_exchange_link_sup_sup, ExpectedId]))
+    after
+        ok = meck:unload(mirrored_supervisor)
+    end.

deps/rabbitmq_queue_federation/src/rabbit_federation_queue_link_sup_sup.erl

@@ -48,6 +48,13 @@ start_child(Q) ->
           ?LOG_WARNING("Federation link for queue ~tp was already started",
                        [rabbit_misc:rs(QueueName)]),
           ok;
+        {error, already_present} ->
+          QueueName = amqqueue:get_name(Q),
+          ?LOG_WARNING("Federation link for queue ~tp had a stale child spec; "
+                       "removing it so the link can be restarted",
+                       [rabbit_misc:rs(QueueName)]),
+          _ = mirrored_supervisor:delete_child(?SUPERVISOR, id(Q)),
+          ok;
         %% A link returned {stop, gone}, the link_sup shut down, that's OK.
         {error, {shutdown, _}} -> ok
     catch

deps/rabbitmq_queue_federation/test/unit_SUITE.erl

@@ -8,6 +8,7 @@
 -module(unit_SUITE).
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
 
 -include("rabbit_queue_federation.hrl").
 
@@ -17,7 +18,8 @@ all() -> [
     reconnect_all_empty_scope,
     reconnect_all_broadcasts_to_members,
     adjust_when_supervisor_not_running,
-    adjust_clear_upstream_when_supervisor_not_running
+    adjust_clear_upstream_when_supervisor_not_running,
+    start_child_handles_already_present
 ].
 
 init_per_suite(Config) ->
@@ -77,3 +79,20 @@ adjust_clear_upstream_when_supervisor_not_running(_Config) ->
     %% adjust/1 with clear_upstream should not fail
     ?assertEqual(ok, rabbit_federation_queue_link_sup_sup:adjust({clear_upstream, <<"/">>, <<"test">>})),
     ?assertEqual(ok, rabbit_federation_queue_link_sup_sup:adjust({clear_upstream_set, <<"test">>})).
+
+start_child_handles_already_present(_Config) ->
+    Name = #resource{virtual_host = <<"/">>, kind = queue, name = <<"q">>},
+    Q = amqqueue:new(Name, none, true, false, none, [], <<"/">>, #{}),
+    ExpectedId = amqqueue:set_policy(amqqueue:set_immutable(Q), amqqueue:get_policy(Q)),
+    ok = meck:new(mirrored_supervisor, [unstick, passthrough]),
+    ok = meck:expect(mirrored_supervisor, start_child,
+                     fun(_Sup, _ChildSpec) -> {error, already_present} end),
+    ok = meck:expect(mirrored_supervisor, delete_child,
+                     fun(_Sup, _Id) -> ok end),
+    try
+        ?assertEqual(ok, rabbit_federation_queue_link_sup_sup:start_child(Q)),
+        ?assert(meck:called(mirrored_supervisor, delete_child,
+                            [rabbit_federation_queue_link_sup_sup, ExpectedId]))
+    after
+        ok = meck:unload(mirrored_supervisor)
+    end.