Merge pull request #14310 from rabbitmq/fix-test-flakes-2025Q2

Collection of test fixes (2025Q2, batch 2)

Author: dumbbell

.github/workflows/test-make-target.yaml

@@ -143,5 +143,6 @@ jobs:
         name: CT logs (${{ inputs.plugin }} ${{ inputs.make_target }} OTP-${{ inputs.erlang_version }} ${{ inputs.metadata_store }}${{ inputs.mixed_clusters && ' mixed' || '' }})
         path: |
           logs/
+          deps/rabbitmq_cli/logs/
 #          !logs/**/log_private
         if-no-files-found: ignore

deps/rabbit/src/rabbit_vhosts.erl

@@ -127,7 +127,7 @@ start_processes_for_all(Nodes) ->
 
 -spec start_processes_for_all() -> 'ok'.
 start_processes_for_all() ->
-    start_processes_for_all(rabbit_nodes:list_reachable()).
+    start_processes_for_all(rabbit_nodes:list_running()).
 
 %% Same as rabbit_vhost_sup_sup:start_on_all_nodes/0.
 -spec start_on_all_nodes(vhost:name(), [node()]) -> 'ok'.

deps/rabbit/test/amqp_client_SUITE.erl

@@ -1867,6 +1867,11 @@ link_target_queue_deleted(QType, Config) ->
     ok = amqp10_client:send_msg(Sender, amqp10_msg:new(DTag1, <<"m1">>, false)),
     ok = wait_for_accepted(DTag1),
 
+    %% Load test module on the broker: we reference an anonymous function
+    %% from it during the configuration of meck.
+    [_ | _] = rabbit_ct_broker_helpers:rpc(
+                Config, ?MODULE, module_info, []),
+
     %% Mock delivery to the target queue to do nothing.
     rabbit_ct_broker_helpers:setup_meck(Config, [?MODULE]),
     Mod = rabbit_queue_type,
@@ -1927,6 +1932,11 @@ target_queues_deleted_accepted(Config) ->
     ok = amqp10_client:send_msg(Sender, amqp10_msg:new(DTag1, <<"m1">>, false)),
     ok = wait_for_accepted(DTag1),
 
+    %% Load test module on the broker: we reference an anonymous function
+    %% from it during the configuration of meck.
+    [_ | _] = rabbit_ct_broker_helpers:rpc(
+                Config, ?MODULE, module_info, []),
+
     %% Mock to deliver only to q1.
     rabbit_ct_broker_helpers:setup_meck(Config, [?MODULE]),
     Mod = rabbit_queue_type,
@@ -4073,7 +4083,7 @@ list_connections(Config) ->
     end,
 
     {ok, StdOut0} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["list_connections", "--silent", "protocol"]),
-    Protocols0 = re:split(StdOut0, <<"\n">>, [trim]),
+    Protocols0 = re:split(string:trim(StdOut0), <<"\n">>, [trim]),
     %% Remove any whitespaces.
     Protocols1 = [binary:replace(Subject, <<" ">>, <<>>, [global]) || Subject <- Protocols0],
     Protocols = lists:sort(Protocols1),
@@ -4726,6 +4736,11 @@ idle_time_out_on_server(Config) ->
         after 30000 -> ct:fail({missing_event, ?LINE})
         end,
 
+        %% Load test module on the broker: we reference an anonymous function
+        %% from it during the configuration of meck.
+        [_ | _] = rabbit_ct_broker_helpers:rpc(
+                    Config, ?MODULE, module_info, []),
+
         %% Mock the server socket to not have received any bytes.
         rabbit_ct_broker_helpers:setup_meck(Config),
         ok = rpc(Config, meck, new, [Mod, [no_link, passthrough]]),

deps/rabbit/test/amqp_filter_sql_SUITE.erl

@@ -26,6 +26,7 @@
          flush/1,
          wait_for_credit/1,
          wait_for_accepts/1,
+         send_message/2,
          send_messages/3,
          detach_link_sync/1,
          end_session_sync/1,
@@ -98,10 +99,10 @@ multiple_sections(Config) ->
     To = rabbitmq_amqp_address:exchange(<<"some exchange">>, <<"routing key">>),
     ReplyTo = rabbitmq_amqp_address:queue(<<"some queue">>),
 
-    ok = amqp10_client:send_msg(
+    ok = send_message(
            Sender,
            amqp10_msg:new(<<"t1">>, <<"m1">>)),
-    ok = amqp10_client:send_msg(
+    ok = send_message(
            Sender,
            amqp10_msg:set_headers(
              #{priority => 200},
@@ -125,7 +126,7 @@ multiple_sections(Config) ->
                    <<"k3">> => true,
                    <<"k4">> => <<"hey👋"/utf8>>},
                  amqp10_msg:new(<<"t2">>, <<"m2">>))))),
-    ok = amqp10_client:send_msg(
+    ok = send_message(
            Sender,
            amqp10_msg:set_properties(
              #{group_id => <<"my group ID">>},
@@ -222,13 +223,13 @@ filter_few_messages_from_many(Config) ->
     {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, Address),
     ok = wait_for_credit(Sender),
 
-    ok = amqp10_client:send_msg(
+    ok = send_message(
            Sender,
            amqp10_msg:set_properties(
              #{group_id => <<"my group ID">>},
              amqp10_msg:new(<<"t1">>, <<"first msg">>))),
     ok = send_messages(Sender, 5000, false),
-    ok = amqp10_client:send_msg(
+    ok = send_message(
            Sender,
            amqp10_msg:set_properties(
              #{group_id => <<"my group ID">>},
@@ -278,7 +279,7 @@ filter_few_messages_from_many(Config) ->
 
     %% We previously set drain=true for Receiver1
     ok = assert_credit_exhausted(Receiver1, ?LINE),
-    ok = amqp10_client:send_msg(
+    ok = send_message(
            Sender,
            amqp10_msg:set_properties(
              #{group_id => <<"my group ID">>},
@@ -328,7 +329,7 @@ sql_and_bloom_filter(Config) ->
     {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, Address),
     ok = wait_for_credit(Sender),
 
-    ok = amqp10_client:send_msg(
+    ok = send_message(
            Sender,
            amqp10_msg:set_message_annotations(
              #{<<"x-stream-filter-value">> => <<"v1">>},

deps/rabbit/test/amqp_jms_SUITE.erl

@@ -52,7 +52,7 @@ groups() ->
 
 suite() ->
     [
-     {timetrap, {minutes, 2}}
+     {timetrap, {minutes, 5}}
     ].
 
 init_per_suite(Config) ->

deps/rabbit/test/amqp_utils.erl

@@ -16,6 +16,7 @@
          flush/1,
          wait_for_credit/1,
          wait_for_accepts/1,
+         send_message/2,
          send_messages/3, send_messages/4,
          detach_link_sync/1,
          end_session_sync/1,
@@ -87,18 +88,10 @@ wait_for_accepts(N) ->
               ct:fail({missing_accepted, N})
     end.
 
-send_messages(Sender, Left, Settled) ->
-    send_messages(Sender, Left, Settled, <<>>).
-
-send_messages(_, 0, _, _) ->
-    ok;
-send_messages(Sender, Left, Settled, BodySuffix) ->
-    Bin = integer_to_binary(Left),
-    Body = <<Bin/binary, BodySuffix/binary>>,
-    Msg = amqp10_msg:new(Bin, Body, Settled),
+send_message(Sender, Msg) ->
     case amqp10_client:send_msg(Sender, Msg) of
         ok ->
-            send_messages(Sender, Left - 1, Settled, BodySuffix);
+            ok;
         {error, insufficient_credit} ->
             ok = wait_for_credit(Sender),
             %% The credited event we just processed could have been received some time ago,
@@ -110,9 +103,21 @@ send_messages(Sender, Left, Settled, BodySuffix) ->
             %%    but do not process the credited event in our mailbox.
             %% So, we must be defensive here and assume that the next amqp10_client:send/2 call might return {error, insufficient_credit}
             %% again causing us then to really wait to receive a credited event (instead of just processing an old credited event).
-            send_messages(Sender, Left, Settled, BodySuffix)
+            send_message(Sender, Msg)
     end.
 
+send_messages(Sender, Left, Settled) ->
+    send_messages(Sender, Left, Settled, <<>>).
+
+send_messages(_, 0, _, _) ->
+    ok;
+send_messages(Sender, Left, Settled, BodySuffix) ->
+    Bin = integer_to_binary(Left),
+    Body = <<Bin/binary, BodySuffix/binary>>,
+    Msg = amqp10_msg:new(Bin, Body, Settled),
+    ok = send_message(Sender, Msg),
+    send_messages(Sender, Left - 1, Settled, BodySuffix).
+
 detach_link_sync(Link) ->
     ok = amqp10_client:detach_link(Link),
     ok = wait_for_link_detach(Link).

deps/rabbit/test/feature_flags_v2_SUITE.erl

@@ -203,7 +203,9 @@ stop_slave_node(Node) ->
             persistent_term:erase({?MODULE, Node}),
 
             ct:pal("- Stopping slave node `~ts`...", [Node]),
-            _ = peer:stop(NodePid)
+            Ret = catch peer:stop(NodePid),
+            ct:pal("  Ret = ~0p", [Ret]),
+            ok
     end.
 
 connect_nodes([FirstNode | OtherNodes] = Nodes) ->

deps/rabbit/test/per_user_connection_channel_tracking_SUITE.erl

@@ -130,7 +130,7 @@ single_node_user_connection_channel_tracking(Config) ->
     [Conn1] = open_connections(Config, [0]),
     [Chan1] = open_channels(Conn1, 1),
     ?awaitMatch(1, count_connections_in(Config, Username), ?A_TOUT),
-    [#tracked_connection{username = Username}] = connections_in(Config, Username),
+    ?awaitMatch([#tracked_connection{username = Username}], connections_in(Config, Username), ?A_TOUT),
     ?awaitMatch(1, count_channels_in(Config, Username), ?A_TOUT),
     [#tracked_channel{username = Username}] = channels_in(Config, Username),
     ?awaitMatch(true, is_process_alive(Conn1), ?A_TOUT),
@@ -147,7 +147,7 @@ single_node_user_connection_channel_tracking(Config) ->
     [Conn2] = open_connections(Config, [{0, Username2}]),
     Chans2  = [_|_] = open_channels(Conn2, 5),
     ?awaitMatch(1, count_connections_in(Config, Username2), ?A_TOUT),
-    [#tracked_connection{username = Username2}] = connections_in(Config, Username2),
+    ?awaitMatch([#tracked_connection{username = Username2}], connections_in(Config, Username2), ?A_TOUT),
     ?awaitMatch(5, count_channels_in(Config, Username2), ?A_TOUT),
     ?awaitMatch(1, tracked_user_connection_count(Config, Username2), ?A_TOUT),
     ?awaitMatch(5, tracked_user_channel_count(Config, Username2), ?A_TOUT),
@@ -157,7 +157,7 @@ single_node_user_connection_channel_tracking(Config) ->
     [Conn3] = open_connections(Config, [0]),
     Chans3 = [_|_] = open_channels(Conn3, 5),
     ?awaitMatch(1, count_connections_in(Config, Username), ?A_TOUT),
-    [#tracked_connection{username = Username}] = connections_in(Config, Username),
+    ?awaitMatch([#tracked_connection{username = Username}], connections_in(Config, Username), ?A_TOUT),
     ?awaitMatch(5, count_channels_in(Config, Username), ?A_TOUT),
     ?awaitMatch(1, tracked_user_connection_count(Config, Username), ?A_TOUT),
     ?awaitMatch(5, tracked_user_channel_count(Config, Username), ?A_TOUT),
@@ -172,7 +172,7 @@ single_node_user_connection_channel_tracking(Config) ->
     [?awaitMatch(true, is_process_alive(Ch), ?A_TOUT) || Ch <- Chans4],
     kill_connections([Conn4]),
     ?awaitMatch(1, count_connections_in(Config, Username), ?A_TOUT),
-    [#tracked_connection{username = Username}] = connections_in(Config, Username),
+    ?awaitMatch([#tracked_connection{username = Username}], connections_in(Config, Username), ?A_TOUT),
     ?awaitMatch(5, count_channels_in(Config, Username), ?A_TOUT),
     ?awaitMatch(1, tracked_user_connection_count(Config, Username), ?A_TOUT),
     ?awaitMatch(5, tracked_user_channel_count(Config, Username), ?A_TOUT),
@@ -182,9 +182,11 @@ single_node_user_connection_channel_tracking(Config) ->
     [Conn5] = open_connections(Config, [0]),
     Chans5  = [_|_] = open_channels(Conn5, 7),
     ?awaitMatch(2, count_connections_in(Config, Username), ?A_TOUT),
-    [Username, Username] =
-        lists:map(fun (#tracked_connection{username = U}) -> U end,
-                  connections_in(Config, Username)),
+    ?awaitMatch(
+       [Username, Username],
+       lists:map(fun (#tracked_connection{username = U}) -> U end,
+                 connections_in(Config, Username)),
+       ?A_TOUT),
     ?awaitMatch(12, count_channels_in(Config, Username), ?A_TOUT),
     ?awaitMatch(12, tracked_user_channel_count(Config, Username), ?A_TOUT),
     ?awaitMatch(2, tracked_user_connection_count(Config, Username), ?A_TOUT),

deps/rabbit/test/per_user_connection_tracking_SUITE.erl

@@ -115,28 +115,42 @@ single_node_list_of_user(Config) ->
 
     [Conn1] = open_connections(Config, [{0, Username1}]),
     ?awaitMatch(1, count_connections_in(Config, Username1), ?AWAIT_TIMEOUT),
-    [#tracked_connection{username = Username1}] = connections_in(Config, Username1),
+    ?awaitMatch(
+       [#tracked_connection{username = Username1}],
+       connections_in(Config, Username1),
+       ?AWAIT_TIMEOUT),
     close_connections([Conn1]),
     ?awaitMatch(0, count_connections_in(Config, Username1), ?AWAIT_TIMEOUT),
 
     [Conn2] = open_connections(Config, [{0, Username2}]),
     ?awaitMatch(1, count_connections_in(Config, Username2), ?AWAIT_TIMEOUT),
-    [#tracked_connection{username = Username2}] = connections_in(Config, Username2),
+    ?awaitMatch(
+       [#tracked_connection{username = Username2}],
+       connections_in(Config, Username2),
+       ?AWAIT_TIMEOUT),
 
     [Conn3] = open_connections(Config, [{0, Username1}]),
     ?awaitMatch(1, count_connections_in(Config, Username1), ?AWAIT_TIMEOUT),
-    [#tracked_connection{username = Username1}] = connections_in(Config, Username1),
+    ?awaitMatch(
+       [#tracked_connection{username = Username1}],
+       connections_in(Config, Username1),
+       ?AWAIT_TIMEOUT),
 
     [Conn4] = open_connections(Config, [{0, Username1}]),
     kill_connections([Conn4]),
     ?awaitMatch(1, count_connections_in(Config, Username1), ?AWAIT_TIMEOUT),
-    [#tracked_connection{username = Username1}] = connections_in(Config, Username1),
+    ?awaitMatch(
+       [#tracked_connection{username = Username1}],
+       connections_in(Config, Username1),
+       ?AWAIT_TIMEOUT),
 
     [Conn5] = open_connections(Config, [{0, Username1}]),
     ?awaitMatch(2, count_connections_in(Config, Username1), ?AWAIT_TIMEOUT),
-    [Username1, Username1] =
-        lists:map(fun (#tracked_connection{username = U}) -> U end,
-                  connections_in(Config, Username1)),
+    ?awaitMatch(
+       [Username1, Username1],
+       lists:map(fun (#tracked_connection{username = U}) -> U end,
+                 connections_in(Config, Username1)),
+       ?AWAIT_TIMEOUT),
 
     close_connections([Conn2, Conn3, Conn5]),
     rabbit_ct_broker_helpers:delete_user(Config, Username2),

deps/rabbit/test/proxy_protocol_SUITE.erl

@@ -10,6 +10,8 @@
 -include_lib("common_test/include/ct.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
 
+-include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
+
 -compile(export_all).
 
 -define(TIMEOUT, 5000).
@@ -65,8 +67,10 @@ proxy_protocol_v1(Config) ->
     {ok, _Packet} = gen_tcp:recv(Socket, 0, ?TIMEOUT),
     ConnectionName = rabbit_ct_broker_helpers:rpc(Config, 0,
         ?MODULE, connection_name, []),
+    ct:pal("Connection name: ~s", [ConnectionName]),
     match = re:run(ConnectionName, <<"^192.168.1.1:80 -> 192.168.1.2:81$">>, [{capture, none}]),
     gen_tcp:close(Socket),
+    wait_for_connection_close(Config),
     ok.
 
 proxy_protocol_v1_tls(Config) ->
@@ -80,8 +84,10 @@ proxy_protocol_v1_tls(Config) ->
     {ok, _Packet} = ssl:recv(SslSocket, 0, ?TIMEOUT),
     ConnectionName = rabbit_ct_broker_helpers:rpc(Config, 0,
         ?MODULE, connection_name, []),
+    ct:pal("Connection name: ~s", [ConnectionName]),
     match = re:run(ConnectionName, <<"^192.168.1.1:80 -> 192.168.1.2:81$">>, [{capture, none}]),
     gen_tcp:close(Socket),
+    wait_for_connection_close(Config),
     ok.
 
 proxy_protocol_v2_local(Config) ->
@@ -97,13 +103,22 @@ proxy_protocol_v2_local(Config) ->
     {ok, _Packet} = gen_tcp:recv(Socket, 0, ?TIMEOUT),
     ConnectionName = rabbit_ct_broker_helpers:rpc(Config, 0,
         ?MODULE, connection_name, []),
+    ct:pal("Connection name: ~s", [ConnectionName]),
     match = re:run(ConnectionName, <<"^127.0.0.1:\\d+ -> 127.0.0.1:\\d+$">>, [{capture, none}]),
     gen_tcp:close(Socket),
+    wait_for_connection_close(Config),
     ok.
 
 connection_name() ->
-    Pids = pg_local:get_members(rabbit_connections),
-    Pid = lists:nth(1, Pids),
+    ?awaitMatch([_], pg_local:get_members(rabbit_connections), 30000),
+    [Pid] = pg_local:get_members(rabbit_connections),
     {dictionary, Dict} = process_info(Pid, dictionary),
     {process_name, {rabbit_reader, ConnectionName}} = lists:keyfind(process_name, 1, Dict),
     ConnectionName.
+
+wait_for_connection_close(Config) ->
+    ?awaitMatch(
+       [],
+       rabbit_ct_broker_helpers:rpc(
+         Config, 0, pg_local, get_members, [rabbit_connnections]),
+       30000).