Khepri: Filter running nodes when selecting a node to cluster with

[Why]
So far, the code that selected the node to use as the "entry point" to
add the local node to a remote cluster assumed that all cluster members
were running and it picked the first node in the cluster members list.

If that node was stopped, the join would fail immediately, even if the
rest of the members were running fine.

[How]
Now the function filters out nodes that are unavailable or don't run the
expected Khepri store. Then it uses the resulting list as before.

The code returns an error if all nodes are stopped or unreachable.

Author: dumbbell

deps/rabbit/src/rabbit_khepri.erl

@@ -373,20 +373,42 @@ add_member(JoiningNode, JoinedNode) when is_atom(JoinedNode) ->
             JoiningNode, rabbit_khepri, do_join, [JoinedNode]),
     post_add_member(JoiningNode, JoinedNode, Ret);
 add_member(JoiningNode, [_ | _] = Cluster) ->
-    JoinedNode = pick_node_in_cluster(Cluster),
-    ?LOG_INFO(
-       "Khepri clustering: Attempt to add node ~p to cluster ~0p "
-       "through node ~p",
-       [JoiningNode, Cluster, JoinedNode],
-       #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
-    %% Recurse with a single node taken in the `Cluster' list.
-    add_member(JoiningNode, JoinedNode).
+    case pick_node_in_cluster(Cluster) of
+        {ok, JoinedNode} ->
+            ?LOG_INFO(
+               "Khepri clustering: Attempt to add node ~p to cluster ~0p "
+               "through node ~p",
+               [JoiningNode, Cluster, JoinedNode],
+               #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+            %% Recurse with a single node taken in the `Cluster' list.
+            add_member(JoiningNode, JoinedNode);
+        {error, _} = Error ->
+            Error
+    end.
 
-pick_node_in_cluster([_ | _] = Cluster) when is_list(Cluster) ->
-    ThisNode = node(),
-    case lists:member(ThisNode, Cluster) of
-        true  -> ThisNode;
-        false -> hd(Cluster)
+pick_node_in_cluster([_ | _] = Cluster) ->
+    RunningNodes = lists:filter(
+                     fun(Node) ->
+                             try
+                                 erpc:call(
+                                   Node,
+                                   khepri_cluster, is_store_running,
+                                   [?STORE_ID])
+                             catch
+                                 _:_ ->
+                                     false
+                             end
+                     end, Cluster),
+    case RunningNodes of
+        [_ | _] ->
+            ThisNode = node(),
+            SelectedNode = case lists:member(ThisNode, RunningNodes) of
+                               true  -> ThisNode;
+                               false -> hd(RunningNodes)
+                           end,
+            {ok, SelectedNode};
+        [] ->
+            {error, {no_nodes_to_cluster_with, Cluster}}
     end.
 
 do_join(RemoteNode) when RemoteNode =/= node() ->