quorum_queue_SUITE: Use Khepri fence before checking number of replicas

[Why]
When `wait_for_messages_ready/3` returns, we are sure that the replicas
are in the expected state. However, the `#amqqueue{}` record is updated
in Khepri, we don't know when all Khepri store members will be
up-to-date.

It can happen that `Server0` is not up-to-date when we query that record
to get the list of replicass, leading to a test failure.

[How]
First, the check is moved to its own function is `queue_utils`.

Then, if Khepri is being used, we use a Khepri fence to ensure previous
operations were applied on the given server. This way, we get a
consistent view of the `#amqqueue{}` record and thus the list of
replicas.

Author: dumbbell

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -1306,27 +1306,20 @@ force_shrink_member_to_current_member(Config) ->
             RaName = ra_name(QQ),
             rabbit_ct_client_helpers:publish(Ch, QQ, 3),
             wait_for_messages_ready([Server0], RaName, 3),
-
-            {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [QQ, <<"/">>]),
-            #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
-            ?assertEqual(3, length(Nodes0)),
+            queue_utils:assert_number_of_replicas(
+              Config, Server0, <<"/">>, QQ, 3),
 
             rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
                                          force_shrink_member_to_current_member, [<<"/">>, QQ]),
 
             wait_for_messages_ready([Server0], RaName, 3),
-
-            {ok, Q1} = rpc:call(Server0, rabbit_amqqueue, lookup, [QQ, <<"/">>]),
-            #{nodes := Nodes1} = amqqueue:get_type_state(Q1),
-            ?assertEqual(1, length(Nodes1)),
+            queue_utils:assert_number_of_replicas(
+              Config, Server0, <<"/">>, QQ, 1),
 
             %% grow queues back to all nodes
             [rpc:call(Server0, rabbit_quorum_queue, grow, [S, <<"/">>, <<".*">>, all]) || S <- [Server1, Server2]],
-
-            wait_for_messages_ready([Server0], RaName, 3),
-            {ok, Q2} = rpc:call(Server0, rabbit_amqqueue, lookup, [QQ, <<"/">>]),
-            #{nodes := Nodes2} = amqqueue:get_type_state(Q2),
-            ?assertEqual(3, length(Nodes2))
+            queue_utils:assert_number_of_replicas(
+              Config, Server0, <<"/">>, QQ, 3)
     end.
 
 force_all_queues_shrink_member_to_current_member(Config) ->
@@ -1351,9 +1344,8 @@ force_all_queues_shrink_member_to_current_member(Config) ->
                  RaName = ra_name(Q),
                  rabbit_ct_client_helpers:publish(Ch, Q, 3),
                  wait_for_messages_ready([Server0], RaName, 3),
-                 {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, <<"/">>]),
-                 #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
-                 ?assertEqual(3, length(Nodes0))
+                 queue_utils:assert_number_of_replicas(
+                   Config, Server0, <<"/">>, Q, 3)
              end || Q <- QQs],
 
             rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
@@ -1362,9 +1354,8 @@ force_all_queues_shrink_member_to_current_member(Config) ->
             [begin
                  RaName = ra_name(Q),
                  wait_for_messages_ready([Server0], RaName, 3),
-                 {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, <<"/">>]),
-                 #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
-                 ?assertEqual(1, length(Nodes0))
+                 queue_utils:assert_number_of_replicas(
+                   Config, Server0, <<"/">>, Q, 1)
              end || Q <- QQs],
 
             %% grow queues back to all nodes
@@ -1373,9 +1364,8 @@ force_all_queues_shrink_member_to_current_member(Config) ->
             [begin
                  RaName = ra_name(Q),
                  wait_for_messages_ready([Server0], RaName, 3),
-                 {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, <<"/">>]),
-                 #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
-                 ?assertEqual(3, length(Nodes0))
+                 queue_utils:assert_number_of_replicas(
+                   Config, Server0, <<"/">>, Q, 3)
              end || Q <- QQs]
     end.
 
@@ -1417,9 +1407,8 @@ force_vhost_queues_shrink_member_to_current_member(Config) ->
                 QQRes = rabbit_misc:r(VHost, queue, Q),
                 {ok, RaName} = rpc:call(Server0, rabbit_queue_type_util, qname_to_internal_name, [QQRes]),
                 wait_for_messages_ready([Server0], RaName, 3),
-                {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, VHost]),
-                #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
-                ?assertEqual(3, length(Nodes0))
+                 queue_utils:assert_number_of_replicas(
+                   Config, Server0, VHost, Q, 3)
             end || Q <- QQs, VHost <- VHosts],
 
             rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
@@ -1429,11 +1418,13 @@ force_vhost_queues_shrink_member_to_current_member(Config) ->
                 QQRes = rabbit_misc:r(VHost, queue, Q),
                 {ok, RaName} = rpc:call(Server0, rabbit_queue_type_util, qname_to_internal_name, [QQRes]),
                 wait_for_messages_ready([Server0], RaName, 3),
-                {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, VHost]),
-                #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
                 case VHost of
-                    VHost1 -> ?assertEqual(3, length(Nodes0));
-                    VHost2 -> ?assertEqual(1, length(Nodes0))
+                    VHost1 ->
+                        queue_utils:assert_number_of_replicas(
+                          Config, Server0, VHost, Q, 3);
+                    VHost2 ->
+                        queue_utils:assert_number_of_replicas(
+                          Config, Server0, VHost, Q, 1)
                 end
             end || Q <- QQs, VHost <- VHosts],
 
@@ -1444,9 +1435,8 @@ force_vhost_queues_shrink_member_to_current_member(Config) ->
                 QQRes = rabbit_misc:r(VHost, queue, Q),
                 {ok, RaName} = rpc:call(Server0, rabbit_queue_type_util, qname_to_internal_name, [QQRes]),
                 wait_for_messages_ready([Server0], RaName, 3),
-                {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, VHost]),
-                #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
-                ?assertEqual(3, length(Nodes0))
+                queue_utils:assert_number_of_replicas(
+                  Config, Server0, VHost, Q, 3)
             end || Q <- QQs, VHost <- VHosts]
     end.
 
@@ -2946,9 +2936,8 @@ delete_member_member_already_deleted(Config) ->
     ?assertEqual(ok,
                  rpc:call(Server, rabbit_quorum_queue, delete_member,
                           [<<"/">>, QQ, Server2])),
-    {ok, Q} = rpc:call(Server, rabbit_amqqueue, lookup, [QQ, <<"/">>]),
-    #{nodes := Nodes} = amqqueue:get_type_state(Q),
-    ?assertEqual(1, length(Nodes)),
+    queue_utils:assert_number_of_replicas(
+      Config, Server, <<"/">>, QQ, 1),
     ok.
 
 delete_member_during_node_down(Config) ->

deps/rabbitmq_ct_helpers/src/queue_utils.erl

@@ -15,7 +15,8 @@
          ra_name/1,
          ra_machines_use_same_version/3,
          wait_for_local_stream_member/4,
-         has_local_stream_member_rpc/1
+         has_local_stream_member_rpc/1,
+         assert_number_of_replicas/5
         ]).
 
 -define(WFM_SLEEP, 256).
@@ -191,3 +192,16 @@ has_local_stream_member_rpc(QName) ->
         {error, _} ->
             false
     end.
+
+assert_number_of_replicas(Config, Server, VHost, QQ, Count) ->
+    _ = case rabbit_ct_broker_helpers:configured_metadata_store(Config) of
+            khepri ->
+                rabbit_ct_broker_helpers:rpc(
+                  Config, Server, rabbit_khepri, fence, [30000]);
+            mnesia ->
+                ok
+        end,
+    {ok, Q} = rabbit_ct_broker_helpers:rpc(
+                Config, Server, rabbit_amqqueue, lookup, [QQ, VHost]),
+    #{nodes := Nodes} = amqqueue:get_type_state(Q),
+    ?assertEqual(Count, length(Nodes)).