Merge pull request #4606 from rabbitmq/route-faster

Route faster in direct exchange

12 files changed, 621 insertions, 38 deletions

diff --git a/deps/rabbit/BUILD.bazel b/deps/rabbit/BUILD.bazel
index cefd487fa5..85e7e6ecd5 100644
--- a/deps/rabbit/BUILD.bazel
+++ b/deps/rabbit/BUILD.bazel
@@ -1058,6 +1058,11 @@ suites = [
             "//deps/rabbit_common:erlang_app",
         ],
     ),
+    rabbitmq_integration_suite(
+        PACKAGE,
+        name = "direct_exchange_routing_v2_SUITE",
+        size = "medium",
+    ),
 ]
 
 assert_suites(
diff --git a/deps/rabbit/src/rabbit_binding.erl b/deps/rabbit/src/rabbit_binding.erl
index 0ee256d389..8a15b302d3 100644
--- a/deps/rabbit/src/rabbit_binding.erl
+++ b/deps/rabbit/src/rabbit_binding.erl
@@ -20,7 +20,8 @@
          remove_for_destination/2, remove_transient_for_destination/1,
          remove_default_exchange_binding_rows_of/1]).
 
--export([implicit_for_destination/1, reverse_binding/1]).
+-export([implicit_for_destination/1, reverse_binding/1,
+         index_route_table_definition/0, populate_index_route_table/0]).
 -export([new/4]).
 
 -define(DEFAULT_EXCHANGE(VHostPath), #resource{virtual_host = VHostPath,
@@ -124,7 +125,7 @@ recover_semi_durable_route_txn(R = #route{binding = B}, X) ->
       fun () ->
               case mnesia:read(rabbit_semi_durable_route, B, read) of
                   [] -> no_recover;
-                  _  -> ok = sync_transient_route(R, fun mnesia:write/3),
+                  _  -> ok = sync_transient_route(R, should_index_table(X), fun mnesia:write/3),
                         rabbit_exchange:serial(X)
               end
       end,
@@ -178,9 +179,8 @@ add(Binding, InnerFun, ActingUser) ->
       end, fun not_found_or_absent_errs/1).
 
 add(Src, Dst, B, ActingUser) ->
-    [SrcDurable, DstDurable] = [durable(E) || E <- [Src, Dst]],
-    ok = sync_route(#route{binding = B}, SrcDurable, DstDurable,
-                    fun mnesia:write/3),
+    ok = sync_route(#route{binding = B}, durable(Src), durable(Dst),
+                    should_index_table(Src), fun mnesia:write/3),
     x_callback(transaction, Src, add_binding, B),
     Serial = rabbit_exchange:serial(Src),
     fun () ->
@@ -221,7 +221,7 @@ remove(Binding, InnerFun, ActingUser) ->
 
 remove(Src, Dst, B, ActingUser) ->
     ok = sync_route(#route{binding = B}, durable(Src), durable(Dst),
-                    fun delete/3),
+                    should_index_table(Src), fun delete/3),
     Deletions = maybe_auto_delete(
                   B#binding.source, [B], new_deletions(), false),
     process_deletions(Deletions, ActingUser).
@@ -400,15 +400,26 @@ has_for_source(SrcName) ->
     contains(rabbit_route, Match) orelse
         contains(rabbit_semi_durable_route, Match).
 
--spec remove_for_source(rabbit_types:binding_source()) -> bindings().
+-spec remove_for_source(rabbit_types:binding_source() |
+                        rabbit_types:exchange()) ->
+    bindings().
 
+remove_for_source(#exchange{name = SrcName} = SrcX) ->
+    remove_for_source(SrcName, should_index_table(SrcX));
 remove_for_source(SrcName) ->
+    remove_for_source(SrcName, undefined).
+
+-spec remove_for_source(rabbit_types:binding_source(),
+                        boolean() | undefined) ->
+    bindings().
+remove_for_source(SrcName, ShouldIndexTable) ->
     lock_resource(SrcName),
     Match = #route{binding = #binding{source = SrcName, _ = '_'}},
     remove_routes(
       lists:usort(
         mnesia:dirty_match_object(rabbit_route, Match) ++
-            mnesia:dirty_match_object(rabbit_semi_durable_route, Match))).
+            mnesia:dirty_match_object(rabbit_semi_durable_route, Match)),
+      ShouldIndexTable).
 
 -spec remove_for_destination
         (rabbit_types:binding_destination(), boolean()) -> deletions().
@@ -438,20 +449,34 @@ binding_action(Binding = #binding{source      = SrcName,
               Fun(Src, Dst, Binding#binding{args = SortedArgs})
       end, ErrFun).
 
-sync_route(Route, true, true, Fun) ->
+sync_route(Route, true, true, ShouldIndexTable, Fun) ->
     ok = Fun(rabbit_durable_route, Route, write),
-    sync_route(Route, false, true, Fun);
+    sync_route(Route, false, true, ShouldIndexTable, Fun);
 
-sync_route(Route, false, true, Fun) ->
+sync_route(Route, false, true, ShouldIndexTable, Fun) ->
     ok = Fun(rabbit_semi_durable_route, Route, write),
-    sync_route(Route, false, false, Fun);
+    sync_route(Route, false, false, ShouldIndexTable, Fun);
 
-sync_route(Route, _SrcDurable, false, Fun) ->
-    sync_transient_route(Route, Fun).
+sync_route(Route, _SrcDurable, false, ShouldIndexTable, Fun) ->
+    sync_transient_route(Route, ShouldIndexTable, Fun).
 
-sync_transient_route(Route, Fun) ->
+sync_transient_route(Route, ShouldIndexTable, Fun) ->
     ok = Fun(rabbit_route, Route, write),
-    ok = Fun(rabbit_reverse_route, reverse_route(Route), write).
+    ok = Fun(rabbit_reverse_route, reverse_route(Route), write),
+    sync_index_route(Route, ShouldIndexTable, Fun).
+
+sync_index_route(Route, true, Fun) ->
+    %% Do not block as blocking will cause a dead lock when
+    %% function rabbit_binding:populate_index_route_table/0
+    %% (i.e. feature flag migration) runs in parallel.
+    case rabbit_feature_flags:is_enabled(direct_exchange_routing_v2, non_blocking) of
+        true ->
+            ok = Fun(rabbit_index_route, index_route(Route), write);
+        _ ->
+            ok
+    end;
+sync_index_route(_, _, _) ->
+    ok.
 
 call_with_source_and_destination(SrcName, DstName, Fun, ErrFun) ->
     SrcTable = table_for_resource(SrcName),
@@ -498,6 +523,9 @@ continue({[_|_], _})         -> true;
 continue({[], Continuation}) -> continue(mnesia:select(Continuation)).
 
 remove_routes(Routes) ->
+    remove_routes(Routes, undefined).
+
+remove_routes(Routes, ShouldIndexTable) ->
     %% This partitioning allows us to suppress unnecessary delete
     %% operations on disk tables, which require an fsync.
     {RamRoutes, DiskRoutes} =
@@ -511,25 +539,40 @@ remove_routes(Routes) ->
     %% Of course the destination might not really be durable but it's
     %% just as easy to try to delete it from the semi-durable table
     %% than check first
-    [ok = sync_route(R, true,  true, fun delete/3) ||
+    [ok = sync_route(R, true,  true, ShouldIndexTable, fun delete/3) ||
         R <- DiskRoutes],
-    [ok = sync_route(R, false, true, fun delete/3) ||
+    [ok = sync_route(R, false, true, ShouldIndexTable, fun delete/3) ||
         R <- SemiDurableRoutes],
-    [ok = sync_route(R, false, false, fun delete/3) ||
+    [ok = sync_route(R, false, false, ShouldIndexTable, fun delete/3) ||
         R <- RamOnlyRoutes],
+    case ShouldIndexTable of
+        B when is_boolean(B) ->
+            ok;
+        undefined ->
+            [begin
+                 case rabbit_exchange:lookup(Src) of
+                     {ok, X} ->
+                         ok = sync_index_route(R, should_index_table(X), fun delete/3);
+                     _ ->
+                         ok
+                 end
+             end || #route{binding = #binding{source = Src}} = R <- Routes]
+    end,
     [R#route.binding || R <- Routes].
 
-
 delete(Tab, #route{binding = B}, LockKind) ->
     mnesia:delete(Tab, B, LockKind);
 delete(Tab, #reverse_route{reverse_binding = B}, LockKind) ->
-    mnesia:delete(Tab, B, LockKind).
+    mnesia:delete(Tab, B, LockKind);
+delete(Tab, #index_route{} = Record, LockKind) ->
+    mnesia:delete_object(Tab, Record, LockKind).
 
 remove_transient_routes(Routes) ->
-    [begin
-         ok = sync_transient_route(R, fun delete/3),
-         R#route.binding
-     end || R <- Routes].
+    lists:map(fun(#route{binding = #binding{source = Src} = Binding} = Route) ->
+                      {ok, X} = rabbit_exchange:lookup(Src),
+                      ok = sync_transient_route(Route, should_index_table(X), fun delete/3),
+                      Binding
+              end, Routes).
 
 remove_for_destination(DstName, OnlyDurable, Fun) ->
     lock_resource(DstName),
@@ -617,6 +660,14 @@ reverse_binding(#binding{source      = SrcName,
                      key         = Key,
                      args        = Args}.
 
+index_route(#route{binding = #binding{source = Source,
+                                      key = Key,
+                                      destination = Destination,
+                                      args = Args}}) ->
+    #index_route{source_key = {Source, Key},
+                 destination = Destination,
+                 args = Args}.
+
 %% ----------------------------------------------------------------------------
 %% Binding / exchange deletion abstraction API
 %% ----------------------------------------------------------------------------
@@ -687,3 +738,50 @@ del_notify(Bs, ActingUser) -> [rabbit_event:notify(
 
 x_callback(Serial, X, F, Bs) ->
     ok = rabbit_exchange:callback(X, F, Serial, [X, Bs]).
+
+-spec index_route_table_definition() -> list(tuple()).
+index_route_table_definition() ->
+    maps:to_list(
+      #{
+        record_name => index_route,
+        attributes  => record_info(fields, index_route),
+        type => bag,
+        ram_copies => rabbit_nodes:all(),
+        storage_properties => [{ets, [{read_concurrency, true}]}]
+       }).
+
+populate_index_route_table() ->
+    rabbit_misc:execute_mnesia_transaction(
+      fun () ->
+              mnesia:lock({table, rabbit_route}, read),
+              mnesia:lock({table, rabbit_index_route}, write),
+              Routes = rabbit_misc:dirty_read_all(rabbit_route),
+              lists:foreach(fun(#route{binding = #binding{source = Exchange}} = Route) ->
+                                    case rabbit_exchange:lookup(Exchange) of
+                                        {ok, X} ->
+                                            case should_index_table(X) of
+                                                true ->
+                                                    mnesia:dirty_write(rabbit_index_route,
+                                                                       index_route(Route));
+                                                false ->
+                                                    ok
+                                            end;
+                                        _ ->
+                                            ok
+                                    end
+                            end, Routes)
+      end).
+
+%% Only the direct exchange type uses the rabbit_index_route table to store its
+%% bindings by table key tuple {SourceExchange, RoutingKey}.
+%% Other built-in exchange types lookup destinations by SourceExchange, and
+%% therefore will not need to read from the rabbit_index_route index 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 = #resource{name = Name},
+                      type = direct})
+  when Name =/= <<>> ->
+    true;
+should_index_table(_) ->
+    false.
diff --git a/deps/rabbit/src/rabbit_core_ff.erl b/deps/rabbit/src/rabbit_core_ff.erl
index 05d5048d4f..661717c7aa 100644
--- a/deps/rabbit/src/rabbit_core_ff.erl
+++ b/deps/rabbit/src/rabbit_core_ff.erl
@@ -14,7 +14,8 @@
          virtual_host_metadata_migration/3,
          maintenance_mode_status_migration/3,
          user_limits_migration/3,
-         stream_single_active_consumer_migration/3]).
+         stream_single_active_consumer_migration/3,
+         direct_exchange_routing_v2_migration/3]).
 
 -rabbit_feature_flag(
    {classic_mirrored_queue_version,
@@ -78,6 +79,14 @@
       migration_fun => {?MODULE, stream_single_active_consumer_migration}
      }}).
 
+-rabbit_feature_flag(
+   {direct_exchange_routing_v2,
+    #{desc          => "v2 direct exchange routing implementation",
+      stability     => stable,
+      depends_on    => [implicit_default_bindings],
+      migration_fun => {?MODULE, direct_exchange_routing_v2_migration}
+     }}).
+
 classic_mirrored_queue_version_migration(_FeatureName, _FeatureProps, _Enable) ->
     ok.
 
@@ -175,7 +184,7 @@ maintenance_mode_status_migration(FeatureName, _FeatureProps, enable) ->
         _ = rabbit_table:create(
               TableName,
               rabbit_maintenance:status_table_definition()),
-        _ = rabbit_table:ensure_table_copy(TableName, node())
+        _ = rabbit_table:ensure_table_copy(TableName, node(), disc_copies)
     catch throw:Reason  ->
         rabbit_log:error(
           "Failed to create maintenance status table: ~p",
@@ -207,3 +216,46 @@ stream_single_active_consumer_migration(_FeatureName, _FeatureProps, enable) ->
     ok;
 stream_single_active_consumer_migration(_FeatureName, _FeatureProps, is_enabled) ->
     undefined.
+
+%% -------------------------------------------------------------------
+%% Direct exchange routing v2.
+%% -------------------------------------------------------------------
+
+direct_exchange_routing_v2_migration(FeatureName, _FeatureProps, enable) ->
+    TableName = rabbit_index_route,
+    ok = rabbit_table:wait([rabbit_route], _Retry = true),
+    try
+        ok = rabbit_table:create(TableName,
+                                 rabbit_binding:index_route_table_definition()),
+        case rabbit_table:ensure_table_copy(TableName, node(), ram_copies) of
+            ok ->
+                rabbit_binding:populate_index_route_table();
+            {error, Err} = Error ->
+                rabbit_log_feature_flags:error("Failed to add copy of table ~s to node ~p: ~p",
+                                               [TableName, node(), Err]),
+                Error
+        end
+    catch throw:Reason  ->
+              rabbit_log_feature_flags:error("Enabling feature flag ~s failed: ~p",
+                                             [FeatureName, Reason]),
+              {error, Reason}
+    end;
+direct_exchange_routing_v2_migration(_FeatureName, _FeatureProps, is_enabled) ->
+    TableName = rabbit_index_route,
+    Enabled = rabbit_table:exists(TableName),
+    case Enabled of
+        true ->
+            %% Always ensure every node has a table copy.
+            %% If we were not to add a table copy here, `make start-cluster NODES=2`
+            %% would result in only `rabbit-1` having a copy.
+            case rabbit_table:ensure_table_copy(TableName, node(), ram_copies) of
+                ok ->
+                    ok;
+                {error, Reason} ->
+                    rabbit_log_feature_flags:warning("Failed to add copy of table ~s to node ~p: ~p",
+                                                     [TableName, node(), Reason])
+            end;
+        _ ->
+            ok
+    end,
+    Enabled.
diff --git a/deps/rabbit/src/rabbit_exchange.erl b/deps/rabbit/src/rabbit_exchange.erl
index 9ec9675230..b344799389 100644
--- a/deps/rabbit/src/rabbit_exchange.erl
+++ b/deps/rabbit/src/rabbit_exchange.erl
@@ -555,7 +555,7 @@ internal_delete(X = #exchange{name = XName}, OnlyDurable, RemoveBindingsForSourc
     ok = mnesia:delete({rabbit_exchange_serial, XName}),
     mnesia:delete({rabbit_durable_exchange, XName}),
     Bindings = case RemoveBindingsForSource of
-        true  -> rabbit_binding:remove_for_source(XName);
+        true  -> rabbit_binding:remove_for_source(X);
         false -> []
     end,
     {deleted, X, Bindings, rabbit_binding:remove_for_destination(
diff --git a/deps/rabbit/src/rabbit_exchange_type_direct.erl b/deps/rabbit/src/rabbit_exchange_type_direct.erl
index 578ebe85d4..0c4c994d39 100644
--- a/deps/rabbit/src/rabbit_exchange_type_direct.erl
+++ b/deps/rabbit/src/rabbit_exchange_type_direct.erl
@@ -33,7 +33,12 @@ serialise_events() -> false.
 
 route(#exchange{name = Name},
       #delivery{message = #basic_message{routing_keys = Routes}}) ->
-    rabbit_router:match_routing_key(Name, Routes).
+    case rabbit_feature_flags:is_enabled(direct_exchange_routing_v2, non_blocking) of
+        true ->
+            route_v2(Name, Routes);
+        _ ->
+            rabbit_router:match_routing_key(Name, Routes)
+    end.
 
 validate(_X) -> ok.
 validate_binding(_X, _B) -> ok.
@@ -44,3 +49,35 @@ add_binding(_Tx, _X, _B) -> ok.
 remove_bindings(_Tx, _X, _Bs) -> ok.
 assert_args_equivalence(X, Args) ->
     rabbit_exchange:assert_args_equivalence(X, Args).
+
+%% rabbit_router:match_routing_key/2 uses ets:select/2 to get destinations.
+%% ets:select/2 is expensive because it needs to compile the match spec every
+%% time and lookup does not happen by a hash key.
+%%
+%% In contrast, route_v2/2 increases end-to-end message sending throughput
+%% (i.e. from RabbitMQ client to the queue process) by up to 35% by using ets:lookup_element/3.
+%% Only the direct exchange type uses the rabbit_index_route table to store its
+%% bindings by table key tuple {SourceExchange, RoutingKey}.
+-spec route_v2(rabbit_types:binding_source(), [rabbit_router:routing_key(), ...]) ->
+    rabbit_router:match_result().
+route_v2(SrcName, [RoutingKey]) ->
+    %% optimization
+    destinations(SrcName, RoutingKey);
+route_v2(SrcName, [_|_] = RoutingKeys) ->
+    lists:flatmap(fun(Key) ->
+                          destinations(SrcName, Key)
+                  end, RoutingKeys).
+
+destinations(SrcName, RoutingKey) ->
+    %% Prefer try-catch block over checking Key existence with ets:member/2.
+    %% The latter reduces throughput by a few thousand messages per second because
+    %% of function db_member_hash in file erl_db_hash.c.
+    %% We optimise for the happy path, that is the binding / table key is present.
+    try
+        ets:lookup_element(rabbit_index_route,
+                           {SrcName, RoutingKey},
+                           #index_route.destination)
+    catch
+        error:badarg ->
+            []
+    end.
diff --git a/deps/rabbit/src/rabbit_feature_flags.erl b/deps/rabbit/src/rabbit_feature_flags.erl
index a162a5c283..3cffec6a88 100644
--- a/deps/rabbit/src/rabbit_feature_flags.erl
+++ b/deps/rabbit/src/rabbit_feature_flags.erl
@@ -764,7 +764,7 @@ init() ->
 %% replaced by a proper registry.
 %%
 %% Once replaced, the registry contains the map of all supported feature
-%% flags and their state. This is makes it very efficient to query a
+%% flags and their state. This makes it very efficient to query a
 %% feature flag state or property.
 %%
 %% The registry is local to all RabbitMQ nodes.
diff --git a/deps/rabbit/src/rabbit_ff_registry.erl b/deps/rabbit/src/rabbit_ff_registry.erl
index dd33e6a23b..b160de060d 100644
--- a/deps/rabbit/src/rabbit_ff_registry.erl
+++ b/deps/rabbit/src/rabbit_ff_registry.erl
@@ -110,13 +110,13 @@ is_supported(FeatureName) ->
 
 -spec is_enabled(rabbit_feature_flags:feature_name()) -> boolean() | state_changing.
 %% @doc
-%% Returns if a feature flag is supported or if its state is changing.
+%% Returns if a feature flag is enabled or if its state is changing.
 %%
 %% Only the informations stored in the local registry is used to answer
 %% this call.
 %%
 %% @param FeatureName The name of the feature flag to be checked.
-%% @returns `true' if the feature flag is supported, `state_changing' if
+%% @returns `true' if the feature flag is enabled, `state_changing' if
 %%   its state is transient, or `false' otherwise.
 
 is_enabled(FeatureName) ->
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
index ce2f5c46d0..bf3ea74fed 100644
--- a/deps/rabbit/src/rabbit_quorum_queue.erl
+++ b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -583,7 +583,7 @@ recover(_Vhost, Queues) ->
                                           "restarted ~w", [Name, Err]),
                        fail
                end,
-         %% we have to ensure the  quorum queue is
+         %% we have to ensure the quorum queue is
          %% present in the rabbit_queue table and not just in
          %% rabbit_durable_queue
          %% So many code paths are dependent on this.
diff --git a/deps/rabbit/src/rabbit_router.erl b/deps/rabbit/src/rabbit_router.erl
index 19541b5f55..dd352bf5fd 100644
--- a/deps/rabbit/src/rabbit_router.erl
+++ b/deps/rabbit/src/rabbit_router.erl
@@ -32,7 +32,7 @@ match_bindings(SrcName, Match) ->
         Routes, Match(Binding)].
 
 -spec match_routing_key(rabbit_types:binding_source(),
-                             [routing_key()] | ['_']) ->
+                        [routing_key(), ...] | ['_']) ->
     match_result().
 
 match_routing_key(SrcName, [RoutingKey]) ->
diff --git a/deps/rabbit/src/rabbit_table.erl b/deps/rabbit/src/rabbit_table.erl
index 839945cb39..c6a439eb14 100644
--- a/deps/rabbit/src/rabbit_table.erl
+++ b/deps/rabbit/src/rabbit_table.erl
@@ -8,7 +8,7 @@
 -module(rabbit_table).
 
 -export([
-    create/0, create/2, ensure_local_copies/1, ensure_table_copy/2,
+    create/0, create/2, ensure_local_copies/1, ensure_table_copy/3,
     wait_for_replicated/1, wait/1, wait/2,
     force_load/0, is_present/0, is_empty/0, needs_default_data/0,
     check_schema_integrity/1, clear_ram_only_tables/0, retry_timeout/0,
@@ -65,10 +65,11 @@ ensure_secondary_index(Table, Field) ->
     {aborted, {already_exists, Table, _}} -> ok
   end.
 
--spec ensure_table_copy(mnesia_table(), node()) -> ok | {error, any()}.
-ensure_table_copy(TableName, Node) ->
+-spec ensure_table_copy(mnesia_table(), node(), ram_copies | disc_copies) ->
+    ok | {error, any()}.
+ensure_table_copy(TableName, Node, StorageType) ->
     rabbit_log:debug("Will add a local schema database copy for table '~s'", [TableName]),
-    case mnesia:add_table_copy(TableName, Node, disc_copies) of
+    case mnesia:add_table_copy(TableName, Node, StorageType) of
         {atomic, ok}                              -> ok;
         {aborted,{already_exists, TableName}}     -> ok;
         {aborted, {already_exists, TableName, _}} -> ok;
diff --git a/deps/rabbit/test/direct_exchange_routing_v2_SUITE.erl b/deps/rabbit/test/direct_exchange_routing_v2_SUITE.erl
new file mode 100644
index 0000000000..cfdb903ecb
--- /dev/null
+++ b/deps/rabbit/test/direct_exchange_routing_v2_SUITE.erl
@@ -0,0 +1,389 @@
+-module(direct_exchange_routing_v2_SUITE).
+
+%% Test suite for the feature flag direct_exchange_routing_v2
+
+-compile([export_all, nowarn_export_all]).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+-include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
+
+-define(FEATURE_FLAG, direct_exchange_routing_v2).
+-define(INDEX_TABLE_NAME, rabbit_index_route).
+
+%%%===================================================================
+%%% Common Test callbacks
+%%%===================================================================
+
+all() ->
+    [
+     {group, cluster_size_1},
+     {group, cluster_size_2}
+    ].
+
+groups() ->
+    [
+     {cluster_size_1, [],
+      [{start_feature_flag_enabled, [], [remove_binding_unbind_queue,
+                                         remove_binding_delete_queue,
+                                         remove_binding_delete_queue_multiple,
+                                         remove_binding_delete_exchange]},
+       {start_feature_flag_disabled, [], [enable_feature_flag]}
+      ]},
+     {cluster_size_2, [],
+      [{start_feature_flag_enabled, [], [remove_binding_node_down_transient_queue,
+                                         remove_binding_node_down_durable_queue
+                                        ]},
+       {start_feature_flag_disabled, [], [enable_feature_flag]}
+      ]}
+    ].
+
+suite() ->
+    [
+     %% If a test hangs, no need to wait for 30 minutes.
+     {timetrap, {minutes, 8}}
+    ].
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    rabbit_ct_helpers:run_setup_steps(Config, []).
+
+end_per_suite(Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config).
+
+init_per_group(cluster_size_1, Config) ->
+    rabbit_ct_helpers:set_config(Config, {rmq_nodes_count, 1});
+init_per_group(cluster_size_2, Config) ->
+    rabbit_ct_helpers:set_config(Config, {rmq_nodes_count, 2});
+
+init_per_group(start_feature_flag_enabled = Group, Config0) ->
+    Config = start_broker(Group, Config0),
+    case rabbit_ct_broker_helpers:enable_feature_flag(Config, ?FEATURE_FLAG) of
+        ok ->
+            Config;
+        {skip, _} = Skip ->
+            end_per_group(Group, Config),
+            Skip
+    end;
+init_per_group(start_feature_flag_disabled = Group, Config0) ->
+    Config1 = rabbit_ct_helpers:merge_app_env(
+                Config0, {rabbit, [{forced_feature_flags_on_init, []}]}),
+    Config = start_broker(Group, Config1),
+    case rabbit_ct_broker_helpers:is_feature_flag_supported(Config, ?FEATURE_FLAG) of
+        true ->
+            Config;
+        false ->
+            end_per_group(Group, Config),
+            {skip, io_lib:format("'~s' feature flag is unsupported", [?FEATURE_FLAG])}
+    end.
+
+start_broker(Group, Config0) ->
+    Size = rabbit_ct_helpers:get_config(Config0, rmq_nodes_count),
+    Config = rabbit_ct_helpers:set_config(Config0, {rmq_nodename_suffix,
+                                                    io_lib:format("cluster_size_~b-~s", [Size, Group])}),
+    rabbit_ct_helpers:run_steps(Config,
+                                rabbit_ct_broker_helpers:setup_steps() ++
+                                rabbit_ct_client_helpers:setup_steps()).
+
+end_per_group(Group, Config)
+  when Group =:= start_feature_flag_enabled;
+       Group =:= start_feature_flag_disabled ->
+    rabbit_ct_helpers:run_steps(Config,
+                                rabbit_ct_client_helpers:teardown_steps() ++
+                                rabbit_ct_broker_helpers:teardown_steps());
+end_per_group(_Group, Config) ->
+    Config.
+
+init_per_testcase(_TestCase, Config) ->
+    Config.
+
+end_per_testcase(_TestCase, Config) ->
+    %% Test that all bindings got removed from the database.
+    ?assertEqual([0,0,0,0,0],
+                 lists:map(fun(Table) ->
+                                   table_size(Config, Table)
+                           end, [rabbit_durable_route,
+                                 rabbit_semi_durable_route,
+                                 rabbit_route,
+                                 rabbit_reverse_route,
+                                 ?INDEX_TABLE_NAME])
+                ).
+
+%%%===================================================================
+%%% Test cases
+%%%===================================================================
+
+remove_binding_unbind_queue(Config) ->
+    {_Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+
+    amqp_channel:call(Ch, #'confirm.select'{}),
+    amqp_channel:register_confirm_handler(Ch, self()),
+    amqp_channel:register_return_handler(Ch, self()),
+
+    X = <<"amq.direct">>,
+    Q = <<"q">>,
+    RKey = <<"k">>,
+
+    declare_queue(Ch, Q, false),
+    bind_queue(Ch, Q, X, RKey),
+    assert_index_table_non_empty(Config),
+    publish(Ch, X, RKey),
+    assert_confirm(),
+
+    unbind_queue(Ch, Q, X, RKey),
+    assert_index_table_empty(Config),
+    publish(Ch, X, RKey),
+    assert_return(),
+    delete_queue(Ch, Q),
+    ok.
+
+remove_binding_delete_queue(Config) ->
+    {_Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+
+    amqp_channel:call(Ch, #'confirm.select'{}),
+    amqp_channel:register_confirm_handler(Ch, self()),
+    amqp_channel:register_return_handler(Ch, self()),
+
+    X = <<"amq.direct">>,
+    Q = <<"q">>,
+    RKey = <<"k">>,
+
+    declare_queue(Ch, Q, false),
+    bind_queue(Ch, Q, X, RKey),
+    assert_index_table_non_empty(Config),
+    publish(Ch, X, RKey),
+    assert_confirm(),
+
+    delete_queue(Ch, Q),
+    assert_index_table_empty(Config),
+    publish(Ch, X, RKey),
+    assert_return(),
+    ok.
+
+remove_binding_delete_queue_multiple(Config) ->
+    {_Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+
+    amqp_channel:call(Ch, #'confirm.select'{}),
+    amqp_channel:register_confirm_handler(Ch, self()),
+    amqp_channel:register_return_handler(Ch, self()),
+
+    X = <<"amq.direct">>,
+    Q1 = <<"q1">>,
+    Q2 = <<"q2">>,
+    RKey = <<"k">>,
+
+    declare_queue(Ch, Q1, false),
+    bind_queue(Ch, Q1, X, RKey),
+    bind_queue(Ch, Q1, <<"amq.fanout">>, RKey),
+    declare_queue(Ch, Q2, true),
+    bind_queue(Ch, Q2, X, RKey),
+
+    %% Table rabbit_index_route stores only bindings
+    %% where the source exchange is a direct exchange.
+    ?assertEqual(2, table_size(Config, ?INDEX_TABLE_NAME)),
+    publish(Ch, X, RKey),
+    assert_confirm(),
+
+    delete_queue(Ch, Q1),
+    assert_index_table_non_empty(Config),
+    publish(Ch, X, RKey),
+    assert_confirm(),
+
+    delete_queue(Ch, Q2),
+    assert_index_table_empty(Config),
+    publish(Ch, X, RKey),
+    assert_return(),
+    ok.
+
+remove_binding_delete_exchange(Config) ->
+    {_Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+
+    amqp_channel:call(Ch, #'confirm.select'{}),
+    amqp_channel:register_confirm_handler(Ch, self()),
+
+    DirectX = <<"x1">>,
+    FanoutX = <<"x2">>,
+    Q = <<"q">>,
+    RKey = <<"k">>,
+
+    declare_exchange(Ch, DirectX, <<"direct">>),
+    declare_exchange(Ch, FanoutX, <<"fanout">>),
+    declare_queue(Ch, Q, false),
+    bind_queue(Ch, Q, DirectX, RKey),
+    bind_queue(Ch, Q, FanoutX, RKey),
+    assert_index_table_non_empty(Config),
+    publish(Ch, DirectX, RKey),
+    assert_confirm(),
+
+    %% Table rabbit_index_route stores only bindings
+    %% where the source exchange is a direct exchange.
+    delete_exchange(Ch, FanoutX),
+    assert_index_table_non_empty(Config),
+
+    delete_exchange(Ch, DirectX),
+    assert_index_table_empty(Config),
+    delete_queue(Ch, Q),
+    ok.
+
+remove_binding_node_down_transient_queue(Config) ->
+    [_Server1, Server2] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    {_Conn1, Ch1} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+    {_Conn2, Ch2} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 1),
+
+    amqp_channel:call(Ch1, #'confirm.select'{}),
+    amqp_channel:register_confirm_handler(Ch1, self()),
+
+    X = <<"amq.direct">>,
+    Q = <<"q">>,
+    RKey = <<"k">>,
+
+    %% transient classic queue on Server2
+    declare_queue(Ch2, Q, false),
+    bind_queue(Ch1, Q, X, RKey),
+    assert_index_table_non_empty(Config),
+    publish(Ch1, X, RKey),
+    assert_confirm(),
+
+    rabbit_control_helper:command(stop_app, Server2),
+    %% We expect no route to transient classic queue when its host node is down.
+    assert_index_table_empty(Config),
+    rabbit_control_helper:command(start_app, Server2),
+    %% We expect route to NOT come back when transient queue's host node comes back.
+    assert_index_table_empty(Config),
+    delete_queue(Ch1, Q),
+    ok.
+
+remove_binding_node_down_durable_queue(Config) ->
+    [_Server1, Server2] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    {_Conn1, Ch1} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+    {_Conn2, Ch2} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 1),
+
+    amqp_channel:call(Ch1, #'confirm.select'{}),
+    amqp_channel:register_confirm_handler(Ch1, self()),
+
+    X = <<"amq.direct">>,
+    Q = <<"q">>,
+    RKey = <<"k">>,
+
+    %% durable classic queue on Server2
+    declare_queue(Ch2, Q, true),
+    bind_queue(Ch1, Q, X, RKey),
+    assert_index_table_non_empty(Config),
+    publish(Ch1, X, RKey),
+    assert_confirm(),
+
+    rabbit_control_helper:command(stop_app, Server2),
+    %% We expect no route to durable classic queue when its host node is down.
+    assert_index_table_empty(Config),
+    rabbit_control_helper:command(start_app, Server2),
+    %% We expect route to come back when durable queue's host node comes back.
+    assert_index_table_non_empty(Config),
+    delete_queue(Ch1, Q),
+    ok.
+
+enable_feature_flag(Config) ->
+    Nodes = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    {_Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+
+    DirectX = <<"amq.direct">>,
+    Q1 = <<"q1">>,
+    Q2 = <<"q2">>,
+    RKey = <<"k">>,
+
+    declare_queue(Ch, Q1, true),
+    bind_queue(Ch, Q1, DirectX, RKey),
+    bind_queue(Ch, Q1, <<"amq.fanout">>, RKey),
+
+    declare_queue(Ch, Q2, false),
+    bind_queue(Ch, Q2, DirectX, RKey),
+
+    amqp_channel:call(Ch, #'confirm.select'{}),
+    amqp_channel:register_confirm_handler(Ch, self()),
+
+    %% Publishing via "direct exchange routing v1" works.
+    publish(Ch, DirectX, RKey),
+    assert_confirm(),
+
+    %% Before the feature flag is enabled, there should not be an index table.
+    Tids = rabbit_ct_broker_helpers:rpc_all(Config, ets, whereis, [?INDEX_TABLE_NAME]),
+    ?assert(lists:all(fun(Tid) -> Tid =:= undefined end, Tids)),
+
+    ok = rabbit_ct_broker_helpers:enable_feature_flag(Config, ?FEATURE_FLAG),
+
+    %% The feature flag migration should have created an index table with a ram copy on all nodes.
+    ?assertEqual(lists:sort(Nodes),
+                 lists:sort(rabbit_ct_broker_helpers:rpc(Config, 0, mnesia, table_info, [?INDEX_TABLE_NAME, ram_copies]))),
+    %% The feature flag migration should have populated the index table with all bindings whose source exchange
+    %% is a direct exchange.
+    ?assertEqual([{rabbit_misc:r(<<"/">>, exchange, DirectX), RKey}],
+                 rabbit_ct_broker_helpers:rpc(Config, 0, mnesia, dirty_all_keys, [?INDEX_TABLE_NAME])),
+    ?assertEqual(2, rabbit_ct_broker_helpers:rpc(Config, 0, mnesia, table_info, [?INDEX_TABLE_NAME, size])),
+
+    %% Publishing via "direct exchange routing v2" works.
+    publish(Ch, DirectX, RKey),
+    assert_confirm(),
+
+    delete_queue(Ch, Q1),
+    delete_queue(Ch, Q2),
+    ok.
+
+%%%===================================================================
+%%% Helpers
+%%%===================================================================
+
+declare_queue(Ch, QName, Durable) ->
+    #'queue.declare_ok'{message_count = 0} = amqp_channel:call(Ch, #'queue.declare'{queue = QName,
+                                                                                    durable = Durable
+                                                                                   }).
+
+delete_queue(Ch, QName) ->
+    #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}).
+
+bind_queue(Ch, Queue, Exchange, RoutingKey) ->
+    #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{queue       = Queue,
+                                                             exchange    = Exchange,
+                                                             routing_key = RoutingKey}).
+
+unbind_queue(Ch, Queue, Exchange, RoutingKey) ->
+    #'queue.unbind_ok'{} = amqp_channel:call(Ch, #'queue.unbind'{queue       = Queue,
+                                                                 exchange    = Exchange,
+                                                                 routing_key = RoutingKey}).
+
+declare_exchange(Ch, XName, Type) ->
+    #'exchange.declare_ok'{} = amqp_channel:call(Ch, #'exchange.declare'{exchange = XName,
+                                                                         type = Type}).
+
+delete_exchange(Ch, XName) ->
+    #'exchange.delete_ok'{} = amqp_channel:call(Ch, #'exchange.delete'{exchange = XName}).
+
+publish(Ch, Exchange, RoutingKey) ->
+    ok = amqp_channel:call(Ch, #'basic.publish'{exchange = Exchange,
+                                                routing_key = RoutingKey,
+                                                mandatory = true},
+                           #amqp_msg{}).
+
+assert_confirm() ->
+    receive
+        #'basic.ack'{} ->
+            ok
+    after 5000 ->
+              throw(missing_confirm)
+    end.
+
+assert_return() ->
+    receive
+        {#'basic.return'{}, _} ->
+            ok
+    after 5000 ->
+              throw(missing_return)
+    end.
+
+assert_index_table_empty(Config) ->
+    ?awaitMatch(0, table_size(Config, ?INDEX_TABLE_NAME), 3000).
+
+assert_index_table_non_empty(Config) ->
+    ?assertNotEqual(0, table_size(Config, ?INDEX_TABLE_NAME)).
+
+table_size(Config, Table) ->
+    rabbit_ct_broker_helpers:rpc(Config, 0, ets, info, [Table, size], 5000).
diff --git a/deps/rabbit_common/include/rabbit.hrl b/deps/rabbit_common/include/rabbit.hrl
index 60d7961094..625ea1fe24 100644
--- a/deps/rabbit_common/include/rabbit.hrl
+++ b/deps/rabbit_common/include/rabbit.hrl
@@ -93,6 +93,7 @@
 %% mnesia doesn't like unary records, so we add a dummy 'value' field
 -record(route, {binding, value = const}).
 -record(reverse_route, {reverse_binding, value = const}).
+-record(index_route, {source_key, destination, args = []}).
 
 -record(binding, {source, key, destination, args = []}).
 -record(reverse_binding, {destination, key, source, args = []}).