Presence server

deps/rabbit/app.bzl

@@ -196,6 +196,7 @@ def all_beam_files(name = "all_beam_files"):
             "src/rabbit_prelaunch_feature_flags.erl",
             "src/rabbit_prelaunch_logging.erl",
             "src/rabbit_prequeue.erl",
+            "src/rabbit_presence.erl",
             "src/rabbit_priority_queue.erl",
             "src/rabbit_process.erl",
             "src/rabbit_queue_consumers.erl",
@@ -461,6 +462,7 @@ def all_test_beam_files(name = "all_test_beam_files"):
             "src/rabbit_prelaunch_feature_flags.erl",
             "src/rabbit_prelaunch_logging.erl",
             "src/rabbit_prequeue.erl",
+            "src/rabbit_presence.erl",
             "src/rabbit_priority_queue.erl",
             "src/rabbit_process.erl",
             "src/rabbit_queue_consumers.erl",
@@ -745,6 +747,7 @@ def all_srcs(name = "all_srcs"):
             "src/rabbit_prelaunch_feature_flags.erl",
             "src/rabbit_prelaunch_logging.erl",
             "src/rabbit_prequeue.erl",
+            "src/rabbit_presence.erl",
             "src/rabbit_priority_queue.erl",
             "src/rabbit_process.erl",
             "src/rabbit_queue_consumers.erl",

deps/rabbit/src/rabbit.erl

@@ -168,6 +168,13 @@
                     {requires,    kernel_ready},
                     {enables,     core_initialized}]}).
 
+-rabbit_boot_step({rabbit_presence,
+                   [{description, "rabbit node presence server"},
+                    {mfa,         {rabbit_sup, start_restartable_child,
+                                   [rabbit_presence]}},
+                    {requires,    [database]},
+                    {enables,     core_initialized}]}).
+
 -rabbit_boot_step({rabbit_node_monitor,
                    [{description, "node monitor"},
                     {mfa,         {rabbit_sup, start_restartable_child,

deps/rabbit/src/rabbit_amqqueue.erl

@@ -40,7 +40,8 @@
 -export([emit_unresponsive/6, emit_unresponsive_local/5, is_unresponsive/2]).
 -export([has_synchronised_mirrors_online/1, is_match/2, is_in_virtual_host/2]).
 -export([is_replicated/1, is_exclusive/1, is_not_exclusive/1, is_dead_exclusive/1]).
--export([list_local_quorum_queues/0, list_local_quorum_queue_names/0, list_local_stream_queues/0,
+-export([list_local_quorum_queues/0, list_local_quorum_queue_names/0,
+         list_local_stream_queues/0, list_stream_queues_on/1,
          list_local_mirrored_classic_queues/0, list_local_mirrored_classic_names/0,
          list_local_leaders/0, list_local_followers/0, get_quorum_nodes/1,
          list_local_mirrored_classic_without_synchronised_mirrors/0,
@@ -1220,9 +1221,12 @@ list_local_quorum_queues() ->
 
 -spec list_local_stream_queues() -> [amqqueue:amqqueue()].
 list_local_stream_queues() ->
-    [ Q || Q <- list_by_type(stream),
-      amqqueue:get_state(Q) =/= crashed,
-      lists:member(node(), get_quorum_nodes(Q))].
+    list_stream_queues_on(node()).
+
+-spec list_stream_queues_on(node()) -> [amqqueue:amqqueue()].
+list_stream_queues_on(Node) when is_atom(Node) ->
+     [Q || Q <- list_by_type(rabbit_stream_queue),
+           lists:member(Node, get_quorum_nodes(Q))].
 
 -spec list_local_leaders() -> [amqqueue:amqqueue()].
 list_local_leaders() ->

deps/rabbit/src/rabbit_channel.erl

@@ -366,8 +366,9 @@ declare_fast_reply_to_v1(EncodedBin) ->
 -spec list() -> [pid()].
 
 list() ->
-    Nodes = rabbit_nodes:list_running(),
-    rabbit_misc:append_rpc_all_nodes(Nodes, rabbit_channel, list_local, [], ?RPC_TIMEOUT).
+    Nodes = rabbit_presence:list_present(),
+    rabbit_misc:append_rpc_all_nodes(Nodes, rabbit_channel, list_local,
+                                     [], ?RPC_TIMEOUT).
 
 -spec list_local() -> [pid()].
 

deps/rabbit/src/rabbit_presence.erl

@@ -0,0 +1,109 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2023 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries.  All rights reserved.
+%%
+
+-module(rabbit_presence).
+
+-behaviour(gen_server).
+
+-export([list_present/0]).
+-export([start_link/0]).
+
+-export([init/1,
+         handle_call/3,
+         handle_cast/2,
+         handle_info/2,
+         terminate/2,
+         code_change/3]).
+
+-define(SERVER, ?MODULE).
+-define(INTERVAL, 1000).
+-define(CUTOFF, ?INTERVAL * 3).
+
+-record(?MODULE, {tbl :: ets:table(),
+                  nodes = [] :: [node()]}).
+
+%%----------------------------------------------------------------------------
+%% A presence server that heartbeats all configured servers with the goal of
+%% providing a very quickly accessible idea of node availability without
+%% having to use rabbit_nodes:all_running/1 which can block for a long time.
+%%----------------------------------------------------------------------------
+
+-spec list_present() -> [node()].
+list_present() ->
+    case whereis(?MODULE) of
+        undefined ->
+            %% TODO: change return type to ok | error?
+            exit(presence_server_not_running);
+        _ ->
+            Cutoff = erlang:system_time(millisecond) - ?CUTOFF,
+            [N || {N, SeenMs} <- ets:tab2list(?MODULE),
+                  %% if it hasn't been seen since the cutoff
+                  SeenMs > Cutoff,
+                  %% if not in nodes() it is also considered not present
+                  lists:member(N, nodes())]
+    end.
+
+-spec start_link() -> rabbit_types:ok_pid_or_error().
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+init([]) ->
+    process_flag(trap_exit, true),
+    Ref = ets:new(?MODULE, [set, named_table, protected]),
+    _ = erlang:send_after(?INTERVAL, self(), beat),
+    Nodes = rabbit_nodes:list_members(),
+    _ = beat_all(Nodes),
+    {ok, #?MODULE{tbl = Ref,
+                  nodes = Nodes}}.
+
+handle_call(_Request, _From, State) ->
+    {noreply, State}.
+
+handle_cast(_Request, State) ->
+    {noreply, State}.
+
+handle_info(beat, #?MODULE{tbl = _Tbl,
+                           nodes = Nodes} = State) ->
+    _ = erlang:send_after(?INTERVAL, self(), beat),
+    _ = beat_all(Nodes),
+    %% this will only be efficient to do this often once list_members
+    %% make use of the ra_leaderboard rather than calling into the local
+    %% khepri process
+    case rabbit_nodes:list_members() of
+        Nodes ->
+            {noreply, State};
+        NewNodes ->
+            {noreply, State#?MODULE{nodes = NewNodes}}
+    end;
+handle_info({hb, Node}, #?MODULE{tbl = Tbl,
+                                 nodes = _Nodes} = State) ->
+    ets:insert(Tbl, {Node, erlang:system_time(millisecond)}),
+    {noreply, State};
+handle_info({terminate, Node}, #?MODULE{tbl = Tbl} = State) ->
+    _ = ets:delete(Tbl, Node),
+    {noreply, State};
+handle_info(_Msg, State) ->
+    {noreply, State}.
+
+terminate(_Reason, #?MODULE{nodes = Nodes}) ->
+    %% TODO: only send terminate if reason is `shutdown`?
+    _ = send_terminate(Nodes),
+    ok.
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%% INTERNAL
+
+beat_all(Nodes) ->
+    [send(N, {hb, node()}) || N <- Nodes, N =/= node()].
+
+send_terminate(Nodes) ->
+    [send(N, {terminate, node()}) || N <- Nodes, N =/= node()].
+
+send(Node, Msg) ->
+    erlang:send({?SERVER, Node}, Msg, [noconnect, nosuspend]).

deps/rabbit/src/rabbit_queue_type_util.erl

@@ -12,7 +12,8 @@
          check_auto_delete/1,
          check_exclusive/1,
          check_non_durable/1,
-         run_checks/2]).
+         run_checks/2,
+         erpc_call/5]).
 
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include("amqqueue.hrl").
@@ -70,3 +71,30 @@ run_checks([C | Checks], Q) ->
         Err ->
             Err
     end.
+
+-spec erpc_call(node(), module(), atom(), list(), non_neg_integer()) ->
+    term() | {error, term()}.
+erpc_call(Node, M, F, A, _Timeout)
+  when Node =:= node()  ->
+    %% Only timeout 'infinity' optimises the local call in OTP 23-25 avoiding a new process being spawned:
+    %% https://github.com/erlang/otp/blob/47f121af8ee55a0dbe2a8c9ab85031ba052bad6b/lib/kernel/src/erpc.erl#L121
+    try erpc:call(Node, M, F, A, infinity) of
+        Result ->
+            Result
+    catch
+        error:Err ->
+            {error, Err}
+    end;
+erpc_call(Node, M, F, A, Timeout) ->
+    case lists:member(Node, nodes()) of
+        true ->
+            try erpc:call(Node, M, F, A, Timeout) of
+                Result ->
+                    Result
+            catch
+                error:Err ->
+                    {error, Err}
+            end;
+        false ->
+            {error, noconnection}
+    end.

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -78,7 +78,9 @@
          force_all_queues_shrink_member_to_current_member/0]).
 
 -import(rabbit_queue_type_util, [args_policy_lookup/3,
-                                 qname_to_internal_name/1]).
+                                 qname_to_internal_name/1,
+                                 erpc_call/5
+                                ]).
 
 -include_lib("stdlib/include/qlc.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
@@ -1676,8 +1678,7 @@ format(Q, Ctx) when ?is_amqqueue(Q) ->
                   #{running_nodes := Running0} ->
                       Running0;
                   _ ->
-                      %% WARN: slow
-                      rabbit_nodes:list_running()
+                      rabbit_presence:list_present()
               end,
     Online = [N || N <- Nodes, lists:member(N, Running)],
     {_, LeaderNode} = amqqueue:get_pid(Q),
@@ -1840,31 +1841,6 @@ notify_decorators(QName, F, A) ->
             ok
     end.
 
-erpc_call(Node, M, F, A, _Timeout)
-  when Node =:= node()  ->
-    %% Only timeout 'infinity' optimises the local call in OTP 23-25 avoiding a new process being spawned:
-    %% https://github.com/erlang/otp/blob/47f121af8ee55a0dbe2a8c9ab85031ba052bad6b/lib/kernel/src/erpc.erl#L121
-    try erpc:call(Node, M, F, A, infinity) of
-        Result ->
-            Result
-    catch
-        error:Err ->
-            {error, Err}
-    end;
-erpc_call(Node, M, F, A, Timeout) ->
-    case lists:member(Node, nodes()) of
-        true ->
-            try erpc:call(Node, M, F, A, Timeout) of
-                Result ->
-                    Result
-            catch
-                error:Err ->
-                    {error, Err}
-            end;
-        false ->
-            {error, noconnection}
-    end.
-
 is_stateful() -> true.
 
 force_shrink_member_to_current_member(VHost, Name) ->