Reduce Khepri test flakes

Test case rabbit_mqtt_qos0_queue_kill_node flaked because after an
MQTT client subscribes on node 0, RabbitMQ returns success
and replicated the new binding to node 0 and node 1, but not
yet to node 2. Another MQTT client then publishes on node 2
without the binding being present yet on node 2, and the
message therefore isn't routed.

This commit attempts to eliminate this flake.
It adds a function to rabbit_ct_broker_helpers which waits until a given
node has caught up with the leader node.
We can reuse that function in future to eliminate more test flakes.

Author: ansd

deps/rabbitmq_ct_helpers/BUILD.bazel

@@ -45,6 +45,7 @@ rabbitmq_app(
         "//deps/rabbit_common:erlang_app",
         "@meck//:erlang_app",
         "@proper//:erlang_app",
+        "@ra//:erlang_app",
     ],
 )
 

deps/rabbitmq_ct_helpers/src/rabbit_ct_broker_helpers.erl

@@ -170,7 +170,8 @@
     test_writer/1,
     user/1,
 
-    configured_metadata_store/1
+    configured_metadata_store/1,
+    await_metadata_store_consistent/2
   ]).
 
 %% Internal functions exported to be used by rpc:call/4.
@@ -990,6 +991,30 @@ enable_khepri_metadata_store(Config, FFs0) ->
                         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, _} ->
+            RaClusterName = rabbit_khepri:get_ra_cluster_name(),
+            Leader = rpc(Config, Node, ra_leaderboard, lookup_leader, [RaClusterName]),
+            LastAppliedLeader = ra_last_applied(Leader),
+
+            NodeName = get_node_config(Config, Node, nodename),
+            ServerId = {RaClusterName, NodeName},
+            rabbit_ct_helpers:eventually(
+              ?_assert(
+                 begin
+                     LastApplied = ra_last_applied(ServerId),
+                     is_integer(LastApplied) andalso LastApplied >= LastAppliedLeader
+                 end))
+    end.
+
+ra_last_applied(ServerId) ->
+    #{last_applied := LastApplied} = ra:key_metrics(ServerId),
+    LastApplied.
+
 rewrite_node_config_file(Config, Node) ->
     NodeConfig = get_node_config(Config, Node),
     I = if

deps/rabbitmq_mqtt/test/shared_SUITE.erl

@@ -27,7 +27,8 @@
          rpc_all/4,
          get_node_config/3,
          drain_node/2,
-         revive_node/2
+         revive_node/2,
+         await_metadata_store_consistent/2
         ]).
 -import(rabbit_ct_helpers,
         [eventually/3,
@@ -1128,6 +1129,7 @@ rabbit_mqtt_qos0_queue_kill_node(Config) ->
     SubscriberId = <<"subscriber">>,
     Sub0 = connect(SubscriberId, Config, 0, []),
     {ok, _, [0]} = emqtt:subscribe(Sub0, Topic1, qos0),
+    ok = await_metadata_store_consistent(Config, 2),
     ok = emqtt:publish(Pub, Topic1, <<"m0">>, qos0),
     ok = expect_publishes(Sub0, Topic1, [<<"m0">>]),