Use rabbit_db_queue for qq leader health check lookups

and introduce rabbit_db_queue:get_all_by_type_and_vhost/2.
Update leader health check timeout to 5s and process limit
threshold to 20% of node's process_limit.

Author: Ayanda-D

deps/rabbit/src/amqqueue.erl

@@ -66,6 +66,7 @@
          pattern_match_on_type/1,
          pattern_match_on_durable/1,
          pattern_match_on_type_and_durable/2,
+         pattern_match_on_type_and_vhost/2,
          reset_decorators/1,
          set_immutable/1,
          qnode/1,
@@ -532,6 +533,12 @@ pattern_match_on_durable(IsDurable) ->
 pattern_match_on_type_and_durable(Type, IsDurable) ->
     #amqqueue{type = Type, durable = IsDurable, _ = '_'}.
 
+-spec pattern_match_on_type_and_vhost(atom(), binary()) ->
+    amqqueue_pattern().
+
+pattern_match_on_type_and_vhost(Type, VHost) ->
+    #amqqueue{type = Type, vhost = VHost, _ = '_'}.
+
 -spec reset_decorators(amqqueue()) -> amqqueue().
 
 reset_decorators(#amqqueue{} = Queue) ->

deps/rabbit/src/rabbit_db_queue.erl

@@ -21,6 +21,7 @@
          get_all/0,
          get_all/1,
          get_all_by_type/1,
+         get_all_by_type_and_vhost/2,
          get_all_by_type_and_node/3,
          list/0,
          count/0,
@@ -829,6 +830,28 @@ get_all_by_type(Type) ->
         khepri => fun() -> get_all_by_pattern_in_khepri(Pattern) end
        }).
 
+%% -------------------------------------------------------------------
+%% get_all_by_type_and_vhost().
+%% -------------------------------------------------------------------
+
+-spec get_all_by_type_and_vhost(Type, VHost) -> [Queue] when
+      Type :: atom(),
+      VHost :: binary(),
+      Queue :: amqqueue:amqqueue().
+
+%% @doc Gets all queues belonging to the given type and vhost
+%%
+%% @returns a list of queue records.
+%%
+%% @private
+
+get_all_by_type_and_vhost(Type, VHost) ->
+    Pattern = amqqueue:pattern_match_on_type_and_vhost(Type, VHost),
+    rabbit_khepri:handle_fallback(
+      #{mnesia => fun() -> get_all_by_pattern_in_mnesia(Pattern) end,
+        khepri => fun() -> get_all_by_pattern_in_khepri(Pattern) end
+       }).
+
 get_all_by_pattern_in_mnesia(Pattern) ->
     rabbit_db:list_in_mnesia(?MNESIA_TABLE, Pattern).
 

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -147,7 +147,7 @@
 -define(SNAPSHOT_INTERVAL, 8192). %% the ra default is 4096
 % -define(UNLIMITED_PREFETCH_COUNT, 2000). %% something large for ra
 -define(MIN_CHECKPOINT_INTERVAL, 8192). %% the ra default is 16384
--define(LEADER_HEALTH_CHECK_TIMEOUT, 1_000).
+-define(LEADER_HEALTH_CHECK_TIMEOUT, 5_000).
 -define(GLOBAL_LEADER_HEALTH_CHECK_TIMEOUT, 60_000).
 
 %%----------- QQ policies ---------------------------------------------------
@@ -2151,19 +2151,19 @@ file_handle_release_reservation() ->
     ok.
 
 leader_health_check(QueueNameOrRegEx, VHost) ->
-    %% Set a process limit threshold to 40% of ErlangVM process limit, beyond which
+    %% Set a process limit threshold to 20% of ErlangVM process limit, beyond which
     %% we cannot spawn any new processes for executing QQ leader health checks.
-    ProcessLimitThreshold = round(0.4 * erlang:system_info(process_limit)),
+    ProcessLimitThreshold = round(0.2 * erlang:system_info(process_limit)),
 
     leader_health_check(QueueNameOrRegEx, VHost, ProcessLimitThreshold).
 
 leader_health_check(QueueNameOrRegEx, VHost, ProcessLimitThreshold) ->
     Qs =
         case VHost of
-            global ->
-                rabbit_amqqueue:list();
+            across_all_vhosts ->
+                rabbit_db_queue:get_all_by_type(?MODULE);
             VHost when is_binary(VHost) ->
-                rabbit_amqqueue:list(VHost)
+                rabbit_db_queue:get_all_by_type_and_vhost(?MODULE, VHost)
         end,
     check_process_limit_safety(length(Qs), ProcessLimitThreshold),
     ParentPID = self(),