Port the rabbit part of #10477

Author: michaelklishin

deps/rabbit/src/rabbit_amqqueue.erl

@@ -404,7 +404,7 @@ maybe_rebalance({true, Id}, Type, VhostSpec, QueueSpec) ->
                     [Type, VhostSpec, QueueSpec]),
     Running = rabbit_maintenance:filter_out_drained_nodes_consistent_read(rabbit_nodes:list_running()),
     NumRunning = length(Running),
-    ToRebalance = [Q || Q <- rabbit_amqqueue:list(),
+    ToRebalance = [Q || Q <- list(),
                         filter_per_type(Type, Q),
                         is_replicated(Q),
                         is_match(amqqueue:get_vhost(Q), VhostSpec) andalso
@@ -1138,7 +1138,7 @@ list() ->
 count() ->
     rabbit_db_queue:count().
 
--spec list_names() -> [rabbit_amqqueue:name()].
+-spec list_names() -> [name()].
 
 list_names() ->
     rabbit_db_queue:list().
@@ -1169,7 +1169,7 @@ is_down(Q) ->
 -spec sample_local_queues() -> [amqqueue:amqqueue()].
 sample_local_queues() -> sample_n_by_name(list_local_names(), 300).
 
--spec sample_n_by_name([rabbit_amqqueue:name()], pos_integer()) -> [amqqueue:amqqueue()].
+-spec sample_n_by_name([name()], pos_integer()) -> [amqqueue:amqqueue()].
 sample_n_by_name([], _N) ->
     [];
 sample_n_by_name(Names, N) when is_list(Names) andalso is_integer(N) andalso N > 0 ->
@@ -1183,7 +1183,7 @@ sample_n_by_name(Names, N) when is_list(Names) andalso is_integer(N) andalso N >
                      end,
          [], lists:seq(1, M)),
     lists:map(fun (Id) ->
-                {ok, Q} = rabbit_amqqueue:lookup(Id),
+                {ok, Q} = lookup(Id),
                 Q
               end,
               lists:usort(Ids)).
@@ -1206,7 +1206,7 @@ list_by_type(stream)  -> list_by_type(rabbit_stream_queue);
 list_by_type(Type) ->
     rabbit_db_queue:get_all_durable_by_type(Type).
 
--spec list_local_quorum_queue_names() -> [rabbit_amqqueue:name()].
+-spec list_local_quorum_queue_names() -> [name()].
 
 list_local_quorum_queue_names() ->
     [ amqqueue:get_name(Q) || Q <- list_by_type(quorum),
@@ -1252,7 +1252,7 @@ list_local_mirrored_classic_queues() ->
         is_local_to_node(amqqueue:get_pid(Q), node()),
         is_replicated(Q)].
 
--spec list_local_mirrored_classic_names() -> [rabbit_amqqueue:name()].
+-spec list_local_mirrored_classic_names() -> [name()].
 list_local_mirrored_classic_names() ->
     [ amqqueue:get_name(Q) || Q <- list(),
            amqqueue:get_state(Q) =/= crashed,
@@ -1671,7 +1671,7 @@ delete_with(QueueName, ConnPid, IfUnused, IfEmpty, Username, CheckExclusive) whe
         {error, not_empty} ->
             rabbit_misc:precondition_failed("~ts not empty", [rabbit_misc:rs(QueueName)]);
         {error, {exit, _, _}} ->
-            %% rabbit_amqqueue:delete()/delegate:invoke might return {error, {exit, _, _}}
+            %% delete()/delegate:invoke might return {error, {exit, _, _}}
             {ok, 0};
         {ok, Count} ->
             {ok, Count};
@@ -2013,8 +2013,8 @@ filter_transient_queues_to_delete(Node) ->
             amqqueue:qnode(Q) == Node andalso
                 not rabbit_process:is_process_alive(amqqueue:get_pid(Q))
                 andalso (not amqqueue:is_classic(Q) orelse not amqqueue:is_durable(Q))
-                andalso (not rabbit_amqqueue:is_replicated(Q)
-                         orelse rabbit_amqqueue:is_dead_exclusive(Q))
+                andalso (not is_replicated(Q)
+                         orelse is_dead_exclusive(Q))
                 andalso amqqueue:get_type(Q) =/= rabbit_mqtt_qos0_queue
     end.
 
@@ -2129,7 +2129,7 @@ queue_names(Queues)
 get_bcc_queue(Q, BCCName) ->
     #resource{virtual_host = VHost} = amqqueue:get_name(Q),
     BCCQueueName = rabbit_misc:r(VHost, queue, BCCName),
-    rabbit_amqqueue:lookup(BCCQueueName).
+    lookup(BCCQueueName).
 
 is_queue_args_combination_permitted(Q) ->
     Durable = amqqueue:is_durable(Q),

deps/rabbit/src/rabbit_auth_backend_internal.erl

@@ -698,8 +698,8 @@ put_user(User, Version, ActingUser) ->
                     throw({error, both_password_and_password_hash_are_provided});
                 %% clear password, update tags if needed
                 _ ->
-                    rabbit_auth_backend_internal:set_tags(Username, Tags, ActingUser),
-                    rabbit_auth_backend_internal:clear_password(Username, ActingUser)
+                    set_tags(Username, Tags, ActingUser),
+                    clear_password(Username, ActingUser)
             end;
         false ->
             case {HasPassword, HasPasswordHash} of
@@ -732,13 +732,13 @@ update_user_password_hash(Username, PasswordHash, Tags, Limits, User, Version) -
 
     Hash = rabbit_misc:b64decode_or_throw(PasswordHash),
     ConvertedTags = [rabbit_data_coercion:to_atom(I) || I <- Tags],
-    rabbit_auth_backend_internal:update_user_with_hash(
+    update_user_with_hash(
       Username, Hash, HashingAlgorithm, ConvertedTags, Limits).
 
 create_user_with_password(_PassedCredentialValidation = true,  Username, Password, Tags, undefined, Limits, ActingUser) ->
-    ok = rabbit_auth_backend_internal:add_user(Username, Password, ActingUser, Limits, Tags);
+    ok = add_user(Username, Password, ActingUser, Limits, Tags);
 create_user_with_password(_PassedCredentialValidation = true,  Username, Password, Tags, PreconfiguredPermissions, Limits, ActingUser) ->
-    ok = rabbit_auth_backend_internal:add_user(Username, Password, ActingUser, Limits, Tags),
+    ok = add_user(Username, Password, ActingUser, Limits, Tags),
     preconfigure_permissions(Username, PreconfiguredPermissions, ActingUser);
 create_user_with_password(_PassedCredentialValidation = false, _Username, _Password, _Tags, _, _, _) ->
     %% we don't log here because
@@ -751,14 +751,14 @@ create_user_with_password_hash(Username, PasswordHash, Tags, User, Version, Prec
     HashingAlgorithm = hashing_algorithm(User, Version),
     Hash             = rabbit_misc:b64decode_or_throw(PasswordHash),
 
-    rabbit_auth_backend_internal:add_user_sans_validation(Username, Hash, HashingAlgorithm, Tags, Limits, ActingUser),
+    add_user_sans_validation(Username, Hash, HashingAlgorithm, Tags, Limits, ActingUser),
     preconfigure_permissions(Username, PreconfiguredPermissions, ActingUser).
 
 preconfigure_permissions(_Username, undefined, _ActingUser) ->
     ok;
 preconfigure_permissions(Username, Map, ActingUser) when is_map(Map) ->
     _ = maps:map(fun(VHost, M) ->
-                     rabbit_auth_backend_internal:set_permissions(Username, VHost,
+                     set_permissions(Username, VHost,
                                                   maps:get(<<"configure">>, M),
                                                   maps:get(<<"write">>,     M),
                                                   maps:get(<<"read">>,      M),

deps/rabbit/src/rabbit_basic.erl

@@ -286,7 +286,7 @@ msg_size(Content) ->
     rabbit_writer:msg_size(Content).
 
 add_header(Name, Type, Value, #basic_message{content = Content0} = Msg) ->
-    Content = rabbit_basic:map_headers(
+    Content = map_headers(
                 fun(undefined) ->
                         rabbit_misc:set_table_value([], Name, Type, Value);
                    (Headers) ->

deps/rabbit/src/rabbit_connection_tracking.erl

@@ -108,7 +108,7 @@ handle_cast({vhost_deleted, Details}) ->
             delete_tracked_connection_vhost_entry, [VHost]),
     rabbit_log_connection:info("Closing all connections in vhost '~ts' because it's being deleted", [VHost]),
     shutdown_tracked_items(
-        rabbit_connection_tracking:list(VHost),
+        list(VHost),
         rabbit_misc:format("vhost '~ts' is deleted", [VHost]));
 %% Note: under normal circumstances this will be called immediately
 %% after the vhost_deleted above. Therefore we should be careful about
@@ -120,7 +120,7 @@ handle_cast({vhost_down, Details}) ->
                                " because the vhost is stopping",
                                [VHost, Node]),
     shutdown_tracked_items(
-        rabbit_connection_tracking:list_on_node(Node, VHost),
+        list_on_node(Node, VHost),
         rabbit_misc:format("vhost '~ts' is down", [VHost]));
 handle_cast({user_deleted, Details}) ->
     Username = pget(name, Details),
@@ -129,7 +129,7 @@ handle_cast({user_deleted, Details}) ->
             delete_tracked_connection_user_entry, [Username]),
     rabbit_log_connection:info("Closing all connections for user '~ts' because the user is being deleted", [Username]),
     shutdown_tracked_items(
-        rabbit_connection_tracking:list_of_user(Username),
+        list_of_user(Username),
         rabbit_misc:format("user '~ts' is deleted", [Username])).
 
 -spec register_tracked(rabbit_types:tracked_connection()) -> ok.

deps/rabbit/src/rabbit_db_binding.erl

@@ -928,7 +928,7 @@ clear_in_mnesia() ->
     ok.
 
 clear_in_khepri() ->
-    Path = rabbit_db_binding:khepri_routes_path(),
+    Path = khepri_routes_path(),
     case rabbit_khepri:delete(Path) of
         ok -> ok;
         Error -> throw(Error)

deps/rabbit/src/rabbit_db_cluster.erl

@@ -78,7 +78,7 @@ can_join_using_khepri(RemoteNode) ->
 
 -spec join(RemoteNode, NodeType) -> Ret when
       RemoteNode :: node(),
-      NodeType :: rabbit_db_cluster:node_type(),
+      NodeType :: node_type(),
       Ret :: Ok | Error,
       Ok :: ok | {ok, already_member},
       Error :: {error, {inconsistent_cluster, string()}}.
@@ -254,7 +254,7 @@ forget_member_using_khepri(Node, false = _RemoveWhenOffline) ->
 %% -------------------------------------------------------------------
 
 -spec change_node_type(NodeType) -> ok when
-      NodeType :: rabbit_db_cluster:node_type().
+      NodeType :: node_type().
 %% @doc Changes the node type to `NodeType'.
 %%
 %% Node types may not all be valid with all databases.
@@ -320,7 +320,7 @@ disc_members_using_mnesia() ->
     rabbit_mnesia:cluster_nodes(disc).
 
 -spec node_type() -> NodeType when
-      NodeType :: rabbit_db_cluster:node_type().
+      NodeType :: node_type().
 %% @doc Returns the type of this node, `disc' or `ram'.
 %%
 %% Node types may not all be relevant with all databases.
@@ -373,7 +373,7 @@ check_consistency_using_khepri() ->
     rabbit_khepri:check_cluster_consistency().
 
 -spec cli_cluster_status() -> ClusterStatus when
-      ClusterStatus :: [{nodes, [{rabbit_db_cluster:node_type(), [node()]}]} |
+      ClusterStatus :: [{nodes, [{node_type(), [node()]}]} |
                         {running_nodes, [node()]} |
                         {partitions, [{node(), [node()]}]}].
 %% @doc Returns information from the cluster for the `cluster_status' CLI

deps/rabbit/src/rabbit_exchange.erl

@@ -220,7 +220,7 @@ list() ->
 count() ->
     rabbit_db_exchange:count().
 
--spec list_names() -> [rabbit_exchange:name()].
+-spec list_names() -> [name()].
 
 list_names() ->
     rabbit_db_exchange:list().

deps/rabbit/src/rabbit_fifo.erl

@@ -4,8 +4,6 @@
 %%
 %% Copyright (c) 2007-2024 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
 
-%% before post gc 1M msg: 203MB, after recovery + gc: 203MB
-
 -module(rabbit_fifo).
 
 -behaviour(ra_machine).
@@ -1129,7 +1127,7 @@ handle_aux(_RaState, {call, _From}, oldest_entry_timestamp,
     end;
 handle_aux(_RaState, {call, _From}, {peek, Pos}, Aux0,
            Log0, MacState) ->
-    case rabbit_fifo:query_peek(Pos, MacState) of
+    case query_peek(Pos, MacState) of
         {ok, ?MSG(Idx, Header)} ->
             %% need to re-hydrate from the log
             {{_, _, {_, _, Cmd, _}}, Log} = ra_log:fetch(Idx, Log0),

deps/rabbit/src/rabbit_mirror_queue_misc.erl

@@ -551,7 +551,7 @@ stop_all_slaves(Reason, SPids, QName, GM, WaitTimeout) ->
                     {'DOWN', MRef, process, _Pid, _Info} ->
                         Acc
                 after WaitTimeout ->
-                        rabbit_mirror_queue_misc:log_warning(
+                        log_warning(
                           QName, "Missing 'DOWN' message from ~tp in"
                           " node ~tp", [Pid, node(Pid)]),
                         [Pid | Acc]
@@ -582,7 +582,7 @@ remove_all_slaves_in_mnesia(QName, PendingSlavePids) ->
         %% ensure old incarnations are stopped using
         %% the pending mirror pids.
         Q3 = amqqueue:set_slave_pids_pending_shutdown(Q2, PendingSlavePids),
-        rabbit_mirror_queue_misc:store_updated_slaves(Q3)
+        store_updated_slaves(Q3)
     end).
 
 remove_all_slaves_in_khepri(QName, PendingSlavePids) ->

deps/rabbit/src/rabbit_mnesia.erl

@@ -462,7 +462,7 @@ cluster_status(WhichNodes) ->
     end.
 
 members() ->
-    case rabbit_mnesia:is_running() andalso rabbit_table:is_present() of
+    case is_running() andalso rabbit_table:is_present() of
         true ->
             %% If Mnesia is running locally and some tables exist, we can know
             %% the database was initialized and we can query the list of

deps/rabbit/src/rabbit_vhost_limit.erl

@@ -84,7 +84,7 @@ list(VHost) ->
 -spec is_over_connection_limit(vhost:name()) -> {true, non_neg_integer()} | false.
 
 is_over_connection_limit(VirtualHost) ->
-    case rabbit_vhost_limit:connection_limit(VirtualHost) of
+    case connection_limit(VirtualHost) of
         %% no limit configured
         undefined                                            -> false;
         %% with limit = 0, no connections are allowed

deps/rabbit/src/rabbit_vhost_sup_sup.erl

@@ -284,7 +284,7 @@ check() ->
     VHosts = rabbit_vhost:list_names(),
     lists:filter(
       fun(V) ->
-              case rabbit_vhost_sup_sup:get_vhost_sup(V) of
+              case get_vhost_sup(V) of
                   {ok, Sup} ->
                       MsgStores = [Pid || {Name, Pid, _, _} <- supervisor:which_children(Sup),
                                          lists:member(Name, [msg_store_persistent,

deps/rabbit/src/vhost.erl

@@ -169,6 +169,6 @@ merge_metadata(VHost, Value) ->
     NewMeta = maps:merge(Meta0, Value),
     VHost#vhost{metadata = NewMeta}.
 
--spec is_tagged_with(vhost:vhost(), tag()) -> boolean().
+-spec is_tagged_with(vhost(), tag()) -> boolean().
 is_tagged_with(VHost, Tag) ->
     lists:member(Tag, get_tags(VHost)).

deps/rabbitmq_ct_helpers/src/rabbit_ct_broker_helpers.erl

@@ -1329,7 +1329,7 @@ rabbitmqctl_list(Config, Node, Args) ->
 
 rabbitmq_queues(Config, Node, Args) ->
     RabbitmqQueues = ?config(rabbitmq_queues_cmd, Config),
-    NodeConfig = rabbit_ct_broker_helpers:get_node_config(Config, Node),
+    NodeConfig = get_node_config(Config, Node),
     Nodename = ?config(nodename, NodeConfig),
     Env0 = [
       {"RABBITMQ_SCRIPTS_DIR", filename:dirname(RabbitmqQueues)},
@@ -1930,38 +1930,38 @@ await_os_pid_death(Pid) ->
     end.
 
 reset_node(Config, Node) ->
-    Name = rabbit_ct_broker_helpers:get_node_config(Config, Node, nodename),
+    Name = get_node_config(Config, Node, nodename),
     rabbit_control_helper:command(reset, Name).
 
 force_reset_node(Config, Node) ->
-    Name = rabbit_ct_broker_helpers:get_node_config(Config, Node, nodename),
+    Name = get_node_config(Config, Node, nodename),
     rabbit_control_helper:command(force_reset, Name).
 
 forget_cluster_node(Config, Node, NodeToForget) ->
     forget_cluster_node(Config, Node, NodeToForget, []).
 forget_cluster_node(Config, Node, NodeToForget, Opts) ->
-    Name = rabbit_ct_broker_helpers:get_node_config(Config, Node, nodename),
+    Name = get_node_config(Config, Node, nodename),
     NameToForget =
-        rabbit_ct_broker_helpers:get_node_config(Config, NodeToForget, nodename),
+        get_node_config(Config, NodeToForget, nodename),
     rabbit_control_helper:command(forget_cluster_node, Name, [NameToForget], Opts).
 
 is_feature_flag_enabled(Config, FeatureName) ->
-    Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
-    rabbit_ct_broker_helpers:rpc(
+    Node = get_node_config(Config, 0, nodename),
+    rpc(
       Config, Node, rabbit_feature_flags, is_enabled, [FeatureName]).
 
 is_feature_flag_supported(Config, FeatureName) ->
-    Nodes = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Nodes = get_node_configs(Config, nodename),
     is_feature_flag_supported(Config, Nodes, FeatureName).
 
 is_feature_flag_supported(Config, [Node1 | _] = _Nodes, FeatureName) ->
-    rabbit_ct_broker_helpers:rpc(
+    rpc(
       Config, Node1,
       rabbit_feature_flags, is_supported,
       [[FeatureName], 60000]).
 
 enable_feature_flag(Config, FeatureName) ->
-    Nodes = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Nodes = get_node_configs(Config, nodename),
     enable_feature_flag(Config, Nodes, FeatureName).
 
 enable_feature_flag(Config, Nodes, FeatureName) ->
@@ -1971,7 +1971,7 @@ enable_feature_flag(Config, Nodes, FeatureName) ->
             %% feature flags on the first one of the list is not enough
             lists:foldl(
               fun(N, ok) ->
-                      case rabbit_ct_broker_helpers:rpc(
+                      case rpc(
                              Config, N, rabbit_feature_flags, enable, [FeatureName]) of
                           {error, unsupported} ->
                               {skip,
@@ -1992,24 +1992,24 @@ enable_feature_flag(Config, Nodes, FeatureName) ->
     end.
 
 mark_as_being_drained(Config, Node) ->
-    rabbit_ct_broker_helpers:rpc(Config, Node, rabbit_maintenance, mark_as_being_drained, []).
+    rpc(Config, Node, rabbit_maintenance, mark_as_being_drained, []).
 unmark_as_being_drained(Config, Node) ->
-    rabbit_ct_broker_helpers:rpc(Config, Node, rabbit_maintenance, unmark_as_being_drained, []).
+    rpc(Config, Node, rabbit_maintenance, unmark_as_being_drained, []).
 
 drain_node(Config, Node) ->
-    rabbit_ct_broker_helpers:rpc(Config, Node, rabbit_maintenance, drain, []).
+    rpc(Config, Node, rabbit_maintenance, drain, []).
 revive_node(Config, Node) ->
-    rabbit_ct_broker_helpers:rpc(Config, Node, rabbit_maintenance, revive, []).
+    rpc(Config, Node, rabbit_maintenance, revive, []).
 
 is_being_drained_consistent_read(Config, Node) ->
-    rabbit_ct_broker_helpers:rpc(Config, Node, rabbit_maintenance, is_being_drained_consistent_read, [Node]).
+    rpc(Config, Node, rabbit_maintenance, is_being_drained_consistent_read, [Node]).
 is_being_drained_local_read(Config, Node) ->
-    rabbit_ct_broker_helpers:rpc(Config, Node, rabbit_maintenance, is_being_drained_local_read, [Node]).
+    rpc(Config, Node, rabbit_maintenance, is_being_drained_local_read, [Node]).
 
 is_being_drained_consistent_read(Config, TargetNode, NodeToCheck) ->
-    rabbit_ct_broker_helpers:rpc(Config, TargetNode, rabbit_maintenance, is_being_drained_consistent_read, [NodeToCheck]).
+    rpc(Config, TargetNode, rabbit_maintenance, is_being_drained_consistent_read, [NodeToCheck]).
 is_being_drained_local_read(Config, TargetNode, NodeToCheck) ->
-    rabbit_ct_broker_helpers:rpc(Config, TargetNode, rabbit_maintenance, is_being_drained_local_read, [NodeToCheck]).
+    rpc(Config, TargetNode, rabbit_maintenance, is_being_drained_local_read, [NodeToCheck]).
 
 %% From a given list of gen_tcp client connections, return the list of
 %% connection handler PID in RabbitMQ.