rabbitmq_*_federation: Stop links during plugin stop

[Why]
Links are started by the plugins but put under the `rabbit` supervision
tree. The federation plugins supervision tree is empty unfortunately...

Links are stopped by a boot step executed by `rabbit`, as a concequence
of unregistering the plugins' parameters.

Unfortunately, links can be terminated if the channel, and implicitly
the connection stops. This happens when the `amqp_client` application
stops.

We end up with a race here:

* Because the federation plugins supervision trees are empty and the
  application stop functions barely stop the pg group (which doesn't
  terminate the group members), nothing waits for the links to stop.
  Therefore, `rabbit` can stop `amqp_client' which is a dependency of
  the federation plugins. Therefore, the links underlying channels and
  connections are stopped.

* `rabbit` unregister the federation parameters, terminating the links.
  The exchange links `terminate/2` function needs the channel to delete
  the remote queue. But the channel and the underlying connection might
  be gone.

This simply logs a `badmatch` exception:

    [error] <0.884.0> Federation link could not create a disposable (one-off) channel due to an error error: {badmatch,
    [error] <0.884.0>                                                                                         {error,
    [error] <0.884.0>                                                                                          {noproc,
    [error] <0.884.0>                                                                                           {gen_server,
    [error] <0.884.0>                                                                                            call,
    [error] <0.884.0>                                                                                            [<0.911.0>,
    [error] <0.884.0>                                                                                             {command,
    [error] <0.884.0>                                                                                              {open_channel,
    [error] <0.884.0>                                                                                               none,
    [error] <0.884.0>                                                                                               {amqp_selective_consumer,
    [error] <0.884.0>                                                                                                []}}},
    [error] <0.884.0>                                                                                             130000]}}}}

[How]
The solution is to make sure links are stopped as part of the stop of
the plugins.

`rabbit_federation_pg:stop_scope/1` is expanded to stop all members of
all groups in this scope, before terminating the pg scope itself. The
new code waits for the stopped processes to exit.

We have to handle the `EXIT` signal in the link processes and change
their restart strategy in their parent supervisor from permanent to
transient. This ensures they are restarted only if they crash. This also
skips a error log message about each stopped link.

Author: dumbbell

deps/rabbitmq_exchange_federation/src/rabbit_federation_exchange_link.erl

@@ -184,6 +184,9 @@ handle_info(check_internal_exchange, State = #state{internal_exchange = IntXName
             {noreply, State#state{internal_exchange_timer = TRef}}
     end;
 
+handle_info({'EXIT', _From, Reason}, State) ->
+    {stop, Reason, State};
+
 handle_info(Msg, State) ->
     {stop, {unexpected_info, Msg}, State}.
 

deps/rabbitmq_federation_common/src/rabbit_federation_link_sup.erl

@@ -99,12 +99,12 @@ specs(LinkMod, XorQ) ->
 
 spec(LinkMod, U = #upstream{reconnect_delay = Delay}, #exchange{name = XName}) ->
     {U, {LinkMod, start_link, [{U, XName}]},
-     {permanent, Delay}, ?WORKER_WAIT, worker,
+     {transient, Delay}, ?WORKER_WAIT, worker,
      [LinkMod]};
 
 spec(LinkMod, Upstream = #upstream{reconnect_delay = Delay}, Q) when ?is_amqqueue(Q) ->
     {Upstream, {LinkMod, start_link, [{Upstream, Q}]},
-     {permanent, Delay}, ?WORKER_WAIT, worker,
+     {transient, Delay}, ?WORKER_WAIT, worker,
      [LinkMod]}.
 
 name(#exchange{name = XName}) -> XName;

deps/rabbitmq_federation_common/src/rabbit_federation_pg.erl

@@ -17,7 +17,29 @@ stop_scope(Scope) ->
   case whereis(Scope) of
       Pid when is_pid(Pid) ->
           rabbit_log_federation:debug("Stopping pg scope ~ts", [Scope]),
+          Groups = pg:which_groups(Scope),
+          lists:foreach(
+            fun(Group) ->
+                    stop_group(Scope, Group)
+            end, Groups),
           exit(Pid, normal);
       _ ->
           ok
   end.
+
+stop_group(Scope, Group) ->
+    Members = pg:get_local_members(Scope, Group),
+    MRefs = [erlang:monitor(process, Member) || Member <- Members],
+    lists:foreach(
+      fun(Member) ->
+              exit(Member, normal)
+      end, Members),
+    lists:foreach(
+      fun(MRef) ->
+              receive
+                  {'DOWN', MRef, process, _Member, _Info} ->
+                      logger:alert("Member ~p stopped: ~0p", [_Member, _Info]),
+                      ok
+              end
+      end, MRefs),
+    ok.

deps/rabbitmq_queue_federation/src/rabbit_federation_queue_link.erl

@@ -162,6 +162,9 @@ handle_info({'DOWN', _Ref, process, Pid, Reason},
     QName = amqqueue:get_name(Q),
     handle_down(Pid, Reason, Ch, DCh, {Upstream, UParams, QName}, State);
 
+handle_info({'EXIT', _From, Reason}, State) ->
+    {stop, Reason, State};
+
 handle_info(Msg, State) ->
     {stop, {unexpected_info, Msg}, State}.