Fix more flakes by adopting eventual matchers

(cherry picked from commit 6d85f8a)

Author: michaelklishin

deps/rabbitmq_cli/test/test_helper.exs

@@ -947,6 +947,16 @@ defmodule TestHelper do
           file
       end
 
+    do_wait_for_log_message(message, log_file, attempts)
+  end
+
+  defp do_wait_for_log_message(message, log_file, 0) do
+    flunk(
+      "Ran out of attempts waiting for #{inspect(message)} to appear in #{inspect(log_file)}"
+    )
+  end
+
+  defp do_wait_for_log_message(message, log_file, attempts) do
     case File.read(log_file) do
       {:ok, data} ->
         case String.match?(data, Regex.compile!(message)) do
@@ -955,12 +965,12 @@ defmodule TestHelper do
 
           false ->
             :timer.sleep(100)
-            wait_for_log_message(message, log_file, attempts - 1)
+            do_wait_for_log_message(message, log_file, attempts - 1)
         end
 
       _ ->
         :timer.sleep(100)
-        wait_for_log_message(message, log_file, attempts - 1)
+        do_wait_for_log_message(message, log_file, attempts - 1)
     end
   end
 end

deps/rabbitmq_stomp/test/command_SUITE.erl

@@ -11,6 +11,7 @@
 -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").
 -include("rabbit_stomp.hrl").
 
 
@@ -74,17 +75,16 @@ run(Config) ->
     StompPort = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_stomp),
 
     {ok, _Client} = rabbit_stomp_client:connect(StompPort),
-    ct:sleep(100),
-
-    [[{session_id, _}]] =
-            'Elixir.Enum':to_list(?COMMAND:run([<<"session_id">>], Opts)),
-
+    %% STOMP connection registration is asynchronous; wait for it to be
+    %% readable (as in consistency).
+    ?awaitMatch([[{session_id, _}]],
+                'Elixir.Enum':to_list(?COMMAND:run([<<"session_id">>], Opts)),
+                30_000),
 
     {ok, _Client2} = rabbit_stomp_client:connect(StompPort),
-    ct:sleep(100),
-
-    [[{session_id, _}], [{session_id, _}]] =
-        'Elixir.Enum':to_list(?COMMAND:run([<<"session_id">>], Opts)),
+    ?awaitMatch([[{session_id, _}], [{session_id, _}]],
+                'Elixir.Enum':to_list(?COMMAND:run([<<"session_id">>], Opts)),
+                30_000),
 
     Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
     start_amqp_connection(network, Node, Port),

deps/rabbitmq_stomp/test/connections_SUITE.erl

@@ -98,8 +98,10 @@ messages_not_dropped_on_disconnect(Config) ->
     StompPort = get_stomp_port(Config),
     N = count_connections(Config),
     {ok, Client} = rabbit_stomp_client:connect(StompPort),
+    %% STOMP connection registration is asynchronous; wait for it to be
+    %% readable (as in consistency).
     N1 = N + 1,
-    N1 = count_connections(Config),
+    ?awaitMatch(N1, count_connections(Config), 30_000),
     [rabbit_stomp_client:send(
        Client, "SEND", [{"destination", ?DESTINATION}],
        [integer_to_list(Count)]) || Count <- lists:seq(1, 1000)],

deps/rabbitmq_stomp/test/proxy_protocol_SUITE.erl

@@ -10,6 +10,7 @@
 
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
 
 -define(TIMEOUT, 5000).
 
@@ -70,9 +71,8 @@ proxy_protocol_v1(Config) ->
     ok = inet:send(Socket, "PROXY TCP4 192.168.1.1 192.168.1.2 80 81\r\n"),
     ok = inet:send(Socket, stomp_connect_frame()),
     {ok, _Packet} = gen_tcp:recv(Socket, 0, ?TIMEOUT),
-    ConnectionName = rabbit_ct_broker_helpers:rpc(Config, 0,
-        ?MODULE, connection_name, []),
-    match = re:run(ConnectionName, <<"^192.168.1.1:80 -> 192.168.1.2:81$">>, [{capture, none}]),
+    await_connection_name_match(
+      Config, <<"^192.168.1.1:80 -> 192.168.1.2:81$">>),
     gen_tcp:close(Socket),
     ok.
 
@@ -85,9 +85,8 @@ proxy_protocol_v1_tls(Config) ->
     {ok, SslSocket} = ssl:connect(Socket, [{verify, verify_none}], ?TIMEOUT),
     ok = ssl:send(SslSocket, stomp_connect_frame()),
     {ok, _Packet} = ssl:recv(SslSocket, 0, ?TIMEOUT),
-    ConnectionName = rabbit_ct_broker_helpers:rpc(Config, 0,
-        ?MODULE, connection_name, []),
-    match = re:run(ConnectionName, <<"^192.168.1.1:80 -> 192.168.1.2:81$">>, [{capture, none}]),
+    await_connection_name_match(
+      Config, <<"^192.168.1.1:80 -> 192.168.1.2:81$">>),
     gen_tcp:close(Socket),
     ok.
 
@@ -102,17 +101,31 @@ proxy_protocol_v2_local(Config) ->
     ok = inet:send(Socket, ranch_proxy_header:header(ProxyInfo)),
     ok = inet:send(Socket, stomp_connect_frame()),
     {ok, _Packet} = gen_tcp:recv(Socket, 0, ?TIMEOUT),
-    ConnectionName = rabbit_ct_broker_helpers:rpc(Config, 0,
-        ?MODULE, connection_name, []),
-    match = re:run(ConnectionName, <<"^127.0.0.1:\\d+ -> 127.0.0.1:\\d+$">>, [{capture, none}]),
+    await_connection_name_match(
+      Config, <<"^127.0.0.1:\\d+ -> 127.0.0.1:\\d+$">>),
     gen_tcp:close(Socket),
     ok.
 
-connection_name() ->
+%% The `connection_created' ETS table is populated asynchronously by
+%% the management agent; wait for an entry whose `name' matches the
+%% pattern.
+await_connection_name_match(Config, Pattern) ->
+    ?awaitMatch(true,
+                rabbit_ct_broker_helpers:rpc(
+                  Config, 0, ?MODULE, has_connection_name_matching, [Pattern]),
+                30_000).
+
+has_connection_name_matching(Pattern) ->
     Connections = ets:tab2list(connection_created),
-    {_Key, Values} = lists:nth(1, Connections),
-    {_, Name} = lists:keyfind(name, 1, Values),
-    Name.
+    lists:any(
+      fun({_Key, Values}) ->
+              case lists:keyfind(name, 1, Values) of
+                  {_, Name} ->
+                      re:run(Name, Pattern, [{capture, none}]) =:= match;
+                  false ->
+                      false
+              end
+      end, Connections).
 
 merge_app_env(StompConfig, Config) ->
     rabbit_ct_helpers:merge_app_env(Config, StompConfig).

deps/rabbitmq_stomp/test/topic_SUITE.erl

@@ -12,6 +12,7 @@
 -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").
 -include("rabbit_stomp.hrl").
 -include("rabbit_stomp_frame.hrl").
 all() ->
@@ -191,9 +192,13 @@ change_default_topic_exchange(Config) ->
                                  {"id", "s0"},
                                  {"durable", "true"}]),
 
-    timer:sleep(500),
-
-    1 = length(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_binding, list_for_source, [#resource{virtual_host= <<"/">>, kind = exchange, name = Ex}])),
+    %% STOMP SUBSCRIBE creates the binding asynchronously; wait for it
+    %% to be readable (as in consistency).
+    ?awaitMatch(1,
+                length(rabbit_ct_broker_helpers:rpc(
+                         Config, 0, rabbit_binding, list_for_source,
+                         [#resource{virtual_host = <<"/">>, kind = exchange, name = Ex}])),
+                30_000),
 
     rabbit_stomp_client:send(
         ClientFoo, "SEND", [{"destination", AuthorisedTopic}], ["ohai there"]),