Bugfixes. Fix metadata_store_phase1_SUITE

With the recursive deletion fixed on khepri projections, there were
some minor tweaks to permission handling. The test suite now tests
rabbit_db_* modules directly so it differs from rabbit_vhost:delete
behaviour

24 files changed, 342 insertions, 198 deletions

diff --git a/deps/rabbit/BUILD.bazel b/deps/rabbit/BUILD.bazel
index a46921bb6e..49fa18daae 100644
--- a/deps/rabbit/BUILD.bazel
+++ b/deps/rabbit/BUILD.bazel
@@ -573,6 +573,10 @@ rabbitmq_integration_suite(
     name = "per_vhost_connection_limit_SUITE",
     size = "medium",
     flaky = True,
+    shard_count = 10,
+    additional_beam = [
+        ":cluster_rename_SUITE_beam_files",
+    ],
 )
 
 rabbitmq_integration_suite(
@@ -626,7 +630,7 @@ rabbitmq_integration_suite(
 rabbitmq_integration_suite(
     name = "queue_master_location_SUITE",
     size = "large",
-    shard_count = 2,
+    shard_count = 3,
 )
 
 rabbitmq_integration_suite(
diff --git a/deps/rabbit/src/rabbit_core_ff.erl b/deps/rabbit/src/rabbit_core_ff.erl
index 2cc3fdf0cf..9b4749e51e 100644
--- a/deps/rabbit/src/rabbit_core_ff.erl
+++ b/deps/rabbit/src/rabbit_core_ff.erl
@@ -355,7 +355,7 @@ migrate_tables_to_khepri(FeatureName, TablesAndOwners) ->
                "Feature flag `~s`:   migration from Mnesia to Khepri "
                "finished",
                [FeatureName]),
-            rabbit_khepri:set_ready(),
+            _ = rabbit_khepri:set_ready(),
             ok;
         {'DOWN', MonitorRef, process, Pid, Info} ->
             ?LOG_ERROR(
diff --git a/deps/rabbit/src/rabbit_db.erl b/deps/rabbit/src/rabbit_db.erl
index 47b980a07a..e9e86ac65d 100644
--- a/deps/rabbit/src/rabbit_db.erl
+++ b/deps/rabbit/src/rabbit_db.erl
@@ -124,7 +124,7 @@ recover_mnesia_tables() ->
     %% the feature flag. See rabbit_core_ff:final_sync_from_mnesia_to_khepri/2
     %% Unlock them here as mnesia is still fully functional.
     Tables = [Table || {Table, _} <- rabbit_table:definitions()],
-    [mnesia:change_table_access_mode(Table, read_write) || Table <- Tables],
+    _ = [mnesia:change_table_access_mode(Table, read_write) || Table <- Tables],
     ok.
 
 init_in_khepri() ->
diff --git a/deps/rabbit/src/rabbit_db_exchange.erl b/deps/rabbit/src/rabbit_db_exchange.erl
index 072a9b4cf4..a41c354f2b 100644
--- a/deps/rabbit/src/rabbit_db_exchange.erl
+++ b/deps/rabbit/src/rabbit_db_exchange.erl
@@ -318,11 +318,13 @@ update_in_khepri(XName, Fun) ->
                     ok;
                 {error, {khepri, mismatching_node, _}} ->
                     update_in_khepri(XName, Fun);
+                {error, {khepri, node_not_found, _}} ->
+                    ok;
                 {error, _} = Error ->
                     Error
             end;
         {error, {khepri, node_not_found, _}} ->
-            not_found;
+            ok;
         {error, _} = Error ->
             Error
     end.
@@ -499,7 +501,7 @@ next_serial_in_mnesia_tx(XName) ->
                       #exchange_serial{name = XName, next = Serial + 1}, write),
     Serial.
 
-next_serial_in_khepri(#exchange{name = XName} = X) ->
+next_serial_in_khepri(XName) ->
     %% Just storing the serial number is enough, no need to keep #exchange_serial{}
     Path = khepri_exchange_serial_path(XName),
     Ret1 = rabbit_khepri:adv_get(Path),
@@ -513,7 +515,7 @@ next_serial_in_khepri(#exchange{name = XName} = X) ->
                 ok ->
                     Serial;
                 {error, {khepri, mismatching_node, _}} ->
-                    next_serial_in_khepri(X);
+                    next_serial_in_khepri(XName);
                 Err ->
                     Err
             end;
@@ -612,7 +614,7 @@ delete_in_khepri(XName, IfUnused) ->
                   {ok, X} -> DeletionFun(X, false);
                   _ -> {error, not_found}
               end
-      end).
+      end, rw).
 
 conditional_delete_in_khepri(X = #exchange{name = XName}, OnlyDurable) ->
     case rabbit_db_binding:has_for_source_in_khepri(XName) of
diff --git a/deps/rabbit/src/rabbit_db_msup.erl b/deps/rabbit/src/rabbit_db_msup.erl
index 7e6812f3d3..58fb70ab9e 100644
--- a/deps/rabbit/src/rabbit_db_msup.erl
+++ b/deps/rabbit/src/rabbit_db_msup.erl
@@ -326,9 +326,9 @@ khepri_delete(Path) ->
 mnesia_write_to_khepri(?TABLE, Specs) ->
     rabbit_khepri:transaction(
       fun() ->
-              [khepri_create_tx(khepri_mirrored_supervisor_path(Group, Id),
+              [khepri_create_tx(khepri_mirrored_supervisor_path(Group, SimpleId),
                                 Spec)
-               || #mirrored_sup_childspec{key = {Group, Id}} = Spec <- Specs]
+               || #mirrored_sup_childspec{key = {Group, {SimpleId, _}}} = Spec <- Specs]
       end, rw).
 
 mnesia_delete_to_khepri(?TABLE, #mirrored_sup_childspec{key = {Group, Id}}) ->
diff --git a/deps/rabbit/src/rabbit_db_rtparams.erl b/deps/rabbit/src/rabbit_db_rtparams.erl
index f8c2e07c49..9b8951892b 100644
--- a/deps/rabbit/src/rabbit_db_rtparams.erl
+++ b/deps/rabbit/src/rabbit_db_rtparams.erl
@@ -203,7 +203,7 @@ get_or_set_in_khepri(Key, Default) ->
                   {ok, undefined} ->
                       Record = #runtime_parameters{key   = Key,
                                                    value = Default},
-                      khepri_tx:put(Path, Record),
+                      ok = khepri_tx:put(Path, Record),
                       Record;
                   {ok, R} ->
                       R
@@ -231,7 +231,7 @@ get_all_in_mnesia() ->
     rabbit_mnesia:dirty_read_all(?MNESIA_TABLE).
 
 get_all_in_khepri() ->
-    Path = khepri_vhost_rp_path(?KHEPRI_WILDCARD_STAR, ?KHEPRI_WILDCARD_STAR, ?KHEPRI_WILDCARD_STAR),
+    Path = khepri_rp_path() ++ [rabbit_khepri:if_has_data_wildcard()],
     {ok, Map} = rabbit_khepri:match(Path),
     maps:values(Map).
 
@@ -282,7 +282,9 @@ get_all_in_khepri_tx(VHost, Component) ->
         %% Inside of a transaction, using `rabbit_vhost:exists` will cause
         %% a deadlock and timeout on the transaction, as it uses `rabbit_khepri:exists`.
         %% The `with` function uses the `khepri_tx` API instead
-        _     -> rabbit_db_vhost:with_fun_in_khepri_tx(VHost, fun() -> ok end)
+        _     ->
+            Fun = rabbit_db_vhost:with_fun_in_khepri_tx(VHost, fun() -> ok end),
+            Fun()
     end,
     case khepri_tx:get_many(Path) of
         {ok, Map} ->
diff --git a/deps/rabbit/src/rabbit_db_user.erl b/deps/rabbit/src/rabbit_db_user.erl
index a576d1cf28..06c110be95 100644
--- a/deps/rabbit/src/rabbit_db_user.erl
+++ b/deps/rabbit/src/rabbit_db_user.erl
@@ -447,7 +447,7 @@ set_user_permissions_in_khepri(Username, VHostName, UserPermission) ->
           VHostName,
           fun() ->
                   set_user_permissions_in_khepri_tx(Username, VHostName, UserPermission)
-          end))).
+          end)), rw).
 
 set_user_permissions_in_khepri_tx(Username, VHostName, UserPermission) ->
     Path = khepri_user_permission_path(
@@ -782,7 +782,7 @@ set_topic_permissions_in_khepri(Username, VHostName, TopicPermission) ->
           VHostName,
           fun() ->
                   set_topic_permissions_in_khepri_tx(Username, VHostName, TopicPermission)
-          end))).
+          end)), rw).
 
 set_topic_permissions_in_khepri_tx(Username, VHostName, TopicPermission) ->
     #topic_permission{topic_permission_key =
diff --git a/deps/rabbit/src/rabbit_mirror_queue_misc.erl b/deps/rabbit/src/rabbit_mirror_queue_misc.erl
index 8c617f5961..f09bea31eb 100644
--- a/deps/rabbit/src/rabbit_mirror_queue_misc.erl
+++ b/deps/rabbit/src/rabbit_mirror_queue_misc.erl
@@ -134,7 +134,7 @@ remove_from_queue_in_mnesia(QueueName, Self, DeadGMPids) ->
                               Q1 = amqqueue:set_pid(Q0, QPid1),
                               Q2 = amqqueue:set_slave_pids(Q1, SPids1),
                               Q3 = amqqueue:set_gm_pids(Q2, AliveGM),
-                              store_updated_slaves_in_mnesia(Q3),
+                              _ = store_updated_slaves_in_mnesia(Q3),
                               %% If we add and remove nodes at the
                               %% same time we might tell the old
                               %% master we need to sync and then
@@ -147,7 +147,7 @@ remove_from_queue_in_mnesia(QueueName, Self, DeadGMPids) ->
                               %% [1].
                               Q1 = amqqueue:set_slave_pids(Q0, Alive),
                               Q2 = amqqueue:set_gm_pids(Q1, AliveGM),
-                              store_updated_slaves_in_mnesia(Q2),
+                              _ = store_updated_slaves_in_mnesia(Q2),
                               {ok, QPid1, DeadPids, []}
                       end
               end
@@ -225,7 +225,7 @@ remove_from_queue_in_khepri(QueueName, Self, DeadGMPids) ->
                               Q1 = amqqueue:set_pid(Q0, QPid1),
                               Q2 = amqqueue:set_slave_pids(Q1, SPids1),
                               Q3 = amqqueue:set_gm_pids(Q2, AliveGM),
-                              store_updated_slaves_in_khepri(Q3, Decorators),
+                              _ = store_updated_slaves_in_khepri(Q3, Decorators),
                               %% If we add and remove nodes at the
                               %% same time we might tell the old
                               %% master we need to sync and then
@@ -239,7 +239,7 @@ remove_from_queue_in_khepri(QueueName, Self, DeadGMPids) ->
                               %% [1].
                               Q1 = amqqueue:set_slave_pids(Q0, Alive),
                               Q2 = amqqueue:set_gm_pids(Q1, AliveGM),
-                              store_updated_slaves_in_khepri(Q2, Decorators),
+                              _ = store_updated_slaves_in_khepri(Q2, Decorators),
                               {ok, QPid1, DeadPids, []}
                       end
               end
@@ -265,7 +265,7 @@ on_vhost_up(VHost) ->
                                    on_vhost_up_in_khepri(VHost)
                            end
                 }),
-    [add_mirror(QName, node(), async) || QName <- QNames],
+    _ = [add_mirror(QName, node(), async) || QName <- QNames],
     ok.
 
 on_vhost_up_in_mnesia(VHost) ->
@@ -469,7 +469,7 @@ store_updated_slaves_in_khepri(Q0, Decorators) ->
     %% HA queues are not supported in Khepri. This update is just enough to make
     %% some of the current tests work, which might start some HA queue.
     %% It will be removed before Khepri is released.
-    rabbit_db_queue:update_in_khepri_tx(amqqueue:get_name(Q0), fun(_) -> Q4 end),
+    _ = rabbit_db_queue:update_in_khepri_tx(amqqueue:get_name(Q0), fun(_) -> Q4 end),
     %% Wake it up so that we emit a stats event
     rabbit_amqqueue:notify_policy_changed(Q3),
     Q3.
diff --git a/deps/rabbit/src/rabbit_mirror_queue_slave.erl b/deps/rabbit/src/rabbit_mirror_queue_slave.erl
index d84bf4eec4..d88ef7bc1b 100644
--- a/deps/rabbit/src/rabbit_mirror_queue_slave.erl
+++ b/deps/rabbit/src/rabbit_mirror_queue_slave.erl
@@ -221,9 +221,9 @@ init_it_in_khepri(Self, GM, Node, QName) ->
             PSPids = amqqueue:get_slave_pids_pending_shutdown(Q),
             %% TODO we can't kill processes!
             case [Pid || Pid <- [QPid | SPids], node(Pid) =:= Node] of
-                []     -> stop_pending_slaves(QName, PSPids),
+                []     -> _ = stop_pending_slaves(QName, PSPids),
                           %% TODO make add_slave_in_khepri and add_slave_in_mnesia
-                          add_slave(Q, Self, GM),
+                          _ = add_slave(Q, Self, GM),
                           {new, QPid, GMPids};
                 %% TODO is_process_alive should never go on a khepri transaction!
                 [QPid] -> case rabbit_mnesia:is_process_alive(QPid) of
@@ -236,7 +236,7 @@ init_it_in_khepri(Self, GM, Node, QName) ->
                                        SPids1 = SPids -- [SPid],
                                        Q1 = amqqueue:set_slave_pids(Q, SPids1),
                                        Q2 = amqqueue:set_gm_pids(Q1, GMPids1),
-                                       add_slave(Q2, Self, GM),
+                                       _ = add_slave(Q2, Self, GM),
                                        {new, QPid, GMPids1}
                           end
             end;
diff --git a/deps/rabbit/src/rabbit_vhost_limit.erl b/deps/rabbit/src/rabbit_vhost_limit.erl
index 6f671a154b..e28accf6a2 100644
--- a/deps/rabbit/src/rabbit_vhost_limit.erl
+++ b/deps/rabbit/src/rabbit_vhost_limit.erl
@@ -199,6 +199,5 @@ get_limit(VirtualHost, Limit) ->
                                         N when N < 0  -> undefined;
                                         N when N >= 0 -> {ok, N}
                                     end
-                   end;
-        {error, {timeout, _}} -> throw({error, {cannot_get_limit, VirtualHost, timeout}})
+                   end
     end.
diff --git a/deps/rabbit/src/rabbit_vhost_process.erl b/deps/rabbit/src/rabbit_vhost_process.erl
index 26c495d922..9e6419491d 100644
--- a/deps/rabbit/src/rabbit_vhost_process.erl
+++ b/deps/rabbit/src/rabbit_vhost_process.erl
@@ -73,10 +73,6 @@ handle_info(check_vhost, VHost) ->
                 fun() ->
                     rabbit_vhost_sup_sup:stop_and_delete_vhost(VHost)
                 end),
-            {noreply, VHost};
-        _ ->
-            %% An error just happened, the node could be down.
-            %% There is nothing we can do, just wait until the next check
             {noreply, VHost}
     end;
 handle_info(_, VHost) ->
diff --git a/deps/rabbit/test/metadata_store_phase1_SUITE.erl b/deps/rabbit/test/metadata_store_phase1_SUITE.erl
index 811f0714b3..454b23da56 100644
--- a/deps/rabbit/test/metadata_store_phase1_SUITE.erl
+++ b/deps/rabbit/test/metadata_store_phase1_SUITE.erl
@@ -86,7 +86,6 @@ groups() ->
        write_non_existing_vhost,
        write_existing_vhost,
        check_vhost_exists,
-       get_existing_vhost_info,
        list_vhost_names,
        list_vhost_objects,
        update_non_existing_vhost,
@@ -1577,27 +1576,39 @@ delete_vhost_and_check_resource_access(_) ->
      ?with(?assertEqual(
               true,
               delete_vhost(_With, VHostName))),
-     ?with(?assertNot(
-              check_vhost_access(_With, Username, VHostName))),
-     ?with(?assertNot(
-              check_resource_access(
-                _With, Username, VHostName, "my-resource", configure))),
      ?with(check_storage(
              _With,
              [{mnesia, rabbit_vhost, []},
               {mnesia, rabbit_user, [User]},
-              {mnesia, rabbit_user_permission, []},
+              {mnesia, rabbit_user_permission, [UserPermission]},
               {khepri, [rabbit_db_vhost],
                #{}},
               {khepri, [rabbit_db_user],
                #{?user_path(Username) => User}}]))
     ],
 
+    %% In mnesia the permissions have to be deleted explicitly
+    %% Khepri permissions have a condition to automatically delete them
+    %% when the vhost is deleted
+    MnesiaTests =
+        [?with(?assert(
+                  check_vhost_access(_With, Username, VHostName))),
+         ?with(?assert(
+                  check_resource_access(
+                    _With, Username, VHostName, "my-resource", configure)))],
+
+    KhepriTests =
+        [?with(?assertNot(
+                  check_vhost_access(_With, Username, VHostName))),
+         ?with(?assertNot(
+                  check_resource_access(
+                    _With, Username, VHostName, "my-resource", configure)))],
+
     ?assertEqual(
        ok,
        eunit:test(
-         [{setup, fun force_mnesia_use/0, [{with, mnesia, Tests}]},
-          {setup, fun force_khepri_use/0, [{with, khepri, Tests}]}],
+         [{setup, fun force_mnesia_use/0, [{with, mnesia, Tests ++ MnesiaTests}]},
+          {setup, fun force_khepri_use/0, [{with, khepri, Tests ++ KhepriTests}]}],
          [verbose])).
 
 %%
@@ -2514,29 +2525,45 @@ delete_vhost_and_check_topic_access(_) ->
                 Context#{routing_key => <<"something-else">>}))),
      ?with(?assert(
               delete_vhost(_With, VHostName))),
-     ?with(?assert(
-              check_topic_access(
-                _With, Username, VHostName, Exchange, read, Context))),
-     ?with(?assert(
-              check_topic_access(
-                _With, Username, VHostName, Exchange, read,
-                Context#{routing_key => <<"something-else">>}))),
      ?with(check_storage(
              _With,
              [{mnesia, rabbit_vhost, []},
               {mnesia, rabbit_user, [User]},
-              {mnesia, rabbit_topic_permission, []},
+              {mnesia, rabbit_topic_permission, [TopicPermission]},
               {khepri, [rabbit_db_vhost],
                #{}},
               {khepri, [rabbit_db_user],
                #{?user_path(Username) => User}}]))
     ],
 
+    %% In mnesia the permissions have to be deleted explicitly
+    %% Khepri permissions have a condition to automatically delete them
+    %% when the vhost is deleted
+    MnesiaTests = 
+        [?with(?assert(
+                  check_topic_access(
+                    _With, Username, VHostName, Exchange, read, Context))),
+         ?with(?assertNot(
+                  check_topic_access(
+                    _With, Username, VHostName, Exchange, read,
+                    Context#{routing_key => <<"something-else">>})))],
+
+    KhepriTests = 
+        [?with(?assertEqual(
+                  undefined,
+                  check_topic_access(
+                    _With, Username, VHostName, Exchange, read, Context))),
+         ?with(?assertEqual(
+                  undefined,
+                  check_topic_access(
+                    _With, Username, VHostName, Exchange, read,
+                    Context#{routing_key => <<"something-else">>})))],
+    
     ?assertEqual(
        ok,
        eunit:test(
-         [{setup, fun force_mnesia_use/0, [{with, mnesia, Tests}]},
-          {setup, fun force_khepri_use/0, [{with, khepri, Tests}]}],
+         [{setup, fun force_mnesia_use/0, [{with, mnesia, Tests ++ MnesiaTests}]},
+          {setup, fun force_khepri_use/0, [{with, khepri, Tests ++ KhepriTests}]}],
          [verbose])).
 
 %% -------------------------------------------------------------------
diff --git a/deps/rabbit/test/rabbit_db_binding_SUITE.erl b/deps/rabbit/test/rabbit_db_binding_SUITE.erl
index ed5376e3ea..99beaa7792 100644
--- a/deps/rabbit/test/rabbit_db_binding_SUITE.erl
+++ b/deps/rabbit/test/rabbit_db_binding_SUITE.erl
@@ -85,8 +85,8 @@ create(Config) ->
 create1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = #exchange{name = XName1, durable = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
     ?assertMatch({error, {resources_missing, [_, _]}},
                  rabbit_db_binding:create(Binding, fun(_, _) -> ok end)),
@@ -105,12 +105,14 @@ exists(Config) ->
 exists1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = #exchange{name = XName1, durable = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
-    ?assertEqual(false, rabbit_db_exchange:exists(Binding)),
+    ?assertMatch({error, {resources_missing, [{not_found, _}, {not_found, _}]}},
+                 rabbit_db_binding:exists(Binding)),
     ?assertMatch({new, #exchange{}}, rabbit_db_exchange:create_or_get(Exchange1)),
     ?assertMatch({new, #exchange{}}, rabbit_db_exchange:create_or_get(Exchange2)),
+    ?assertEqual(false, rabbit_db_binding:exists(Binding)),
     ?assertMatch(ok, rabbit_db_binding:create(Binding, fun(_, _) -> ok end)),
     ?assertEqual(true, rabbit_db_binding:exists(Binding)),
     passed.
@@ -121,8 +123,8 @@ delete(Config) ->
 delete1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true, auto_delete = false},
-    Exchange2 = #exchange{name = XName2, durable = true, auto_delete = false},
+    Exchange1 = #exchange{name = XName1, durable = true, auto_delete = false, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, auto_delete = false, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
     ?assertEqual(ok, rabbit_db_binding:delete(Binding, fun(_, _) -> ok end)),
     ?assertMatch({new, #exchange{}}, rabbit_db_exchange:create_or_get(Exchange1)),
@@ -142,8 +144,8 @@ auto_delete(Config) ->
 auto_delete1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true, auto_delete = true},
-    Exchange2 = #exchange{name = XName2, durable = true, auto_delete = false},
+    Exchange1 = #exchange{name = XName1, durable = true, auto_delete = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, auto_delete = false, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
     ?assertEqual(ok, rabbit_db_binding:delete(Binding, fun(_, _) -> ok end)),
     ?assertMatch({new, #exchange{}}, rabbit_db_exchange:create_or_get(Exchange1)),
@@ -163,8 +165,8 @@ get_all(Config) ->
 get_all1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = #exchange{name = XName1, durable = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
     ?assertEqual([], rabbit_db_binding:get_all()),
     ?assertMatch({new, #exchange{}}, rabbit_db_exchange:create_or_get(Exchange1)),
@@ -179,8 +181,8 @@ get_all_by_vhost(Config) ->
 get_all_by_vhost1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = #exchange{name = XName1, durable = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
     ?assertEqual([], rabbit_db_binding:get_all(?VHOST)),
     ?assertMatch({new, #exchange{}}, rabbit_db_exchange:create_or_get(Exchange1)),
@@ -197,8 +199,8 @@ get_all_for_source(Config) ->
 get_all_for_source1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = #exchange{name = XName1, durable = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
     ?assertEqual([], rabbit_db_binding:get_all_for_source(XName1)),
     ?assertEqual([], rabbit_db_binding:get_all_for_source(XName2)),
@@ -216,8 +218,8 @@ get_all_for_destination(Config) ->
 get_all_for_destination1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = #exchange{name = XName1, durable = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
     ?assertEqual([], rabbit_db_binding:get_all_for_destination(XName1)),
     ?assertEqual([], rabbit_db_binding:get_all_for_destination(XName2)),
@@ -235,8 +237,8 @@ get_all_for_source_and_destination(Config) ->
 get_all_for_source_and_destination1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = #exchange{name = XName1, durable = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
     ?assertEqual([], rabbit_db_binding:get_all(XName1, XName2, false)),
     ?assertEqual([], rabbit_db_binding:get_all(XName2, XName1, false)),
@@ -256,8 +258,8 @@ get_all_for_source_and_destination_reverse(Config) ->
 get_all_for_source_and_destination_reverse1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = #exchange{name = XName1, durable = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
     ?assertEqual([], rabbit_db_binding:get_all(XName1, XName2, true)),
     ?assertEqual([], rabbit_db_binding:get_all(XName2, XName1, true)),
@@ -276,8 +278,8 @@ fold(Config) ->
 fold1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = #exchange{name = XName1, durable = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
     ?assertEqual([], rabbit_db_binding:fold(fun(B, Acc) -> [B | Acc] end, [])),
     ?assertMatch({new, #exchange{}}, rabbit_db_exchange:create_or_get(Exchange1)),
@@ -292,8 +294,8 @@ match(Config) ->
 match1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = #exchange{name = XName1, durable = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2,
                        args = #{foo => bar}},
     ?assertEqual([], rabbit_db_binding:match(XName1, fun(#binding{args = Args}) ->
@@ -318,8 +320,8 @@ match_routing_key(Config) ->
 match_routing_key1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = #exchange{name = XName1, durable = true, decorators = {[], []}},
+    Exchange2 = #exchange{name = XName2, durable = true, decorators = {[], []}},
     Binding = #binding{source = XName1, key = <<"*.*">>, destination = XName2,
                        args = #{foo => bar}},
     ?assertEqual([], rabbit_db_binding:match_routing_key(XName1, [<<"a.b.c">>], false)),
diff --git a/deps/rabbit/test/rabbit_db_exchange_SUITE.erl b/deps/rabbit/test/rabbit_db_exchange_SUITE.erl
index 33982e8b34..97b0c1f645 100644
--- a/deps/rabbit/test/rabbit_db_exchange_SUITE.erl
+++ b/deps/rabbit/test/rabbit_db_exchange_SUITE.erl
@@ -92,8 +92,8 @@ create_or_get1(_Config) ->
     XName = rabbit_misc:r(?VHOST, exchange, <<"test-exchange">>),
     Exchange0 = #exchange{name = XName, durable = true},
     Exchange = rabbit_exchange_decorator:set(Exchange0),
-    ?assertMatch({new, Exchange}, rabbit_db_exchange:create_or_get(Exchange0)),
-    ?assertEqual({existing, Exchange}, rabbit_db_exchange:create_or_get(Exchange0)),
+    ?assertMatch({new, Exchange}, rabbit_db_exchange:create_or_get(Exchange)),
+    ?assertEqual({existing, Exchange}, rabbit_db_exchange:create_or_get(Exchange)),
     passed.
 
 get(Config) ->
@@ -104,7 +104,7 @@ get1(_Config) ->
     Exchange0 = #exchange{name = XName, durable = true},
     Exchange = rabbit_exchange_decorator:set(Exchange0),
     ?assertEqual({error, not_found}, rabbit_db_exchange:get(XName)),
-    ?assertEqual({new, Exchange}, rabbit_db_exchange:create_or_get(Exchange0)),
+    ?assertEqual({new, Exchange}, rabbit_db_exchange:create_or_get(Exchange)),
     ?assertEqual({ok, Exchange}, rabbit_db_exchange:get(XName)),
     passed.
 
@@ -116,7 +116,7 @@ get_many1(_Config) ->
     Exchange0 = #exchange{name = XName, durable = true},
     Exchange = rabbit_exchange_decorator:set(Exchange0),
     ?assertEqual([], rabbit_db_exchange:get_many([XName])),
-    ?assertEqual({new, Exchange}, rabbit_db_exchange:create_or_get(Exchange0)),
+    ?assertEqual({new, Exchange}, rabbit_db_exchange:create_or_get(Exchange)),
     ?assertEqual([Exchange], rabbit_db_exchange:get_many([XName])),
     passed.
 
@@ -132,7 +132,7 @@ get_all1(_Config) ->
     Exchange2 = rabbit_exchange_decorator:set(Exchange2_0),
     All = lists:sort([Exchange1, Exchange2]),
     ?assertEqual([], rabbit_db_exchange:get_all()),
-    create([Exchange1_0, Exchange2_0]),
+    create([Exchange1, Exchange2]),
     ?assertEqual(All, lists:sort(rabbit_db_exchange:get_all())),
     passed.
 
@@ -148,7 +148,7 @@ get_all_by_vhost1(_Config) ->
     Exchange2 = rabbit_exchange_decorator:set(Exchange2_0),
     All = lists:sort([Exchange1, Exchange2]),
     ?assertEqual([], rabbit_db_exchange:get_all(?VHOST)),
-    create([Exchange1_0, Exchange2_0]),
+    create([Exchange1, Exchange2]),
     ?assertEqual(All, lists:sort(rabbit_db_exchange:get_all(?VHOST))),
     ?assertEqual([], lists:sort(rabbit_db_exchange:get_all(<<"other-vhost">>))),
     passed.
@@ -216,7 +216,6 @@ set1(_Config) ->
     XName = rabbit_misc:r(?VHOST, exchange, <<"test-exchange">>),
     Exchange = #exchange{name = XName, durable = true},
     ?assertEqual(ok, rabbit_db_exchange:set([Exchange])),
-    ?assertEqual({error, not_found}, rabbit_db_exchange:get(XName)),
     ?assertEqual([Exchange], rabbit_db_exchange:get_all_durable()),
     passed.
 
@@ -275,8 +274,8 @@ delete_if_unused(Config) ->
 delete_if_unused1(_Config) ->
     XName1 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange1">>),
     XName2 = rabbit_misc:r(?VHOST, exchange, <<"test-exchange2">>),
-    Exchange1 = #exchange{name = XName1, durable = true},
-    Exchange2 = #exchange{name = XName2, durable = true},
+    Exchange1 = rabbit_exchange_decorator:set(#exchange{name = XName1, durable = true}),
+    Exchange2 = rabbit_exchange_decorator:set(#exchange{name = XName2, durable = true}),
     Binding = #binding{source = XName1, key = <<"">>, destination = XName2, args = #{}},
     ?assertMatch({error, not_found}, rabbit_db_exchange:delete(XName1, true)),
     create([Exchange1, Exchange2]),
@@ -316,9 +315,8 @@ recover(Config) ->
 
 recover1(_Config) ->
     XName = rabbit_misc:r(?VHOST, exchange, <<"test-exchange">>),
-    Exchange = #exchange{name = XName, durable = true},
+    Exchange = rabbit_exchange_decorator:set(#exchange{name = XName, durable = true}),
     ?assertEqual(ok, rabbit_db_exchange:set([Exchange])),
-    ?assertEqual({error, not_found}, rabbit_db_exchange:get(XName)),
     ?assertEqual([Exchange], rabbit_db_exchange:get_all_durable()),
     ?assertMatch([Exchange], rabbit_db_exchange:recover(?VHOST)),
     ?assertMatch({ok, #exchange{name = XName}}, rabbit_db_exchange:get(XName)),
diff --git a/deps/rabbit/test/rabbit_db_msup_SUITE.erl b/deps/rabbit/test/rabbit_db_msup_SUITE.erl
index 20ca9f1fbf..46e8caf1d4 100644
--- a/deps/rabbit/test/rabbit_db_msup_SUITE.erl
+++ b/deps/rabbit/test/rabbit_db_msup_SUITE.erl
@@ -83,9 +83,12 @@ create_or_update1(_Config) ->
     Overall = spawn(fun() -> ok end),
     Spec = #{id => id, start => {m, f, args}},
     ?assertEqual(start,
-                 rabbit_db_msup:create_or_update(group, Overall, undefined, Spec, id)),
+                 rabbit_db_msup:create_or_update(group, Overall, undefined, Spec, id(id))),
     passed.
 
+id(Id) ->
+    {[Id], Id}.
+
 find_mirror(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, find_mirror1, [Config]).
 
@@ -93,8 +96,8 @@ find_mirror1(_Config) ->
     Overall = spawn(fun() -> ok end),
     Spec = #{id => id, start => {m, f, args}},
     ?assertEqual(start, rabbit_db_msup:create_or_update(group, Overall, undefined,
-                                                        Spec, id)),
-    ?assertEqual({ok, Overall}, rabbit_db_msup:find_mirror(group, id)),
+                                                        Spec, id(id))),
+    ?assertEqual({ok, Overall}, rabbit_db_msup:find_mirror(group, id(id))),
     passed.
 
 delete(Config) ->
@@ -104,9 +107,9 @@ delete1(_Config) ->
     Overall = spawn(fun() -> ok end),
     Spec = #{id => id, start => {m, f, args}},
     ?assertEqual(start, rabbit_db_msup:create_or_update(group, Overall, undefined,
-                                                        Spec, id)),
-    ?assertEqual(ok, rabbit_db_msup:delete(group, id)),
-    ?assertEqual({error, not_found}, rabbit_db_msup:find_mirror(group, id)),
+                                                        Spec, id(id))),
+    ?assertEqual(ok, rabbit_db_msup:delete(group, id(id))),
+    ?assertEqual({error, not_found}, rabbit_db_msup:find_mirror(group, id(id))),
     passed.
 
 delete_all(Config) ->
@@ -116,9 +119,9 @@ delete_all1(_Config) ->
     Overall = spawn(fun() -> ok end),
     Spec = #{id => id, start => {m, f, args}},
     ?assertEqual(start, rabbit_db_msup:create_or_update(group, Overall, undefined,
-                                                        Spec, id)),
+                                                        Spec, id(id))),
     ?assertEqual(ok, rabbit_db_msup:delete_all(group)),
-    ?assertEqual({error, not_found}, rabbit_db_msup:find_mirror(group, id)),
+    ?assertEqual({error, not_found}, rabbit_db_msup:find_mirror(group, id(id))),
     passed.
 
 update_all(Config) ->
@@ -129,8 +132,8 @@ update_all1(_Config) ->
     Overall = spawn(fun() -> ok end),
     Spec = #{id => id, start => {m, f, args}},
     ?assertEqual(start, rabbit_db_msup:create_or_update(group, OldOverall, undefined,
-                                                        Spec, id)),
-    ?assertEqual({ok, OldOverall}, rabbit_db_msup:find_mirror(group, id)),
+                                                        Spec, id(id))),
+    ?assertEqual({ok, OldOverall}, rabbit_db_msup:find_mirror(group, id(id))),
     ?assertEqual([Spec], rabbit_db_msup:update_all(Overall, OldOverall)),
-    ?assertEqual({ok, Overall}, rabbit_db_msup:find_mirror(group, id)),
+    ?assertEqual({ok, Overall}, rabbit_db_msup:find_mirror(group, id(id))),
     passed.
diff --git a/deps/rabbit/test/rabbit_db_topic_exchange_SUITE.erl b/deps/rabbit/test/rabbit_db_topic_exchange_SUITE.erl
index 4f57c200d1..1d61616b21 100644
--- a/deps/rabbit/test/rabbit_db_topic_exchange_SUITE.erl
+++ b/deps/rabbit/test/rabbit_db_topic_exchange_SUITE.erl
@@ -17,11 +17,13 @@
 
 all() ->
     [
-     {group, all_tests}
+     {group, all_tests},
+     {group, mnesia_store}
     ].
 
 groups() ->
     [
+     {mnesia_store, [], mnesia_tests()},
      {all_tests, [], all_tests()},
      {benchmarks, [], benchmarks()}
     ].
@@ -31,7 +33,11 @@ all_tests() ->
      set,
      delete,
      delete_all_for_exchange,
-     match,
+     match
+    ].
+
+mnesia_tests() ->
+    [
      build_key_from_topic_trie_binding_record,
      build_key_from_deletion_events,
      build_key_from_binding_deletion_event,
@@ -50,7 +56,13 @@ init_per_suite(Config) ->
 end_per_suite(Config) ->
     rabbit_ct_helpers:run_teardown_steps(Config).
 
+init_per_group(mnesia_store = Group, Config0) ->
+    Config = rabbit_ct_helpers:set_config(Config0, [{metadata_store, mnesia}]),
+    init_per_group_common(Group, Config);
 init_per_group(Group, Config) ->
+    init_per_group_common(Group, Config).
+
+init_per_group_common(Group, Config) ->
     Config1 = rabbit_ct_helpers:set_config(Config, [
         {rmq_nodename_suffix, Group},
         {rmq_nodes_count, 1}
diff --git a/deps/rabbitmq_consistent_hash_exchange/BUILD.bazel b/deps/rabbitmq_consistent_hash_exchange/BUILD.bazel
index 627bffa5f7..75a46f352c 100644
--- a/deps/rabbitmq_consistent_hash_exchange/BUILD.bazel
+++ b/deps/rabbitmq_consistent_hash_exchange/BUILD.bazel
@@ -80,7 +80,7 @@ broker_for_integration_suites()
 
 rabbitmq_integration_suite(
     name = "rabbit_exchange_type_consistent_hash_SUITE",
-    shard_count = 3,
+    shard_count = 7,
 )
 
 assert_suites()
diff --git a/deps/rabbitmq_mqtt/BUILD.bazel b/deps/rabbitmq_mqtt/BUILD.bazel
index 554db4c181..d76fb6dabe 100644
--- a/deps/rabbitmq_mqtt/BUILD.bazel
+++ b/deps/rabbitmq_mqtt/BUILD.bazel
@@ -198,6 +198,9 @@ rabbitmq_suite(
         "//deps/amqp_client:erlang_app",
         "//deps/rabbit_common:erlang_app",
     ],
+    runtime_deps = [
+        "@meck//:erlang_app",
+    ],
 )
 
 rabbitmq_integration_suite(
diff --git a/deps/rabbitmq_mqtt/test/processor_SUITE.erl b/deps/rabbitmq_mqtt/test/processor_SUITE.erl
index 15ae0dd537..8083ab4dff 100644
--- a/deps/rabbitmq_mqtt/test/processor_SUITE.erl
+++ b/deps/rabbitmq_mqtt/test/processor_SUITE.erl
@@ -43,9 +43,17 @@ init_per_testcase(get_vhost, Config) ->
     mnesia:create_table(rabbit_runtime_parameters, [
         {attributes, record_info(fields, runtime_parameters)},
         {record_name, runtime_parameters}]),
+    meck:new(rabbit_feature_flags, [passthrough, no_link]),
+    meck:expect(
+      rabbit_feature_flags, is_enabled,
+      fun
+          (raft_based_metadata_store_phase1, _) -> false;
+          (FeatureNames, _)           -> meck:passthrough([FeatureNames])
+      end),
     Config;
 init_per_testcase(_, Config) -> Config.
 end_per_testcase(get_vhost, Config) ->
+    meck:unload(rabbit_feature_flags),
     mnesia:stop(),
     Config;
 end_per_testcase(_, Config) -> Config.
diff --git a/deps/rabbitmq_mqtt/test/shared_SUITE.erl b/deps/rabbitmq_mqtt/test/shared_SUITE.erl
index de94bdee7a..aacd409fe6 100644
--- a/deps/rabbitmq_mqtt/test/shared_SUITE.erl
+++ b/deps/rabbitmq_mqtt/test/shared_SUITE.erl
@@ -101,15 +101,20 @@ subgroups() ->
      {cluster_size_3, [],
       [
        queue_down_qos1,
-       consuming_classic_mirrored_queue_down,
        consuming_classic_queue_down,
-       flow_classic_mirrored_queue,
        flow_quorum_queue,
        flow_stream,
        rabbit_mqtt_qos0_queue,
        cli_list_queues,
        maintenance,
        delete_create_queue,
+       publish_to_all_non_deprecated_queue_types_qos0,
+       publish_to_all_non_deprecated_queue_types_qos1
+      ]},
+     {mnesia_store, [],
+      [
+       consuming_classic_mirrored_queue_down,
+       flow_classic_mirrored_queue,
        publish_to_all_queue_types_qos0,
        publish_to_all_queue_types_qos1,
        duplicate_client_id
@@ -140,6 +145,11 @@ init_per_group(cluster_size_1, Config) ->
 init_per_group(cluster_size_3 = Group, Config) ->
     init_per_group0(Group,
                     rabbit_ct_helpers:set_config(Config, [{rmq_nodes_count, 3}]));
+init_per_group(mnesia_store = Group, Config) ->
+    init_per_group0(
+      Group,
+      rabbit_ct_helpers:set_config(Config, [{metadata_store, mnesia},
+                                            {rmq_nodes_count, 3}]));
 init_per_group(Group, Config)
   when Group =:= global_counters orelse
        Group =:= tests ->
@@ -371,6 +381,65 @@ publish_to_all_queue_types(Config, QoS) ->
     ?awaitMatch([],
                 all_connection_pids(Config), 10_000, 1000).
 
+publish_to_all_non_deprecated_queue_types_qos0(Config) ->
+    publish_to_all_non_deprecated_queue_types(Config, qos0).
+
+publish_to_all_non_deprecated_queue_types_qos1(Config) ->
+    publish_to_all_non_deprecated_queue_types(Config, qos1).
+
+publish_to_all_non_deprecated_queue_types(Config, QoS) ->
+    Ch = rabbit_ct_client_helpers:open_channel(Config),
+
+    CQ = <<"classic-queue">>,
+    QQ = <<"quorum-queue">>,
+    SQ = <<"stream-queue">>,
+    Topic = <<"mytopic">>,
+
+    declare_queue(Ch, CQ, []),
+    bind(Ch, CQ, Topic),
+
+    declare_queue(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}]),
+    bind(Ch, QQ, Topic),
+
+    declare_queue(Ch, SQ, [{<<"x-queue-type">>, longstr, <<"stream">>}]),
+    bind(Ch, SQ, Topic),
+
+    NumMsgs = 2000,
+    C = connect(?FUNCTION_NAME, Config, [{retry_interval, 2}]),
+    lists:foreach(fun(N) ->
+                          case emqtt:publish(C, Topic, integer_to_binary(N), QoS) of
+                              ok ->
+                                  ok;
+                              {ok, _} ->
+                                  ok;
+                              Other ->
+                                  ct:fail("Failed to publish: ~p", [Other])
+                          end
+                  end, lists:seq(1, NumMsgs)),
+
+    eventually(?_assert(
+                  begin
+                      L = rabbitmqctl_list(Config, 0, ["list_queues", "messages", "--no-table-headers"]),
+                      length(L) =:= 3 andalso
+                      lists:all(fun([Bin]) ->
+                                        N = binary_to_integer(Bin),
+                                        case QoS of
+                                            qos0 ->
+                                                N =:= NumMsgs;
+                                            qos1 ->
+                                                %% Allow for some duplicates when client resends
+                                                %% a message that gets acked at roughly the same time.
+                                                N >= NumMsgs andalso
+                                                N < NumMsgs * 2
+                                        end
+                                end, L)
+                  end), 2000, 10),
+
+    delete_queue(Ch, [CQ, QQ, SQ]),
+    ok = emqtt:disconnect(C),
+    ?awaitMatch([],
+                all_connection_pids(Config), 10_000, 1000).
+
 flow_classic_mirrored_queue(Config) ->
     QueueName = <<"flow">>,
     ok = rabbit_ct_broker_helpers:set_ha_policy(Config, 0, QueueName, <<"all">>),
@@ -1272,24 +1341,15 @@ clean_session_node_down(NodeDown, Config) ->
             ?assertEqual(0, length(QsQos0)),
             ?assertEqual(2, length(QsClassic))
     end,
-    Tables = [rabbit_durable_queue,
-              rabbit_queue,
-              rabbit_durable_route,
-              rabbit_semi_durable_route,
-              rabbit_route,
-              rabbit_reverse_route,
-              rabbit_topic_trie_node,
-              rabbit_topic_trie_edge,
-              rabbit_topic_trie_binding],
-    [?assertNotEqual(0, rpc(Config, ets, info, [T, size])) || T <- Tables],
+    ?assertEqual(2, rpc(Config, rabbit_amqqueue, count, [])),
 
     unlink(C),
     ok = rabbit_ct_broker_helpers:NodeDown(Config, 0),
     ok = rabbit_ct_broker_helpers:start_node(Config, 0),
 
-    %% After terminating a clean session by either node crash or graceful node shutdown, we
-    %% expect any session state to be cleaned up on the server once the server finished booting.
-    [?assertEqual(0, rpc(Config, ets, info, [T, size])) || T <- Tables].
+    %% After terminating a clean session by a node crash, we expect any session
+    %% state to be cleaned up on the server once the server comes back up.
+    ?assertEqual(0, rpc(Config, rabbit_amqqueue, count, [])).
 
 rabbit_status_connection_count(Config) ->
     _Pid = rabbit_ct_client_helpers:open_connection(Config, 0),
diff --git a/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl b/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl
index 76f0b141ea..28e3813202 100644
--- a/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl
+++ b/deps/rabbitmq_recent_history_exchange/src/rabbit_db_rh_exchange.erl
@@ -49,11 +49,11 @@ setup_schema() ->
        }).
 
 setup_schema_in_mnesia() ->
-    mnesia:create_table(?RH_TABLE,
-                        [{attributes, record_info(fields, cached)},
-                         {record_name, cached},
-                         {type, set}]),
-    mnesia:add_table_copy(?RH_TABLE, node(), ram_copies),
+    _ = mnesia:create_table(?RH_TABLE,
+                            [{attributes, record_info(fields, cached)},
+                             {record_name, cached},
+                             {type, set}]),
+    _ = mnesia:add_table_copy(?RH_TABLE, node(), ram_copies),
     rabbit_table:wait([?RH_TABLE]),
     ok.
 
diff --git a/deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl b/deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl
index daa2c59521..c0b58a5323 100644
--- a/deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl
+++ b/deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl
@@ -34,6 +34,14 @@
          forward/4
         ]).
 
+%% Function references should not be stored on the metadata store.
+%% They are only valid for the version of the module they were created
+%% from and can break with the next upgrade. It should not be used by
+%% another one that the one who created it or survive a node restart.
+%% Thus, function references have been replace by the following MFA.
+-export([decl_fun/3, publish_fun/4, props_fun_timestamp_header/4,
+         props_fun_forward_header/5]).
+
 -define(MAX_CONNECTION_CLOSE_TIMEOUT, 10000).
 
 parse(_Name, {source, Source}) ->
@@ -77,9 +85,9 @@ init_source(Conf = #{ack_mode := AckMode,
                      source := #{queue := Queue,
                                  current := {Conn, Chan, _},
                                  prefetch_count := Prefetch,
-                                 resource_decl := Decl,
+                                 resource_decl := {M, F, MFArgs},
                                  consumer_args := Args} = Src}) ->
-    Decl(Conn, Chan),
+    apply(M, F, MFArgs ++ [Conn, Chan]),
 
     NoAck = AckMode =:= no_ack,
     case NoAck of
@@ -108,9 +116,9 @@ connect_dest(Conf = #{name := Name, dest := #{uris := Uris} = Dst}) ->
 
 init_dest(Conf = #{ack_mode := AckMode,
                    dest := #{current := {Conn, Chan, _},
-                             resource_decl := Decl} = Dst}) ->
+                             resource_decl := {M, F, MFArgs}} = Dst}) ->
 
-    Decl(Conn, Chan),
+    apply(M, F, MFArgs ++ [Conn, Chan]),
 
     case AckMode of
         on_confirm ->
@@ -187,16 +195,16 @@ forward(IncomingTag, Props, Payload, State) ->
     end.
 
 do_forward(IncomingTag, Props, Payload,
-           State0 = #{dest := #{props_fun := PropsFun,
+           State0 = #{dest := #{props_fun := {M, F, Args},
                                 current := {_, _, DstUri},
-                                fields_fun := FieldsFun}}) ->
+                                fields_fun := {Mf, Ff, Argsf}}}) ->
     SrcUri = rabbit_shovel_behaviour:source_uri(State0),
     % do publish
     Exchange = maps:get(exchange, Props, undefined),
     RoutingKey = maps:get(routing_key, Props, undefined),
     Method = #'basic.publish'{exchange = Exchange, routing_key = RoutingKey},
-    Method1 = FieldsFun(SrcUri, DstUri, Method),
-    Msg1 = #amqp_msg{props = PropsFun(SrcUri, DstUri, props_from_map(Props)),
+    Method1 = apply(Mf, Ff, Argsf ++ [SrcUri, DstUri, Method]),
+    Msg1 = #amqp_msg{props = apply(M, F, Args ++ [SrcUri, DstUri, props_from_map(Props)]),
                      payload = Payload},
     publish(IncomingTag, Method1, Msg1, State0).
 
@@ -519,11 +527,7 @@ make_publish_fun(Fields, ValidFields) when is_list(Fields) ->
     case SuppliedFields -- ValidFields of
         [] ->
             FieldIndices = make_field_indices(ValidFields, Fields),
-            fun (_SrcUri, _DestUri, Publish) ->
-                    lists:foldl(fun ({Pos1, Value}, Pub) ->
-                                        setelement(Pos1, Pub, Value)
-                                end, Publish, FieldIndices)
-            end;
+            {?MODULE, publish_fun, [FieldIndices]};
         Unexpected ->
             fail({invalid_parameter_value, publish_properties,
                   {unexpected_fields, Unexpected, ValidFields}})
@@ -532,6 +536,11 @@ make_publish_fun(Fields, _) ->
     fail({invalid_parameter_value, publish_properties,
           {require_list, Fields}}).
 
+publish_fun(FieldIndices, _SrcUri, _DestUri, Publish) ->
+    lists:foldl(fun ({Pos1, Value}, Pub) ->
+                        setelement(Pos1, Pub, Value)
+                end, Publish, FieldIndices).
+
 make_field_indices(Valid, Fields) ->
     make_field_indices(Fields, field_map(Valid, 2), []).
 
@@ -551,23 +560,25 @@ field_map(Fields, Idx0) ->
 fail(Reason) -> throw({error, Reason}).
 
 add_forward_headers_fun(Name, true, PubProps) ->
-    fun(SrcUri, DestUri, Props) ->
-            rabbit_shovel_util:update_headers(
-              [{<<"shovelled-by">>, rabbit_nodes:cluster_name()},
-               {<<"shovel-type">>,  <<"static">>},
-               {<<"shovel-name">>,  list_to_binary(atom_to_list(Name))}],
-              [], SrcUri, DestUri, PubProps(SrcUri, DestUri, Props))
-    end;
+   {?MODULE, props_fun_forward_header, [Name, PubProps]};
 add_forward_headers_fun(_Name, false, PubProps) ->
     PubProps.
 
+props_fun_forward_header(Name, {M, F, Args}, SrcUri, DestUri, Props) ->
+    rabbit_shovel_util:update_headers(
+      [{<<"shovelled-by">>, rabbit_nodes:cluster_name()},
+       {<<"shovel-type">>,  <<"static">>},
+       {<<"shovel-name">>,  list_to_binary(atom_to_list(Name))}],
+      [], SrcUri, DestUri, apply(M, F, Args ++ [SrcUri, DestUri, Props])).
+
 add_timestamp_header_fun(true, PubProps) ->
-    fun(SrcUri, DestUri, Props) ->
-        rabbit_shovel_util:add_timestamp_header(
-            PubProps(SrcUri, DestUri, Props))
-    end;
+    {?MODULE, props_fun_timestamp_header, [PubProps]};
 add_timestamp_header_fun(false, PubProps) -> PubProps.
 
+props_fun_timestamp_header({M, F, Args}, SrcUri, DestUri, Props) ->
+    rabbit_shovel_util:add_timestamp_header(
+      apply(M, F, Args ++ [SrcUri, DestUri, Props])).
+
 parse_declaration({[], Acc}) ->
     Acc;
 parse_declaration({[{Method, Props} | Rest], Acc}) when is_list(Props) ->
@@ -596,11 +607,12 @@ parse_declaration({[Method | Rest], Acc}) ->
 decl_fun(Endpoint) ->
     Decl = parse_declaration({proplists:get_value(declarations, Endpoint, []),
                               []}),
-    fun (_Conn, Ch) ->
-            [begin
-                 amqp_channel:call(Ch, M)
-             end || M <- lists:reverse(Decl)]
-    end.
+    {?MODULE, decl_fun, [Decl]}.
+
+decl_fun(Decl, _Conn, Ch) ->
+    [begin
+         amqp_channel:call(Ch, M)
+     end || M <- lists:reverse(Decl)].
 
 parse_parameter(Param, Fun, Value) ->
     try
diff --git a/deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl b/deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl
index b6682c25b8..df7df4f375 100644
--- a/deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl
+++ b/deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl
@@ -15,6 +15,14 @@
 -export([register/0, unregister/0, parse/3]).
 -export([obfuscate_uris_in_definition/1]).
 
+%% Function references should not be stored on the metadata store.
+%% They are only valid for the version of the module they were created
+%% from and can break with the next upgrade. It should not be used by
+%% another one that the one who created it or survive a node restart.
+%% Thus, function references have been replace by the following MFA.
+-export([dest_decl/4, src_decl_exchange/4, src_decl_queue/4,
+         fields_fun/5, props_fun/9]).
+
 -import(rabbit_misc, [pget/2, pget/3, pset/3]).
 
 -rabbit_boot_step({?MODULE,
@@ -321,12 +329,7 @@ parse_amqp091_dest({VHost, Name}, ClusterName, Def, SourceHeaders) ->
     DestXKey  = pget(<<"dest-exchange-key">>, Def, none),
     DestQ     = pget(<<"dest-queue">>,        Def, none),
     DestQArgs = pget(<<"dest-queue-args">>,   Def, #{}),
-    DestDeclFun = fun (Conn, _Ch) ->
-                      case DestQ of
-                          none -> ok;
-                          _ -> ensure_queue(Conn, DestQ, rabbit_misc:to_amqp_table(DestQArgs))
-                      end
-              end,
+    DestDeclFun = {?MODULE, dest_decl, [DestQ, DestQArgs]},
     {X, Key} = case DestQ of
                    none -> {DestX, DestXKey};
                    _    -> {<<>>,  DestQ}
@@ -335,16 +338,6 @@ parse_amqp091_dest({VHost, Name}, ClusterName, Def, SourceHeaders) ->
                                    {<<"dest-exchange-key">>, DestXKey},
                                    {<<"dest-queue">>,        DestQ}],
                         V =/= none],
-    PubFun = fun (_SrcURI, _DestURI, P0) ->
-                     P1 = case X of
-                              none -> P0;
-                              _    -> P0#'basic.publish'{exchange = X}
-                          end,
-                     case Key of
-                         none -> P1;
-                         _    -> P1#'basic.publish'{routing_key = Key}
-                     end
-             end,
     AddHeadersLegacy = pget(<<"add-forward-headers">>, Def, false),
     AddHeaders = pget(<<"dest-add-forward-headers">>, Def, AddHeadersLegacy),
     Table0 = [{<<"shovelled-by">>, ClusterName},
@@ -357,19 +350,6 @@ parse_amqp091_dest({VHost, Name}, ClusterName, Def, SourceHeaders) ->
     AddTimestampHeaderLegacy = pget(<<"add-timestamp-header">>, Def, false),
     AddTimestampHeader = pget(<<"dest-add-timestamp-header">>, Def,
                               AddTimestampHeaderLegacy),
-    PubPropsFun = fun (SrcURI, DestURI, P0) ->
-                      P  = set_properties(P0, SetProps),
-                      P1 = case AddHeaders of
-                               true -> rabbit_shovel_util:update_headers(
-                                          Table0, SourceHeaders ++ Table2,
-                                          SrcURI, DestURI, P);
-                               false -> P
-                      end,
-                      case AddTimestampHeader of
-                          true  -> rabbit_shovel_util:add_timestamp_header(P1);
-                          false -> P1
-                      end
-                  end,
     %% Details are only used for status report in rabbitmqctl, as vhost is not
     %% available to query the runtime parameters.
     Details = maps:from_list([{K, V} || {K, V} <- [{dest_exchange, DestX},
@@ -379,10 +359,42 @@ parse_amqp091_dest({VHost, Name}, ClusterName, Def, SourceHeaders) ->
     maps:merge(#{module => rabbit_amqp091_shovel,
                  uris => DestURIs,
                  resource_decl => DestDeclFun,
-                 fields_fun => PubFun,
-                 props_fun => PubPropsFun
+                 fields_fun => {?MODULE, fields_fun, [X, Key]},
+                 props_fun => {?MODULE, props_fun, [Table0, Table2, SetProps,
+                                                    AddHeaders, SourceHeaders,
+                                                    AddTimestampHeader]}
                 }, Details).
 
+fields_fun(X, Key, _SrcURI, _DestURI, P0) ->
+    P1 = case X of
+             none -> P0;
+             _    -> P0#'basic.publish'{exchange = X}
+         end,
+    case Key of
+        none -> P1;
+        _    -> P1#'basic.publish'{routing_key = Key}
+    end.
+
+props_fun(Table0, Table2, SetProps, AddHeaders, SourceHeaders, AddTimestampHeader,
+          SrcURI, DestURI, P0) ->
+    P  = set_properties(P0, SetProps),
+    P1 = case AddHeaders of
+             true -> rabbit_shovel_util:update_headers(
+                       Table0, SourceHeaders ++ Table2,
+                       SrcURI, DestURI, P);
+             false -> P
+         end,
+    case AddTimestampHeader of
+        true  -> rabbit_shovel_util:add_timestamp_header(P1);
+        false -> P1
+    end.
+
+dest_decl(DestQ, DestQArgs, Conn, _Ch) ->
+    case DestQ of
+        none -> ok;
+        _ -> ensure_queue(Conn, DestQ, rabbit_misc:to_amqp_table(DestQArgs))
+    end.
+
 parse_amqp10_source(Def) ->
     Uris = deobfuscated_uris(<<"src-uri">>, Def),
     Address = pget(<<"src-address">>, Def),
@@ -405,16 +417,11 @@ parse_amqp091_source(Def) ->
     SrcCArgs = rabbit_misc:to_amqp_table(pget(<<"src-consumer-args">>, Def, [])),
     {SrcDeclFun, Queue, DestHeaders} =
     case SrcQ of
-        none -> {fun (_Conn, Ch) ->
-                         Ms = [#'queue.declare'{exclusive = true},
-                               #'queue.bind'{routing_key = SrcXKey,
-                                             exchange    = SrcX}],
-                         [amqp_channel:call(Ch, M) || M <- Ms]
-                 end, <<>>, [{<<"src-exchange">>,     SrcX},
-                             {<<"src-exchange-key">>, SrcXKey}]};
-        _ -> {fun (Conn, _Ch) ->
-                      ensure_queue(Conn, SrcQ, rabbit_misc:to_amqp_table(SrcQArgs))
-              end, SrcQ, [{<<"src-queue">>, SrcQ}]}
+        none -> {{?MODULE, src_decl_exchange, [SrcX, SrcXKey]}, <<>>,
+                 [{<<"src-exchange">>,     SrcX},
+                  {<<"src-exchange-key">>, SrcXKey}]};
+        _ -> {{?MODULE, src_decl_queue, [SrcQ, SrcQArgs]},
+              SrcQ, [{<<"src-queue">>, SrcQ}]}
     end,
     DeleteAfter = pget(<<"src-delete-after">>, Def,
                        pget(<<"delete-after">>, Def, <<"never">>)),
@@ -434,6 +441,15 @@ parse_amqp091_source(Def) ->
                   consumer_args => SrcCArgs
                  }, Details), DestHeaders}.
 
+src_decl_exchange(SrcX, SrcXKey, _Conn, Ch) ->
+    Ms = [#'queue.declare'{exclusive = true},
+          #'queue.bind'{routing_key = SrcXKey,
+                        exchange    = SrcX}],
+    [amqp_channel:call(Ch, M) || M <- Ms].
+
+src_decl_queue(SrcQ, SrcQArgs, Conn, _Ch) ->
+    ensure_queue(Conn, SrcQ, rabbit_misc:to_amqp_table(SrcQArgs)).
+
 get_uris(Key, Def) ->
     URIs = case pget(Key, Def) of
                B when is_binary(B) -> [B];
diff --git a/deps/rabbitmq_shovel/test/parameters_SUITE.erl b/deps/rabbitmq_shovel/test/parameters_SUITE.erl
index 837faea7bc..913ee24cec 100644
--- a/deps/rabbitmq_shovel/test/parameters_SUITE.erl
+++ b/deps/rabbitmq_shovel/test/parameters_SUITE.erl
@@ -159,7 +159,7 @@ test_parse_amqp091(Params) ->
       reconnect_delay := 1001,
       dest := #{module := rabbit_amqp091_shovel,
                 uris := ["amqp://remotehost:5672"],
-                props_fun := PropsFun
+                props_fun := {M, F, Args}
                },
       source := #{module := rabbit_amqp091_shovel,
                   uris := ["amqp://localhost:5672"],
@@ -170,9 +170,9 @@ test_parse_amqp091(Params) ->
 
     #'P_basic'{headers = ActualHeaders,
                delivery_mode = 2,
-               cluster_id = <<"x">>} = PropsFun("amqp://localhost:5672",
-                                                "amqp://remotehost:5672",
-                                                #'P_basic'{headers = undefined}),
+               cluster_id = <<"x">>} = apply(M, F, Args ++ ["amqp://localhost:5672",
+                                                            "amqp://remotehost:5672",
+                                                            #'P_basic'{headers = undefined}]),
     assert_amqp901_headers(ActualHeaders),
     ok.
 
@@ -185,7 +185,7 @@ test_parse_amqp091_with_blank_proprties(Params) ->
       reconnect_delay := 1001,
       dest := #{module := rabbit_amqp091_shovel,
                 uris := ["amqp://remotehost:5672"],
-                props_fun := PropsFun
+                props_fun := {M, F, Args}
                },
       source := #{module := rabbit_amqp091_shovel,
                   uris := ["amqp://localhost:5672"],
@@ -194,9 +194,9 @@ test_parse_amqp091_with_blank_proprties(Params) ->
                   delete_after := 'queue-length'}
      } = Result,
 
-    #'P_basic'{headers = ActualHeaders} = PropsFun("amqp://localhost:5672",
-                                                   "amqp://remotehost:5672",
-                                                   #'P_basic'{headers = undefined}),
+    #'P_basic'{headers = ActualHeaders} = apply(M, F, Args ++ ["amqp://localhost:5672",
+                                                               "amqp://remotehost:5672",
+                                                               #'P_basic'{headers = undefined}]),
     assert_amqp901_headers(ActualHeaders),
     ok.