Merge pull request #16997 from rabbitmq/rabbitmq-server-16991

Safer federation upstream deletion

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),