Safer federation upstream deletion, closes #16991

If an exchange shared the same (virtual host, name)
tuple as a federated queue deleted by the federation
plugin, `rabbit_exchange:update_scratch/3` could
unintentionally end up being called with the
conflicting resource (exchange) virtual host
and name, overriding its decorator list.

The operation in question does not make sense
for federated queues, so let's only perform it
for exchanges.

Updating ` rabbit_db_exchange:update/2` is another
option but it has backwards-compatibility risks
and would not fully resolve the issue at hand.

(cherry picked from commit 9522020)

Author: michaelklishin

deps/rabbitmq_federation_common/src/rabbit_federation_link_sup.erl

@@ -52,10 +52,16 @@ adjust(Sup, LinkMod, XorQ, {upstream, UpstreamName}) ->
     _ = [stop(Sup, OldUpstream, XorQ) || OldUpstream <- OldUpstreams],
     [start(Sup, LinkMod, NewUpstream, XorQ) || NewUpstream <- NewUpstreams];
 
-adjust(Sup, _LinkMod, XorQ, {clear_upstream, UpstreamName}) ->
+adjust(Sup, _LinkMod, X = #exchange{}, {clear_upstream, UpstreamName}) ->
+    %% The federation scratch only exists for exchanges. Pruning it for a
+    %% queue passes the queue name to `update_scratch/3`, which resolves it to
+    %% a same-named exchange (the Khepri path ignores the resource kind) and
+    %% overwrites that exchange's decorators. See rabbitmq/rabbitmq-server#16991.
     ok = rabbit_federation_db:prune_scratch(
-           name(XorQ), rabbit_federation_upstream:for(XorQ)),
-    [stop(Sup, Upstream, XorQ) || Upstream <- children(Sup, UpstreamName)];
+           name(X), rabbit_federation_upstream:for(X)),
+    [stop(Sup, Upstream, X) || Upstream <- children(Sup, UpstreamName)];
+adjust(Sup, _LinkMod, Q, {clear_upstream, UpstreamName}) when ?is_amqqueue(Q) ->
+    [stop(Sup, Upstream, Q) || Upstream <- children(Sup, UpstreamName)];
 
 adjust(Sup, LinkMod, X = #exchange{name = XName}, {upstream_set, _Set}) ->
     _ = adjust(Sup, LinkMod, X, everything),

deps/rabbitmq_queue_federation/test/queue_SUITE.erl

@@ -60,7 +60,8 @@ federation_mechanics_tests() ->
                                                        dynamic_reconfiguration,
                                                        federate_unfederate,
                                                        dynamic_plugin_stop_start,
-                                                       supervisor_shutdown_concurrency_safety
+                                                       supervisor_shutdown_concurrency_safety,
+                                                       clear_upstream_leaves_same_named_exchange_intact
                                                       ]}
                                 ]}
     ].
@@ -399,6 +400,49 @@ supervisor_shutdown_concurrency_safety(Config) ->
 
           expect_federation(Ch, UpQ1, DownQ1, ?EXPECT_FEDERATION_TIMEOUT)
       end, upstream_downstream(Config) ++ [q(DownQ2, Args)]).
+
+%% Clearing a queue-federation upstream must leave a same-named exchange
+%% untouched. See rabbitmq/rabbitmq-server#16991.
+clear_upstream_leaves_same_named_exchange_intact(Config) ->
+    %% The federated queue and a fanout exchange share {vhost, name}.
+    Name = <<"fed1.downstream">>,
+    XName = rabbit_misc:r(<<"/">>, exchange, Name),
+    with_ch(Config,
+      fun (Ch) ->
+              #'exchange.declare_ok'{} =
+                  amqp_channel:call(Ch, #'exchange.declare'{exchange = Name,
+                                                            type = <<"fanout">>,
+                                                            durable = true}),
+              #'queue.bind_ok'{} =
+                  amqp_channel:call(Ch, #'queue.bind'{queue = Name,
+                                                      exchange = Name}),
+
+              await_running_federation(Config,
+                [{Name, <<"upstream">>}],
+                ?EXPECT_FEDERATION_TIMEOUT),
+              expect_federation(Ch, <<"upstream">>, Name, ?EXPECT_FEDERATION_TIMEOUT),
+
+              %% Before clearing, the decorators are a valid {Route, NoRoute} tuple.
+              {_, _} = exchange_decorators(Config, XName),
+
+              clear_upstream(Config, 0, <<"localhost">>),
+
+              %% After clearing they must still be a tuple, not a bare [].
+              {_, _} = exchange_decorators(Config, XName),
+
+              %% Publishing to the exchange still routes to the queue rather
+              %% than crashing the channel with a decorator function_clause.
+              Payload = <<"after-clear">>,
+              publish(Ch, Name, <<>>, Payload),
+              expect(Ch, Name, [Payload])
+      end, upstream_downstream(Config)).
+
+%% #exchange.decorators
+exchange_decorators(Config, XName) ->
+    {ok, X} = rabbit_ct_broker_helpers:rpc(Config, 0,
+                                           rabbit_exchange, lookup, [XName]),
+    element(11, X).
+
 restart_upstream(Config) ->
     [Rabbit, Hare] = rabbit_ct_broker_helpers:get_node_configs(Config,
       nodename),