rabbitmq_ct_helpers: Change how Mnesia/Khepri is selected

[Why]
Once `khepr_db` is enabled by default, we need another way to disable it
to select Mnesia instead.

[How]
We use the new relative forced feature flags mechanism to indicate if we
want to explicitly enable or disable `khepri_db`. This way, we don't
touch other stable feature flags and only mess with Khepri.

However, this mechanism is not supported by RabbitMQ 4.0.x and older.
They will ignore the setting. Therefore, to make this work in
mixed-version testing, after a node has been started, we try to enable
`khepri_db` if we detect that it doesn't support this mechanism.

At the end, we compare the effective metadata store to the expected one.
If they don't match, we skip the test.

While here, change `rjms_topic_selector_SUITE` to only choose Khepri
without specifying any feature flags.

deps/rabbit/test/bindings_SUITE.erl

@@ -72,7 +72,7 @@ end_per_suite(Config) ->
 %     init_per_group_common(Group, Config, 1);
 init_per_group(khepri_migration = Group, Config) ->
     case rabbit_ct_broker_helpers:configured_metadata_store(Config) of
-        {khepri, _} ->
+        khepri ->
             {skip, "skip khepri migration test when khepri already configured"};
         mnesia ->
             init_per_group_common(Group, Config, 1)

deps/rabbit/test/clustering_management_SUITE.erl

@@ -138,7 +138,7 @@ init_per_group(khepri_store, Config) ->
     end;
 init_per_group(mnesia_store, Config) ->
     case rabbit_ct_broker_helpers:configured_metadata_store(Config) of
-        {khepri, _} ->
+        khepri ->
             {skip, "These tests target mnesia"};
         _ ->
             Config

deps/rabbit/test/clustering_recovery_SUITE.erl

@@ -80,7 +80,7 @@ init_per_group(khepri_store, Config) ->
     end;
 init_per_group(mnesia_store, Config) ->
     case rabbit_ct_broker_helpers:configured_metadata_store(Config) of
-        {khepri, _} ->
+        khepri ->
             {skip, "These tests target mnesia"};
         _ ->
             Config

deps/rabbitmq_ct_helpers/src/rabbit_ct_broker_helpers.erl

@@ -215,19 +215,19 @@ setup_steps() ->
                 fun rabbit_ct_helpers:ensure_rabbitmqctl_app/1,
                 fun rabbit_ct_helpers:ensure_rabbitmq_plugins_cmd/1,
                 fun set_lager_flood_limit/1,
+                fun configure_metadata_store/1,
                 fun start_rabbitmq_nodes/1,
-                fun share_dist_and_proxy_ports_map/1,
-                fun configure_metadata_store/1
+                fun share_dist_and_proxy_ports_map/1
             ];
         _ ->
             [
                 fun rabbit_ct_helpers:ensure_rabbitmqctl_cmd/1,
                 fun rabbit_ct_helpers:load_rabbitmqctl_app/1,
                 fun rabbit_ct_helpers:ensure_rabbitmq_plugins_cmd/1,
                 fun set_lager_flood_limit/1,
+                fun configure_metadata_store/1,
                 fun start_rabbitmq_nodes/1,
-                fun share_dist_and_proxy_ports_map/1,
-                fun configure_metadata_store/1
+                fun share_dist_and_proxy_ports_map/1
             ]
     end.
 
@@ -441,8 +441,24 @@ start_rabbitmq_node(Master, Config, NodeConfig, I) ->
               {failed_boot_attempts, Attempts + 1}),
             start_rabbitmq_node(Master, Config, NodeConfig5, I);
         NodeConfig4 ->
-            Master ! {self(), I, NodeConfig4},
-            unlink(Master)
+            case uses_expected_metadata_store(Config, NodeConfig4) of
+                {MetadataStore, MetadataStore} ->
+                    Master ! {self(), I, NodeConfig4},
+                    unlink(Master);
+                {ExpectedMetadataStore, UsedMetadataStore} ->
+                    %% If the active metadata store is not the one expected, we
+                    %% stop the node and skip the test.
+                    _ = stop_rabbitmq_node(Config, NodeConfig4),
+                    Nodename = ?config(nodename, NodeConfig4),
+                    Error = {skip,
+                             rabbit_misc:format(
+                               "Node ~s is using the ~s metadata store, "
+                               "~s was expected",
+                               [Nodename, UsedMetadataStore,
+                                ExpectedMetadataStore])},
+                    Master ! {self(), Error},
+                    unlink(Master)
+            end
     end.
 
 run_node_steps(Config, NodeConfig, I, [Step | Rest]) ->
@@ -883,6 +899,67 @@ query_node(Config, NodeConfig) ->
     cover_add_node(Nodename),
     rabbit_ct_helpers:set_config(NodeConfig, Vars).
 
+uses_expected_metadata_store(Config, NodeConfig) ->
+    %% We want to verify if the active metadata store matches the expected one.
+    %%
+    %% Before that, we may need to enable `khepri_db' on the node because if it
+    %% doesn't support the relative forced feature flags mechanism, it will
+    %% ignore the setting from `configure_metadata_store/1'. In this case, we
+    %% have to enable it at runtime.
+    Nodename = ?config(nodename, NodeConfig),
+    ExpectedMetadataStore = rabbit_ct_helpers:get_config(
+                              Config, metadata_store),
+    case ExpectedMetadataStore of
+        mnesia -> ok;
+        khepri -> maybe_enable_khepri_as_expected(Config, Nodename)
+    end,
+    IsKhepriEnabled = rpc(Config, Nodename, rabbit_khepri, is_enabled, []),
+    UsedMetadataStore = case IsKhepriEnabled of
+                            true  -> khepri;
+                            false -> mnesia
+                        end,
+    ct:pal(
+      "Metadata store on ~s: expected=~s, used=~s",
+      [Nodename, UsedMetadataStore, ExpectedMetadataStore]),
+    {ExpectedMetadataStore, UsedMetadataStore}.
+
+maybe_enable_khepri_as_expected(Config, Nodename) ->
+    %% We detect if the remote node supports the relative forced feature flags
+    %% mechanism by checking if `rabbit_feature_flags:get_require_level/1' is
+    %% exported. This is unrelated to the mechanism but it was committed around
+    %% the same time and was not backported either.
+    RelativeForcedFeatureFlagsUnsupported = (
+      rpc(Config, Nodename,
+          erlang, function_exported,
+          [rabbit_feature_flags, get_require_level, 1])
+     ),
+    case RelativeForcedFeatureFlagsUnsupported of
+        true ->
+            ok;
+        false ->
+            ct:pal(
+              "Relative forced feature flags unsupported on ~s, "
+              "enable Khepri now to enable expected metadata store",
+              [Nodename]),
+            Ret = enable_feature_flag(Config, [Nodename], khepri_db),
+            case Ret of
+                ok ->
+                    ok;
+                {error, missing_clustered_nodes} ->
+                    ct:pal(
+                      "Tried to enable Khepri on ~s as expected, but node "
+                      "not ready; retrying in 1 second",
+                      [Nodename]),
+                    timer:sleep(1000),
+                    maybe_enable_khepri_as_expected(Config, Nodename);
+                _ ->
+                    ct:pal(
+                      "Tried to enable Khepri on ~s as expected: ~0p",
+                      [Nodename, Ret]),
+                    ok
+            end
+    end.
+
 maybe_cluster_nodes(Config) ->
     Clustered0 = rabbit_ct_helpers:get_config(Config, rmq_nodes_clustered),
     Clustered = case Clustered0 of
@@ -999,57 +1076,71 @@ share_dist_and_proxy_ports_map(Config) ->
 configured_metadata_store(Config) ->
     case rabbit_ct_helpers:get_config(Config, metadata_store) of
         khepri ->
-            {khepri, []};
-        {khepri, _FFs0} = Khepri ->
-            Khepri;
+            khepri;
         mnesia ->
             mnesia;
         _ ->
             case os:getenv("RABBITMQ_METADATA_STORE") of
-                "khepri" ->
-                    {khepri, []};
-                _ ->
-                    mnesia
+                "khepri" -> khepri;
+                _        -> mnesia
             end
     end.
 
 configure_metadata_store(Config) ->
     ct:log("Configuring metadata store..."),
-    case configured_metadata_store(Config) of
-        {khepri, FFs0} ->
-            case enable_khepri_metadata_store(Config, FFs0) of
-                {skip, _} = Skip ->
-                    _ = stop_rabbitmq_nodes(Config),
-                    Skip;
-                Config1 ->
-                    Config1
+    Value = rabbit_ct_helpers:get_app_env(
+              Config, rabbit, forced_feature_flags_on_init, undefined),
+    MetadataStore = configured_metadata_store(Config),
+    Config1 = rabbit_ct_helpers:set_config(
+                Config, {metadata_store, MetadataStore}),
+    %% To enabled or disable `khepri_db', we use the relative forced feature
+    %% flags mechanism. This allows us to select the state of Khepri without
+    %% having to worry about other feature flags.
+    %%
+    %% However, RabbitMQ 4.0.x and older don't support it. See the
+    %% `uses_expected_metadata_store/2' check to see how Khepri is enabled in
+    %% this case.
+    case MetadataStore of
+        khepri ->
+            ct:log("Enabling Khepri metadata store"),
+            case Value of
+                undefined ->
+                    rabbit_ct_helpers:merge_app_env(
+                      Config1,
+                      {rabbit,
+                       [{forced_feature_flags_on_init,
+                         {rel, [khepri_db], []}}]});
+                _ ->
+                    rabbit_ct_helpers:merge_app_env(
+                      Config1,
+                      {rabbit,
+                       [{forced_feature_flags_on_init,
+                         [khepri_db | Value]}]})
             end;
         mnesia ->
             ct:log("Enabling Mnesia metadata store"),
-            Config
+            case Value of
+                undefined ->
+                    rabbit_ct_helpers:merge_app_env(
+                      Config1,
+                      {rabbit,
+                       [{forced_feature_flags_on_init,
+                         {rel, [], [khepri_db]}}]});
+                _ ->
+                    rabbit_ct_helpers:merge_app_env(
+                      Config1,
+                      {rabbit,
+                       [{forced_feature_flags_on_init,
+                         Value -- [khepri_db]}]})
+            end
     end.
 
-enable_khepri_metadata_store(Config, FFs0) ->
-    ct:log("Enabling Khepri metadata store"),
-    FFs = [khepri_db | FFs0],
-    lists:foldl(fun(_FF, {skip, _Reason} = Skip) ->
-                        Skip;
-                   (FF, C) ->
-                        case enable_feature_flag(C, FF) of
-                            ok ->
-                                C;
-                            {skip, _} = Skip ->
-                                ct:pal("Enabling metadata store failed: ~p", [Skip]),
-                                Skip
-                        end
-                end, Config, FFs).
-
 %% Waits until the metadata store replica on Node is up to date with the leader.
 await_metadata_store_consistent(Config, Node) ->
     case configured_metadata_store(Config) of
         mnesia ->
             ok;
-        {khepri, _} ->
+        khepri ->
             RaClusterName = rabbit_khepri:get_ra_cluster_name(),
             Leader = rpc(Config, Node, ra_leaderboard, lookup_leader, [RaClusterName]),
             LastAppliedLeader = ra_last_applied(Leader),

deps/rabbitmq_jms_topic_exchange/test/rjms_topic_selector_SUITE.erl

@@ -56,7 +56,7 @@ init_per_group(mnesia_store = Group, Config0) ->
 init_per_group(khepri_store = Group, Config0) ->
     Config = rabbit_ct_helpers:set_config(
                Config0,
-               [{metadata_store, {khepri, [khepri_db]}}]),
+               [{metadata_store, khepri}]),
     init_per_group_common(Group, Config);
 init_per_group(khepri_migration = Group, Config0) ->
     Config = rabbit_ct_helpers:set_config(Config0, [{metadata_store, mnesia}]),

deps/rabbitmq_recent_history_exchange/test/system_SUITE.erl

@@ -59,8 +59,8 @@ end_per_suite(Config) ->
 
 init_per_group(mnesia_store, Config) ->
     case rabbit_ct_broker_helpers:configured_metadata_store(Config) of
-        {khepri, _} -> {skip, "These tests target Mnesia"};
-        _           -> Config
+        khepri -> {skip, "These tests target Mnesia"};
+        _      -> Config
     end;
 init_per_group(khepri_store, Config) ->
     case rabbit_ct_broker_helpers:configured_metadata_store(Config) of