Revert "Merge pull request #5394 from rabbitmq/mergify/bp/v3.11.x/pr-5390"

Two reasons (like commit dcff9a219532b14119de93658a93c553b90cd3f2):
1. It was too early to backport.
2. This one should be backported last to avoid conflicts.

This reverts commit 97e9bb1a52cb048383637131d342bb68cc7007ef, reversing
changes made to 4b9e76b8a991ea99d24445424a2b17fd7a94fb2b.

3 files changed, 67 insertions, 32 deletions

diff --git a/deps/rabbit/src/rabbit_binding.erl b/deps/rabbit/src/rabbit_binding.erl
index dbc4634f13..9c875bdf53 100644
--- a/deps/rabbit/src/rabbit_binding.erl
+++ b/deps/rabbit/src/rabbit_binding.erl
@@ -18,9 +18,10 @@
 -export([info_keys/0, info/1, info/2, info_all/1, info_all/2, info_all/4]).
 %% these must all be run inside a mnesia tx
 -export([has_for_source/1, remove_for_source/1,
-         remove_for_destination/2, remove_transient_for_destination/1]).
+         remove_for_destination/2, remove_transient_for_destination/1,
+         remove_default_exchange_binding_rows_of/1]).
 
--export([populate_index_route_table/0]).
+-export([implicit_for_destination/1, reverse_binding/1, populate_index_route_table/0]).
 -export([new/4]).
 
 -define(DEFAULT_EXCHANGE(VHostPath), #resource{virtual_host = VHostPath,
@@ -225,14 +226,35 @@ remove(Src, Dst, B, ActingUser) ->
                   B#binding.source, [B], new_deletions(), false),
     process_deletions(Deletions, ActingUser).
 
+%% Implicit bindings are implicit as of rabbitmq/rabbitmq-server#1721.
+remove_default_exchange_binding_rows_of(Dst = #resource{}) ->
+    case implicit_for_destination(Dst) of
+        [Binding] ->
+            mnesia:dirty_delete(rabbit_durable_route, Binding),
+            mnesia:dirty_delete(rabbit_semi_durable_route, Binding),
+            mnesia:dirty_delete(rabbit_reverse_route,
+                                reverse_binding(Binding)),
+            mnesia:dirty_delete(rabbit_route, Binding);
+        _ ->
+            %% no binding to remove or
+            %% a competing tx has beaten us to it?
+            ok
+    end,
+    ok.
+
 -spec list_explicit() -> bindings().
 
 list_explicit() ->
     mnesia:async_dirty(
-      fun () ->
-              AllRoutes = mnesia:dirty_match_object(rabbit_route, #route{_ = '_'}),
-              [B || #route{binding = B} <- AllRoutes]
-      end).
+        fun () ->
+            AllRoutes = mnesia:dirty_match_object(rabbit_route, #route{_ = '_'}),
+            %% if there are any default exchange bindings left after an upgrade
+            %% of a pre-3.8 database, filter them out
+            AllBindings = [B || #route{binding = B} <- AllRoutes],
+            lists:filter(fun(#binding{source = S}) ->
+                            not (S#resource.kind =:= exchange andalso S#resource.name =:= <<>>)
+                         end, AllBindings)
+            end).
 
 -spec list(rabbit_types:vhost()) -> bindings().
 
@@ -242,9 +264,14 @@ list(VHostPath) ->
                                       destination = VHostResource,
                                       _           = '_'},
                    _       = '_'},
-    ExplicitBindings = [B || #route{binding = B} <- mnesia:dirty_match_object(rabbit_route,
-                                                                              Route)],
-    implicit_bindings(VHostPath) ++ ExplicitBindings.
+    %% if there are any default exchange bindings left after an upgrade
+    %% of a pre-3.8 database, filter them out
+    AllBindings = [B || #route{binding = B} <- mnesia:dirty_match_object(rabbit_route,
+                                                                         Route)],
+    Filtered    = lists:filter(fun(#binding{source = S}) ->
+                                       S =/= ?DEFAULT_EXCHANGE(VHostPath)
+                               end, AllBindings),
+    implicit_bindings(VHostPath) ++ Filtered.
 
 -spec list_for_source
         (rabbit_types:binding_source()) -> bindings().
@@ -262,35 +289,41 @@ list_for_source(SrcName) ->
 -spec list_for_destination
         (rabbit_types:binding_destination()) -> bindings().
 
-list_for_destination(DstName = #resource{}) ->
-    ExplicitBindings = mnesia:async_dirty(
-                         fun() ->
-                                 Route = #route{binding = #binding{destination = DstName,
-                                                                   _ = '_'}},
-                                 [reverse_binding(B) ||
-                                  #reverse_route{reverse_binding = B} <-
-                                  mnesia:match_object(rabbit_reverse_route,
-                                                      reverse_route(Route), read)]
-                         end),
-    implicit_for_destination(DstName) ++ ExplicitBindings.
+list_for_destination(DstName = #resource{virtual_host = VHostPath}) ->
+    AllBindings = mnesia:async_dirty(
+          fun() ->
+                  Route = #route{binding = #binding{destination = DstName,
+                                                    _ = '_'}},
+                  [reverse_binding(B) ||
+                      #reverse_route{reverse_binding = B} <-
+                          mnesia:match_object(rabbit_reverse_route,
+                                              reverse_route(Route), read)]
+          end),
+    Filtered    = lists:filter(fun(#binding{source = S}) ->
+                                       S =/= ?DEFAULT_EXCHANGE(VHostPath)
+                               end, AllBindings),
+    implicit_for_destination(DstName) ++ Filtered.
 
 -spec list_between(
     rabbit_types:binding_source(),
     rabbit_types:binding_destination()) -> bindings().
 
-list_between(SrcName = #resource{}, DstName = #resource{}) ->
-    mnesia:async_dirty(
+list_between(SrcName = #resource{virtual_host = SVH}, DstName = #resource{}) ->
+    AllBindings = mnesia:async_dirty(
       fun() ->
               Route = #route{binding = #binding{
-                                          source = SrcName,
-                                          destination = DstName,
-                                          _ = '_'}
-                            },
+                  source = SrcName,
+                  destination = DstName,
+                  _ = '_'}
+              },
               Durable     = [B || #route{binding = B} <- mnesia:match_object(rabbit_durable_route, Route, read)],
               Transient   = [B || #route{binding = B} <- mnesia:match_object(rabbit_route, Route, read)],
               SemiDurable = [B || #route{binding = B} <- mnesia:match_object(rabbit_semi_durable_route, Route, read)],
               Durable ++ Transient ++ SemiDurable
-      end).
+      end),
+    lists:filter(fun(#binding{source = S}) ->
+                       S =/= ?DEFAULT_EXCHANGE(SVH)
+                 end, AllBindings).
 
 -spec has_any_between(rabbit_types:binding_source(),
     rabbit_types:binding_destination()) -> boolean().
@@ -774,9 +807,9 @@ populate_index_route_table() ->
 %% Therefore, we avoid inserting and deleting into rabbit_index_route for other exchange
 %% types. This reduces write lock conflicts on the same tuple {SourceExchange, RoutingKey}
 %% reducing the number of restarted Mnesia transactions.
-should_index_table(#exchange{name = ?DEFAULT_EXCHANGE(_)}) ->
-    false;
-should_index_table(#exchange{type = direct}) ->
+should_index_table(#exchange{name = #resource{name = Name},
+                             type = direct})
+  when Name =/= <<>> ->
     true;
 should_index_table(_) ->
     false.
diff --git a/deps/rabbit/src/rabbit_fifo_dlx_worker.erl b/deps/rabbit/src/rabbit_fifo_dlx_worker.erl
index c151f99e11..414a0e628e 100644
--- a/deps/rabbit/src/rabbit_fifo_dlx_worker.erl
+++ b/deps/rabbit/src/rabbit_fifo_dlx_worker.erl
@@ -460,8 +460,9 @@ redeliver0(#pending{delivery = #delivery{message = BasicMsg} = Delivery0,
   when is_list(DLRKeys) ->
     Delivery = Delivery0#delivery{message = BasicMsg#basic_message{exchange_name = DLXRef,
                                                                    routing_keys  = DLRKeys}},
-    %% Because of implicit default bindings rabbit_exchange:route/2 can route to target
-    %% queues that do not exist. Therefore, filter out non-existent target queues.
+    %% rabbit_exchange:route/2 can route to target queues that do not exist
+    %% (feature flag implicit_default_bindings).
+    %% Therefore, filter out non-existent target queues.
     RouteToQs0 = queue_names(
                    rabbit_amqqueue:prepend_extra_bcc(
                      rabbit_amqqueue:lookup(
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl b/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl
index 0f31481143..9004c47ef6 100644
--- a/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl
+++ b/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl
@@ -82,6 +82,7 @@ init_per_group(Group, Config)
                                                     {rabbit,
                                                      [{forced_feature_flags_on_init,
                                                        [classic_mirrored_queue_version,
+                                                        implicit_default_bindings,
                                                         maintenance_mode_status,
                                                         user_limits,
                                                         virtual_host_metadata,